chore(tests): remove Langfuse scoring from preprocess tests

Scoring is only meaningful for LLM-backed steps. Preprocess tests are
deterministic Python, so scores add no value. Kept only for detect tests.

- test_preprocess: drop _lf_score call, simplify _run_assertions return type
- cases.yaml: remove score_name from all op=preprocess entries

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.env.example

@@ -39,6 +39,13 @@ QDRANT_URL=
 QDRANT_API_KEY=
 # For local Qdrant (homelab): QDRANT_URL=http://qdrant:6333
 
+# ── Langfuse (leave empty to disable observability) ───────────────────────────
+LANGFUSE_SECRET_KEY=
+LANGFUSE_PUBLIC_KEY=
+# LANGFUSE_HOST=https://cloud.langfuse.com        # EU (default)
+# LANGFUSE_HOST=https://us.cloud.langfuse.com     # US
+# LANGFUSE_HOST=http://localhost:3000             # Self-hosted
+
 # ── CORS ──────────────────────────────────────────────────────────────────────
 # Comma-separated list parsed by Settings (override default if needed)
 # CORS_ORIGINS=["app://.","http://localhost:3000"]

alembic/versions/9a1f2d0b6c7e_deprecate_backend_agent_config_tables.py

@@ -0,0 +1,92 @@
+"""Deprecate backend agent config tables.
+
+The Electron client is now the source of truth for agent configuration
+(directory, extract targets, batch interval, custom prompt). Backend keeps
+billing checks and trigger/run logs only.
+
+Revision ID: 9a1f2d0b6c7e
+Revises: 818478c251dc
+Create Date: 2026-03-16
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "9a1f2d0b6c7e"
+down_revision: Union[str, None] = "818478c251dc"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    existing = set(inspector.get_table_names())
+
+    if "cloud_agent_configs" in existing:
+        op.drop_index("ix_cloud_agent_configs_user_id", table_name="cloud_agent_configs")
+        op.drop_table("cloud_agent_configs")
+
+    if "local_agent_configs" in existing:
+        op.drop_index("ix_local_agent_configs_user_id", table_name="local_agent_configs")
+        op.drop_table("local_agent_configs")
+
+
+def downgrade() -> None:
+    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"])
+
+    op.execute(
+        """
+        DO $$ BEGIN
+            CREATE TYPE cloud_provider AS ENUM ('gmail', 'teams', 'outlook');
+        EXCEPTION WHEN duplicate_object THEN NULL;
+        END $$;
+        """
+    )
+
+    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"])

app/agents/__init__.py

@@ -1,5 +1,5 @@
-"""Agent tool modules — imported by deep_agent.py to build sub-agent graphs."""
+"""Expose tool modules used by deep orchestrator-worker graphs."""
 
-from app.agents import timeline_agent, note_agent, project_agent, task_agent
+from app.agents import filesystem_agent, timeline_agent, note_agent, project_agent, task_agent
 
-__all__ = ["timeline_agent", "note_agent", "project_agent", "task_agent"]
+__all__ = ["filesystem_agent", "timeline_agent", "note_agent", "project_agent", "task_agent"]

app/agents/filesystem_agent.py

@@ -0,0 +1,85 @@
+"""Filesystem agent — tools for reading local directories and files on Electron.
+
+These tools delegate to the Electron client via ``execute_on_client()`` using
+the same WS tool-call round-trip pattern as CRUD tools.  The Electron app
+handles actual disk I/O and responds with ``tool_result`` frames.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from langchain_core.tools import tool
+
+from app.core.ws_context import execute_on_client
+
+
+@tool
+async def list_directory(path: str) -> str:
+    """List files and folders in a local directory on the user's device.
+
+    Returns a formatted listing of entries with name, type (file/directory),
+    and full path.
+    """
+    result = await execute_on_client(
+        action="list_directory",
+        data={"path": path},
+    )
+    entries: list[dict[str, Any]] = result.get("entries", [])
+    if not entries:
+        return f"Directory '{path}' is empty or does not exist."
+    lines: list[str] = []
+    for entry in entries:
+        entry_type = entry.get("type", "unknown")
+        entry_name = entry.get("name", "")
+        entry_path = entry.get("path", "")
+        lines.append(f"- [{entry_type}] {entry_name}  ({entry_path})")
+    return f"Directory listing for '{path}' ({len(entries)} entries):\n" + "\n".join(lines)
+
+
+@tool
+async def read_file_content(path: str) -> str:
+    """Read the text content of a local file on the user's device.
+
+    Returns the file content as a string.  Large files may be truncated
+    by the Electron client.
+    """
+    result = await execute_on_client(
+        action="read_file_content",
+        data={"path": path},
+    )
+    content: str = result.get("content", "")
+    if not content:
+        return f"File '{path}' is empty or could not be read."
+    return content
+
+
+@tool
+async def get_file_metadata(path: str) -> str:
+    """Get metadata for a local file: size, creation date, modification date, extension.
+
+    Returns a formatted summary of the file's metadata.
+    """
+    result = await execute_on_client(
+        action="get_file_metadata",
+        data={"path": path},
+    )
+    size = result.get("size", "unknown")
+    created = result.get("createdAt", "unknown")
+    modified = result.get("modifiedAt", "unknown")
+    extension = result.get("extension", "unknown")
+    name = result.get("name", path)
+    return (
+        f"File: {name}\n"
+        f"  Extension: {extension}\n"
+        f"  Size: {size} bytes\n"
+        f"  Created: {created}\n"
+        f"  Modified: {modified}"
+    )
+
+
+FILESYSTEM_TOOLS: list[Any] = [
+    list_directory,
+    read_file_content,
+    get_file_metadata,
+]

app/agents/note_agent.py

@@ -1,7 +1,8 @@
-"""Note agent — tool definitions for Markdown note CRUD."""
+"""Note agent — Markdown note management (list, get, create, update, delete)."""
 
 from __future__ import annotations
 
+import re
 from typing import Any
 
 from langchain_core.tools import tool
@@ -9,14 +10,38 @@ from langchain_core.tools import tool
 from app.core.llm import embed
 from app.core.ws_context import execute_on_client
 
+_UUID_RE = re.compile(
+    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$"
+)
+
+
+def _is_uuid(value: str) -> bool:
+    return bool(_UUID_RE.match(value))
+
+NOTE_SYSTEM_PROMPT = (
+    "You are a note-taking assistant. You help users create, retrieve, update,\n"
+    "and delete Markdown notes in their workspace.\n\n"
+    "Rules:\n"
+    "  - content is always Markdown; preserve formatting when updating\n"
+    "  - project_id is optional; link a note to a project when mentioned\n"
+    "  - When updating, call get_note first if you need to read existing content\n"
+    "    before appending or replacing sections\n"
+    "  - list_notes without project_id returns all notes; scope with project_id\n"
+    "    when the user is working within a specific project\n"
+    "  - project_id must be a UUID; if you only know a project name, do not pass it as project_id\n"
+    "  - Do not fabricate note content — reflect what the user provides or what\n"
+    "    is already in the note (retrieved via get_note)."
+)
+
 
 @tool
 async def list_notes(project_id: str = "") -> str:
     """List notes, optionally scoped to a project by project_id."""
+    normalized_project_id = project_id if (project_id and _is_uuid(project_id)) else ""
     result = await execute_on_client(
         action="select",
         table="notes",
-        filters={"projectId": project_id or None},
+        filters={"projectId": normalized_project_id or None},
     )
     rows = result.get("rows", [])
     if not rows:
@@ -105,4 +130,10 @@ async def delete_note(note_id: str) -> str:
     return f"Note {note_id} deleted."
 
 
-
+NOTE_TOOLS: list[Any] = [
+    list_notes,
+    get_note,
+    create_note,
+    update_note,
+    delete_note,
+]

app/agents/project_agent.py

@@ -1,4 +1,4 @@
-"""Project agent — tool definitions for project lifecycle CRUD."""
+"""Project agent — full lifecycle management (list, get, create, update, archive, delete)."""
 
 from __future__ import annotations
 
@@ -8,6 +8,22 @@ from langchain_core.tools import tool
 
 from app.core.ws_context import execute_on_client
 
+PROJECT_SYSTEM_PROMPT = (
+    "You are a project management assistant. You help users create, find,\n"
+    "update, and archive projects in their workspace.\n\n"
+    "Rules:\n"
+    "  - status must be one of: active, archived\n"
+    "  - client_id is optional; link to a client only when explicitly mentioned\n"
+    "  - ai_summary is populated only when the user asks for a project summary;\n"
+    "    derive it from context data — do not fabricate content\n"
+    "  - Use list_projects for scoped queries; list_all_projects only when the\n"
+    "    user wants a complete cross-client view including archived projects\n"
+    "  - get_project requires a project UUID; resolve the ID first by calling\n"
+    "    list_projects if you only have a project name\n"
+    "  - Prefer archiving (update_project status=archived) over deletion;\n"
+    "    only call delete_project when the user explicitly confirms deletion."
+)
+
 
 @tool
 async def list_projects(
@@ -117,4 +133,11 @@ async def delete_project(project_id: str) -> str:
     return f"Project {project_id} permanently deleted."
 
 
-
+PROJECT_TOOLS: list[Any] = [
+    list_projects,
+    list_all_projects,
+    get_project,
+    create_project,
+    update_project,
+    delete_project,
+]

app/agents/task_agent.py

@@ -1,14 +1,40 @@
-"""Task agent — tool definitions for task and task comment CRUD."""
+"""Task agent — full CRUD for tasks and task comments."""
 
 from __future__ import annotations
 
 from datetime import datetime, timezone
+import re
 from typing import Any
 
 from langchain_core.tools import tool
 
 from app.core.ws_context import execute_on_client
 
+_UUID_RE = re.compile(
+    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$"
+)
+
+
+def _is_uuid(value: str) -> bool:
+    return bool(_UUID_RE.match(value))
+
+TASK_SYSTEM_PROMPT = (
+    "You are a task management assistant for a project workspace.\n"
+    "You create, update, list, and track tasks and their comments.\n\n"
+    "Rules:\n"
+    "  - status must be one of: todo, in_progress, done\n"
+    "  - priority must be one of: high, medium, low\n"
+    "  - due_date is a Unix timestamp in milliseconds; convert human dates\n"
+    "  - assignees is a JSON-encoded array of strings (e.g. '[\"Alice\",\"Bob\"]')\n"
+    "  - project_id is optional; link to a project when the user mentions one\n"
+    "  - is_ai_suggested: 1 only when proactively proposing a task the user\n"
+    "    did not explicitly request; 0 otherwise\n"
+    "  - is_ai_suggested: 1 only when proactively proposing a task the user did not explicitly request; 0 otherwise\n"
+    "  - Use list_tasks_due_today for 'what's due today' queries\n"
+    "  - For update_task, use -1 for integer fields you do not want to change\n"
+    "  - Always confirm the action in plain, user-friendly language."
+)
+
 
 # ── Task tools ────────────────────────────────────────────────────────
 
@@ -22,11 +48,12 @@ async def list_tasks(
 ) -> str:
     """List tasks, optionally filtered by project_id, status (todo|in_progress|done),
     a search string, or an order_by field name (dueDate|priority|createdAt)."""
+    normalized_project_id = project_id if (project_id and _is_uuid(project_id)) else ""
     result = await execute_on_client(
         action="select",
         table="tasks",
         filters={
-            "projectId": project_id or None,
+            "projectId": normalized_project_id or None,
             "status": status or None,
             "search": search or None,
             "orderBy": order_by or None,
@@ -52,7 +79,6 @@ async def create_task(
     due_date: int = 0,
     project_id: str = "",
     is_ai_suggested: int = 0,
-    is_approved: int = 0,
 ) -> str:
     """Create a new task.
     title: task title (required)
@@ -63,7 +89,6 @@ async def create_task(
     due_date: Unix timestamp in milliseconds; 0 means no due date
     project_id: optional UUID of the parent project
     is_ai_suggested: 1 if proactively suggested, 0 if user-requested
-    is_approved: 0 until the user confirms; 1 when confirmed
     """
     result = await execute_on_client(
         action="insert",
@@ -77,7 +102,6 @@ async def create_task(
             "dueDate": due_date or None,
             "projectId": project_id or None,
             "isAiSuggested": is_ai_suggested,
-            "isApproved": is_approved,
         },
     )
     row = result["row"]
@@ -97,12 +121,10 @@ async def update_task(
     assignees: str = "",
     due_date: int = -1,
     project_id: str = "",
-    is_approved: int = -1,
 ) -> str:
     """Update fields on an existing task. Only pass fields you want to change.
     task_id: the task's UUID (required)
     due_date: -1 means unchanged; 0 clears the due date; any positive value sets it
-    is_approved: -1 means unchanged; 0 or 1 sets the value
     """
     updates: dict[str, Any] = {}
     if title:
@@ -119,8 +141,6 @@ async def update_task(
         updates["dueDate"] = due_date or None
     if project_id:
         updates["projectId"] = project_id
-    if is_approved != -1:
-        updates["isApproved"] = is_approved
     result = await execute_on_client(
         action="update",
         table="tasks",
@@ -188,8 +208,12 @@ async def add_task_comment(task_id: str, author: str, content: str) -> str:
         table="taskComments",
         data={"taskId": task_id, "author": author, "content": content},
     )
-    row = result["row"]
+    row = result.get("row", {})
-    return f"Comment added by {row['author']} on task {row['taskId']} (comment id: {row['id']})."
+    row_author = row.get("author", author)
+    # Electron payloads can vary (taskId vs task_id). Fall back to input task_id.
+    row_task_id = row.get("taskId") or row.get("task_id") or task_id
+    row_comment_id = row.get("id", "unknown")
+    return f"Comment added by {row_author} on task {row_task_id} (comment id: {row_comment_id})."
 
 
 @tool
@@ -199,4 +223,16 @@ async def delete_task_comment(comment_id: str) -> str:
     return f"Comment {comment_id} deleted."
 
 
+# ── Agent ─────────────────────────────────────────────────────────────
 
+
+TASK_TOOLS: list[Any] = [
+    list_tasks,
+    create_task,
+    update_task,
+    delete_task,
+    list_tasks_due_today,
+    list_task_comments,
+    add_task_comment,
+    delete_task_comment,
+]

app/agents/timeline_agent.py

@@ -1,21 +1,45 @@
-"""Timeline agent — tool definitions for project milestone CRUD."""
+"""Timeline agent — project milestone management (list, create, update, delete)."""
 
 from __future__ import annotations
 
+import re
 from typing import Any
 
 from langchain_core.tools import tool
 
 from app.core.ws_context import execute_on_client
 
+_UUID_RE = re.compile(
+    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$"
+)
+
+
+def _is_uuid(value: str) -> bool:
+    return bool(_UUID_RE.match(value))
+
+TIMELINE_SYSTEM_PROMPT = (
+    "You are a project timeline assistant. Timelines are milestone dates that\n"
+    "track progress on a project — they are not calendar events.\n\n"
+    "Rules:\n"
+    "  - project_id is REQUIRED for every create; confirm with the user if unknown\n"
+    "  - For listing, project_id must be a UUID; never pass plain names as project_id\n"
+    "  - date is a Unix timestamp in milliseconds; convert human-readable dates\n"
+    "  - is_ai_suggested: 1 when proactively proposing a timeline, 0 otherwise\n"
+    "  - is_ai_suggested: 1 when proactively proposing a timeline, 0 otherwise\n"
+    "  - For update_timeline, use -1 for integer fields you do not want to change\n"
+    "  - Listing without a project_id returns all timelines across projects\n"
+    "  - Always echo the title and formatted date in your confirmation."
+)
+
 
 @tool
 async def list_timelines(project_id: str = "") -> str:
     """List timelines. Provide project_id to scope to a specific project."""
+    normalized_project_id = project_id if (project_id and _is_uuid(project_id)) else ""
     result = await execute_on_client(
         action="select",
         table="timelines",
-        filters={"projectId": project_id or None},
+        filters={"projectId": normalized_project_id or None},
     )
     rows = result.get("rows", [])
     if not rows:
@@ -30,14 +54,12 @@ async def create_timeline(
     title: str,
     date: int,
     is_ai_suggested: int = 0,
-    is_approved: int = 0,
 ) -> str:
     """Create a project timeline (milestone).
     project_id: REQUIRED UUID of the parent project
     title: descriptive name for the milestone
     date: Unix timestamp in milliseconds
     is_ai_suggested: 1 if proactively suggested, 0 if user-requested
-    is_approved: 0 until the user confirms
     """
     result = await execute_on_client(
         action="insert",
@@ -47,7 +69,6 @@ async def create_timeline(
             "title": title,
             "date": date,
             "isAiSuggested": is_ai_suggested,
-            "isApproved": is_approved,
         },
     )
     row = result["row"]
@@ -59,20 +80,16 @@ async def update_timeline(
     timeline_id: str,
     title: str = "",
     date: int = -1,
-    is_approved: int = -1,
 ) -> str:
     """Update a timeline. Only pass fields that should change.
     timeline_id: UUID of the timeline (required)
     date: -1 means unchanged; any other value sets the new date (ms timestamp)
-    is_approved: -1 means unchanged; 0 or 1 sets the approval state
     """
     updates: dict[str, Any] = {}
     if title:
         updates["title"] = title
     if date != -1:
         updates["date"] = date
-    if is_approved != -1:
-        updates["isApproved"] = is_approved
     result = await execute_on_client(
         action="update",
         table="timelines",
@@ -89,4 +106,9 @@ async def delete_timeline(timeline_id: str) -> str:
     return f"Timeline {timeline_id} deleted."
 
 
-
+TIMELINE_TOOLS: list[Any] = [
+    list_timelines,
+    create_timeline,
+    update_timeline,
+    delete_timeline,
+]

app/api/middleware/auth.py

@@ -55,12 +55,15 @@ async def get_current_user(
         raise credentials_exc
 
     # Live tier lookup — subscription row is the authoritative source.
+    # In dev, fall back to 'power' (unlimited) so quota limits don't
+    # block local development when no Stripe subscription exists.
     from app.models import Subscription, User  # noqa: PLC0415
 
     result = await db.execute(
         select(Subscription.tier).where(Subscription.user_id == user_id)
     )
-    tier: str = result.scalar_one_or_none() or "free"
+    default_tier = "power" if settings.ENV == "dev" else "free"
+    tier: str = result.scalar_one_or_none() or default_tier
 
     # Fetch name/surname from user row.
     user_result = await db.execute(

app/api/routes/agent_setup.py

@@ -1,54 +1,42 @@
-"""Chatbot Journey endpoints — guided conversation to build an agent prompt_template.
+"""Chatbot Journey — WS-based guided conversation to build an agent prompt_template.
 
-Endpoints:
+The journey is driven entirely through WebSocket frames (no REST endpoints).
-  POST /agents/journey/start    — start a new journey session
+The device WS handler dispatches ``journey_start`` and ``journey_message``
-  POST /agents/journey/message  — continue the conversation
+frames to the functions exported here.
-
-Sessions are stored in-memory with a 30-minute TTL.  Stale entries are
-cleaned up lazily on access.  Upgrade to Redis for multi-instance deployments.
 
 Journey flow:
-  1. Client sends ``{ agent_type, agent_id? }`` to ``/start``.
+  1. FE sends ``journey_start`` frame with basic agent config (directory,
-  2. Server creates a session, calls the LLM with a contextual system prompt,
+     data_types, schedule).
-     and returns the first question.
+  2. Server creates an in-memory session, sets up a WS executor so the
-  3. Client sends follow-up messages to ``/message``.
+     setup LLM can use file-system tools, does a first directory scrape,
-  4. After 3-5 turns the LLM wraps up by emitting a ``prompt_template`` block
+     and sends back a ``journey_reply`` with the first question.
-     delimited by ``PROMPT_TEMPLATE_START`` / ``PROMPT_TEMPLATE_END``.
+  3. FE sends ``journey_message`` frames for each user reply.
-  5. Server parses the block, sets ``done=True``, and returns the template.
+  4. Server appends the user message, calls the LLM (which may read files
-
+     via tools), and sends back a ``journey_reply``.
-The ``prompt_template`` from the final response is meant to be stored in
+  5. After 3-5 turns the LLM wraps up by emitting a ``prompt_template``
-``LocalAgentConfig.prompt_template`` or ``CloudAgentConfig.prompt_template``
+     block delimited by ``PROMPT_TEMPLATE_START`` / ``PROMPT_TEMPLATE_END``.
-by the Electron client (via the agent CRUD endpoints).
+  6. Server parses the block, sends ``journey_reply`` with ``done=True``
+     and the template.  FE stores it locally.
 """
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 import uuid
 from dataclasses import dataclass, field
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
-from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
-from sqlalchemy import select
-from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.api.deps import get_current_user
+from app.agents.filesystem_agent import FILESYSTEM_TOOLS
+from app.config.settings import settings
+from app.core.langfuse_client import extract_usage, get_langfuse, get_prompt_or_fallback
 from app.core.llm import get_llm
-from app.db import get_session
-from app.models import CloudAgentConfig, LocalAgentConfig
-from app.schemas import (
-    JourneyMessageRequest,
-    JourneyResponse,
-    JourneyStartRequest,
-    UserProfile,
-)
 
 logger = logging.getLogger(__name__)
 
-router = APIRouter(prefix="/agents/journey", tags=["agents"])
-
 # ── Session TTL ───────────────────────────────────────────────────────────
 
 _SESSION_TTL_SECONDS: int = 1800  # 30 minutes
@@ -57,18 +45,26 @@ _SESSION_TTL_SECONDS: int = 1800  # 30 minutes
 _TEMPLATE_START = "PROMPT_TEMPLATE_START"
 _TEMPLATE_END = "PROMPT_TEMPLATE_END"
 
-# Maximum number of conversation turns before the LLM is nudged to wrap up.
+# Minimum turns before we consider nudging the LLM to wrap up.
-_MAX_TURNS: int = 5
+_MIN_TURNS_BEFORE_NUDGE: int = 3
+# Hard cap to avoid infinite loops (safety net, not the primary stopping criterion).
+_MAX_TURNS: int = 15
+# Max tool-calling steps per LLM invocation.
+_MAX_TOOL_STEPS: int = 6
 
 # ── In-memory session store ───────────────────────────────────────────────
 
 
 @dataclass
-class _JourneySession:
+class JourneySession:
     session_id: str
     user_id: str
     agent_type: str  # "local" | "cloud"
+    directory: str
+    data_types: list[str]
     history: list[dict[str, Any]] = field(default_factory=list)
+    system_prompt: str = ""
+    langfuse_prompt: Any = None
     created_at: float = field(default_factory=time.monotonic)
 
     def is_expired(self) -> bool:
@@ -76,84 +72,102 @@ class _JourneySession:
 
 
 # session_id → session
-_sessions: dict[str, _JourneySession] = {}
+_sessions: dict[str, JourneySession] = {}
 
 
-def _get_session(session_id: str, user_id: str) -> _JourneySession:
+def get_journey_session(session_id: str, user_id: str) -> JourneySession | None:
-    """Retrieve session; raise 404 on missing, expired, or wrong owner."""
+    """Retrieve session; return None on missing, expired, or wrong owner."""
     s = _sessions.get(session_id)
     if s is None or s.is_expired():
         _sessions.pop(session_id, None)
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Journey session not found or expired")
+        return None
     if s.user_id != user_id:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Journey session not found or expired")
+        return None
     return s
 
 
 # ── System prompt builder ─────────────────────────────────────────────────
 
-_LOCAL_PREAMBLE = """\
+_JOURNEY_SYSTEM_PROMPT = """\
-What kind of files are in the directories you want to monitor? \
-(for example: emails saved as .eml, documents in .pdf or .txt, markdown notes, etc.)"""
-
-_CLOUD_PREAMBLE = """\
-What kind of emails or messages should I look for? \
-(for example: client communications, invoices, meeting notes, project updates, etc.)"""
-
-_SYSTEM_PROMPT_TEMPLATE = """\
 You are a friendly assistant helping a freelancer configure a data-extraction agent.
-Your job is to understand exactly what data the user wants to extract from their {source_description} \
+Your job is to understand exactly what data the user wants to extract from their
-and produce a detailed prompt_template that a separate AI will use as its instruction set.
+local directory and produce a detailed prompt_template that a separate AI will use
+as its instruction set.
 
-Ask concise, focused questions one at a time.  Cover these topics (not necessarily in this order):
+The extraction agent already has this base behaviour built in:
-  1. The type and format of the source content.
+  - Reads each file using file-system tools.
-  2. Which data types to extract: tasks, notes, timelines, and/or projects.
+  - Creates records (tasks, notes, timelines, projects) via CRUD tools.
-  3. How fields should be mapped (e.g. email subject → task title).
+  - Sets isAiSuggested=1 on every new record.
-  4. Priority or status rules (e.g. "urgent" keyword → high priority).
+  - Only extracts data explicitly present in the files — it never invents information.
-  5. Any special handling, date extraction, or exclusions.
+The user's custom prompt is appended AFTER this base behaviour, so focus on
+what to look for and how to map it — not on the general extraction mechanics.
 
-After 3-5 questions (when you have enough information), output the final prompt_template between \
+You have access to file-system tools to explore the user's directory:
-these exact markers on their own lines:
+- list_directory: to see folder structure
+- read_file_content: to peek at file contents
+- get_file_metadata: to check file info
+
+The user's configured directory is: {directory}
+Target data types: {data_types}
+
+IMPORTANT — project assignment is handled automatically by the main agent runner
+before the custom prompt is ever used.  You MUST NOT ask the user about projects,
+projectId, or how to link records to projects.  Never include projectId logic or
+project creation instructions in the generated prompt_template.
+
+Start by exploring the directory to understand its structure.  Then ask concise,
+focused questions one at a time.  Cover these topics (not necessarily in this order):
+  1. The type and format of the source content (confirmed by your exploration).
+  2. How fields should be mapped (e.g. filename → task title).
+  3. Priority or status rules (e.g. "urgent" keyword → high priority).
+  4. Any special handling, date extraction, or exclusions.
+
+Once you reach 90% confidence, output the final prompt_template between these exact
+markers on their own lines:
 
 {template_start}
 <the complete extraction prompt here>
 {template_end}
 
-The prompt_template must be a self-contained instruction for an AI that receives a document/email/message \
+The prompt_template must be a self-contained instruction for an AI that reads files
-and must return a JSON array of records in this shape:
+and must perform CRUD operations using tools to create records.  It should specify:
-  [{{ "table": "<tasks|notes|timelines|projects>", "data": {{ <field: value> }} }}, ...]
+  - What entity types to create (tasks, notes, timelines) — never projects.
+  - How to map file content to record fields (camelCase: title, status, priority,
+    dueDate, content, etc.) — never include projectId.
+  - That isAiSuggested must be set to 1 on every new record.
+  - Concrete examples of mappings based on what you discovered in the directory.
 
-Rules for the generated template:
-  - Be explicit about field names (camelCase: title, status, priority, dueDate, projectId, content, etc.).
-  - Include concrete examples of mappings.
-  - Mention that Electron adds id/createdAt/updatedAt automatically.
-  - Set isAiSuggested: true and isApproved: false on every record.
 {existing_section}\
-Do not ask more than {max_turns} questions total. Start with your first question now.\
+Keep asking clarifying questions until you are at least 90% confident you have
+enough information to generate an accurate prompt_template.  Once you reach that
+confidence level, stop asking and produce the final template immediately.
+Begin by exploring the directory, then ask your first question.\
 """
 
 
-def _build_system_prompt(agent_type: str, existing_template: str | None) -> str:
+def _build_system_prompt(
-    source_description = (
+    directory: str,
-        "files in local directories" if agent_type == "local" else "emails and messages from cloud providers"
+    data_types: list[str],
-    )
+    existing_template: str | None = None,
+) -> tuple[str, Any]:
+    """Return ``(compiled_system_prompt, langfuse_prompt_obj_or_None)``."""
     existing_section = (
         f"\nThe user already has the following prompt_template — refine it based on their answers:\n"
         f"---\n{existing_template}\n---\n"
         if existing_template
         else ""
     )
-    return _SYSTEM_PROMPT_TEMPLATE.format(
+    template, prompt_obj = get_prompt_or_fallback(
-        source_description=source_description,
+        "journey_system", _JOURNEY_SYSTEM_PROMPT
+    )
+    compiled = template.format(
+        directory=directory,
+        data_types=", ".join(data_types),
         template_start=_TEMPLATE_START,
         template_end=_TEMPLATE_END,
         existing_section=existing_section,
-        max_turns=_MAX_TURNS,
     )
-
+    return compiled, prompt_obj
-
-def _first_question(agent_type: str) -> str:
-    return _LOCAL_PREAMBLE if agent_type == "local" else _CLOUD_PREAMBLE
 
 
 # ── Template extraction ───────────────────────────────────────────────────
@@ -168,11 +182,42 @@ def _extract_template(text: str) -> str | None:
     return text[start_idx:end_idx].strip() or None
 
 
-# ── LLM call ─────────────────────────────────────────────────────────────
+# ── LLM call with tool support ───────────────────────────────────────────
 
 
-async def _call_llm(system_prompt: str, history: list[dict[str, Any]]) -> str:
+def _as_text(content: Any) -> str:
-    """Build LangChain messages from history and invoke the LLM."""
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for item in content:
+            if isinstance(item, str):
+                parts.append(item)
+            elif isinstance(item, dict):
+                text = item.get("text")
+                if isinstance(text, str):
+                    parts.append(text)
+        return "".join(parts)
+    return str(content)
+
+
+async def _call_llm_with_tools(
+    system_prompt: str,
+    history: list[dict[str, Any]],
+    tools: list[Any],
+    *,
+    user_id: str = "",
+    session_id: str = "",
+    langfuse_prompt: Any = None,
+) -> str:
+    """Build LangChain messages from history and invoke the LLM with tools.
+
+    Handles tool-calling loops: if the LLM calls tools, execute them and
+    continue until a final text response is produced.
+    """
+    lf = get_langfuse()
     messages: list[Any] = [SystemMessage(content=system_prompt)]
     for turn in history:
         if turn["role"] == "user":
@@ -181,137 +226,242 @@ async def _call_llm(system_prompt: str, history: list[dict[str, Any]]) -> str:
             messages.append(AIMessage(content=turn["content"]))
 
     llm = get_llm(model=None, temperature=0.4)
-    response = await llm.ainvoke(messages)
+    llm_with_tools = llm.bind_tools(tools)
-    return response.content  # type: ignore[return-value]
+    tool_map = {tool_def.name: tool_def for tool_def in tools}
+
+    _span_ctx = (
+        lf.start_as_current_observation(
+            as_type="span",
+            name="journey-setup",
+            user_id=user_id or None,
+            session_id=session_id or None,
+            input=history[-1]["content"] if history else "",
+        )
+        if lf else None
+    )
+    _span = _span_ctx.__enter__() if _span_ctx else None
+
+    try:
+        for _ in range(_MAX_TOOL_STEPS):
+            _gen_ctx = (
+                lf.start_as_current_observation(
+                    as_type="generation",
+                    name="journey-setup-llm",
+                    model=settings.LLM_MODEL,
+                    prompt=langfuse_prompt,
+                    input=messages,
+                )
+                if lf else None
+            )
+            _gen = _gen_ctx.__enter__() if _gen_ctx else None
+            response: AIMessage = await llm_with_tools.ainvoke(messages)
+            if _gen_ctx:
+                _gen.update(output=_as_text(response.content), usage=extract_usage(response))
+                _gen_ctx.__exit__(None, None, None)
+
+            messages.append(response)
+
+            if not response.tool_calls:
+                if _span:
+                    _span.update(output=_as_text(response.content))
+                return _as_text(response.content)
+
+            for call in response.tool_calls:
+                call_name = str(call.get("name", ""))
+                call_args = call.get("args", {})
+                logger.info(
+                    "agent_setup: journey tool_call name=%s args=%s",
+                    call_name,
+                    json.dumps(call_args, ensure_ascii=True)[:500],
+                )
+
+                tool_fn = tool_map.get(call_name)
+                if tool_fn is None:
+                    tool_output = f"Unknown tool: {call_name}"
+                else:
+                    tool_output = await tool_fn.ainvoke(call_args)
+
+                logger.info(
+                    "agent_setup: journey tool_result name=%s output=%s",
+                    call_name,
+                    str(tool_output)[:800],
+                )
+                messages.append(ToolMessage(content=str(tool_output), tool_call_id=call["id"]))
+
+        # Fallback: exceeded max steps.
+        final = await llm.ainvoke(messages)
+        final_text = _as_text(final.content)
+        if _span:
+            _span.update(output=final_text)
+        return final_text
+    finally:
+        if _span_ctx:
+            _span_ctx.__exit__(None, None, None)
+        if lf:
+            lf.flush()
 
 
-# ── Existing-config loader ────────────────────────────────────────────────
+# ── Journey handlers (called from device_ws.py) ──────────────────────────
 
 
-async def _load_existing_template(
+async def handle_journey_start(
-    agent_id: str,
     user_id: str,
-    db: AsyncSession,
+    frame: dict[str, Any],
-) -> str | None:
+) -> dict[str, Any]:
-    """Return the prompt_template of an existing agent config, or None."""
+    """Handle a ``journey_start`` WS frame.
-    # Try local first, then cloud.
-    local_result = await db.execute(
-        select(LocalAgentConfig).where(
-            LocalAgentConfig.id == agent_id,
-            LocalAgentConfig.user_id == user_id,
-        )
-    )
-    local = local_result.scalar_one_or_none()
-    if local is not None:
-        return local.prompt_template
 
-    cloud_result = await db.execute(
+    Creates a session, runs the setup LLM with directory exploration,
-        select(CloudAgentConfig).where(
+    and returns the ``journey_reply`` payload.
-            CloudAgentConfig.id == agent_id,
-            CloudAgentConfig.user_id == user_id,
-        )
-    )
-    cloud = cloud_result.scalar_one_or_none()
-    return cloud.prompt_template if cloud is not None else None
-
-
-# ── Routes ────────────────────────────────────────────────────────────────
-
-
-@router.post("/start", response_model=JourneyResponse, status_code=status.HTTP_200_OK)
-async def start_journey(
-    body: JourneyStartRequest,
-    current_user: UserProfile = Depends(get_current_user),
-    db: AsyncSession = Depends(get_session),
-) -> JourneyResponse:
-    """Start a new Chatbot Journey session.
-
-    If ``agent_id`` is provided the session is pre-seeded with the existing
-    agent's ``prompt_template`` so the user can refine it.
     """
-    # Load existing template (may be None).
+    agent_type = frame.get("agent_type", "local")
-    existing_template: str | None = None
+    directory = frame.get("directory", "")
-    if body.agent_id:
+    data_types = frame.get("data_types", [])
-        existing_template = await _load_existing_template(body.agent_id, current_user.id, db)
+    existing_template = frame.get("existing_template")
-        # If agent_id was given but not found, proceed without seeding (don't 404 —
-        # the user may be starting a fresh journey for a not-yet-persisted config).
 
-    system_prompt = _build_system_prompt(body.agent_type, existing_template)
+    # Use the session_id provided by the FE so the reply matches the
-    first_question = _first_question(body.agent_type)
+    # listener key; fall back to a generated one if absent.
+    session_id = frame.get("session_id") or str(uuid.uuid4())
+    system_prompt, langfuse_prompt = _build_system_prompt(directory, data_types, existing_template)
 
-    session_id = str(uuid.uuid4())
+    session = JourneySession(
-    session = _JourneySession(
         session_id=session_id,
-        user_id=current_user.id,
+        user_id=user_id,
-        agent_type=body.agent_type,
+        agent_type=agent_type,
-        # Seed history with the AI's first question so it stays consistent.
+        directory=directory,
-        history=[{"role": "assistant", "content": first_question}],
+        data_types=data_types,
+        system_prompt=system_prompt,
+        langfuse_prompt=langfuse_prompt,
     )
-    # Store the system prompt inside the session for reuse in /message.
+
-    session.__dict__["_system_prompt"] = system_prompt  # type: ignore[index]
+    # The LLM will explore the directory using FILESYSTEM_TOOLS via the
+    # ws_context executor (already set by the WS handler before calling us).
+    # Seed with an initial user message — some providers (e.g. GitHub Copilot)
+    # require at least one user/input message to be present.
+    seed_history: list[dict[str, Any]] = [
+        {"role": "user", "content": "Hi, I'm ready to set up my agent. Please explore my directory and ask me your first question."},
+    ]
+    ai_reply = await _call_llm_with_tools(
+        system_prompt=system_prompt,
+        history=seed_history,
+        tools=list(FILESYSTEM_TOOLS),
+        user_id=user_id,
+        session_id=session_id,
+        langfuse_prompt=langfuse_prompt,
+    )
+
+    session.history.extend(seed_history)
+    session.history.append({"role": "assistant", "content": ai_reply})
     _sessions[session_id] = session
 
-    logger.info("Journey session %s started for user %s (agent_type=%s)", session_id, current_user.id, body.agent_type)
+    logger.info(
-    return JourneyResponse(session_id=session_id, message=first_question, done=False)
+        "agent_setup: journey session %s started for user %s (directory=%s)",
+        session_id,
+        user_id,
+        directory,
+    )
 
-
+    # Check if the LLM produced the template on the first turn (unlikely but possible).
-@router.post("/message", response_model=JourneyResponse, status_code=status.HTTP_200_OK)
-async def send_journey_message(
-    body: JourneyMessageRequest,
-    current_user: UserProfile = Depends(get_current_user),
-    db: AsyncSession = Depends(get_session),
-) -> JourneyResponse:
-    """Send a message in an existing Chatbot Journey session.
-
-    The server appends the user's message to the conversation history,
-    calls the LLM, and appends the AI reply.  When the LLM wraps up with a
-    ``prompt_template`` block the response includes ``done=True`` and the
-    extracted template.
-    """
-    session = _get_session(body.session_id, current_user.id)
-    system_prompt: str = session.__dict__.get("_system_prompt", _build_system_prompt(session.agent_type, None))  # type: ignore[assignment]
-
-    # Append user turn to history.
-    session.history.append({"role": "user", "content": body.message})
-
-    # Call the LLM with the full conversation so far.
-    ai_reply = await _call_llm(system_prompt, session.history)
-
-    # Append AI turn.
-    session.history.append({"role": "assistant", "content": ai_reply})
-
-    # Check if the LLM produced the final template.
     prompt_template = _extract_template(ai_reply)
     done = prompt_template is not None
 
-    # Strip the sentinel markers from the message shown to the user.
     display_message = ai_reply
     if done:
         display_message = (
             ai_reply[: ai_reply.index(_TEMPLATE_START)].strip()
             or "Here is your agent configuration. You can save it or continue refining."
         )
+        _sessions.pop(session_id, None)
 
-    if done:
+    return {
-        logger.info("Journey session %s completed for user %s", body.session_id, current_user.id)
+        "type": "journey_reply",
-        # Clean up the session immediately on completion.
+        "session_id": session_id,
-        _sessions.pop(body.session_id, None)
+        "message": display_message,
-    else:
+        "done": done,
-        # Nudge the LLM to wrap up after max turns.
+        "prompt_template": prompt_template,
+    }
+
+
+async def handle_journey_message(
+    user_id: str,
+    frame: dict[str, Any],
+) -> dict[str, Any]:
+    """Handle a ``journey_message`` WS frame.
+
+    Appends the user message, calls the LLM, and returns the
+    ``journey_reply`` payload.
+    """
+    session_id = frame.get("session_id", "")
+    message = frame.get("message", "")
+
+    session = get_journey_session(session_id, user_id)
+    if session is None:
+        return {
+            "type": "journey_reply",
+            "session_id": session_id,
+            "message": "Journey session not found or expired. Please start a new setup.",
+            "done": True,
+            "prompt_template": None,
+        }
+
+    # Append user turn.
+    session.history.append({"role": "user", "content": message})
+
+    # Call the LLM with tools.
+    ai_reply = await _call_llm_with_tools(
+        system_prompt=session.system_prompt,
+        history=session.history,
+        tools=list(FILESYSTEM_TOOLS),
+        user_id=session.user_id,
+        session_id=session_id,
+        langfuse_prompt=session.langfuse_prompt,
+    )
+
+    session.history.append({"role": "assistant", "content": ai_reply})
+
+    # Check if the LLM produced the final template.
+    prompt_template = _extract_template(ai_reply)
+    done = prompt_template is not None
+
+    # If the LLM didn't produce a template, nudge it once it has asked enough
+    # questions (>= _MIN_TURNS_BEFORE_NUDGE) or hits the hard safety cap.
+    if not done:
         turns = sum(1 for t in session.history if t["role"] == "user")
         if turns >= _MAX_TURNS:
-            # Add a system-level nudge as a hidden user message.
+            nudge_content = (
-            session.history.append({
-                "role": "user",
-                "content": (
                 "[System: You have enough information. Please generate the final "
                 f"prompt_template now, wrapped in {_TEMPLATE_START} / {_TEMPLATE_END} markers.]"
-                ),
-            })
-
-    return JourneyResponse(
-        session_id=body.session_id,
-        message=display_message,
-        done=done,
-        prompt_template=prompt_template,
             )
+            session.history.append({"role": "user", "content": nudge_content})
+
+            nudge_reply = await _call_llm_with_tools(
+                system_prompt=session.system_prompt,
+                history=session.history,
+                tools=list(FILESYSTEM_TOOLS),
+                user_id=session.user_id,
+                session_id=session_id,
+                langfuse_prompt=session.langfuse_prompt,
+            )
+            session.history.append({"role": "assistant", "content": nudge_reply})
+
+            prompt_template = _extract_template(nudge_reply)
+            if prompt_template is not None:
+                done = True
+                ai_reply = nudge_reply
+
+    display_message = ai_reply
+    if done:
+        display_message = (
+            ai_reply[: ai_reply.index(_TEMPLATE_START)].strip()
+            if _TEMPLATE_START in ai_reply
+            else "Here is your agent configuration. You can save it or continue refining."
+        )
+        _sessions.pop(session_id, None)
+        logger.info("agent_setup: journey session %s completed for user %s", session_id, user_id)
+
+    return {
+        "type": "journey_reply",
+        "session_id": session_id,
+        "message": display_message,
+        "done": done,
+        "prompt_template": prompt_template,
+    }

app/api/routes/agents.py

@@ -1,45 +1,36 @@
-"""Agent CRUD routes: local directory agents and cloud connector agents.
+"""Agent routes.
 
-Endpoints:
+Backend responsibilities are intentionally minimal:
-  GET    /agents/catalog            — hardcoded agent type catalog
+    GET  /agents/catalog         — static catalog for UI display
-  GET    /agents/local              — list user's local agent configs
+    POST /agents/can-create      — billing eligibility check
-  POST   /agents/local              — create local agent (tier-gated)
+    POST /agents/trigger         — trigger a local agent run
-  PUT    /agents/local/{agent_id}   — partial update (ownership check)
+
-  DELETE /agents/local/{agent_id}   — delete + cascade run logs
+Agent configuration is owned by the Electron app and is not persisted
-  GET    /agents/cloud              — list user's cloud agent configs
+in backend agent-config tables.
-  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
+import uuid
-from typing import Any
+from datetime import datetime, timedelta, timezone
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, status
-from pydantic import BaseModel
+from sqlalchemy import func, select
-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.agent_runner import is_agent_running, 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.models import AgentRunLog, LocalAgentConfig
 from app.schemas import (
     AgentCatalogItem,
+    AgentCreationCheckRequest,
+    AgentCreationCheckResponse,
     AgentRunLogResponse,
-    CloudAgentConfigCreate,
+    AgentTriggerRequest,
-    CloudAgentConfigResponse,
-    CloudAgentConfigUpdate,
-    LocalAgentConfigCreate,
-    LocalAgentConfigResponse,
-    LocalAgentConfigUpdate,
     UserProfile,
 )
 
@@ -56,39 +47,21 @@ def _dt_ms_opt(dt: datetime | None) -> int | None:
     return int(dt.timestamp() * 1000) if dt else None
 
 
-# ── Model → schema converters ─────────────────────────────────────────
+def _to_data_types(values: list[str]) -> list[str]:
-
+    normalize = {
-def _to_local_response(a: LocalAgentConfig) -> LocalAgentConfigResponse:
+        "task": "tasks",           "tasks": "tasks",
-    return LocalAgentConfigResponse(
+        "note": "notes",           "notes": "notes",
-        id=a.id,
+        "timeline": "timelines",   "timelines": "timelines",   "timelineEvents": "timelines",
-        name=a.name,
+        "project": "projects",     "projects": "projects",
-        device_id=a.device_id,
+    }
-        directory_paths=a.directory_paths,
+    seen: set[str] = set()
-        data_types=a.data_types,
+    result: list[str] = []
-        prompt_template=a.prompt_template,
+    for v in values:
-        file_extensions=a.file_extensions,
+        mapped = normalize.get(v)
-        schedule_cron=a.schedule_cron,
+        if mapped and mapped not in seen:
-        enabled=a.enabled,
+            seen.add(mapped)
-        last_run_at=_dt_ms_opt(a.last_run_at),
+            result.append(mapped)
-        created_at=_dt_ms(a.created_at),
+    return result
-        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:
@@ -105,77 +78,42 @@ def _to_run_log_response(log: AgentRunLog) -> AgentRunLogResponse:
     )
 
 
-# ── Ownership-checked lookups ─────────────────────────────────────────
+def _enforce_agent_limit(tier: str, current_count: int) -> int:
-
-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.",
         )
+    return limit
 
 
-# ── Local page schema (used by runs endpoint) ─────────────────────────
+async def _enforce_run_frequency(
+    tier: str,
+    user_id: str,
+    db: AsyncSession,
+) -> None:
+    """Raise HTTP 402 if the user has exceeded their daily batch run limit."""
+    limit: int = FEATURES.get(tier, FEATURES["free"])["batch_runs_per_day"]
+    if limit == -1:
+        return  # unlimited
 
-class _RunsPage(BaseModel):
+    today_start = datetime.now(timezone.utc).replace(
-    total: int
+        hour=0, minute=0, second=0, microsecond=0
-    page: int
+    )
-    limit: int
+    result = await db.execute(
-    items: list[AgentRunLogResponse]
+        select(func.count(AgentRunLog.id)).where(
+            AgentRunLog.user_id == user_id,
+            AgentRunLog.started_at >= today_start,
+        )
+    )
+    runs_today: int = result.scalar_one()
+
+    if runs_today >= limit:
+        raise HTTPException(
+            status_code=status.HTTP_402_PAYMENT_REQUIRED,
+            detail=f"Daily batch run limit ({limit}) reached for your tier. Upgrade for more runs.",
+        )
 
 
 # ── Catalog ───────────────────────────────────────────────────────────
@@ -209,229 +147,61 @@ async def get_agent_catalog(
     ]
 
 
-# ── Local agent CRUD ──────────────────────────────────────────────────
+@router.post("/can-create", response_model=AgentCreationCheckResponse)
-
+async def can_create_agent(
-@router.get("/local", response_model=list[LocalAgentConfigResponse])
+    body: AgentCreationCheckRequest,
-async def list_local_agents(
     current_user: UserProfile = Depends(get_current_user),
-    db: AsyncSession = Depends(get_session),
+) -> AgentCreationCheckResponse:
-) -> list[LocalAgentConfigResponse]:
+    """Check if the user can create one more agent based on billing tier.
-    """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()]
 
-
+    Since configuration is client-owned, the Electron app sends its current
-@router.post("/local", response_model=LocalAgentConfigResponse, status_code=status.HTTP_201_CREATED)
+    active agent count and the backend applies tier limits.
-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))
+    limit: int = FEATURES.get(current_user.tier, FEATURES["free"])["batch_active"]
-    agent = LocalAgentConfig(
+    allowed = limit == -1 or body.active_agents < limit
-        user_id=current_user.id,
+    return AgentCreationCheckResponse(
-        name=body.name,
+        allowed=allowed,
-        device_id=body.device_id,
+        tier=current_user.tier,
-        directory_paths=body.directory_paths,
+        active_agents=body.active_agents,
-        data_types=body.data_types,
+        limit=limit,
-        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)
+@router.post("/trigger", response_model=AgentRunLogResponse, status_code=status.HTTP_202_ACCEPTED)
-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,
+    body: AgentTriggerRequest,
     current_user: UserProfile = Depends(get_current_user),
     db: AsyncSession = Depends(get_session),
 ) -> AgentRunLogResponse:
-    """Manually trigger an agent run.
+    """Trigger a local agent run using client-provided configuration."""
+    _enforce_agent_limit(current_user.tier, body.active_agents)
+    await _enforce_run_frequency(current_user.tier, current_user.id, db)
 
-    Looks up the agent config (local or cloud) by ID with ownership check,
+    config = LocalAgentConfig(
-    creates a run log entry with ``status="running"``, and returns it.
+        id=str(uuid.uuid4()),
+        user_id=current_user.id,
+        device_id=body.device_id,
+        name="Local Directory Monitor",
+        directory_paths=[body.directory],
+        data_types=_to_data_types(body.what_to_extract),
+        prompt_template=body.custom_agent_prompt,
+        file_extensions=[],
+        schedule_cron=body.batch_interval,
+        enabled=True,
+    )
 
-    Actual dispatch to the agent runner is wired in Step 3.4 once
+    # Use the FE's stable agent_id if provided, fall back to the ephemeral config id.
-    ``DeviceConnectionManager`` and ``agent_runner`` are available.
+    stable_agent_id = body.agent_id or config.id
-    """
-    # 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(
+    if is_agent_running(stable_agent_id):
-        select(LocalAgentConfig).where(
+        raise HTTPException(
-            LocalAgentConfig.id == agent_id,
+            status_code=status.HTTP_409_CONFLICT,
-            LocalAgentConfig.user_id == current_user.id,
+            detail="Agent is already running. Only one run per agent is allowed at a time.",
         )
-    )
-    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_id=stable_agent_id,
-        agent_type=agent_type,
+        agent_type="local",
         user_id=current_user.id,
         status="running",
     )
@@ -439,14 +209,14 @@ async def trigger_agent_run(
     await db.commit()
     await db.refresh(run_log)
 
-    # Dispatch the run as a background task — returns 202 immediately.
+    run_context = {
-    if agent_type == "local" and local_config is not None:
+        "type": "agent_batch",
+        "run_id": run_log.id,
+        "agent_id": stable_agent_id,
+    }
+
     asyncio.create_task(
-            run_local_agent(current_user.id, local_config, run_log, device_manager)
+        run_local_agent(current_user.id, config, run_log, device_manager, run_context)
-        )
-    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/chat.py

@@ -10,9 +10,7 @@ from fastapi.responses import JSONResponse
 
 from app.api.deps import get_current_user
 from app.core.deep_agent import run_home
-from app.core.memory_middleware import MemoryMiddleware
+from app.schemas import ChatRequest, UserProfile
-from app.db import async_session
-from app.schemas import ChatRequest, ChatResponse, UserProfile
 
 router = APIRouter(prefix="/chat", tags=["chat"])
 
@@ -22,21 +20,10 @@ async def chat(
     body: ChatRequest,
     current_user: UserProfile = Depends(get_current_user),
 ) -> JSONResponse:
-    """Route a chat message through the Home deep agent (non-streaming)."""
+    """REST fallback for home chat when websocket streaming is unavailable."""
-    async with async_session() as db:
+    response = await run_home(
-        memory = MemoryMiddleware(db)
-        memory_context = await memory.enrich_context(current_user.id, body.message)
-
-    context = {
-        **body.context.model_dump(),
-        **memory_context,
-    }
-
-    response_text = await run_home(
         user_id=current_user.id,
         message=body.message,
-        context=context,
+        context=body.context.model_dump(),
-        db_session_factory=async_session,
     )
-    result = ChatResponse(response=response_text)
+    return JSONResponse(content={"response": response})
-    return JSONResponse(content=result.model_dump())

app/api/routes/device_ws.py

@@ -15,8 +15,8 @@ Protocol:
 
 Incoming frame dispatch:
   - ``tool_result``      → resolves a pending tool-call Future.
-  - ``agent_data``     → enqueued in the per-run agent data queue.
+  - ``journey_start``    → starts a guided setup journey session.
-  - ``agent_complete`` → sends None sentinel to close the queue stream.
+  - ``journey_message``  → continues a journey conversation.
   - ``pong``             → heartbeat acknowledgement (updates last-seen).
   - unknown types        → logged, ignored.
 
@@ -39,12 +39,13 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 from jose import JWTError, jwt
 from sqlalchemy import update
 
+from app.api.routes.agent_setup import handle_journey_message, handle_journey_start
 from app.config.settings import settings
 from app.core.agent_runner import trigger_pending_runs
+from app.core.deep_agent import run_floating_stream, run_home_stream
 from app.core.device_manager import device_manager
 from app.core.memory_middleware import MemoryMiddleware
-from app.core.deep_agent import run_home_stream, run_floating_stream
+from app.core.output_formatter import StreamFormatter
-from app.core.output_formatter import HomeFormatter, FloatingFormatter
 from app.core.ws_context import clear_client_executor, set_client_executor
 from app.db import async_session
 from app.models import AgentRunLog
@@ -147,37 +148,6 @@ async def _message_loop(websocket: WebSocket, user_id: str) -> None:
                     "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 == WsFrameType.home_request:
             asyncio.create_task(
                 _handle_home_request(websocket, user_id, frame)
@@ -188,6 +158,16 @@ async def _message_loop(websocket: WebSocket, user_id: str) -> None:
                 _handle_floating_request(websocket, user_id, frame)
             )
 
+        elif frame_type == WsFrameType.journey_start:
+            asyncio.create_task(
+                _handle_journey_start(websocket, user_id, frame)
+            )
+
+        elif frame_type == WsFrameType.journey_message:
+            asyncio.create_task(
+                _handle_journey_message(websocket, user_id, frame)
+            )
+
         elif frame_type == "pong":
             # Heartbeat ack — nothing to do, connection is alive.
             pass
@@ -200,35 +180,13 @@ async def _message_loop(websocket: WebSocket, user_id: str) -> None:
 
 # ── v3 Chat Handlers ──────────────────────────────────────────────────
 
-_WS_TOOL_CALL_TIMEOUT = 30  # seconds to wait for Electron tool_result
-
-
 async def _make_ws_executor(websocket: WebSocket, user_id: str):
     """Return a callback that sends tool_call frames and awaits tool_result."""
     async def _executor(payload: dict) -> dict:
         payload["type"] = WsFrameType.tool_call
-        call_id = payload["id"]
-        logger.info("ws_executor: sending tool_call id=%s action=%s", call_id, payload.get("action"))
         await websocket.send_text(json.dumps(payload))
-        future = device_manager.create_pending_call(user_id, call_id)
+        future = device_manager.create_pending_call(user_id, payload["id"])
-        try:
+        return await future
-            result = await asyncio.wait_for(future, timeout=_WS_TOOL_CALL_TIMEOUT)
-        except asyncio.TimeoutError:
-            logger.error(
-                "ws_executor: timeout waiting for tool_result id=%s action=%s user=%s",
-                call_id, payload.get("action"), user_id,
-            )
-            # Clean up the pending future so it doesn't leak
-            conn = device_manager._connections.get(user_id)
-            if conn:
-                conn.pending_calls.pop(call_id, None)
-            return {"error": f"Tool call timed out after {_WS_TOOL_CALL_TIMEOUT}s", "rows": []}
-        logger.info("ws_executor: tool_result id=%s result_type=%s result_keys=%s",
-                     call_id, type(result).__name__,
-                     list(result.keys()) if isinstance(result, dict) else "N/A")
-        if result is None:
-            logger.error("ws_executor: future resolved to None for call_id=%s user=%s", call_id, user_id)
-        return result
     return _executor
 
 
@@ -241,14 +199,27 @@ async def _handle_home_request(
     request_id = frame.get("request_id") or str(uuid4())
     message: str = frame.get("message", "")
     session_id: str = frame.get("session_id") or str(uuid4())
+    logger.info(
+        "device_ws: home_request_start user=%s req=%s session=%s msg=%s",
+        user_id,
+        request_id,
+        session_id,
+        message[:200],
+    )
 
     # ── Memory: enrich context before LLM call ────────────────────────
     async with async_session() as db:
         memory = MemoryMiddleware(db)
-        memory_context = await memory.enrich_context(user_id, message)
+        memory_context = await memory.enrich_context(
+            user_id,
+            message,
+            trace_id=request_id,
+            session_id=session_id,
+        )
 
     context: dict = {
         "conversation_history": frame.get("conversation_history", []),
+        "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
         **memory_context,
     }
 
@@ -256,12 +227,11 @@ async def _handle_home_request(
     set_client_executor(executor)
     response_chunks: list[str] = []
     try:
-        event_stream = run_home_stream(
+        event_stream = run_home_stream(user_id, message, context)
-            user_id, message, context, db_session_factory=async_session
+        formatter = StreamFormatter(request_id=request_id)
-        )
-        formatter = HomeFormatter(request_id=request_id)
         async for ws_frame in formatter.format(event_stream):
             await websocket.send_text(ws_frame.model_dump_json())
+            # Collect text chunks to build the full response for episode storage
             if ws_frame.type == "stream_text":  # type: ignore[union-attr]
                 response_chunks.append(ws_frame.chunk)  # type: ignore[union-attr]
     except Exception as exc:
@@ -276,7 +246,14 @@ async def _handle_home_request(
     async with async_session() as db:
         memory = MemoryMiddleware(db)
         await memory.store_episode(
-            user_id, session_id, message, "".join(response_chunks)
+            user_id, session_id, message, "".join(response_chunks), trace_id=request_id
+        )
+    logger.info(
+        "device_ws: home_request_end user=%s req=%s session=%s response_chars=%d",
+        user_id,
+        request_id,
+        session_id,
+        len("".join(response_chunks)),
     )
 
 
@@ -290,23 +267,37 @@ async def _handle_floating_request(
     message: str = frame.get("message", "")
     session_id: str = frame.get("session_id") or str(uuid4())
     scope: dict = frame.get("scope", {})
+    logger.info(
+        "device_ws: floating_request_start user=%s req=%s session=%s scope=%s msg=%s",
+        user_id,
+        request_id,
+        session_id,
+        json.dumps(scope, ensure_ascii=True)[:200],
+        message[:200],
+    )
 
     # ── Memory: enrich context before LLM call ────────────────────────
     async with async_session() as db:
         memory = MemoryMiddleware(db)
-        memory_context = await memory.enrich_context(user_id, message)
+        memory_context = await memory.enrich_context(
+            user_id,
+            message,
+            trace_id=request_id,
+            session_id=session_id,
+        )
 
-    context: dict = {"scope": scope, **memory_context}
+    context: dict = {
+        "scope": scope,
+        "_debug": {"request_id": request_id, "session_id": session_id, "user_id": user_id},
+        **memory_context,
+    }
 
     executor = await _make_ws_executor(websocket, user_id)
     set_client_executor(executor)
     response_chunks: list[str] = []
     try:
-        event_stream = run_floating_stream(
+        event_stream = run_floating_stream(user_id, message, context)
-            user_id, message, context, scope=scope,
+        formatter = StreamFormatter(request_id=request_id)
-            db_session_factory=async_session,
-        )
-        formatter = FloatingFormatter(request_id=request_id)
         async for ws_frame in formatter.format(event_stream):
             await websocket.send_text(ws_frame.model_dump_json())
             if ws_frame.type == "stream_text":  # type: ignore[union-attr]
@@ -323,8 +314,72 @@ async def _handle_floating_request(
     async with async_session() as db:
         memory = MemoryMiddleware(db)
         await memory.store_episode(
-            user_id, session_id, message, "".join(response_chunks)
+            user_id, session_id, message, "".join(response_chunks), trace_id=request_id
         )
+    logger.info(
+        "device_ws: floating_request_end user=%s req=%s session=%s response_chars=%d",
+        user_id,
+        request_id,
+        session_id,
+        len("".join(response_chunks)),
+    )
+
+
+# ── v4 Journey Handlers ─────────────────────────────────────────────
+
+
+async def _handle_journey_start(
+    websocket: WebSocket,
+    user_id: str,
+    frame: dict,
+) -> None:
+    """Handle a journey_start frame — explores directory and sends first question."""
+    executor = await _make_ws_executor(websocket, user_id)
+    set_client_executor(executor)
+    try:
+        reply = await handle_journey_start(user_id, frame)
+        await websocket.send_text(json.dumps(reply))
+    except Exception as exc:
+        logger.error(
+            "device_ws: journey_start failed user=%s: %s", user_id, exc
+        )
+        await websocket.send_text(json.dumps({
+            "type": "journey_reply",
+            "session_id": frame.get("session_id", ""),
+            "message": f"Failed to start journey: {exc}",
+            "done": True,
+            "prompt_template": None,
+        }))
+    finally:
+        clear_client_executor()
+
+
+async def _handle_journey_message(
+    websocket: WebSocket,
+    user_id: str,
+    frame: dict,
+) -> None:
+    """Handle a journey_message frame — continues the journey conversation."""
+    executor = await _make_ws_executor(websocket, user_id)
+    set_client_executor(executor)
+    try:
+        reply = await handle_journey_message(user_id, frame)
+        await websocket.send_text(json.dumps(reply))
+    except Exception as exc:
+        session_id = frame.get("session_id", "")
+        logger.error(
+            "device_ws: journey_message failed user=%s session=%s: %s",
+            user_id, session_id, exc,
+        )
+        await websocket.send_text(json.dumps({
+            "type": "journey_reply",
+            "session_id": session_id,
+            "message": f"Journey error: {exc}",
+            "done": True,
+            "prompt_template": None,
+        }))
+    finally:
+        clear_client_executor()
 
 
 # ── Heartbeat ─────────────────────────────────────────────────────────
@@ -360,6 +415,3 @@ async def _mark_runs_disconnected(user_id: str) -> None:
             user_id,
             exc,
         )
-
-
-

app/billing/tier_manager.py

@@ -21,6 +21,7 @@ FEATURES: dict[str, dict[str, Any]] = {
     "free": {
         "agents": 3,
         "batch_active": 2,
+        "batch_runs_per_day": 5,
         "cloud_storage_gb": 0,
         "backup_gb": 0,
         "providers": 1,
@@ -31,6 +32,7 @@ FEATURES: dict[str, dict[str, Any]] = {
     "pro": {
         "agents": -1,           # unlimited
         "batch_active": 10,
+        "batch_runs_per_day": 50,
         "cloud_storage_gb": 5,
         "backup_gb": 5,
         "providers": -1,
@@ -41,6 +43,7 @@ FEATURES: dict[str, dict[str, Any]] = {
     "power": {
         "agents": -1,
         "batch_active": -1,     # unlimited
+        "batch_runs_per_day": -1,  # unlimited
         "cloud_storage_gb": 25,
         "backup_gb": 25,
         "providers": -1,
@@ -51,6 +54,7 @@ FEATURES: dict[str, dict[str, Any]] = {
     "team": {
         "agents": -1,
         "batch_active": -1,
+        "batch_runs_per_day": -1,  # unlimited
         "cloud_storage_gb": -1,  # unlimited
         "backup_gb": -1,         # unlimited
         "providers": -1,
@@ -77,16 +81,18 @@ class TierManager:
     async def get_tier(self, user_id: str, db: AsyncSession) -> BillingTier:
         """Return the current billing tier for ``user_id`` from the DB.
 
-        Falls back to ``'free'`` when no subscription row exists.
+        Falls back to ``'power'`` in dev (unlimited) or ``'free'`` in prod
+        when no subscription row exists.
         """
         from app.models import Subscription  # noqa: PLC0415
+        from app.config.settings import settings  # noqa: PLC0415
 
         result = await db.execute(
             select(Subscription.tier).where(Subscription.user_id == user_id)
         )
         tier: str | None = result.scalar_one_or_none()
         if tier is None or tier not in FEATURES:
-            return "free"
+            return "power" if settings.ENV == "dev" else "free"
         return tier  # type: ignore[return-value]
 
     # ── Feature access ───────────────────────────────────────────────────

app/config/settings.py

@@ -52,6 +52,10 @@ class Settings(BaseSettings):
 
     CORS_ORIGINS: list[str] = ["app://.", "http://localhost:3000", "http://localhost:5173"]
 
+    LANGFUSE_SECRET_KEY: str = ""
+    LANGFUSE_PUBLIC_KEY: str = ""
+    LANGFUSE_HOST: str = "https://cloud.langfuse.com"
+
     ENV: Literal["dev", "prod"] = "dev"
 
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

app/core/agent_registry.py

@@ -0,0 +1,30 @@
+"""Minimal agent base types retained for compatibility with batch runners."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class BaseAgent(ABC):
+    """Common base for non-chat agents still using the old base contract."""
+
+    def __init__(
+        self,
+        user_id: str = "",
+        shared_memory: dict[str, Any] | None = None,
+        vector_store_context: list[str] | None = None,
+    ) -> None:
+        self.user_id = user_id
+        self.shared_memory: dict[str, Any] = shared_memory or {}
+        self.vector_store_context: list[str] = vector_store_context or []
+
+    @abstractmethod
+    def get_name(self) -> str: ...
+
+    @abstractmethod
+    def get_description(self) -> str: ...
+
+    @property
+    def skills(self) -> list[str]:
+        return []

app/core/device_manager.py

@@ -3,20 +3,15 @@
 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:
+The manager handles the **tool-call round-trip** pattern:
-
+  - Backend sends ``tool_call`` frame → Electron executes the action →
-1. **Tool-call round-trip** (bidirectional CRUD):
+    returns ``tool_result`` frame.
-   - 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):
+This pattern is used by all tools (CRUD, file-system, etc.) via
-   - Backend sends ``agent_run`` frame → Electron reads files and sends
+``execute_on_client()`` in ``ws_context.py``.
-     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.
@@ -42,8 +37,6 @@ class DeviceConnection:
     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:
@@ -153,31 +146,6 @@ class DeviceConnectionManager:
         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/core/langfuse_client.py

@@ -0,0 +1,114 @@
+"""Langfuse observability — singleton client and prompt helpers.
+
+If LANGFUSE_SECRET_KEY / LANGFUSE_PUBLIC_KEY are not set,
+all helpers are no-ops so the app works without Langfuse configured.
+
+Usage
+-----
+Tracing::
+
+    from app.core.langfuse_client import get_langfuse
+
+    lf = get_langfuse()
+    if lf:
+        with lf.start_as_current_observation(as_type="span", name="my-agent") as span:
+            span.update(input=user_message)
+            # ... do work ...
+            span.update(output=result)
+        lf.flush()
+
+Prompt management::
+
+    from app.core.langfuse_client import get_prompt_or_fallback
+
+    text, prompt_obj = get_prompt_or_fallback("home_system", FALLBACK_PROMPT)
+    # Use text as the system prompt; pass prompt_obj to generations for linking.
+
+Linking a prompt to a generation::
+
+    with lf.start_as_current_observation(
+        as_type="generation",
+        name="llm-call",
+        model="gpt-4o",
+        prompt=prompt_obj,   # links generation → prompt version in the UI
+        input=messages,
+    ) as gen:
+        response = await llm.ainvoke(messages)
+        gen.update(output=response.content, usage=_usage(response))
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_client: Any = None
+_initialized: bool = False
+
+
+def get_langfuse() -> Any | None:
+    """Return the Langfuse singleton, or ``None`` when not configured."""
+    global _client, _initialized
+    if _initialized:
+        return _client
+    _initialized = True
+
+    from app.config.settings import settings  # local import to avoid circular deps
+
+    if not settings.LANGFUSE_SECRET_KEY or not settings.LANGFUSE_PUBLIC_KEY:
+        logger.debug("langfuse: not configured — observability disabled")
+        return None
+
+    try:
+        from langfuse import Langfuse
+
+        _client = Langfuse(
+            secret_key=settings.LANGFUSE_SECRET_KEY,
+            public_key=settings.LANGFUSE_PUBLIC_KEY,
+            host=settings.LANGFUSE_HOST,
+        )
+        logger.info("langfuse: client initialized host=%s", settings.LANGFUSE_HOST)
+    except Exception as exc:
+        logger.warning("langfuse: failed to initialize: %s", exc)
+        _client = None
+
+    return _client
+
+
+def get_prompt_or_fallback(name: str, fallback: str) -> tuple[str, Any]:
+    """Fetch a text prompt from Langfuse; fall back to ``fallback`` on any error.
+
+    Returns ``(prompt_text, prompt_obj_or_None)``.
+
+    * ``prompt_text`` — the raw template string (variables not yet substituted).
+      Callers perform variable substitution with Python's ``.format()``.
+    * ``prompt_obj`` — the Langfuse prompt object, or ``None`` when Langfuse is
+      unavailable / the fetch failed.  Pass this to generation observations so
+      Langfuse links the generation to the exact prompt version in the UI.
+    """
+    lf = get_langfuse()
+    if lf is None:
+        return fallback, None
+
+    try:
+        prompt = lf.get_prompt(name, label="production", fallback=fallback)
+        # For text-type prompts .prompt holds the raw template string.
+        raw = prompt.prompt if hasattr(prompt, "prompt") and isinstance(prompt.prompt, str) else fallback
+        return raw, prompt
+    except Exception as exc:
+        logger.warning("langfuse: get_prompt %r failed: %s — using fallback", name, exc)
+        return fallback, None
+
+
+def extract_usage(response: Any) -> dict[str, int]:
+    """Extract token usage from a LangChain AI message into Langfuse format."""
+    meta = getattr(response, "usage_metadata", None)
+    if not meta:
+        return {}
+    return {
+        "input": int(meta.get("input_tokens", 0)),
+        "output": int(meta.get("output_tokens", 0)),
+        "total": int(meta.get("total_tokens", 0)),
+    }

app/core/llm.py

@@ -1,6 +1,6 @@
 """LLM factory — centralised model instantiation via LiteLLM.
 
-Every agent and the deep-agent supervisors call ``get_llm()`` or ``get_router_llm()``
+Every agent and the orchestrator call ``get_llm()`` or ``get_router_llm()``
 instead of directly constructing a provider-specific class.  The model string
 follows the `LiteLLM model naming convention
 <https://docs.litellm.ai/docs/providers>`_:
@@ -18,6 +18,7 @@ Switch providers by changing **LLM_MODEL** / **LLM_ROUTER_MODEL** in ``.env``
 from __future__ import annotations
 
 import os
+import warnings
 
 from openai import AsyncOpenAI
 import litellm
@@ -32,6 +33,14 @@ from app.config.settings import settings
 # Drop them silently instead of raising UnsupportedParamsError.
 litellm.drop_params = True
 
+# Some provider responses include a plain dict in the `usage` field where a
+# richer Pydantic model is expected. This warning is noisy but non-fatal.
+warnings.filterwarnings(
+    "ignore",
+    message=r"PydanticSerializationUnexpectedValue\(Expected `ResponseAPIUsage`",
+    category=UserWarning,
+)
+
 
 def _api_key_for_model(model: str) -> str | None:
     """Return the most appropriate API key for the given LiteLLM model string."""

app/core/memory_middleware.py

@@ -43,15 +43,21 @@ _PROACTIVE_CONFIDENCE_THRESHOLD = 0.6
 
 
 class MemoryMiddleware:
-    """Enrich agent context with memory and persist interactions after."""
+    """Enrich orchestrator context with memory and persist interactions after."""
 
     def __init__(self, db: AsyncSession) -> None:
         self._db = db
 
     # ── Public API ────────────────────────────────────────────────────────────
 
-    async def enrich_context(self, user_id: str, message: str) -> dict[str, Any]:
+    async def enrich_context(
-        """Build memory context dict to inject into the agent before LLM call.
+        self,
+        user_id: str,
+        message: str,
+        trace_id: str | None = None,
+        session_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Build memory context dict to inject into the orchestrator before LLM call.
 
         Returns a dict with keys:
           core_memory        — {key: plaintext_value, ...}
@@ -65,9 +71,21 @@ class MemoryMiddleware:
 
         core = await self._load_core(user_id, fernet)
         associative = await self._load_associative(user_id, message, fernet)
-        episodic = await self._load_episodic(user_id, fernet)
+        episodic = await self._load_episodic(user_id, fernet, session_id=session_id)
         proactive = await self._load_proactive(user_id, fernet)
 
+        user_dbg = await self._get_user_debug(user_id)
+        logger.info(
+            "memory: enrich_context trace=%s user=%s tier=%s core=%d associative=%d episodic=%d proactive=%d",
+            trace_id or "-",
+            user_id,
+            user_dbg.get("tier") or "-",
+            len(core),
+            len(associative),
+            len(episodic),
+            len(proactive),
+        )
+
         return {
             "core_memory": core,
             "associative_memory": associative,
@@ -81,6 +99,7 @@ class MemoryMiddleware:
         session_id: str,
         message: str,
         response: str,
+        trace_id: str | None = None,
     ) -> None:
         """Summarise and store a completed interaction in episodic memory.
 
@@ -103,11 +122,19 @@ class MemoryMiddleware:
         self._db.add(row)
         try:
             await self._db.commit()
+            user_dbg = await self._get_user_debug(user_id)
+            logger.info(
+                "memory: store_episode trace=%s user=%s tier=%s session=%s",
+                trace_id or "-",
+                user_id,
+                user_dbg.get("tier") or "-",
+                session_id,
+            )
         except Exception as exc:
             logger.error("memory: store_episode failed user=%s: %s", user_id, exc)
             await self._db.rollback()
 
-    async def update_core(self, user_id: str, key: str, value: str) -> None:
+    async def update_core(self, user_id: str, key: str, value: str, trace_id: str | None = None) -> None:
         """Upsert a core memory key/value for a user."""
         fernet = await self._get_fernet(user_id)
         if fernet is None:
@@ -133,10 +160,176 @@ class MemoryMiddleware:
             ))
         try:
             await self._db.commit()
+            user_dbg = await self._get_user_debug(user_id)
+            logger.info(
+                "memory: update_core trace=%s user=%s tier=%s key=%s",
+                trace_id or "-",
+                user_id,
+                user_dbg.get("tier") or "-",
+                key,
+            )
         except Exception as exc:
             logger.error("memory: update_core failed user=%s key=%s: %s", user_id, key, exc)
             await self._db.rollback()
 
+    async def list_core_blocks(self, user_id: str) -> list[dict[str, str]]:
+        """Return core memory as editable blocks (label/value)."""
+        fernet = await self._get_fernet(user_id)
+        if fernet is None:
+            return []
+
+        result = await self._db.execute(
+            select(MemoryCore)
+            .where(MemoryCore.user_id == user_id)
+            .order_by(MemoryCore.key.asc())
+        )
+        rows = result.scalars().all()
+        out: list[dict[str, str]] = []
+        for row in rows:
+            plaintext = _safe_decrypt(fernet, row.value_encrypted)
+            if plaintext is not None:
+                out.append({"label": row.key, "value": plaintext})
+        logger.debug("memory: list_core_blocks user=%s count=%d", user_id, len(out))
+        return out
+
+    async def get_core_block(self, user_id: str, label: str) -> str | None:
+        """Return a single core memory block value by label."""
+        fernet = await self._get_fernet(user_id)
+        if fernet is None:
+            return None
+
+        result = await self._db.execute(
+            select(MemoryCore).where(
+                MemoryCore.user_id == user_id,
+                MemoryCore.key == label,
+            )
+        )
+        row = result.scalar_one_or_none()
+        if row is None:
+            logger.debug("memory: get_core_block user=%s label=%s found=0", user_id, label)
+            return None
+        value = _safe_decrypt(fernet, row.value_encrypted)
+        logger.debug("memory: get_core_block user=%s label=%s found=%d", user_id, label, 1 if value is not None else 0)
+        return value
+
+    async def delete_core(self, user_id: str, label: str) -> bool:
+        """Delete a core memory block by label. Returns True if deleted."""
+        result = await self._db.execute(
+            select(MemoryCore).where(
+                MemoryCore.user_id == user_id,
+                MemoryCore.key == label,
+            )
+        )
+        row = result.scalar_one_or_none()
+        if row is None:
+            logger.debug("memory: delete_core user=%s label=%s found=0", user_id, label)
+            return False
+
+        await self._db.delete(row)
+        try:
+            await self._db.commit()
+            logger.info("memory: delete_core user=%s label=%s", user_id, label)
+            return True
+        except Exception as exc:
+            logger.error("memory: delete_core failed user=%s label=%s: %s", user_id, label, exc)
+            await self._db.rollback()
+            return False
+
+    async def append_core(self, user_id: str, label: str, content: str) -> None:
+        """Append content to a core block, creating it if missing."""
+        current = await self.get_core_block(user_id, label)
+        if current is None:
+            await self.update_core(user_id, label, content)
+            logger.info("memory: append_core user=%s label=%s created=1", user_id, label)
+            return
+        await self.update_core(user_id, label, f"{current}\n{content}")
+        logger.info("memory: append_core user=%s label=%s created=0", user_id, label)
+
+    async def replace_core(self, user_id: str, label: str, old: str, new: str) -> bool:
+        """Replace one exact string inside a core block. Returns False if not found."""
+        current = await self.get_core_block(user_id, label)
+        if current is None or old not in current:
+            logger.debug("memory: replace_core user=%s label=%s changed=0", user_id, label)
+            return False
+        await self.update_core(user_id, label, current.replace(old, new, 1))
+        logger.info("memory: replace_core user=%s label=%s changed=1", user_id, label)
+        return True
+
+    async def insert_archival(self, user_id: str, content: str, source: str = "manual") -> None:
+        """Insert a long-term archival memory entry."""
+        fernet = await self._get_fernet(user_id)
+        if fernet is None:
+            return
+
+        encrypted = _encrypt(fernet, content)
+        row = MemoryAssociative(
+            id=str(uuid.uuid4()),
+            user_id=user_id,
+            content_encrypted=encrypted,
+            embedding=None,
+            entity_type=source,
+            entity_id=None,
+        )
+        self._db.add(row)
+        try:
+            await self._db.commit()
+            logger.info("memory: insert_archival user=%s source=%s", user_id, source)
+        except Exception as exc:
+            logger.error("memory: insert_archival failed user=%s: %s", user_id, exc)
+            await self._db.rollback()
+
+    async def search_archival(self, user_id: str, query: str, top_k: int = 5) -> list[str]:
+        """Search archival memory (keyword fallback; semantic ranking can replace this)."""
+        fernet = await self._get_fernet(user_id)
+        if fernet is None:
+            return []
+
+        result = await self._db.execute(
+            select(MemoryAssociative)
+            .where(MemoryAssociative.user_id == user_id)
+            .order_by(MemoryAssociative.updated_at.desc())
+            .limit(100)
+        )
+        rows = result.scalars().all()
+        needle = query.strip().lower()
+        out: list[str] = []
+        for row in rows:
+            plaintext = _safe_decrypt(fernet, row.content_encrypted)
+            if plaintext is None:
+                continue
+            if not needle or needle in plaintext.lower():
+                out.append(plaintext)
+            if len(out) >= max(top_k, 1):
+                break
+        logger.info("memory: search_archival user=%s query=%s hits=%d", user_id, query[:80], len(out))
+        return out
+
+    async def search_recall(self, user_id: str, query: str, top_k: int = 5) -> list[str]:
+        """Search recall memory (episodic summaries) by keyword."""
+        fernet = await self._get_fernet(user_id)
+        if fernet is None:
+            return []
+
+        result = await self._db.execute(
+            select(MemoryEpisodic)
+            .where(MemoryEpisodic.user_id == user_id)
+            .order_by(MemoryEpisodic.created_at.desc())
+            .limit(100)
+        )
+        rows = result.scalars().all()
+        needle = query.strip().lower()
+        out: list[str] = []
+        for row in rows:
+            plaintext = _safe_decrypt(fernet, row.summary_encrypted)
+            if plaintext is None:
+                continue
+            if not needle or needle in plaintext.lower():
+                out.append(plaintext)
+            if len(out) >= max(top_k, 1):
+                break
+        logger.info("memory: search_recall user=%s query=%s hits=%d", user_id, query[:80], len(out))
+        return out
+
     # ── Private helpers ───────────────────────────────────────────────────────
 
     async def _get_fernet(self, user_id: str) -> Fernet | None:
@@ -148,6 +341,16 @@ class MemoryMiddleware:
             return None
         return Fernet(user.encryption_key.encode())
 
+    async def _get_user_debug(self, user_id: str) -> dict[str, str | None]:
+        """Load lightweight user debug fields for trace logs."""
+        result = await self._db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
+        if user is None:
+            return {"tier": None}
+        return {
+            "tier": user.tier,
+        }
+
     async def _load_core(self, user_id: str, fernet: Fernet) -> dict[str, str]:
         result = await self._db.execute(
             select(MemoryCore).where(MemoryCore.user_id == user_id)
@@ -183,10 +386,17 @@ class MemoryMiddleware:
                 out.append(plaintext)
         return out
 
-    async def _load_episodic(self, user_id: str, fernet: Fernet) -> list[str]:
+    async def _load_episodic(
+        self,
+        user_id: str,
+        fernet: Fernet,
+        session_id: str | None = None,
+    ) -> list[str]:
+        query = select(MemoryEpisodic).where(MemoryEpisodic.user_id == user_id)
+        if session_id:
+            query = query.where(MemoryEpisodic.session_id == session_id)
         result = await self._db.execute(
-            select(MemoryEpisodic)
+            query
-            .where(MemoryEpisodic.user_id == user_id)
             .order_by(MemoryEpisodic.created_at.desc())
             .limit(_EPISODIC_RECENT_N)
         )

app/core/output_formatter.py

@@ -1,141 +1,47 @@
-"""Output Formatter — transforms deep-agent event streams into WS frame sequences.
+"""Output formatter for deep-agent stream events."""
-
-Consumes ``(event_type, data)`` tuples yielded by ``deep_agent.run_*_stream()``:
-  * ``("token", str)``       — supervisor text token
-  * ``("tool_end", dict)``   — sub-agent finished: ``{name, result}``
-  * ``("mutations", list)``  — collected CRUD mutations for ``stream_end``
-
-HomeFormatter:
-  * Streams text tokens as-is → emits ``WsStreamText``
-    (text may contain inline ``<type>[id,...]</type>`` entity tags
-    for the frontend to parse and render as interactive components)
-  * Attaches mutations        → injects into ``WsStreamEnd``
-
-FloatingFormatter:
-  * Sniffs first ``tool_end`` name → emits ``WsFloatingDomain``
-  * Streams text tokens             → emits ``WsStreamText``
-  * Attaches mutations              → injects into ``WsStreamEnd``
-"""
 
 from __future__ import annotations
 
-import logging
 from collections.abc import AsyncGenerator
 from typing import Any
 
-from app.schemas import (
+from app.schemas import WsFloatingDomain, WsStreamEnd, WsStreamStart, WsStreamText
-    WsFloatingDomain,
-    WsStreamEnd,
-    WsStreamStart,
-    WsStreamText,
-)
-
-logger = logging.getLogger(__name__)
-
-# Map sub-agent tool name → floating domain / entity type
-_AGENT_DOMAIN: dict[str, str] = {
-    "task_agent": "tasks",
-    "timeline_agent": "timelines",
-    "note_agent": "notes",
-    "project_agent": "projects",
-}
 
 WsFrame = WsStreamStart | WsStreamText | WsStreamEnd | WsFloatingDomain
 
 
-class HomeFormatter:
+class StreamFormatter:
-    """Consumes a deep-agent event stream and yields WS frames for the Home view.
+    """Convert `(event_type, data)` stream events into websocket frame models."""
-
-    Text tokens are forwarded as-is via ``WsStreamText``.  The supervisor
-    embeds ``<type>[id1,id2]</type>`` entity tags inline — the frontend
-    is responsible for parsing those and rendering interactive components.
-    Mutations are attached to ``WsStreamEnd``.
-    """
 
     def __init__(self, request_id: str) -> None:
         self.request_id = request_id
-        self._mutations: list[dict] = []
 
     async def format(
         self,
         event_stream: AsyncGenerator[tuple[str, Any], None],
     ) -> AsyncGenerator[WsFrame, None]:
-        yield WsStreamStart(request_id=self.request_id)
+        started = False
 
         async for event_type, data in event_stream:
-            if event_type == "token":
+            if event_type == "floating_domain":
-                if data:
+                if isinstance(data, dict):
-                    yield WsStreamText(request_id=self.request_id, chunk=data)
-
-            elif event_type == "mutations":
-                self._mutations = data or []
-
-        yield WsStreamEnd(
-            request_id=self.request_id,
-            mutations=[
-                {"action": m["action"], "table": m["table"], "data": m["data"]}
-                for m in self._mutations
-            ],
-        )
-
-
-class FloatingFormatter:
-    """Consumes a deep-agent event stream and yields WS frames for the Floating view.
-
-    Sniffs the first ``tool_end`` event name to derive the domain (e.g.
-    ``task_agent`` → ``"tasks"``), then streams text tokens as plain
-    ``WsStreamText``.  No block parsing for floating context.
-    """
-
-    def __init__(self, request_id: str) -> None:
-        self.request_id = request_id
-        self._mutations: list[dict] = []
-
-    async def format(
-        self,
-        event_stream: AsyncGenerator[tuple[str, Any], None],
-    ) -> AsyncGenerator[WsFrame, None]:
-        domain_sent = False
-
-        async for event_type, data in event_stream:
-            if event_type == "tool_end" and not domain_sent:
-                # Sniff domain from the first sub-agent that completes
-                name = data.get("name", "")
-                domain = _AGENT_DOMAIN.get(name, "tasks")
                     yield WsFloatingDomain(
                         request_id=self.request_id,
-                    domain=domain,  # type: ignore[arg-type]
+                        domain=data,
                     )
+                continue
+
+            if event_type != "token":
+                continue
+
+            if not started:
                 yield WsStreamStart(request_id=self.request_id)
-                domain_sent = True
+                started = True
 
-            elif event_type == "token":
+            text = str(data or "")
-                if not domain_sent:
+            if text:
-                    # First token arrived before any tool_end — default domain
+                yield WsStreamText(request_id=self.request_id, chunk=text)
-                    yield WsFloatingDomain(
+
-                        request_id=self.request_id,
+        if not started:
-                        domain="tasks",  # type: ignore[arg-type]
-                    )
             yield WsStreamStart(request_id=self.request_id)
-                    domain_sent = True
+        yield WsStreamEnd(request_id=self.request_id)
-                if data:
-                    yield WsStreamText(request_id=self.request_id, chunk=data)
-
-            elif event_type == "mutations":
-                self._mutations = data or []
-
-        # If no events triggered domain_sent (edge case), still emit structure
-        if not domain_sent:
-            yield WsFloatingDomain(
-                request_id=self.request_id,
-                domain="tasks",  # type: ignore[arg-type]
-            )
-            yield WsStreamStart(request_id=self.request_id)
-
-        yield WsStreamEnd(
-            request_id=self.request_id,
-            mutations=[
-                {"action": m["action"], "table": m["table"], "data": m["data"]}
-                for m in self._mutations
-            ],
-        )

app/core/preprocessors/__init__.py

@@ -0,0 +1,104 @@
+"""Preprocessor registry: detect content type and dispatch to handlers.
+
+Public API
+----------
+detect_content_type(filename, raw_content) -> str
+    Heuristic detection based on file extension and content patterns.
+
+preprocess(content_type, raw_content) -> PreprocessResult
+    Dispatch to the appropriate handler.
+"""
+
+from __future__ import annotations
+
+import re
+
+from app.core.preprocessors.base import PreprocessResult
+
+# ── Heuristics ────────────────────────────────────────────────────────
+
+# Patterns that strongly suggest an email HTML file
+_EMAIL_SIGNALS = re.compile(
+    r"(Subject:|From:|To:|Date:|Sent:|MIME-Version:|Content-Type:\s*text/html)",
+    re.IGNORECASE,
+)
+
+# Patterns that suggest a generic HTML page (not an email)
+_GENERIC_HTML_SIGNALS = re.compile(
+    r"<(nav|main|header|footer|article|section)\b",
+    re.IGNORECASE,
+)
+
+
+def detect_content_type(filename: str, raw_content: str) -> str:
+    """Return a content-type string for the given file.
+
+    Supported types: ``"email_html"``, ``"generic_html"``,
+    ``"plain_text"``, ``"unknown"``.
+    """
+    ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+
+    if ext == "txt":
+        return "plain_text"
+
+    if ext in ("html", "htm", "eml", "mhtml", "mht"):
+        # Prefer email detection over generic HTML
+        if _EMAIL_SIGNALS.search(raw_content[:4096]):
+            return "email_html"
+        if _GENERIC_HTML_SIGNALS.search(raw_content[:4096]) or "<html" in raw_content[:200].lower():
+            return "generic_html"
+        # .html without clear signals — check for any email header
+        if re.search(r"^(From|To|Subject|Date):", raw_content[:2048], re.MULTILINE | re.IGNORECASE):
+            return "email_html"
+        return "generic_html"
+
+    # Plain text files with email headers
+    if ext in ("", "txt") or not ext:
+        if _EMAIL_SIGNALS.search(raw_content[:4096]):
+            return "email_html"
+
+    # Detect binary content
+    try:
+        raw_content.encode("utf-8")
+    except (UnicodeEncodeError, AttributeError):
+        return "unknown"
+
+    # Non-text bytes heuristic: high ratio of non-printable chars
+    sample = raw_content[:512]
+    non_printable = sum(1 for c in sample if ord(c) < 32 and c not in "\r\n\t")
+    if len(sample) > 0 and non_printable / len(sample) > 0.1:
+        return "unknown"
+
+    return "unknown"
+
+
+# ── Generic fallback handler ──────────────────────────────────────────
+
+def _preprocess_generic(raw_content: str, content_type: str) -> PreprocessResult:
+    """Strip HTML tags if present, return text as-is."""
+    try:
+        from bs4 import BeautifulSoup
+        text = BeautifulSoup(raw_content, "html.parser").get_text(separator="\n")
+    except ImportError:
+        # No BeautifulSoup — strip tags with a simple regex
+        text = re.sub(r"<[^>]+>", "", raw_content)
+
+    text = re.sub(r"\n{3,}", "\n\n", text).strip()
+    return PreprocessResult(content_type=content_type, clean_text=text, metadata={})
+
+
+# ── Dispatch ──────────────────────────────────────────────────────────
+
+def preprocess(content_type: str, raw_content: str) -> PreprocessResult:
+    """Dispatch *raw_content* to the handler registered for *content_type*.
+
+    Falls back to the generic handler for unknown types.
+    """
+    if content_type == "email_html":
+        from app.core.preprocessors.email_html import preprocess_email_html
+        return preprocess_email_html(raw_content)
+
+    return _preprocess_generic(raw_content, content_type)
+
+
+__all__ = ["detect_content_type", "preprocess", "PreprocessResult"]

app/core/preprocessors/base.py

@@ -0,0 +1,25 @@
+"""Base types for the preprocessor system."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+
+@dataclass
+class PreprocessResult:
+    """Output of a preprocessor handler.
+
+    Attributes
+    ----------
+    content_type:
+        The detected content type (e.g. ``"email_html"``, ``"plain_text"``).
+    clean_text:
+        Human-readable text stripped of markup/binary noise.
+    metadata:
+        Dict of extracted metadata (keys vary by handler).
+        Common keys: ``subject``, ``from``, ``to``, ``date``, ``filename``.
+    """
+
+    content_type: str
+    clean_text: str
+    metadata: dict = field(default_factory=dict)

app/core/preprocessors/email_html.py

@@ -0,0 +1,111 @@
+"""Preprocessor for email HTML files.
+
+Handles:
+- HTML stripping via BeautifulSoup
+- Metadata extraction (Subject, From, To, Date)
+- Thread splitting — isolates the latest reply
+"""
+
+from __future__ import annotations
+
+import re
+from typing import TYPE_CHECKING
+
+from app.core.preprocessors.base import PreprocessResult
+
+if TYPE_CHECKING:
+    pass
+
+# ── Thread split markers ──────────────────────────────────────────────
+
+# Matches patterns like:
+#   "On Mon, Apr 7, 2026 at 10:00 AM, Alice <alice@co.com> wrote:"
+#   "-----Original Message-----"
+#   "> " (plain-text quote prefix)
+_THREAD_PATTERNS = [
+    re.compile(r"^On\s+.+wrote\s*:", re.IGNORECASE | re.MULTILINE),
+    re.compile(r"^-{3,}\s*(original message|forwarded message)\s*-{3,}", re.IGNORECASE | re.MULTILINE),
+    re.compile(r"^>{1,}\s+\S", re.MULTILINE),
+    re.compile(r"^From:\s+.+\nSent:\s+", re.IGNORECASE | re.MULTILINE),
+]
+
+# ── Metadata patterns (applied on raw HTML / plain fallback) ──────────
+
+_META_PATTERNS: dict[str, list[re.Pattern]] = {
+    "subject": [
+        re.compile(r"<title>(.+?)</title>", re.IGNORECASE | re.DOTALL),
+        re.compile(r"Subject:\s*(.+)", re.IGNORECASE),
+    ],
+    "from": [
+        re.compile(r'<meta[^>]+name=["\']?from["\']?[^>]+content=["\']([^"\']+)["\']', re.IGNORECASE),
+        re.compile(r"From:\s*(.+)", re.IGNORECASE),
+    ],
+    "to": [
+        re.compile(r'<meta[^>]+name=["\']?to["\']?[^>]+content=["\']([^"\']+)["\']', re.IGNORECASE),
+        re.compile(r"To:\s*(.+)", re.IGNORECASE),
+    ],
+    "date": [
+        re.compile(r'<meta[^>]+name=["\']?date["\']?[^>]+content=["\']([^"\']+)["\']', re.IGNORECASE),
+        re.compile(r"Date:\s*(.+)", re.IGNORECASE),
+        re.compile(r"Sent:\s*(.+)", re.IGNORECASE),
+    ],
+}
+
+
+def _extract_metadata(raw_html: str, text: str) -> dict:
+    """Extract Subject/From/To/Date from raw HTML or plain text."""
+    metadata: dict[str, str] = {}
+    for field, patterns in _META_PATTERNS.items():
+        for pat in patterns:
+            m = pat.search(raw_html) or pat.search(text)
+            if m:
+                metadata[field] = m.group(1).strip()
+                break
+    return metadata
+
+
+def _split_thread(text: str) -> str:
+    """Return only the latest message in a threaded email."""
+    earliest_pos: int | None = None
+    for pat in _THREAD_PATTERNS:
+        m = pat.search(text)
+        if m and (earliest_pos is None or m.start() < earliest_pos):
+            earliest_pos = m.start()
+
+    if earliest_pos is not None and earliest_pos > 0:
+        return text[:earliest_pos].strip()
+    return text.strip()
+
+
+def preprocess_email_html(raw_content: str) -> PreprocessResult:
+    """Strip HTML, extract metadata, split thread from an email HTML file."""
+    try:
+        from bs4 import BeautifulSoup  # lazy import — optional dep
+    except ImportError as exc:
+        raise ImportError(
+            "beautifulsoup4 is required for email_html preprocessing. "
+            "Install it with: pip install beautifulsoup4"
+        ) from exc
+
+    # Parse with lxml if available, fall back to html.parser
+    try:
+        soup = BeautifulSoup(raw_content, "lxml")
+    except Exception:
+        soup = BeautifulSoup(raw_content, "html.parser")
+
+    # Remove noise tags
+    for tag in soup(["style", "script", "head", "noscript"]):
+        tag.decompose()
+
+    clean_text = soup.get_text(separator="\n")
+    # Collapse excessive blank lines
+    clean_text = re.sub(r"\n{3,}", "\n\n", clean_text).strip()
+
+    metadata = _extract_metadata(raw_content, clean_text)
+    latest_message = _split_thread(clean_text)
+
+    return PreprocessResult(
+        content_type="email_html",
+        clean_text=latest_message,
+        metadata=metadata,
+    )

app/core/ws_context.py

@@ -7,21 +7,18 @@ The callback sends a `tool_call` WS frame and awaits the `tool_result`.
 
 from __future__ import annotations
 
-import logging
 from contextvars import ContextVar
 from typing import Any, Callable, Coroutine
 from uuid import uuid4
 
-logger = logging.getLogger(__name__)
-
 # Holds the execute callback for the current WS session.
-# Set by the chat WS handler before the deep agent runs; cleared after.
+# Set by the chat WS handler before the orchestrator runs; cleared after.
 _client_executor: ContextVar[Callable[[dict], Coroutine[Any, Any, dict]]] = ContextVar(
     "_client_executor"
 )
 
 # Optional collector that captures raw execute_on_client results.
-# Set by the deep agent tool loop to capture CRUD mutations.
+# Set by _tool_loop / _tool_loop_stream to populate ChatAgent.tool_results.
 _tool_result_collector: ContextVar[list[dict] | None] = ContextVar(
     "_tool_result_collector", default=None
 )
@@ -84,17 +81,12 @@ async def execute_on_client(
     if limit is not None:
         payload["limit"] = limit
 
-    logger.info("execute_on_client: sending payload action=%s table=%s id=%s", action, table, payload["id"])
     result = await callback(payload)
-    if result is None:
-        logger.error("execute_on_client: callback returned None for action=%s table=%s id=%s", action, table, payload["id"])
-    else:
-        logger.info("execute_on_client: got result type=%s keys=%s", type(result).__name__, list(result.keys()) if isinstance(result, dict) else "N/A")
     collector = _tool_result_collector.get(None)
-    if collector is not None and action in ("insert", "update", "delete"):
+    if collector is not None:
         collector.append({
             "action": action,
             "table": table,
-            "data": data or {},
+            "data": result,
         })
     return result

app/main.py

@@ -18,7 +18,9 @@ from app.config.settings import settings
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # Startup: initialise DB connection pool
+    # Startup: ensure agent tool modules are loaded.
+    import app.agents  # noqa: F401
+
     yield
 
     # Shutdown: dispose SQLAlchemy connection pool
@@ -48,7 +50,7 @@ def create_app() -> FastAPI:
     app.add_middleware(SanitizerMiddleware)
     app.add_middleware(TierRateLimitMiddleware)
 
-    from app.api.routes import agent_setup, agents, auth, backup, billing, chat, device_ws, plugins, storage, vectors
+    from app.api.routes import agents, auth, backup, billing, chat, device_ws, plugins, storage, vectors
 
     app.include_router(auth.router,       prefix="/api/v1")
     app.include_router(chat.router,       prefix="/api/v1")
@@ -58,7 +60,6 @@ def create_app() -> FastAPI:
     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(agent_setup.router, prefix="/api/v1")
     app.include_router(device_ws.router,  prefix="/api/v1")
 
     @app.get("/api/v1/health", tags=["health"])

app/schemas.py

@@ -142,9 +142,6 @@ 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"
     # ── v3 frame types ─────────────────────────────────────────────────
     home_request = "home_request"
@@ -156,6 +153,10 @@ class WsFrameType(str, Enum):
     data_request = "data_request"
     data_response = "data_response"
     mutation = "mutation"
+    # ── v4 journey frame types ────────────────────────────────────────
+    journey_start = "journey_start"
+    journey_message = "journey_message"
+    journey_reply = "journey_reply"
 
 
 class WsToolCall(BaseModel):
@@ -208,31 +209,6 @@ class WsDeviceHello(BaseModel):
     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)
-
 
 # ── WebSocket v3 Frame Models ─────────────────────────────────────────
 
@@ -279,7 +255,14 @@ class WsStreamEnd(BaseModel):
 
     type: Literal[WsFrameType.stream_end] = WsFrameType.stream_end
     request_id: str
-    mutations: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class WsDomain(BaseModel):
+    """Structured floating domain payload for UI routing decisions."""
+
+    type: Literal["task", "timeline", "project", "node"]
+    id: str | None = None
+    section: Literal["task", "timeline", "note"] | None = None
 
 
 class WsFloatingDomain(BaseModel):
@@ -287,7 +270,7 @@ class WsFloatingDomain(BaseModel):
 
     type: Literal[WsFrameType.floating_domain] = WsFrameType.floating_domain
     request_id: str
-    domain: Literal["tasks", "timelines", "notes", "projects"]
+    domain: WsDomain
 
 
 # ── Agent Catalog ─────────────────────────────────────────────────────
@@ -296,84 +279,28 @@ class AgentCatalogItem(BaseModel):
     type: str
     name: str
     description: str
-    config_schema: dict[str, Any] = Field(default_factory=dict)
 
 
-# ── Local Agent Config ────────────────────────────────────────────────
+class AgentCreationCheckRequest(BaseModel):
-
+    active_agents: int = Field(ge=0, default=0)
-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):
+class AgentCreationCheckResponse(BaseModel):
-    name: str | None = None
+    allowed: bool
-    device_id: str | None = None
+    tier: BillingTier
-    directory_paths: list[str] | None = None
+    active_agents: int
-    data_types: list[str] | None = None
+    limit: int
-    prompt_template: str | None = None
-    file_extensions: list[str] | None = None
-    schedule_cron: str | None = None
-    enabled: bool | None = None
 
 
-class LocalAgentConfigResponse(BaseModel):
+class AgentTriggerRequest(BaseModel):
-    id: str
+    directory: str = Field(min_length=1)
-    name: str
+    device_id: str = Field(default="")
-    device_id: str
+    agent_id: str | None = None  # FE stable agent ID (electron-store UUID)
-    directory_paths: list[str]
+    what_to_extract: list[str] = Field(min_length=1)
-    data_types: list[str]
+    actions_by_type: dict[str, list[str]] | None = None
-    prompt_template: str
+    batch_interval: str = Field(min_length=1)
-    file_extensions: list[str]
+    custom_agent_prompt: str = Field(min_length=1)
-    schedule_cron: str
+    active_agents: int = Field(ge=0, default=0)
-    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 ─────────────────────────────────────────────────────
@@ -392,18 +319,3 @@ class AgentRunLogResponse(BaseModel):
 
 # ── Chatbot Journey ───────────────────────────────────────────────────
 
-class JourneyStartRequest(BaseModel):
-    agent_type: Literal["local", "cloud"]
-    agent_id: str | None = None
-
-
-class JourneyMessageRequest(BaseModel):
-    session_id: str
-    message: str
-
-
-class JourneyResponse(BaseModel):
-    session_id: str
-    message: str
-    done: bool
-    prompt_template: str | None = None

docs/MICROSERVICES_ARCHITECTURE.md

@@ -0,0 +1,941 @@
+# Adiuva — Architettura Microservizi (MVP)
+
+## Panoramica
+
+Il monolite viene suddiviso in **4 servizi MVP** + un **API Gateway (Traefik)**, orchestrati con Docker Compose su un singolo VPS raggiungibile via Cloudflare.
+
+> **Fuori dall'MVP**: Storage Service (S3/backup CRUD) e Plugin Service (marketplace). Verranno aggiunti come servizi indipendenti in una fase successiva.
+
+```
+                          ┌──────────────┐
+                          │  Cloudflare  │
+                          │  (DNS + CDN) │
+                          └──────┬───────┘
+                                 │ HTTPS / WSS
+                          ┌──────▼───────┐
+                          │   Traefik    │
+                          │ API Gateway  │
+                          │  (routing,   │
+                          │   TLS, rate  │
+                          │   limiting)  │
+                          └──────┬───────┘
+                                 │
+          ┌──────────┬───────────┼───────────┐
+          │          │           │           │
+    ┌─────▼────┐ ┌───▼───┐ ┌────▼────┐ ┌────▼───┐
+    │  Auth    │ │  Chat │ │  Agent  │ │Billing │
+    │ Service  │ │Service│ │ Service │ │Service │
+    └─────┬────┘ └───┬───┘ └────┬────┘ └────┬───┘
+          │          │          │           │
+    ┌─────▼──────────▼──────────▼───────────▼────┐
+    │              Infrastruttura                 │
+    │  PostgreSQL  │  Redis  │  Qdrant            │
+    └─────────────────────────────────────────────┘
+```
+
+---
+
+## 1. Suddivisione dei Servizi
+
+### 1.1 Auth Service (`auth-service`)
+
+**Responsabilità**: Registrazione, login, refresh token, profilo utente, encryption key.
+
+| Endpoint originale | Metodo |
+|---|---|
+| `/api/v1/auth/register` | POST |
+| `/api/v1/auth/login` | POST |
+| `/api/v1/auth/refresh` | POST |
+| `/api/v1/auth/me` | GET / PUT |
+
+**Database**: Tabelle `users`, `refresh_tokens` (PostgreSQL condiviso, schema `auth`).
+
+**Modifica chiave — JWT con RS256**:
+Il monolite usa un `SECRET_KEY` simmetrico (HS256). Con i microservizi, passare a **RS256** (asimmetrico):
+- L'Auth Service firma i JWT con la **chiave privata**.
+- Tutti gli altri servizi verificano i JWT con la **chiave pubblica** senza mai contattare l'Auth Service.
+- La chiave pubblica viene esposta via `GET /api/v1/auth/.well-known/jwks.json` oppure montata come volume condiviso.
+
+```python
+# auth-service/app/auth/jwt.py
+from cryptography.hazmat.primitives.asymmetric import rsa
+from jose import jwt
+
+PRIVATE_KEY = ...  # Da env/secret
+PUBLIC_KEY = ...   # Derivata o da env
+
+def create_access_token(user_id: str, tier: str) -> str:
+    return jwt.encode(
+        {"sub": user_id, "tier": tier, "exp": ...},
+        PRIVATE_KEY,
+        algorithm="RS256",
+    )
+```
+
+```python
+# shared/auth.py  (usato da tutti gli altri servizi)
+from jose import jwt
+
+PUBLIC_KEY = ...  # Volume montato o fetched da JWKS endpoint
+
+def verify_token(token: str) -> dict:
+    return jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
+```
+
+**Scaling**: 2 repliche sufficienti, stateless. Rate-limit dedicato su `/login` e `/register`.
+
+---
+
+### 1.2 Chat Service (`chat-service`) ⭐ Real-time
+
+**Responsabilità**: WebSocket device connection, home chat, floating chat, memory middleware, streaming LLM responses verso il client.
+
+Questo servizio gestisce la **connessione persistente** con l'app Electron e le interazioni **real-time** dell'utente (chat home, floating chat). È il proprietario della WebSocket.
+
+| Endpoint | Tipo |
+|---|---|
+| `/api/v1/ws/device` | WebSocket (connessione persistente) |
+| `/api/v1/chat` | POST (REST fallback) |
+
+**Moduli inclusi**: `deep_agent`, `memory_middleware`, `ws_context`, `device_manager` (Redis-backed), `output_formatter`, `llm`, tutti gli agent tools (`task_agent`, `project_agent`, `note_agent`, `timeline_agent`).
+
+**Perché separato dall'Agent Service**: Il Chat Service tiene la WebSocket aperta e risponde in tempo reale (streaming). Scalare aggiungendo repliche è semplice con sticky sessions + Redis pub/sub per il cross-instance routing dei tool_call.
+
+**Scaling**: 2–N repliche. Sticky cookies per le WS + Redis per cross-instance.
+
+---
+
+### 1.3 Agent Service (`agent-service`) ⭐ Batch
+
+**Responsabilità**: Batch agent processing (directory scanning, file classification, entity extraction), agent setup journeys, agent configuration CRUD.
+
+Questo servizio gestisce i processi **long-running** e **CPU-intensive**: scansione filesystem, classificazione file con LLM, estrazione entità in batch. Non possiede la WebSocket — comunica con il device dell'utente tramite **Redis pub/sub** passando per il Chat Service.
+
+| Endpoint | Tipo |
+|---|---|
+| `/api/v1/agents/catalog` | GET |
+| `/api/v1/agents/can-create` | POST |
+| `/api/v1/agents/trigger` | POST |
+| `/api/v1/agents/journey/start` | POST (o WS relay) |
+| `/api/v1/agents/journey/message` | POST (o WS relay) |
+
+**Moduli inclusi**: `agent_runner`, `agent_registry`, `filesystem_agent`, `llm`.
+
+**Flusso tool-call cross-service** (l'Agent Service non ha la WS):
+
+```
+┌──────────────┐            ┌──────────────┐            ┌──────────┐
+│ Agent Service│            │    Redis     │            │  Chat    │
+│ (batch run)  │            │              │            │ Service  │
+│              │            │              │            │ (ha WS)  │
+│ 1. Needs to  │  PUBLISH   │              │ SUBSCRIBE  │          │
+│    read file ├───────────►│tool_call:u123├───────────►│ 2. Invia │
+│    from      │            │              │            │    al    │
+│    device    │            │              │            │    device│
+│              │            │              │            │    via WS│
+│              │  SUBSCRIBE │              │  PUBLISH   │          │
+│ 4. Riceve   ◄────────────┤tool_result:id│◄───────────┤ 3. Device│
+│    risultato │            │              │            │    reply │
+└──────────────┘            └──────────────┘            └──────────┘
+```
+
+**Scaling**: 1–N repliche. Completamente stateless, scala indipendentemente dalla chat. Ogni replica processa batch job diversi. Può essere scalato a 0 se non ci sono agent attivi (risparmio risorse).
+
+**Vantaggio dello split**: Se 50 utenti triggerano agenti batch contemporaneamente, il Chat Service non ne risente — le risposte real-time rimangono veloci.
+
+---
+
+### 1.4 Billing Service (`billing-service`)
+
+**Responsabilità**: Stripe checkout, webhook, subscription management.
+
+| Endpoint originale | Metodo |
+|---|---|
+| `/api/v1/billing/checkout` | POST |
+| `/api/v1/billing/webhook` | POST |
+| `/api/v1/billing/subscription` | GET / DELETE |
+
+**Database**: Tabelle `subscriptions` (schema `billing`).
+
+**Comunicazione inter-servizio**: Quando Stripe invia un webhook e il tier cambia, il Billing Service pubblica un evento su **Redis pub/sub** channel `tier_changed:{user_id}`. L'Auth Service aggiorna il campo `tier` nella tabella users. Al prossimo token refresh il JWT conterrà il tier aggiornato.
+
+**Scaling**: 1 replica sufficiente. Basso traffico.
+
+---
+
+### 1.5 Servizi esclusi dall'MVP
+
+I seguenti servizi verranno aggiunti post-MVP come servizi indipendenti:
+
+| Servizio | Responsabilità | Note |
+|---|---|---|
+| **Storage Service** | S3 blobs CRUD, vector ops, backup | Le funzionalità vector/embed possono restare nel Chat Service per il MVP |
+| **Plugin Service** | Marketplace, install, revenue split | Feature non critica per il lancio |
+
+---
+
+## 2. Tier Check — Dove e Come
+
+Il tier dell'utente (free/pro/power/team) determina rate-limiting, quote e accesso a funzionalità. Con i microservizi, **ogni servizio controlla il tier autonomamente** senza chiamare l'Auth Service.
+
+### Strategia: Tier nel JWT
+
+L'Auth Service include il `tier` come claim nel JWT al momento del login/refresh:
+
+```json
+{
+  "sub": "user_123",
+  "tier": "pro",
+  "exp": 1742515200,
+  "iat": 1742511600
+}
+```
+
+Ogni servizio:
+1. Decodifica il JWT con la chiave pubblica (già lo fa per l'auth)
+2. Legge `payload["tier"]` — **zero chiamate extra**
+3. Applica le sue regole di enforcement localmente
+
+```python
+# shared/auth.py — dependency FastAPI condivisa
+from fastapi import Depends, HTTPException, Request
+from jose import jwt
+
+PUBLIC_KEY = ...
+
+class CurrentUser:
+    def __init__(self, user_id: str, tier: str):
+        self.user_id = user_id
+        self.tier = tier
+
+async def get_current_user(request: Request) -> CurrentUser:
+    token = request.headers.get("Authorization", "").removeprefix("Bearer ")
+    payload = jwt.decode(token, PUBLIC_KEY, algorithms=["RS256"])
+    return CurrentUser(user_id=payload["sub"], tier=payload["tier"])
+
+def require_tier(*allowed_tiers: str):
+    """Dependency che blocca se il tier non è tra quelli ammessi."""
+    async def check(user: CurrentUser = Depends(get_current_user)):
+        if user.tier not in allowed_tiers:
+            raise HTTPException(403, "Tier insufficient")
+        return user
+    return check
+```
+
+### Cosa succede quando il tier cambia (upgrade/downgrade)?
+
+```
+┌──────────┐  Stripe webhook   ┌──────────┐  tier_changed   ┌──────────┐
+│  Stripe  │ ─────────────────►│ Billing  │ ───────────────►│   Auth   │
+│          │                    │ Service  │  (Redis pub/sub) │ Service  │
+└──────────┘                    └──────────┘                  └────┬─────┘
+                                                                   │
+                                                          UPDATE users
+                                                          SET tier = 'power'
+                                                                   │
+                                                    Al prossimo /refresh
+                                                    il JWT conterrà tier='power'
+```
+
+**Latenza del cambio**: Il tier si propaga al prossimo token refresh (tipicamente 15–30 min, o il client può forzare un refresh immediato dopo il checkout). Per il billing webhook, il downgrade può essere forzato invalidando il refresh token su Redis → il client è obbligato a ri-autenticarsi.
+
+### Dove si applica in ciascun servizio
+
+| Servizio | Enforcement |
+|---|---|
+| **Auth Service** | Nessuno (è lui che scrive il tier) |
+| **Chat Service** | Rate-limit per tier (req/min), quota messaggi |
+| **Agent Service** | Max agent configs, max runs/day, max concurrent batches |
+| **Billing Service** | Nessuno (gestisce i tier, non li consuma) |
+
+### Rate-limit distribuito via Redis
+
+Poiché ogni servizio ha le sue repliche, il rate-limiting deve essere **condiviso** via Redis:
+
+```python
+# shared/middleware/rate_limit.py
+import redis.asyncio as aioredis
+
+class DistributedRateLimiter:
+    def __init__(self, redis: aioredis.Redis):
+        self._redis = redis
+
+    async def check(self, user_id: str, tier: str, service: str) -> bool:
+        limits = {"free": 20, "pro": 60, "power": 120, "team": 200}
+        max_req = limits.get(tier, 20)
+        key = f"rate:{service}:{user_id}"
+
+        pipe = self._redis.pipeline()
+        pipe.incr(key)
+        pipe.expire(key, 60)
+        count, _ = await pipe.execute()
+
+        return count <= max_req
+```
+
+---
+
+## 3. WebSocket con Scaling Orizzontale — Il Problema Chiave
+
+`DeviceConnectionManager` è un **singleton in-memory**:
+
+```python
+class DeviceConnectionManager:
+    def __init__(self):
+        self._connections: dict[str, DeviceConnection] = {}  # ← In-memory!
+```
+
+Con N istanze del Chat Service, il device si connette a **una sola** istanza. Quando un'altra istanza deve inviare un `tool_call` a quel device (es. un agent trigger da un'API call), non trova la connessione.
+
+### La soluzione: Redis Pub/Sub + Registry
+
+```
+┌──────────────────────────────────────────────────────────────┐
+│                     Redis                                    │
+│                                                              │
+│  Hash: ws:connections                                        │
+│    user_123 → instance_A                                     │
+│    user_456 → instance_B                                     │
+│                                                              │
+│  Pub/Sub channels:                                           │
+│    tool_call:{user_id}  → tool call payloads                 │
+│    tool_result:{call_id} → tool result payloads              │
+│    stream:{user_id}     → text_chunk streaming               │
+└──────────────────────────────────────────────────────────────┘
+
+ Instance A (ha WS di user_123)     Instance B (deve chiamare tool su user_123)
+ ┌───────────────────────┐          ┌───────────────────────┐
+ │  1. Sottoscrive a     │          │  1. Lookup Redis Hash │
+ │     tool_call:user_123│          │     → user_123 è su A │
+ │                       │          │                       │
+ │  2. Riceve tool_call  │◄─────────│  2. PUBLISH           │
+ │     da Redis channel  │          │    tool_call:user_123 │
+ │                       │          │    {id, action, ...}  │
+ │  3. Invia al device   │          │                       │
+ │     via WS            │          │  4. SUBSCRIBE         │
+ │                       │          │    tool_result:{id}   │
+ │  4. Device risponde   │          │                       │
+ │     tool_result       │──────────│► 5. Riceve risultato  │
+ │                       │          │                       │
+ │  5. PUBLISH           │          │                       │
+ │    tool_result:{id}   │          │                       │
+ └───────────────────────┘          └───────────────────────┘
+```
+
+### Implementazione: `RedisDeviceManager`
+
+```python
+# chat-service/app/core/device_manager.py
+
+import asyncio
+import json
+import os
+import redis.asyncio as aioredis
+from dataclasses import dataclass, field
+from fastapi import WebSocket
+
+INSTANCE_ID = os.environ.get("INSTANCE_ID", os.urandom(8).hex())
+
+@dataclass
+class LocalConnection:
+    ws: WebSocket
+    device_id: str
+    pending_calls: dict[str, asyncio.Future[dict]] = field(default_factory=dict)
+
+
+class RedisDeviceManager:
+    """Device manager backed by Redis for cross-instance communication."""
+
+    def __init__(self, redis_url: str = "redis://redis:6379"):
+        self._redis = aioredis.from_url(redis_url)
+        self._pubsub = self._redis.pubsub()
+        self._local: dict[str, LocalConnection] = {}  # Solo connessioni locali
+        self._remote_futures: dict[str, asyncio.Future[dict]] = {}
+
+    async def start(self):
+        """Avvia il listener Redis per tool_call in arrivo."""
+        asyncio.create_task(self._listen_tool_calls())
+
+    # ── Registrazione ──
+
+    async def register(self, user_id: str, device_id: str, ws: WebSocket):
+        # Registra localmente
+        self._local[user_id] = LocalConnection(ws=ws, device_id=device_id)
+        # Registra in Redis quale istanza ha la connessione
+        await self._redis.hset("ws:connections", user_id, INSTANCE_ID)
+        # Sottoscrivi ai tool_call per questo utente
+        await self._pubsub.subscribe(f"tool_call:{user_id}")
+
+    async def unregister(self, user_id: str):
+        conn = self._local.pop(user_id, None)
+        if conn:
+            for fut in conn.pending_calls.values():
+                if not fut.done():
+                    fut.cancel()
+        await self._redis.hdel("ws:connections", user_id)
+        await self._pubsub.unsubscribe(f"tool_call:{user_id}")
+
+    # ── Presenza ──
+
+    async def is_online(self, user_id: str) -> bool:
+        return await self._redis.hexists("ws:connections", user_id)
+
+    # ── Tool-call round-trip (cross-instance) ──
+
+    async def execute_tool_call(self, user_id: str, payload: dict) -> dict:
+        """
+        Invia un tool_call al device dell'utente.
+        Funziona sia che la WS sia locale che su un'altra istanza.
+        """
+        call_id = payload["id"]
+
+        # Caso 1: connessione locale → invio diretto
+        if user_id in self._local:
+            conn = self._local[user_id]
+            loop = asyncio.get_event_loop()
+            fut: asyncio.Future[dict] = loop.create_future()
+            conn.pending_calls[call_id] = fut
+            await conn.ws.send_text(json.dumps({"type": "tool_call", **payload}))
+            return await asyncio.wait_for(fut, timeout=30.0)
+
+        # Caso 2: connessione remota → Redis pub/sub
+        loop = asyncio.get_event_loop()
+        fut = loop.create_future()
+        self._remote_futures[call_id] = fut
+
+        # Sottoscrivi al canale di risposta
+        result_channel = f"tool_result:{call_id}"
+        await self._pubsub.subscribe(result_channel)
+
+        # Pubblica il tool_call
+        await self._redis.publish(
+            f"tool_call:{user_id}",
+            json.dumps(payload),
+        )
+
+        try:
+            return await asyncio.wait_for(fut, timeout=30.0)
+        finally:
+            self._remote_futures.pop(call_id, None)
+            await self._pubsub.unsubscribe(result_channel)
+
+    # ── Risoluzione tool_result (da WS locale) ──
+
+    def resolve_local(self, user_id: str, call_id: str, result: dict):
+        conn = self._local.get(user_id)
+        if conn:
+            fut = conn.pending_calls.pop(call_id, None)
+            if fut and not fut.done():
+                fut.set_result(result)
+
+    async def resolve_and_publish(self, user_id: str, call_id: str, result: dict):
+        """Chiamato quando il device locale invia un tool_result."""
+        self.resolve_local(user_id, call_id, result)
+        # Pubblica anche su Redis per l'istanza remota che aspetta
+        await self._redis.publish(
+            f"tool_result:{call_id}",
+            json.dumps(result),
+        )
+
+    # ── Listener Redis ──
+
+    async def _listen_tool_calls(self):
+        """Loop che ascolta i tool_call in arrivo da altre istanze."""
+        async for message in self._pubsub.listen():
+            if message["type"] != "message":
+                continue
+            channel = message["channel"]
+            if isinstance(channel, bytes):
+                channel = channel.decode()
+
+            data = json.loads(message["data"])
+
+            if channel.startswith("tool_call:"):
+                # Un'altra istanza vuole che inviamo un tool_call al nostro device
+                user_id = channel.split(":", 1)[1]
+                conn = self._local.get(user_id)
+                if conn:
+                    await conn.ws.send_text(json.dumps({"type": "tool_call", **data}))
+
+            elif channel.startswith("tool_result:"):
+                # Risposta a un tool_call che abbiamo inviato tramite Redis
+                call_id = channel.split(":", 1)[1]
+                fut = self._remote_futures.pop(call_id, None)
+                if fut and not fut.done():
+                    fut.set_result(data)
+
+    # ── Stream cross-instance ──
+
+    async def publish_stream_chunk(self, user_id: str, chunk: dict):
+        """Pubblica un chunk di streaming su Redis (per REST→WS relay)."""
+        await self._redis.publish(f"stream:{user_id}", json.dumps(chunk))
+```
+
+---
+
+## 4. Struttura Directory Proposta (MVP)
+
+```
+adiuva-api/
+├── docker-compose.yml          # Orchestrazione completa
+├── docker-compose.dev.yml      # Override per sviluppo locale
+├── shared/                     # Codice condiviso (montato come volume)
+│   ├── auth.py                 # JWT verification (chiave pubblica)
+│   ├── schemas.py              # Pydantic schemas condivisi
+│   ├── middleware/
+│   │   ├── rate_limit.py       # DistributedRateLimiter (Redis)
+│   │   └── sanitizer.py
+│   └── models/
+│       └── base.py             # SQLAlchemy base condivisa
+│
+├── auth-service/
+│   ├── Dockerfile
+│   ├── requirements.txt
+│   └── app/
+│       ├── main.py
+│       ├── config.py
+│       ├── db.py
+│       ├── models.py           # users, refresh_tokens
+│       ├── routes/
+│       │   └── auth.py
+│       └── services/
+│           ├── jwt_service.py  # RS256 signing
+│           └── user_service.py
+│
+├── chat-service/
+│   ├── Dockerfile
+│   ├── requirements.txt
+│   └── app/
+│       ├── main.py
+│       ├── config.py
+│       ├── db.py
+│       ├── models.py           # memory_*
+│       ├── routes/
+│       │   ├── device_ws.py    # WS connection owner
+│       │   └── chat.py         # REST fallback
+│       ├── core/
+│       │   ├── device_manager.py   # RedisDeviceManager
+│       │   ├── deep_agent.py       # Home + floating chat
+│       │   ├── memory_middleware.py
+│       │   ├── ws_context.py
+│       │   ├── output_formatter.py
+│       │   └── llm.py
+│       └── agents/                 # Tool definitions (used by deep_agent)
+│           ├── task_agent.py
+│           ├── project_agent.py
+│           ├── note_agent.py
+│           └── timeline_agent.py
+│
+├── agent-service/
+│   ├── Dockerfile
+│   ├── requirements.txt
+│   └── app/
+│       ├── main.py
+│       ├── config.py
+│       ├── db.py
+│       ├── models.py           # agent_run_logs, local/cloud_agent_configs
+│       ├── routes/
+│       │   ├── agents.py       # catalog, can-create, trigger
+│       │   └── agent_setup.py  # journey start/message
+│       ├── core/
+│       │   ├── agent_runner.py     # Batch classify → process
+│       │   ├── agent_registry.py
+│       │   ├── redis_executor.py   # execute_on_client via Redis pub/sub
+│       │   └── llm.py
+│       └── agents/
+│           ├── task_agent.py       # Tool definitions (batch context)
+│           ├── project_agent.py
+│           ├── note_agent.py
+│           ├── timeline_agent.py
+│           └── filesystem_agent.py
+│
+├── billing-service/
+│   ├── Dockerfile
+│   ├── requirements.txt
+│   └── app/
+│       ├── main.py
+│       ├── config.py
+│       ├── db.py
+│       ├── models.py           # subscriptions
+│       ├── routes/
+│       │   └── billing.py
+│       └── services/
+│           ├── stripe_service.py
+│           └── tier_manager.py
+│
+└── infra/
+    ├── traefik/
+    │   └── traefik.yml
+    ├── keys/
+    │   ├── jwt_private.pem     # Solo auth-service
+    │   └── jwt_public.pem      # Tutti i servizi
+    └── alembic/                # Migrazioni condivise o per-servizio
+```
+
+---
+
+## 5. Docker Compose — Configurazione MVP
+
+```yaml
+# docker-compose.yml
+
+services:
+
+  # ══════════════════════════════════════════════════════════
+  # API Gateway
+  # ══════════════════════════════════════════════════════════
+  traefik:
+    image: traefik:v3.2
+    command:
+      - "--api.insecure=true"
+      - "--providers.docker=true"
+      - "--providers.docker.exposedbydefault=false"
+      - "--entrypoints.web.address=:80"
+      - "--entrypoints.websecure.address=:443"
+      - "--entrypoints.web.http.redirections.entrypoint.to=websecure"
+    ports:
+      - "80:80"
+      - "443:443"
+      - "8080:8080"   # Dashboard Traefik (disabilitare in prod)
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock:ro
+      - ./infra/certs:/certs:ro
+    restart: unless-stopped
+
+  # ══════════════════════════════════════════════════════════
+  # Auth Service (2 repliche)
+  # ══════════════════════════════════════════════════════════
+  auth-service:
+    build: ./auth-service
+    deploy:
+      replicas: 2
+    env_file: .env
+    environment:
+      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/adiuva
+      REDIS_URL: redis://redis:6379
+      JWT_PRIVATE_KEY_FILE: /run/secrets/jwt_private_key
+      SERVICE_NAME: auth
+    secrets:
+      - jwt_private_key
+      - jwt_public_key
+    labels:
+      - "traefik.enable=true"
+      - "traefik.http.routers.auth.rule=PathPrefix(`/api/v1/auth`)"
+      - "traefik.http.services.auth.loadbalancer.server.port=8000"
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
+  # ══════════════════════════════════════════════════════════
+  # Chat Service — Real-time WS + Chat (scalabile)
+  # ══════════════════════════════════════════════════════════
+  chat-service:
+    build: ./chat-service
+    deploy:
+      replicas: 2
+    env_file: .env
+    environment:
+      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/adiuva
+      REDIS_URL: redis://redis:6379
+      JWT_PUBLIC_KEY_FILE: /run/secrets/jwt_public_key
+      SERVICE_NAME: chat
+    secrets:
+      - jwt_public_key
+    labels:
+      - "traefik.enable=true"
+      # REST chat endpoint
+      - "traefik.http.routers.chat.rule=PathPrefix(`/api/v1/chat`)"
+      - "traefik.http.services.chat.loadbalancer.server.port=8000"
+      # WebSocket route con sticky session
+      - "traefik.http.routers.ws.rule=PathPrefix(`/api/v1/ws`)"
+      - "traefik.http.routers.ws.service=chat-ws"
+      - "traefik.http.services.chat-ws.loadbalancer.server.port=8000"
+      - "traefik.http.services.chat-ws.loadbalancer.sticky.cookie.name=ws_affinity"
+      - "traefik.http.services.chat-ws.loadbalancer.sticky.cookie.httpOnly=true"
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
+  # ══════════════════════════════════════════════════════════
+  # Agent Service — Batch processing (scalabile indipendentemente)
+  # ══════════════════════════════════════════════════════════
+  agent-service:
+    build: ./agent-service
+    deploy:
+      replicas: 2
+    env_file: .env
+    environment:
+      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/adiuva
+      REDIS_URL: redis://redis:6379
+      JWT_PUBLIC_KEY_FILE: /run/secrets/jwt_public_key
+      SERVICE_NAME: agent
+    secrets:
+      - jwt_public_key
+    labels:
+      - "traefik.enable=true"
+      - "traefik.http.routers.agents.rule=PathPrefix(`/api/v1/agents`)"
+      - "traefik.http.services.agents.loadbalancer.server.port=8000"
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
+  # ══════════════════════════════════════════════════════════
+  # Billing Service (1 replica)
+  # ══════════════════════════════════════════════════════════
+  billing-service:
+    build: ./billing-service
+    deploy:
+      replicas: 1
+    env_file: .env
+    environment:
+      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/adiuva
+      REDIS_URL: redis://redis:6379
+      JWT_PUBLIC_KEY_FILE: /run/secrets/jwt_public_key
+      SERVICE_NAME: billing
+    secrets:
+      - jwt_public_key
+    labels:
+      - "traefik.enable=true"
+      - "traefik.http.routers.billing.rule=PathPrefix(`/api/v1/billing`)"
+      - "traefik.http.services.billing.loadbalancer.server.port=8000"
+    depends_on:
+      db:
+        condition: service_healthy
+      redis:
+        condition: service_healthy
+
+  # ══════════════════════════════════════════════════════════
+  # Infrastruttura
+  # ══════════════════════════════════════════════════════════
+  db:
+    image: pgvector/pgvector:pg16
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: adiuva
+    volumes:
+      - postgres_data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 5s
+      timeout: 5s
+      retries: 5
+    restart: unless-stopped
+
+  redis:
+    image: redis:7-alpine
+    command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
+    volumes:
+      - redis_data:/data
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 5s
+      timeout: 3s
+      retries: 5
+    restart: unless-stopped
+
+  qdrant:
+    image: qdrant/qdrant:latest
+    volumes:
+      - qdrant_data:/qdrant/storage
+    restart: unless-stopped
+
+secrets:
+  jwt_private_key:
+    file: ./infra/keys/jwt_private.pem
+  jwt_public_key:
+    file: ./infra/keys/jwt_public.pem
+
+volumes:
+  postgres_data:
+  redis_data:
+  qdrant_data:
+```
+
+---
+
+## 6. Configurazione Cloudflare + VPS
+
+### 6.1 DNS
+
+```
+api.tuodominio.com  →  A record  →  IP del VPS
+                    →  Proxy: ON (orange cloud)
+```
+
+### 6.2 Cloudflare Settings
+
+| Setting | Valore | Motivo |
+|---------|--------|--------|
+| SSL/TLS mode | **Full (Strict)** | Cloudflare ↔ VPS con certificato valido |
+| WebSocket | **ON** | Necessario per `/api/v1/ws/device` |
+| Proxy timeout | **100s** (Enterprise) o default | Le LLM calls possono durare 30s+ |
+| Under Attack Mode | Off (attivare se necessario) | |
+
+### 6.3 TLS sul VPS
+
+Due opzioni:
+- **Opzione A (consigliata)**: Cloudflare Origin Certificate → montato in Traefik
+- **Opzione B**: Let's Encrypt via Traefik (con DNS challenge Cloudflare)
+
+```yaml
+# traefik.yml — con Cloudflare Origin Certificate
+entryPoints:
+  websecure:
+    address: ":443"
+
+tls:
+  certificates:
+    - certFile: /certs/origin.pem
+      keyFile: /certs/origin-key.pem
+```
+
+### 6.4 Rete VPS
+
+```bash
+# UFW firewall — solo Cloudflare può raggiungere le porte 80/443
+# https://www.cloudflare.com/ips/
+ufw default deny incoming
+ufw allow from 173.245.48.0/20 to any port 443
+ufw allow from 103.21.244.0/22 to any port 443
+# ... (tutti gli IP range di Cloudflare)
+ufw allow ssh
+ufw enable
+```
+
+---
+
+## 7. Comunicazione Inter-Servizio
+
+### 7.1 Redis Pub/Sub — Event Bus
+
+```
+┌──────────┐  tier_changed:user_123   ┌──────────┐
+│ Billing  │ ────────────────────────► │   Auth   │
+│ Service  │                           │ Service  │
+└──────────┘                           └──────────┘
+
+┌──────────┐  tool_call:user_123      ┌──────────┐
+│  Agent   │ ────────────────────────► │   Chat   │
+│ Service  │                           │ Service  │
+│ (batch)  │ ◄────────────────────────│ (ha WS)  │
+└──────────┘  tool_result:{call_id}    └──────────┘
+```
+
+### 7.2 Health Checks e Service Discovery
+
+Traefik gestisce automaticamente il service discovery via Docker labels. I servizi non devono conoscersi tra loro — comunicano solo via:
+- **Redis pub/sub** (tool-call cross-instance, tier events)
+- **Redis hash** (stato condiviso: `ws:connections`, rate-limit counters)
+- **PostgreSQL** (dati persistenti condivisi)
+
+---
+
+## 8. Piano di Migrazione Incrementale (MVP)
+
+### Fase 1 — Preparazione (nel monolite attuale)
+1. Aggiungere Redis al `docker-compose.yml` attuale
+2. Migrare JWT da HS256 → RS256 (backward-compatible: accetta entrambi per un periodo)
+3. Implementare `RedisDeviceManager` come drop-in replacement del singleton in-memory
+4. Estrarre `shared/` con auth verification, schemas, middleware
+
+### Fase 2 — Auth Service (primo split)
+1. Estrarre `auth.py` routes + models in `auth-service/`
+2. Verificare che i JWT firmati da `auth-service` vengano validati dal monolite
+3. Aggiungere Traefik e routare `/api/v1/auth/*` al nuovo servizio
+4. Il monolite continua a servire tutto il resto
+
+### Fase 3 — Billing Service
+1. Estrarre billing routes, Stripe service, tier manager
+2. Configurare Redis pub/sub per `tier_changed` events
+3. Routare via Traefik
+
+### Fase 4 — Split Chat + Agent (il più delicato)
+1. Il monolite residuo contiene WS + chat + agents
+2. Separare Agent Service: estrarre `agent_runner`, `agent_registry`, `agent_setup`, route `/agents/*`
+3. Implementare `redis_executor.py` nell'Agent Service per tool-call via Redis
+4. Il Chat Service resta proprietario della WS e sottoscrive i canali `tool_call:{user_id}`
+5. Testare: trigger agent dall'Agent Service → tool_call via Redis → Chat Service → WS → device → risposta
+
+### Fase 5 — Scaling test
+1. Scalare Chat Service a 2 repliche, verificare sticky sessions
+2. Scalare Agent Service a 2 repliche, verificare batch processing distribuito
+3. Monitoring (Prometheus + Grafana) per ogni servizio
+
+---
+
+## 9. Monitoraggio e Logging
+
+```yaml
+# Aggiungere al docker-compose.yml
+
+  prometheus:
+    image: prom/prometheus:latest
+    volumes:
+      - ./infra/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
+    restart: unless-stopped
+
+  grafana:
+    image: grafana/grafana:latest
+    ports:
+      - "3000:3000"
+    volumes:
+      - grafana_data:/var/lib/grafana
+    restart: unless-stopped
+
+  loki:
+    image: grafana/loki:latest
+    restart: unless-stopped
+```
+
+Ogni servizio espone `/metrics` (Prometheus) e scrive log strutturati (JSON) raccolti da Loki.
+
+---
+
+## 10. Sizing VPS Minimo Consigliato (MVP)
+
+| Componente | CPU | RAM | Note |
+|---|---|---|---|
+| Traefik | 0.25 | 128MB | |
+| Auth Service ×2 | 0.25 ×2 | 128MB ×2 | Stateless, leggero |
+| Chat Service ×2 | 1.0 ×2 | 1GB ×2 | WS + streaming LLM |
+| Agent Service ×2 | 0.75 ×2 | 512MB ×2 | Batch LLM, CPU-bound |
+| Billing Service | 0.25 | 128MB | |
+| PostgreSQL | 1.0 | 1GB | |
+| Redis | 0.25 | 256MB | |
+| Qdrant | 0.5 | 512MB | |
+| **Totale MVP** | **~5.5 vCPU** | **~5 GB** | |
+
+**Raccomandazione**: VPS con **8 vCPU / 16 GB RAM** per avere margine. Hetzner CPX41 (~€30/mese) o equivalente. Senza Storage/Plugin si risparmia ~1 vCPU e 512MB rispetto alla versione completa.
+
+---
+
+## Riepilogo Architettura MVP
+
+| Servizio | Repliche | Proprietario di |
+|---|---|---|
+| **Traefik** | 1 | Routing, TLS, sticky sessions |
+| **Auth Service** | 2 | JWT RS256, registrazione, login, profilo |
+| **Chat Service** | 2–N | WebSocket, home/floating chat, streaming |
+| **Agent Service** | 2–N | Batch processing, directory scan, agent setup |
+| **Billing Service** | 1 | Stripe, subscriptions, tier management |
+
+| Decisione | Scelta | Motivazione |
+|---|---|---|
+| API Gateway | Traefik | Nativo Docker, WebSocket support, service discovery automatico |
+| JWT | RS256 (asimmetrico) | Verifica distribuita senza contattare Auth Service |
+| Tier check | Claim nel JWT | Ogni servizio verifica localmente, zero roundtrip |
+| WebSocket scaling | Redis pub/sub + sticky cookies | Cross-instance tool-call routing |
+| Chat ↔ Agent split | Servizi separati | Batch CPU-bound non impatta real-time chat |
+| Agent → Device comms | Redis pub/sub via Chat Service | Agent non possiede la WS, usa un relay |
+| Rate limiting | Redis contatori distribuiti | Sliding window condivisa tra repliche |
+| Database | PostgreSQL condiviso | Semplicità MVP; split DB futuro facile |
+| TLS | Cloudflare Origin Certificate | Zero maintenance |
+| Orchestrazione | Docker Compose | Sufficiente per un singolo VPS |
+| Storage / Plugin | Post-MVP | Non critici per il lancio |

requirements.txt

@@ -4,8 +4,6 @@ gunicorn>=22.0.0
 langchain>=0.3.0
 langchain-openai>=0.3.0
 langchain-litellm>=0.1.0
-langgraph>=0.3.0
-deepagents>=0.4.10
 litellm>=1.50.0
 pydantic>=2.10.0
 pydantic-settings>=2.7.0
@@ -34,4 +32,8 @@ google-auth-oauthlib>=1.2.0
 google-auth-httplib2>=0.2.0
 msal>=1.28.0
 cryptography>=42.0.0
+langfuse>=2.0.0
+beautifulsoup4>=4.12.0
+lxml>=5.0.0
+PyYAML>=6.0.0
 ruff>=0.8.0

tests/fixtures/preprocessors/cases.yaml

@@ -0,0 +1,121 @@
+# Preprocessor test cases — Step 1 (Local Agent V2)
+#
+# Schema per caso:
+#   id: "1.N"
+#   description: str
+#   score_name: str                     # nome score inviato a Langfuse
+#
+# Sorgente contenuto (una delle due):
+#   file: <nome file in data/>          # letto come testo UTF-8
+#   generate: binary_noise              # contenuto generato dal runner (per test binari)
+#
+# Per op=detect:
+#   op: detect
+#   input_filename: str                 # filename passato a detect_content_type
+#   expected_content_type: str
+#
+# Per op=preprocess:
+#   op: preprocess
+#   input_content_type: str             # content_type passato a preprocess()
+#   assertions:
+#     no_html_tags: bool
+#     min_length: int
+#     compression_ratio_lt: float       # len(clean) / len(raw) < soglia
+#     metadata_keys: [str, ...]         # chiavi che devono essere in metadata
+#     contains: str | [str, ...]        # substring(s) presenti in clean_text
+#     not_contains: str | [str, ...]    # substring(s) assenti da clean_text
+#     content_type: str                 # valore atteso di result.content_type
+
+cases:
+
+  # ── Detection tests ────────────────────────────────────────────────
+
+  - id: "1.1"
+    description: "Detect email HTML"
+    score_name: preprocess.detect_email
+    file: email_action.html
+    op: detect
+    input_filename: email_export.html
+    expected_content_type: email_html
+
+  - id: "1.2"
+    description: "Detect generic HTML"
+    score_name: preprocess.detect_generic
+    file: generic_page.html
+    op: detect
+    input_filename: index.html
+    expected_content_type: generic_html
+
+  - id: "1.3"
+    description: "Detect plain text"
+    score_name: preprocess.detect_text
+    file: notes.txt
+    op: detect
+    input_filename: notes.txt
+    expected_content_type: plain_text
+
+  - id: "1.4"
+    description: "Detect unknown (binary-like content)"
+    score_name: preprocess.detect_unknown
+    generate: binary_noise
+    op: detect
+    input_filename: archive.xyz
+    expected_content_type: unknown
+
+  # ── Preprocess tests ───────────────────────────────────────────────
+
+  - id: "1.5"
+    description: "Email: strip HTML tags"
+    file: email_action.html
+    op: preprocess
+    input_content_type: email_html
+    assertions:
+      no_html_tags: true
+      min_length: 50
+      compression_ratio_lt: 0.8
+
+  - id: "1.6"
+    description: "Email: extract metadata (Subject + From)"
+    file: email_action.html
+    op: preprocess
+    input_content_type: email_html
+    assertions:
+      metadata_keys: [subject, from]
+
+  - id: "1.7"
+    description: "Email: split thread — solo ultimo messaggio"
+    file: email_thread.html
+    op: preprocess
+    input_content_type: email_html
+    assertions:
+      contains: "Sure, I'll handle the deploy"
+      not_contains: "Let's plan the deploy"
+
+  - id: "1.8"
+    description: "Email: singolo messaggio senza thread"
+    file: email_single.html
+    op: preprocess
+    input_content_type: email_html
+    assertions:
+      contains: "deploy is done"
+
+  - id: "1.9"
+    description: "Email: HTML pesante con table layout"
+    file: email_heavy.html
+    op: preprocess
+    input_content_type: email_html
+    assertions:
+      no_html_tags: true
+      min_length: 30
+      not_contains:
+        - "border-collapse"
+        - "font-size"
+
+  - id: "1.10"
+    description: "Fallback: file sconosciuto → testo restituito"
+    file: fallback.txt
+    op: preprocess
+    input_content_type: unknown
+    assertions:
+      min_length: 1
+      content_type: unknown

tests/fixtures/preprocessors/data/email_action.html

@@ -0,0 +1,25 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <title>Fix the login bug</title>
+  <style>
+    body { font-family: Arial, sans-serif; color: #333; margin: 0; padding: 20px; }
+    .header { background: #f5f5f5; padding: 10px; border-bottom: 1px solid #ddd; }
+    .body { padding: 20px; }
+  </style>
+</head>
+<body>
+  <div class="header">
+    <p><strong>From:</strong> boss@company.com</p>
+    <p><strong>To:</strong> dev@company.com</p>
+    <p><strong>Subject:</strong> Fix the login bug</p>
+    <p><strong>Date:</strong> Mon, 7 Apr 2026 09:00:00 +0200</p>
+  </div>
+  <div class="body">
+    <p>Hi,</p>
+    <p>Please fix the login bug by Friday. It is blocking the release.</p>
+    <p>Priority: high. Let me know if you need anything.</p>
+    <p>Thanks,<br>Boss</p>
+  </div>
+</body>
+</html>

tests/fixtures/preprocessors/data/email_heavy.html

@@ -0,0 +1,49 @@
+<!DOCTYPE html>
+<html>
+<head>
+<style>
+  table { border-collapse: collapse; width: 100%; max-width: 600px; margin: 0 auto; }
+  td { padding: 8px 12px; border: 1px solid #dddddd; font-size: 12px; color: #444444; }
+  .header-row { background-color: #003366; color: #ffffff; font-weight: bold; }
+  .label-col { background-color: #f0f0f0; width: 80px; font-weight: bold; }
+  .footer-row { font-size: 10px; color: #999999; text-align: center; }
+</style>
+</head>
+<body bgcolor="#eeeeee">
+<center>
+<table cellpadding="0" cellspacing="0">
+  <tr class="header-row">
+    <td colspan="2">Company Internal Update</td>
+  </tr>
+  <tr>
+    <td class="label-col">From:</td>
+    <td>newsletter@corp.com</td>
+  </tr>
+  <tr>
+    <td class="label-col">Subject:</td>
+    <td>Q1 Results Update</td>
+  </tr>
+  <tr>
+    <td class="label-col">Date:</td>
+    <td>Apr 7, 2026</td>
+  </tr>
+  <tr>
+    <td colspan="2">
+      <table width="100%" cellpadding="10">
+        <tr>
+          <td>
+            <p style="font-size:14px; font-weight:bold;">Dear Team,</p>
+            <p>Q1 results are in. Revenue up 15% year-over-year.</p>
+            <p>Please review the attached report and share any feedback by EOW.</p>
+          </td>
+        </tr>
+      </table>
+    </td>
+  </tr>
+  <tr class="footer-row">
+    <td colspan="2">Confidential — do not forward outside the company.</td>
+  </tr>
+</table>
+</center>
+</body>
+</html>

tests/fixtures/preprocessors/data/email_single.html

@@ -0,0 +1,8 @@
+<!DOCTYPE html>
+<html><body>
+  <p><strong>From:</strong> alice@co.com</p>
+  <p><strong>To:</strong> team@co.com</p>
+  <p><strong>Subject:</strong> Quick update</p>
+  <p><strong>Date:</strong> Tue, 7 Apr 2026 10:30:00 +0200</p>
+  <p>The deploy is done. Everything looks good. No issues so far.</p>
+</body></html>

tests/fixtures/preprocessors/data/email_thread.html

@@ -0,0 +1,24 @@
+<!DOCTYPE html>
+<html><body>
+  <div class="message-latest">
+    <p><strong>From:</strong> alice@co.com</p>
+    <p><strong>Subject:</strong> Re: Re: Deploy plan</p>
+    <p>Sure, I'll handle the deploy.</p>
+  </div>
+
+  <p>On Mon, Apr 6, 2026 at 3:00 PM, Bob &lt;bob@co.com&gt; wrote:</p>
+  <blockquote>
+    <p>From: bob@co.com</p>
+    <p>Can you handle the deploy?</p>
+    <p>On Sun, Apr 5, 2026 at 1:00 PM, Alice &lt;alice@co.com&gt; wrote:</p>
+    <blockquote>
+      <p>From: alice@co.com</p>
+      <p>Let's plan the deploy for Monday.</p>
+      <p>On Sat, Apr 4, 2026 at 11:00 AM, Charlie &lt;charlie@co.com&gt; wrote:</p>
+      <blockquote>
+        <p>From: charlie@co.com</p>
+        <p>We need to schedule the deploy. What day works?</p>
+      </blockquote>
+    </blockquote>
+  </blockquote>
+</body></html>

tests/fixtures/preprocessors/data/fallback.txt

@@ -0,0 +1,3 @@
+random text content without any structure
+line two with some words
+line three and more content here

tests/fixtures/preprocessors/data/generic_page.html

@@ -0,0 +1,35 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+  <meta charset="UTF-8">
+  <title>My Web App</title>
+  <link rel="stylesheet" href="styles.css">
+</head>
+<body>
+  <nav>
+    <a href="/">Home</a>
+    <a href="/about">About</a>
+    <a href="/contact">Contact</a>
+  </nav>
+  <main>
+    <header>
+      <h1>Welcome to My App</h1>
+    </header>
+    <article>
+      <p>This is a generic web page with no email headers.</p>
+      <p>It has navigation, main content, and a footer.</p>
+    </article>
+    <section>
+      <h2>Features</h2>
+      <ul>
+        <li>Fast</li>
+        <li>Reliable</li>
+        <li>Secure</li>
+      </ul>
+    </section>
+  </main>
+  <footer>
+    <p>&copy; 2026 My App</p>
+  </footer>
+</body>
+</html>

tests/fixtures/preprocessors/data/notes.txt

@@ -0,0 +1,15 @@
+Meeting notes - April 7, 2026
+
+Attendees: Alice, Bob, Charlie
+
+Discussion points:
+- Deploy scheduled for Friday
+- Bug fix for login must be completed by Thursday
+- Review Q1 numbers before EOW
+
+Action items:
+- Alice: fix login bug
+- Bob: prepare deploy checklist
+- Charlie: send Q1 report
+
+Next meeting: April 14, 2026

tests/test_agent_runner.py

@@ -10,13 +10,13 @@ Coverage:
     - 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        — skipped when config is client-owned
     - 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/can-create     — billing eligibility check
-    - POST /agents/{id}/run       — creates run log + dispatches background task
+        - POST /agents/trigger        — creates run log + dispatches background task
 """
 
 from __future__ import annotations
@@ -373,7 +373,7 @@ async def test_run_local_agent_happy_path():
     assert kwargs["items_processed"] == 1
     assert kwargs["items_created"] == 1
     assert kwargs["errors"] == []
-    assert kwargs["update_config_last_run"] is True
+    assert kwargs["update_config_last_run"] is False
 
     # Verify agent_run frame was sent.
     agent_run_frames = [f for f in sent_frames if f.get("type") == "agent_run"]
@@ -690,31 +690,11 @@ async def test_finalize_run_updates_cloud_config_last_run_at():
 
 @pytest.mark.asyncio
 async def test_trigger_pending_runs_no_overdue():
-    """If no agents are overdue trigger_pending_runs does nothing."""
+    """Pending-run scan is skipped because agent config is client-owned."""
-    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, \
+    with patch("app.core.agent_runner.run_local_agent", new_callable=AsyncMock) as mock_run:
-         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()
@@ -722,31 +702,11 @@ async def test_trigger_pending_runs_no_overdue():
 
 @pytest.mark.asyncio
 async def test_trigger_pending_runs_device_id_filter():
-    """Local agents are only triggered for the matching device_id."""
+    """Device filtering is no longer backend-managed in pending runs."""
-    # 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, \
+    with patch("app.core.agent_runner.run_local_agent", new_callable=AsyncMock) as mock_run:
-         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()
@@ -754,56 +714,18 @@ async def test_trigger_pending_runs_device_id_filter():
 
 @pytest.mark.asyncio
 async def test_trigger_pending_runs_dispatches_overdue():
-    """Overdue local agent triggers run_local_agent sequentially."""
+    """No pending runs are dispatched by backend after config deprecation."""
-    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] = []
+    with patch("app.core.agent_runner.run_local_agent", new_callable=AsyncMock) as mock_run:
-
-    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"]
+    mock_run.assert_not_called()
 
 
 # ---------------------------------------------------------------------------
-# Integration: POST /agents/{id}/run
+# Integration: POST /agents/can-create and /agents/trigger
 # ---------------------------------------------------------------------------
 
 
@@ -820,50 +742,67 @@ def _override_db(db_session):
 
 
 @pytest.mark.asyncio
-async def test_trigger_run_unknown_agent(client):
+async def test_can_create_agent_allows_when_under_limit(client):
-    """POST /agents/{id}/run returns 404 for unknown agent id."""
+    """POST /agents/can-create returns allowed=True when under tier limit."""
     resp = client.post(
-        f"/api/v1/agents/{uuid.uuid4()}/run",
+        "/api/v1/agents/can-create",
-        headers=auth_header("power"),
+        json={"active_agents": 0},
+        headers=auth_header("free"),
     )
-    assert resp.status_code == 404
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["allowed"] is True
+    assert body["tier"] == "free"
+    assert body["active_agents"] == 0
+    assert body["limit"] == 2
+
+
+@pytest.mark.asyncio
+async def test_can_create_agent_denies_when_at_limit(client):
+    """POST /agents/can-create returns allowed=False at free-tier limit."""
+    resp = client.post(
+        "/api/v1/agents/can-create",
+        json={"active_agents": 2},
+        headers=auth_header("free"),
+    )
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["allowed"] is False
+    assert body["limit"] == 2
 
 
 @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."""
+    """POST /agents/trigger creates a local run log and dispatches background task."""
-    # Create the local agent config in the DB.
+    dispatched: list[tuple[str, str]] = []
-    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))
 
+    def _fake_create_task(coro):
+        coro.close()
+        return MagicMock()
+
     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:
+        mock_create_task.side_effect = _fake_create_task
         resp = client.post(
-            f"/api/v1/agents/{config.id}/run",
+            "/api/v1/agents/trigger",
+            json={
+                "directory": "/home/user/docs",
+                "what_to_extract": ["task", "note"],
+                "actions_by_type": {"task": ["add", "update"], "note": ["add"]},
+                "batch_interval": "0 */6 * * *",
+                "custom_agent_prompt": "Extract tasks and notes.",
+                "active_agents": 0,
+            },
             headers=auth_header("power"),
         )
 
     assert resp.status_code == 202
     data = resp.json()
-    assert data["agent_id"] == config.id
+    assert isinstance(data["agent_id"], str)
+    assert data["agent_id"]
     assert data["status"] == "running"
     assert data["agent_type"] == "local"
 

tests/test_classify_file.py

@@ -0,0 +1,184 @@
+"""Unit tests for Step 1 file classification (_classify_file).
+
+These tests call the real LLM so they require OPENAI_API_KEY / LLM env vars.
+Run with: pytest tests/test_classify_file.py -v
+
+To run a quick manual check against a real file without the full UI:
+    python -m tests.test_classify_file <path/to/file.txt> [project_name...]
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+
+import pytest
+
+from app.core.agent_runner import _classify_file
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────
+
+PROJECTS_SAMPLE = [
+    {
+        "id": "aaaa-0001-0000-0000-000000000001",
+        "name": "ARPA Sicilia POC",
+        "status": "active",
+        "aiSummary": "Proof of concept for AI features targeting ARPA Sicilia agency.",
+    },
+    {
+        "id": "bbbb-0002-0000-0000-000000000002",
+        "name": "SNAM AI Meeting Prep",
+        "status": "active",
+        "aiSummary": "AI-assisted preparation of meeting materials for SNAM.",
+    },
+    {
+        "id": "cccc-0003-0000-0000-000000000003",
+        "name": "SFERA+ Wave 2",
+        "status": "active",
+        "aiSummary": "Second wave of the SFERA+ whitelist project.",
+    },
+]
+
+ARPA_EMAIL = """\
+to: roberto.musso@hpe.com; luca.tondin@hpecds.com
+isImportance: normal
+hasAttachment: True
+---
+## Body
+Buongiorno,
+
+In riferimento alla riunione di ieri sul POC ARPA Sicilia, vi invio il riassunto
+dei deliverable concordati:
+- Preparare demo entro il 30 marzo
+- Condividere documentazione tecnica con il team ARPA
+- Fissare call di follow-up la prossima settimana
+
+Cordiali saluti
+Roberto Marchetti
+"""
+
+SNAM_EMAIL = """\
+to: roberto.musso@hpe.com
+isImportance: high
+hasAttachment: False
+---
+## Body
+Ciao,
+ti invio l'agenda per la riunione SNAM di domani.
+Per favore conferma la tua presenza.
+"""
+
+UNRELATED_EMAIL = """\
+to: roberto.musso@hpe.com
+isImportance: normal
+---
+## Body
+Benvenuto nel programma HPE Employee Learning Series.
+Completa la formazione richiesta entro la fine del trimestre.
+"""
+
+
+# ── Tests ─────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_classify_arpa_matches_existing():
+    project_id, domains, new_name = await _classify_file(
+        file_path="arpa_email.txt",
+        file_content=ARPA_EMAIL,
+        projects=PROJECTS_SAMPLE,
+        config_data_types=["tasks", "notes", "timelines"],
+    )
+    assert project_id == "aaaa-0001-0000-0000-000000000001", (
+        f"Expected ARPA project, got project_id={project_id!r} new_name={new_name!r}"
+    )
+    assert new_name is None
+
+
+@pytest.mark.asyncio
+async def test_classify_snam_matches_existing():
+    project_id, domains, new_name = await _classify_file(
+        file_path="snam_email.txt",
+        file_content=SNAM_EMAIL,
+        projects=PROJECTS_SAMPLE,
+        config_data_types=["tasks", "notes"],
+    )
+    assert project_id == "bbbb-0002-0000-0000-000000000002", (
+        f"Expected SNAM project, got project_id={project_id!r} new_name={new_name!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_classify_unrelated_returns_new():
+    project_id, domains, new_name = await _classify_file(
+        file_path="learning_email.txt",
+        file_content=UNRELATED_EMAIL,
+        projects=PROJECTS_SAMPLE,
+        config_data_types=["tasks", "notes"],
+    )
+    assert project_id == "new"
+    assert new_name is not None  # LLM should suggest a name
+
+
+@pytest.mark.asyncio
+async def test_classify_empty_file_returns_new():
+    project_id, domains, new_name = await _classify_file(
+        file_path="empty.txt",
+        file_content="   ",
+        projects=PROJECTS_SAMPLE,
+        config_data_types=["tasks"],
+    )
+    assert project_id == "new"
+
+
+@pytest.mark.asyncio
+async def test_classify_no_projects_returns_new():
+    project_id, domains, new_name = await _classify_file(
+        file_path="arpa_email.txt",
+        file_content=ARPA_EMAIL,
+        projects=[],
+        config_data_types=["tasks", "notes"],
+    )
+    assert project_id == "new"
+    assert new_name is not None
+
+
+# ── CLI quick-test runner ─────────────────────────────────────────────────
+
+
+async def _cli_test(file_path: str, project_names: list[str]) -> None:
+    """Run Step 1 classification against a real file from the CLI."""
+    import json
+    from pathlib import Path
+
+    content = Path(file_path).read_text(encoding="utf-8", errors="replace")
+    projects = [
+        {"id": f"test-id-{i:04d}", "name": name, "status": "active", "aiSummary": ""}
+        for i, name in enumerate(project_names)
+    ]
+
+    print(f"\nClassifying: {file_path}")
+    print(f"Projects in context: {[p['name'] for p in projects]}\n")
+
+    project_id, domains, new_name = await _classify_file(
+        file_path=file_path,
+        file_content=content,
+        projects=projects,
+        config_data_types=["tasks", "notes", "timelines"],
+    )
+
+    result = {
+        "project_id": project_id,
+        "matched_name": next((p["name"] for p in projects if p["id"] == project_id), None),
+        "new_project_name": new_name,
+        "domains": domains,
+    }
+    print(json.dumps(result, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: python -m tests.test_classify_file <file_path> [project_name ...]")
+        sys.exit(1)
+    asyncio.run(_cli_test(sys.argv[1], sys.argv[2:]))

tests/test_deep_agent.py

@@ -0,0 +1,288 @@
+"""Unit tests for single-agent deep_agent flows with mocked tool results."""
+
+from __future__ import annotations
+
+from datetime import date, timedelta
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from langchain_core.messages import AIMessage, ToolMessage
+
+from app.core.deep_agent import (
+    _infer_floating_domain,
+    _normalize_tagged_list_lines,
+    run_floating,
+    run_floating_stream,
+    run_home,
+)
+
+
+class _FakeTool:
+    name = "list_tasks"
+
+    async def ainvoke(self, args):
+        return {"rows": [{"id": "task-1", "title": "Mock Task"}], "echo": args}
+
+
+class _FakeLLM:
+    def __init__(self) -> None:
+        self.agent_calls = 0
+
+    def bind_tools(self, _tools):
+        return self
+
+    async def ainvoke(self, messages):
+        system_prompt = str(getattr(messages[0], "content", "")) if messages else ""
+        if "strict domain classifier" in system_prompt:
+            return AIMessage(content='{"type":"timeline","id":"tl-1","section":null}')
+
+        self.agent_calls += 1
+        if self.agent_calls == 1:
+            return AIMessage(
+                content="",
+                tool_calls=[
+                    {
+                        "id": "call-1",
+                        "name": "list_tasks",
+                        "args": {"project_id": "proj-1"},
+                    }
+                ],
+            )
+
+        tool_messages = [m for m in messages if isinstance(m, ToolMessage)]
+        assert tool_messages, "Expected at least one tool message"
+        return AIMessage(content=f"Final answer from mocked tool: {tool_messages[-1].content}")
+
+    async def astream(self, _messages):
+        yield SimpleNamespace(content="stream-")
+        yield SimpleNamespace(content="ok")
+
+
+@pytest.mark.asyncio
+async def test_run_home_uses_mocked_tool_result():
+    fake_llm = _FakeLLM()
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._all_tools", return_value=[_FakeTool()]
+    ):
+        out = await run_home("user-1", "list my tasks", {})
+
+    assert "Final answer from mocked tool" in out
+    assert "Mock Task" in out
+
+
+@pytest.mark.asyncio
+async def test_run_floating_stream_emits_domain_then_tokens_with_mocked_tool_result():
+    fake_llm = _FakeLLM()
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._all_tools", return_value=[_FakeTool()]
+    ):
+        events = []
+        async for event in run_floating_stream(
+            "user-1",
+            "show me timeline updates",
+            {"scope": {"type": "timeline", "id": "tl-1"}},
+        ):
+            events.append(event)
+
+    assert events[0] == (
+        "floating_domain",
+        {"type": "timeline", "id": "tl-1", "section": None},
+    )
+    assert ("token", "stream-") in events
+    assert ("token", "ok") in events
+
+
+@pytest.mark.asyncio
+async def test_infer_floating_domain_prefers_message_intent_over_scope_type():
+    class _ClassifierOnlyLLM:
+        async def ainvoke(self, _messages):
+            return AIMessage(
+                content='{"type":"project","id":"213213-312321-312312-421321","section":"task"}'
+            )
+
+    with patch("app.core.deep_agent.get_llm", return_value=_ClassifierOnlyLLM()):
+        domain = await _infer_floating_domain(
+            "Quali sono i miei task per il progetto X",
+            {
+                "scope": {"type": "timeline"},
+                "resolved_project_id": "213213-312321-312312-421321",
+            },
+        )
+
+    assert domain == {
+        "type": "project",
+        "id": "213213-312321-312312-421321",
+        "section": "task",
+    }
+
+
+def test_normalize_tagged_list_lines_rewrites_mixed_task_lines_to_tag_only_lines():
+    raw = (
+        "Certo!\n\n"
+        "1. **Task A** — priorita high <task>[task-1]</task>\n"
+        "2. **Task B** — priorita medium <task>[task-2]</task>\n"
+    )
+
+    out = _normalize_tagged_list_lines(raw, "quali sono le prossime attivita?")
+
+    assert "<task>[task-1]</task>" in out
+    assert "<task>[task-2]</task>" in out
+    assert "Task A" not in out
+    assert "Task B" not in out
+
+
+def test_normalize_tagged_list_lines_filters_upcoming_timeline_query_to_current_month_future_only():
+    today = date.today()
+    tomorrow = today + timedelta(days=1)
+    yesterday = today - timedelta(days=1)
+    next_month = (today.replace(day=28) + timedelta(days=5)).replace(day=1)
+
+    raw = "\n".join(
+        [
+            f"- Milestone old — {yesterday.strftime('%d/%m/%Y')} <timeline>[tl-old]</timeline>",
+            f"- Milestone next — {tomorrow.strftime('%d/%m/%Y')} <timeline>[tl-next]</timeline>",
+            f"- Milestone future — {next_month.strftime('%d/%m/%Y')} <timeline>[tl-future]</timeline>",
+        ]
+    )
+
+    out = _normalize_tagged_list_lines(raw, "invece i miei eventi prossimi?")
+
+    assert "<timeline>[tl-next]</timeline>" in out
+    assert "<timeline>[tl-old]</timeline>" not in out
+    assert "<timeline>[tl-future]</timeline>" not in out
+
+
+@pytest.mark.asyncio
+async def test_run_floating_strips_xml_like_tags_from_final_text():
+    fake_llm = _FakeLLM()
+
+    async def _fake_run_single_agent(**_kwargs):
+        return (
+            "Hai 1 task:\\n"
+            "Mail barra in prod <task>[180faff3-507d-4d88-aba8-66f204eb59ef]</task>"
+        )
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._run_single_agent", side_effect=_fake_run_single_agent
+    ):
+        text, _domain = await run_floating(
+            "user-1",
+            "quali task ho?",
+            {"scope": {"type": "task"}},
+        )
+
+    assert "<task>" not in text
+    assert "</task>" not in text
+    assert "[180faff3-507d-4d88-aba8-66f204eb59ef]" not in text
+
+
+@pytest.mark.asyncio
+async def test_run_floating_stream_strips_xml_like_tags_from_streamed_text():
+    fake_llm = _FakeLLM()
+
+    async def _fake_stream(**_kwargs):
+        yield "token", "Hai 1 task:\\n"
+        yield "token", "Mail barra in prod <task>[180faff3-507d-4d88-aba8-66f204eb59ef]</task>"
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._run_single_agent_stream", side_effect=_fake_stream
+    ):
+        events = []
+        async for event in run_floating_stream(
+            "user-1",
+            "quali task ho?",
+            {"scope": {"type": "task"}},
+        ):
+            events.append(event)
+
+    token_events = [str(data) for event_type, data in events if event_type == "token"]
+    combined = "".join(token_events)
+    assert "<task>" not in combined
+    assert "</task>" not in combined
+    assert "[180faff3-507d-4d88-aba8-66f204eb59ef]" not in combined
+
+
+@pytest.mark.asyncio
+async def test_run_floating_stream_falls_back_to_final_response_content_when_astream_is_empty():
+    class _NoChunkLLM:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        def bind_tools(self, _tools):
+            return self
+
+        async def ainvoke(self, _messages):
+            self.calls += 1
+            if self.calls == 1:
+                return AIMessage(
+                    content="",
+                    tool_calls=[
+                        {
+                            "id": "call-1",
+                            "name": "list_tasks",
+                            "args": {},
+                        }
+                    ],
+                )
+            return AIMessage(content="No notes found.")
+
+        async def astream(self, _messages):
+            if False:
+                yield None
+
+    with patch("app.core.deep_agent.get_llm", return_value=_NoChunkLLM()), patch(
+        "app.core.deep_agent._all_tools", return_value=[_FakeTool()]
+    ):
+        events = []
+        async for event in run_floating_stream(
+            "user-1",
+            "quali sono le note?",
+            {"scope": {"type": "note"}},
+        ):
+            events.append(event)
+
+    assert events[0][0] == "floating_domain"
+    assert ("token", "No notes found.") in events
+
+
+@pytest.mark.asyncio
+async def test_run_floating_returns_fallback_when_sanitization_would_empty_text():
+    fake_llm = _FakeLLM()
+
+    async def _fake_run_single_agent(**_kwargs):
+        return "<task>[180faff3-507d-4d88-aba8-66f204eb59ef]</task>"
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._run_single_agent", side_effect=_fake_run_single_agent
+    ):
+        text, _domain = await run_floating(
+            "user-1",
+            "quali task ho?",
+            {"scope": {"type": "task"}},
+        )
+
+    assert text == "No results found."
+
+
+@pytest.mark.asyncio
+async def test_run_floating_stream_returns_fallback_when_sanitization_would_empty_text():
+    fake_llm = _FakeLLM()
+
+    async def _fake_stream(**_kwargs):
+        yield "token", "<task>[180faff3-507d-4d88-aba8-66f204eb59ef]</task>"
+
+    with patch("app.core.deep_agent.get_llm", return_value=fake_llm), patch(
+        "app.core.deep_agent._run_single_agent_stream", side_effect=_fake_stream
+    ):
+        events = []
+        async for event in run_floating_stream(
+            "user-1",
+            "quali task ho?",
+            {"scope": {"type": "task"}},
+        ):
+            events.append(event)
+
+    assert ("token", "No results found.") in events

tests/test_memory_middleware.py

@@ -110,6 +110,32 @@ async def test_enrich_context_returns_episodic_memory(db_session, user_with_key)
     assert any("Q1 tasks" in s for s in ctx["episodic_memory"])
 
 
+@pytest.mark.asyncio
+async def test_enrich_context_filters_episodic_by_session_id(db_session, user_with_key):
+    target_session = str(uuid.uuid4())
+    other_session = str(uuid.uuid4())
+    db_session.add(MemoryEpisodic(
+        id=str(uuid.uuid4()),
+        user_id=USER_ID,
+        summary_encrypted=_enc("Target session memory"),
+        session_id=target_session,
+    ))
+    db_session.add(MemoryEpisodic(
+        id=str(uuid.uuid4()),
+        user_id=USER_ID,
+        summary_encrypted=_enc("Other session memory"),
+        session_id=other_session,
+    ))
+    await db_session.commit()
+
+    middleware = MemoryMiddleware(db_session)
+    ctx = await middleware.enrich_context(USER_ID, "any message", session_id=target_session)
+
+    episodic = ctx.get("episodic_memory", [])
+    assert any("Target session" in s for s in episodic)
+    assert not any("Other session" in s for s in episodic)
+
+
 @pytest.mark.asyncio
 async def test_enrich_context_returns_proactive_hints(db_session, user_with_key):
     # Add one pattern above threshold and one below
@@ -229,6 +255,40 @@ async def test_update_core_upsert(db_session, user_with_key):
     assert _dec(rows[0].value_encrypted) == "fr"
 
 
+@pytest.mark.asyncio
+async def test_core_block_edit_ops(db_session, user_with_key):
+    middleware = MemoryMiddleware(db_session)
+
+    await middleware.update_core(USER_ID, "human", "Name: Roberto")
+    await middleware.append_core(USER_ID, "human", "Timezone: Europe/Rome")
+    replaced = await middleware.replace_core(USER_ID, "human", "Roberto", "Robert")
+
+    blocks = await middleware.list_core_blocks(USER_ID)
+    human = next(b for b in blocks if b["label"] == "human")
+
+    assert replaced is True
+    assert "Name: Robert" in human["value"]
+    assert "Timezone: Europe/Rome" in human["value"]
+
+    deleted = await middleware.delete_core(USER_ID, "human")
+    assert deleted is True
+    assert await middleware.get_core_block(USER_ID, "human") is None
+
+
+@pytest.mark.asyncio
+async def test_archival_and_recall_search_helpers(db_session, user_with_key):
+    middleware = MemoryMiddleware(db_session)
+
+    await middleware.insert_archival(USER_ID, "Project whitelist has release risk", source="assistant")
+    await middleware.store_episode(USER_ID, str(uuid.uuid4()), "How is whitelist?", "Whitelist is delayed")
+
+    arch = await middleware.search_archival(USER_ID, "whitelist", top_k=3)
+    rec = await middleware.search_recall(USER_ID, "delayed", top_k=3)
+
+    assert any("whitelist" in item.lower() for item in arch)
+    assert any("delayed" in item.lower() for item in rec)
+
+
 # ── End-to-end WS: memory middleware is called during home_request ────────────
 
 def test_home_request_calls_memory_middleware(client):
@@ -240,21 +300,20 @@ def test_home_request_calls_memory_middleware(client):
         def __init__(self, db):
             pass
 
-        async def enrich_context(self, user_id, message):
+        async def enrich_context(self, user_id, message, **kwargs):
             enrich_calls.append((user_id, message))
             return {"core_memory": {"tz": "UTC"}}
 
-        async def store_episode(self, user_id, session_id, message, response):
+        async def store_episode(self, user_id, session_id, message, response, **kwargs):
             store_calls.append((user_id, session_id, message, response))
 
     token = make_jwt("power", user_id=USER_ID)
     session_id = str(uuid.uuid4())
 
-    async def _mock_stream(user_id, message, context, db_session_factory=None):
+    async def _mock_stream(user_id, message, context):
         # Verify memory context was injected
         assert context.get("core_memory") == {"tz": "UTC"}
-        yield ("token", "Done")
+        yield "token", "Done"
-        yield ("mutations", [])
 
     with (
         patch("app.api.routes.device_ws.MemoryMiddleware", _MockMiddleware),

tests/test_output_formatter.py

@@ -1,214 +1,82 @@
-"""Tests for app.core.output_formatter — HomeFormatter and FloatingFormatter."""
+"""Tests for app.core.output_formatter.StreamFormatter."""
 
 from __future__ import annotations
 
 import pytest
 
-from app.core.output_formatter import HomeFormatter, FloatingFormatter
+from app.core.output_formatter import StreamFormatter
-from app.schemas import (
+from app.schemas import WsFloatingDomain, WsStreamEnd, WsStreamStart, WsStreamText
-    WsFloatingDomain,
-    WsStreamEnd,
-    WsStreamStart,
-    WsStreamText,
-)
 
 
-# ── helpers ───────────────────────────────────────────────────────────────────
-
 async def _stream(*events: tuple[str, object]):
-    """Async generator that yields (event_type, data) tuples."""
     for event in events:
         yield event
 
 
-async def collect(formatter, event_stream):
+async def _collect(formatter: StreamFormatter, event_stream):
     frames = []
     async for frame in formatter.format(event_stream):
         frames.append(frame)
     return frames
 
 
-# ── HomeFormatter ─────────────────────────────────────────────────────────────
-
 @pytest.mark.asyncio
-async def test_home_formatter_plain_text():
+async def test_stream_formatter_text_stream() -> None:
-    req_id = "req-1"
+    formatter = StreamFormatter(request_id="req-1")
-    events = [
+    frames = await _collect(
-        ("token", "Hello world"),
+        formatter,
-        ("mutations", []),
+        _stream(("token", "Hello"), ("token", " world")),
-    ]
+    )
-    formatter = HomeFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(*events))
 
     assert isinstance(frames[0], WsStreamStart)
-    assert frames[0].request_id == req_id
+    assert isinstance(frames[1], WsStreamText)
-    text_frames = [f for f in frames if isinstance(f, WsStreamText)]
+    assert frames[1].chunk == "Hello"
-    assert any("Hello world" in f.chunk for f in text_frames)
+    assert isinstance(frames[2], WsStreamText)
+    assert frames[2].chunk == " world"
     assert isinstance(frames[-1], WsStreamEnd)
 
 
 @pytest.mark.asyncio
-async def test_home_formatter_entity_tags_passed_through():
+async def test_stream_formatter_floating_domain_first() -> None:
-    """Entity tags are streamed as-is — the frontend parses them."""
+    formatter = StreamFormatter(request_id="req-2")
-    req_id = "req-2"
+    frames = await _collect(
-    events = [
+        formatter,
-        ("token", "Here is your project:\n<project>[abc-123]</project>\nAll good."),
+        _stream(
-        ("mutations", []),
+            (
-    ]
+                "floating_domain",
-    formatter = HomeFormatter(request_id=req_id)
+                {"type": "node", "id": "n-1", "section": None},
-    frames = await collect(formatter, _stream(*events))
+            ),
-
-    text = "".join(f.chunk for f in frames if isinstance(f, WsStreamText))
-    assert "<project>[abc-123]</project>" in text
-    assert "Here is your project:" in text
-    assert "All good." in text
-
-
-@pytest.mark.asyncio
-async def test_home_formatter_multiple_tags_passed_through():
-    req_id = "req-3"
-    events = [
-        ("token", "<project>[p1]</project>\nText\n<task>[t1,t2]</task>"),
-        ("mutations", []),
-    ]
-    formatter = HomeFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(*events))
-
-    text = "".join(f.chunk for f in frames if isinstance(f, WsStreamText))
-    assert "<project>[p1]</project>" in text
-    assert "<task>[t1,t2]</task>" in text
-
-
-@pytest.mark.asyncio
-async def test_home_formatter_tool_end_ignored():
-    """tool_end events are silently ignored by HomeFormatter."""
-    req_id = "req-4"
-    events = [
-        ("tool_end", {"name": "task_agent", "result": "3 tasks"}),
-        ("token", "No tags here."),
-        ("mutations", []),
-    ]
-    formatter = HomeFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(*events))
-
-    text = "".join(f.chunk for f in frames if isinstance(f, WsStreamText))
-    assert text == "No tags here."
-
-
-@pytest.mark.asyncio
-async def test_home_formatter_mutations_in_stream_end():
-    req_id = "req-5"
-    muts = [{"action": "insert", "table": "tasks", "data": {"id": "t1"}}]
-    events = [
-        ("token", "Done"),
-        ("mutations", muts),
-    ]
-    formatter = HomeFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(*events))
-
-    end_frame = frames[-1]
-    assert isinstance(end_frame, WsStreamEnd)
-    assert len(end_frame.mutations) == 1
-    assert end_frame.mutations[0]["action"] == "insert"
-
-
-@pytest.mark.asyncio
-async def test_home_formatter_frame_order():
-    """stream_start is first, stream_end is last."""
-    req_id = "req-6"
-    formatter = HomeFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(("token", "Hi"), ("mutations", [])))
-    assert isinstance(frames[0], WsStreamStart)
-    assert isinstance(frames[-1], WsStreamEnd)
-
-
-# ── FloatingFormatter ─────────────────────────────────────────────────────────
-
-@pytest.mark.asyncio
-async def test_floating_formatter_domain_from_tool_end():
-    req_id = "pop-1"
-    formatter = FloatingFormatter(request_id=req_id)
-    events = [
-        ("tool_end", {"name": "task_agent", "result": "ok"}),
-        ("token", "Hello"),
-        ("mutations", []),
-    ]
-    frames = await collect(formatter, _stream(*events))
-
-    assert isinstance(frames[0], WsFloatingDomain)
-    assert frames[0].domain == "tasks"
-    assert frames[0].request_id == req_id
-
-
-@pytest.mark.asyncio
-async def test_floating_formatter_text_only():
-    req_id = "pop-2"
-    formatter = FloatingFormatter(request_id=req_id)
-    events = [
-        ("tool_end", {"name": "timeline_agent", "result": "done"}),
             ("token", "Summary"),
-        ("mutations", []),
+        ),
-    ]
+    )
-    frames = await collect(formatter, _stream(*events))
 
     assert isinstance(frames[0], WsFloatingDomain)
-    assert frames[0].domain == "timelines"
+    assert frames[0].domain.type == "node"
+    assert frames[0].domain.id == "n-1"
+    assert isinstance(frames[1], WsStreamStart)
+    assert isinstance(frames[2], WsStreamText)
+    assert frames[2].chunk == "Summary"
+    assert isinstance(frames[-1], WsStreamEnd)
+
+
+@pytest.mark.asyncio
+async def test_stream_formatter_ignores_unknown_events() -> None:
+    formatter = StreamFormatter(request_id="req-3")
+    frames = await _collect(
+        formatter,
+        _stream(("tool_end", {"name": "x"}), ("token", "ok")),
+    )
+
     text_frames = [f for f in frames if isinstance(f, WsStreamText)]
     assert len(text_frames) == 1
-    assert text_frames[0].chunk == "Summary"
+    assert text_frames[0].chunk == "ok"
 
 
 @pytest.mark.asyncio
-async def test_floating_formatter_no_entity_tags():
+async def test_stream_formatter_empty_stream_still_brackets() -> None:
-    """FloatingFormatter never emits entity tag blocks."""
+    formatter = StreamFormatter(request_id="req-4")
-    req_id = "pop-3"
+    frames = await _collect(formatter, _stream())
-    formatter = FloatingFormatter(request_id=req_id)
-    events = [
-        ("tool_end", {"name": "note_agent", "result": "data"}),
-        ("token", "some text"),
-        ("mutations", []),
-    ]
-    frames = await collect(formatter, _stream(*events))
-    # Only expected frame types
-    for f in frames:
-        assert isinstance(f, (WsFloatingDomain, WsStreamStart, WsStreamText, WsStreamEnd))
 
-
+    assert len(frames) == 2
-@pytest.mark.asyncio
+    assert isinstance(frames[0], WsStreamStart)
-async def test_floating_formatter_end_frame():
+    assert isinstance(frames[1], WsStreamEnd)
-    req_id = "pop-4"
-    formatter = FloatingFormatter(request_id=req_id)
-    events = [
-        ("tool_end", {"name": "project_agent", "result": "ok"}),
-        ("token", "Done"),
-        ("mutations", []),
-    ]
-    frames = await collect(formatter, _stream(*events))
-    assert isinstance(frames[-1], WsStreamEnd)
-
-
-@pytest.mark.asyncio
-async def test_floating_formatter_default_domain_on_early_token():
-    """When the first event is a token (no tool_end yet), default to 'tasks'."""
-    req_id = "pop-5"
-    formatter = FloatingFormatter(request_id=req_id)
-    events = [("token", "hi"), ("mutations", [])]
-    frames = await collect(formatter, _stream(*events))
-    assert isinstance(frames[0], WsFloatingDomain)
-    assert frames[0].domain == "tasks"
-
-
-@pytest.mark.asyncio
-async def test_floating_formatter_mutations_in_stream_end():
-    req_id = "pop-6"
-    muts = [{"action": "update", "table": "tasks", "data": {"id": "t2"}}]
-    events = [
-        ("token", "Updated"),
-        ("mutations", muts),
-    ]
-    formatter = FloatingFormatter(request_id=req_id)
-    frames = await collect(formatter, _stream(*events))
-
-    end_frame = frames[-1]
-    assert isinstance(end_frame, WsStreamEnd)
-    assert len(end_frame.mutations) == 1

tests/test_plugins.py

@@ -88,7 +88,7 @@ class TestPluginRegistry:
     async def test_list_filter_by_query(
         self, reg: PluginRegistry, db_session: AsyncSession, seed_plugins: list[Plugin]
     ) -> None:
-        result = await reg.list_plugins(db_session, query="time tracker")
+        result = await reg.list_plugins(db_session, query="time")
         assert result.total == 1
         assert result.plugins[0].id == "plugin-time-tracker"
 

tests/test_preprocessors.py

@@ -0,0 +1,171 @@
+"""Tests for the preprocessor system (Step 1 — Local Agent V2).
+
+Fixtures are driven by:
+  tests/fixtures/preprocessors/cases.yaml   — test case definitions
+  tests/fixtures/preprocessors/data/        — input files (HTML, txt, ...)
+
+Run:
+    pytest tests/test_preprocessors.py -v
+
+    # Only detection tests
+    pytest tests/test_preprocessors.py -v -k detect
+
+    # Only preprocess tests
+    pytest tests/test_preprocessors.py -v -k preprocess
+
+Langfuse scores are sent when LANGFUSE_SECRET_KEY / LANGFUSE_PUBLIC_KEY are set.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+import yaml
+
+from app.core.langfuse_client import get_langfuse
+from app.core.preprocessors import detect_content_type, preprocess
+
+# ── Paths ──────────────────────────────────────────────────────────────
+
+_FIXTURES_DIR = Path(__file__).parent / "fixtures" / "preprocessors"
+_DATA_DIR = _FIXTURES_DIR / "data"
+_CASES_FILE = _FIXTURES_DIR / "cases.yaml"
+
+# ── Content generators ─────────────────────────────────────────────────
+
+_GENERATORS: dict[str, str] = {
+    # High ratio of non-printable chars → triggers "unknown" heuristic
+    "binary_noise": "some\x00\x01\x02\x03\x04\x05content" * 20,
+}
+
+
+def _load_cases() -> list[dict]:
+    with _CASES_FILE.open(encoding="utf-8") as f:
+        return yaml.safe_load(f)["cases"]
+
+
+def _read_content(case: dict) -> str:
+    if "generate" in case:
+        key = case["generate"]
+        if key not in _GENERATORS:
+            raise ValueError(f"Unknown generator '{key}' in case {case['id']}")
+        return _GENERATORS[key]
+    file_path = _DATA_DIR / case["file"]
+    return file_path.read_text(encoding="utf-8")
+
+
+# ── Langfuse helper ───────────────────────────────────────────────────
+
+def _lf_score(score_name: str, value: float, comment: str = "") -> None:
+    lf = get_langfuse()
+    if lf:
+        trace = lf.trace(name=f"eval-{score_name}")
+        lf.score(
+            trace_id=trace.id,
+            name=score_name,
+            value=value,
+            data_type="NUMERIC",
+            comment=comment,
+        )
+        lf.flush()
+
+
+# ── Assertion engine ──────────────────────────────────────────────────
+
+def _run_assertions(assertions: dict[str, Any], result: Any, raw: str) -> list[str]:
+    """Run all assertions declared in the YAML case. Returns failure messages."""
+    failures: list[str] = []
+
+    if assertions.get("no_html_tags"):
+        if re.search(r"<[^>]+>", result.clean_text):
+            failures.append("clean_text still contains HTML tags")
+
+    min_len = assertions.get("min_length")
+    if min_len is not None:
+        if len(result.clean_text) < min_len:
+            failures.append(
+                f"clean_text too short: {len(result.clean_text)} < {min_len}"
+            )
+
+    ratio_lt = assertions.get("compression_ratio_lt")
+    if ratio_lt is not None and len(raw) > 0:
+        ratio = len(result.clean_text) / len(raw)
+        if ratio >= ratio_lt:
+            failures.append(f"compression ratio {ratio:.2f} >= {ratio_lt}")
+
+    meta_keys = assertions.get("metadata_keys", [])
+    for key in meta_keys:
+        if not result.metadata.get(key):
+            failures.append(f"metadata missing key '{key}' (got {result.metadata})")
+
+    contains = assertions.get("contains")
+    if contains:
+        items = [contains] if isinstance(contains, str) else contains
+        for item in items:
+            if item not in result.clean_text:
+                failures.append(f"clean_text missing expected substring: {item!r}")
+
+    not_contains = assertions.get("not_contains")
+    if not_contains:
+        items = [not_contains] if isinstance(not_contains, str) else not_contains
+        for item in items:
+            if item in result.clean_text:
+                failures.append(f"clean_text contains forbidden substring: {item!r}")
+
+    expected_ct = assertions.get("content_type")
+    if expected_ct and result.content_type != expected_ct:
+        failures.append(
+            f"content_type mismatch: expected {expected_ct!r}, got {result.content_type!r}"
+        )
+
+    return failures
+
+
+# ── Parametrized: detect ──────────────────────────────────────────────
+
+_detect_cases = [c for c in _load_cases() if c["op"] == "detect"]
+
+
+@pytest.mark.parametrize(
+    "case",
+    _detect_cases,
+    ids=[c["id"] for c in _detect_cases],
+)
+def test_detect(case: dict) -> None:
+    raw = _read_content(case)
+    ct = detect_content_type(case["input_filename"], raw)
+
+    expected = case["expected_content_type"]
+    score = 1.0 if ct == expected else 0.0
+    _lf_score(case["score_name"], score, f"got={ct}, expected={expected}")
+
+    assert ct == expected, (
+        f"[{case['id']}] {case['description']}: "
+        f"expected content_type={expected!r}, got {ct!r}"
+    )
+
+
+# ── Parametrized: preprocess ──────────────────────────────────────────
+
+_preprocess_cases = [c for c in _load_cases() if c["op"] == "preprocess"]
+
+
+@pytest.mark.parametrize(
+    "case",
+    _preprocess_cases,
+    ids=[c["id"] for c in _preprocess_cases],
+)
+def test_preprocess(case: dict) -> None:
+    raw = _read_content(case)
+    result = preprocess(case["input_content_type"], raw)
+
+    assertions = case.get("assertions", {})
+    failures = _run_assertions(assertions, result, raw)
+
+    assert not failures, (
+        f"[{case['id']}] {case['description']} — {len(failures)} assertion(s) failed:\n"
+        + "\n".join(f"  • {f}" for f in failures)
+    )

tests/test_schemas_v3.py

@@ -4,6 +4,7 @@ import pytest
 from pydantic import ValidationError
 
 from app.schemas import (
+    WsDomain,
     WsFrameType,
     WsHomeRequest,
     WsFloatingDomain,
@@ -178,23 +179,15 @@ def test_stream_text_deserializes():
 def test_stream_end_defaults():
     frame = WsStreamEnd(request_id="r1")
     assert frame.type == WsFrameType.stream_end
-    assert frame.mutations == []
-
-
-def test_stream_end_with_mutations():
-    mutations = [{"action": "create", "table": "tasks", "data": {"title": "New task"}}]
-    frame = WsStreamEnd(request_id="r1", mutations=mutations)
-    assert len(frame.mutations) == 1
-    assert frame.mutations[0]["action"] == "create"
 
 
 def test_stream_end_serializes():
     data = WsStreamEnd(request_id="r2").model_dump()
-    assert data == {"type": "stream_end", "request_id": "r2", "mutations": []}
+    assert data == {"type": "stream_end", "request_id": "r2"}
 
 
 def test_stream_end_deserializes():
-    raw = {"type": "stream_end", "request_id": "r3", "mutations": []}
+    raw = {"type": "stream_end", "request_id": "r3"}
     frame = WsStreamEnd.model_validate(raw)
     assert frame.request_id == "r3"
 
@@ -203,28 +196,47 @@ def test_stream_end_deserializes():
 
 
 def test_floating_domain_tasks():
-    frame = WsFloatingDomain(request_id="r1", domain="tasks")
+    frame = WsFloatingDomain(request_id="r1", domain=WsDomain(type="task"))
     assert frame.type == WsFrameType.floating_domain
-    assert frame.domain == "tasks"
+    assert frame.domain.type == "task"
 
 
-@pytest.mark.parametrize("domain", ["tasks", "timelines", "notes", "projects"])
+def test_floating_domain_valid_domains():
-def test_floating_domain_valid_domains(domain: str):
+    frame = WsFloatingDomain(
-    frame = WsFloatingDomain(request_id="r1", domain=domain)  # type: ignore[arg-type]
+        request_id="r1",
-    assert frame.domain == domain
+        domain=WsDomain(type="project", id="213213-312321-312312-421321", section="task"),
+    )
+    assert frame.domain.type == "project"
+    assert frame.domain.id == "213213-312321-312312-421321"
+    assert frame.domain.section == "task"
 
 
-def test_floating_domain_invalid():
+def test_floating_domain_object_valid():
-    with pytest.raises(ValidationError):
+    frame = WsFloatingDomain(
-        WsFloatingDomain(request_id="r1", domain="invalid")  # type: ignore[arg-type]
+        request_id="r1",
+        domain=WsDomain(type="project", id="p1", section="task"),
+    )
+    assert frame.domain.type == "project"
 
 
 def test_floating_domain_serializes():
-    d = WsFloatingDomain(request_id="r1", domain="notes").model_dump()
+    d = WsFloatingDomain(
-    assert d == {"type": "floating_domain", "request_id": "r1", "domain": "notes"}
+        request_id="r1",
+        domain=WsDomain(type="timeline"),
+    ).model_dump()
+    assert d == {
+        "type": "floating_domain",
+        "request_id": "r1",
+        "domain": {"type": "timeline", "id": None, "section": None},
+    }
 
 
 def test_floating_domain_deserializes():
-    raw = {"type": "floating_domain", "request_id": "r1", "domain": "projects"}
+    raw = {
+        "type": "floating_domain",
+        "request_id": "r1",
+        "domain": {"type": "node", "id": "n-1", "section": None},
+    }
     frame = WsFloatingDomain.model_validate(raw)
-    assert frame.domain == "projects"
+    assert frame.domain.type == "node"
+    assert frame.domain.id == "n-1"

tests/test_ws_unified.py

@@ -45,15 +45,13 @@ def _recv_until_end(ws, max_frames: int = 20) -> list[dict]:
     return frames
 
 
-async def _mock_home_stream(user_id, message, context, db_session_factory=None):
+async def _mock_home_stream(user_id, message, context):
-    yield "token", "Here are your tasks:\n<task>[t1,t2]</task>"
+    yield "token", "Hello"
-    yield "mutations", []
 
 
-async def _mock_floating_stream(user_id, message, context, scope=None, db_session_factory=None):
+async def _mock_floating_stream(user_id, message, context):
-    yield "tool_end", {"name": "task_agent", "result": "ok"}
+    yield "floating_domain", {"type": "task", "id": None, "section": None}
     yield "token", "Here is a summary"
-    yield "mutations", []
 
 
 # ── tests ─────────────────────────────────────────────────────────────────────
@@ -104,7 +102,7 @@ def test_floating_request_produces_domain_frame(client):
     assert types.index(WsFrameType.floating_domain) < types.index(WsFrameType.stream_end)
 
     domain_frame = next(f for f in frames if f["type"] == WsFrameType.floating_domain)
-    assert domain_frame["domain"] == "tasks"
+    assert domain_frame["domain"]["type"] == "task"
     assert domain_frame["request_id"] == "p1"
 
 
@@ -113,9 +111,8 @@ def test_home_request_request_id_propagated(client):
     token = make_jwt("power", user_id=USER_ID)
     req_id = "my-unique-req-id"
 
-    async def _stream(user_id, message, context, db_session_factory=None):
+    async def _stream(user_id, message, context):
         yield "token", "ok"
-        yield "mutations", []
 
     with patch("app.api.routes.device_ws.run_home_stream", side_effect=_stream):
         with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws: