Add Gmail, Google Drive, and Jira connector route slices.

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/agent.py

@@ -0,0 +1,54 @@
+"""`gmail` route: ``SubAgent`` spec for deepagents."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from deepagents import SubAgent
+from langchain_core.language_models import BaseChatModel
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.md_file_reader import (
+    read_md_file,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+    merge_tools_permissions,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.subagent_builder import (
+    pack_subagent,
+)
+
+from .tools.index import load_tools
+
+NAME = "gmail"
+
+
+def build_subagent(
+    *,
+    dependencies: dict[str, Any],
+    model: BaseChatModel | None = None,
+    extra_middleware: Sequence[Any] | None = None,
+    extra_tools_bucket: ToolsPermissions | None = None,
+) -> SubAgent:
+    buckets = load_tools(dependencies=dependencies)
+    merged_tools_bucket = merge_tools_permissions(buckets, extra_tools_bucket)
+    tools = [
+        row["tool"]
+        for row in (*merged_tools_bucket["allow"], *merged_tools_bucket["ask"])
+        if row.get("tool") is not None
+    ]
+    interrupt_on = {r["name"]: True for r in merged_tools_bucket["ask"] if r.get("name")}
+    description = read_md_file(__package__, "description").strip()
+    if not description:
+        description = "Handles gmail tasks for this workspace."
+    system_prompt = read_md_file(__package__, "system_prompt").strip()
+    return pack_subagent(
+        name=NAME,
+        description=description,
+        system_prompt=system_prompt,
+        tools=tools,
+        interrupt_on=interrupt_on,
+        model=model,
+        extra_middleware=extra_middleware,
+    )

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/description.md

@@ -0,0 +1 @@
+Use for Gmail inbox actions: search/read emails, draft or update replies, send messages, and trash emails.

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/system_prompt.md

@@ -0,0 +1,82 @@
+You are the Gmail operations sub-agent.
+You receive delegated instructions from a supervisor agent and return structured results for supervisor synthesis.
+
+<goal>
+Execute Gmail operations accurately: search/read emails, prepare drafts, send, and trash.
+</goal>
+
+<available_tools>
+- `search_gmail`: find candidate emails with query constraints.
+- `read_gmail_email`: read one message in full detail.
+- `create_gmail_draft`: create a new draft.
+- `update_gmail_draft`: modify an existing draft.
+- `send_gmail_email`: send an email.
+- `trash_gmail_email`: move an email to trash.
+</available_tools>
+
+<tool_policy>
+- Use only tools in `<available_tools>`.
+- Build precise search queries using Gmail operators when possible (`from:`, `to:`, `subject:`, `after:`, `before:`, `has:attachment`, `is:unread`, `label:`).
+- Resolve relative dates against runtime timestamp; prefer narrower interpretation.
+- For reply requests, identify the target thread/email via search + read before drafting.
+- If required fields are missing or target selection is ambiguous, return `status=blocked` with `missing_fields` and disambiguation candidates.
+- Never invent IDs, recipients, timestamps, quoted text, or tool outcomes.
+</tool_policy>
+
+<out_of_scope>
+- Do not perform non-Gmail work.
+- Filing operations not represented in `<available_tools>` (archive/label/mark-read/move-folder) are unsupported here.
+</out_of_scope>
+
+<safety>
+- For send: verify draft `to`, `subject`, and `body` match delegated instructions.
+- If any send-critical field was inferred, do not send; return `status=blocked` with inferred values in `assumptions`.
+- For trash: ensure explicit target match before deletion.
+- If a destructive action appears already completed this session, do not repeat; return prior evidence.
+</safety>
+
+<failure_policy>
+- On tool failure, return `status=error` with concise recovery `next_step`.
+- If search has no strong match, return `status=blocked` with suggested tighter filters.
+- If multiple strong candidates remain for risky actions, return `status=blocked` with top options.
+</failure_policy>
+
+<output_contract>
+Return **only** one JSON object (no markdown/prose):
+{
+  "status": "success" | "partial" | "blocked" | "error",
+  "action_summary": string,
+  "evidence": {
+    "email_id": string | null,
+    "thread_id": string | null,
+    "subject": string | null,
+    "sender": string | null,
+    "recipients": string[] | null,
+    "received_at": string (ISO 8601 with timezone) | null,
+    "sent_message": {
+      "id": string,
+      "to": string[],
+      "subject": string | null,
+      "sent_at": string (ISO 8601 with timezone) | null
+    } | null,
+    "matched_candidates": [
+      {
+        "email_id": string,
+        "subject": string | null,
+        "sender": string | null,
+        "received_at": string (ISO 8601 with timezone) | null
+      }
+    ] | null
+  },
+  "next_step": string | null,
+  "missing_fields": string[] | null,
+  "assumptions": string[] | null
+}
+
+Rules:
+- `status=success` -> `next_step=null`, `missing_fields=null`.
+- `status=partial|blocked|error` -> `next_step` must be non-null.
+- `status=blocked` due to missing required inputs -> `missing_fields` must be non-null.
+- For blocked ambiguity, include options in `evidence.matched_candidates`.
+- For trash actions, `evidence.email_id` is the trashed message.
+</output_contract>

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/__init__.py

@@ -0,0 +1,27 @@
+from app.agents.new_chat.tools.gmail.create_draft import (
+    create_create_gmail_draft_tool,
+)
+from app.agents.new_chat.tools.gmail.read_email import (
+    create_read_gmail_email_tool,
+)
+from app.agents.new_chat.tools.gmail.search_emails import (
+    create_search_gmail_tool,
+)
+from app.agents.new_chat.tools.gmail.send_email import (
+    create_send_gmail_email_tool,
+)
+from app.agents.new_chat.tools.gmail.trash_email import (
+    create_trash_gmail_email_tool,
+)
+from app.agents.new_chat.tools.gmail.update_draft import (
+    create_update_gmail_draft_tool,
+)
+
+__all__ = [
+    "create_create_gmail_draft_tool",
+    "create_read_gmail_email_tool",
+    "create_search_gmail_tool",
+    "create_send_gmail_email_tool",
+    "create_trash_gmail_email_tool",
+    "create_update_gmail_draft_tool",
+]

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/create_draft.py

@@ -0,0 +1,313 @@
+import asyncio
+import base64
+import logging
+from datetime import datetime
+from email.mime.text import MIMEText
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.services.gmail import GmailToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_create_gmail_draft_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def create_gmail_draft(
+        to: str,
+        subject: str,
+        body: str,
+        cc: str | None = None,
+        bcc: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a draft email in Gmail.
+
+        Use when the user asks to draft, compose, or prepare an email without
+        sending it.
+
+        Args:
+            to: Recipient email address.
+            subject: Email subject line.
+            body: Email body content.
+            cc: Optional CC recipient(s), comma-separated.
+            bcc: Optional BCC recipient(s), comma-separated.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", or "error"
+            - draft_id: Gmail draft ID (if success)
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined the action.
+              Respond with a brief acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry the action.
+
+        Examples:
+            - "Draft an email to alice@example.com about the meeting"
+            - "Compose a reply to Bob about the project update"
+        """
+        logger.info(f"create_gmail_draft called: to='{to}', subject='{subject}'")
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Gmail tool not properly configured. Please contact support.",
+            }
+
+        try:
+            metadata_service = GmailToolMetadataService(db_session)
+            context = await metadata_service.get_creation_context(
+                search_space_id, user_id
+            )
+
+            if "error" in context:
+                logger.error(f"Failed to fetch creation context: {context['error']}")
+                return {"status": "error", "message": context["error"]}
+
+            accounts = context.get("accounts", [])
+            if accounts and all(a.get("auth_expired") for a in accounts):
+                logger.warning("All Gmail accounts have expired authentication")
+                return {
+                    "status": "auth_error",
+                    "message": "All connected Gmail accounts need re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "gmail",
+                }
+
+            logger.info(
+                f"Requesting approval for creating Gmail draft: to='{to}', subject='{subject}'"
+            )
+            result = request_approval(
+                action_type="gmail_draft_creation",
+                tool_name="create_gmail_draft",
+                params={
+                    "to": to,
+                    "subject": subject,
+                    "body": body,
+                    "cc": cc,
+                    "bcc": bcc,
+                    "connector_id": None,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The draft was not created. Do not ask again or suggest alternatives.",
+                }
+
+            final_to = result.params.get("to", to)
+            final_subject = result.params.get("subject", subject)
+            final_body = result.params.get("body", body)
+            final_cc = result.params.get("cc", cc)
+            final_bcc = result.params.get("bcc", bcc)
+            final_connector_id = result.params.get("connector_id")
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _gmail_types = [
+                SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+            ]
+
+            if final_connector_id is not None:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.id == final_connector_id,
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_gmail_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "Selected Gmail connector is invalid or has been disconnected.",
+                    }
+                actual_connector_id = connector.id
+            else:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_gmail_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "No Gmail connector found. Please connect Gmail in your workspace settings.",
+                    }
+                actual_connector_id = connector.id
+
+            logger.info(
+                f"Creating Gmail draft: to='{final_to}', subject='{final_subject}', connector={actual_connector_id}"
+            )
+
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    creds = build_composio_credentials(cca_id)
+                else:
+                    return {
+                        "status": "error",
+                        "message": "Composio connected account ID not found for this Gmail connector.",
+                    }
+            else:
+                from google.oauth2.credentials import Credentials
+
+                from app.config import config
+                from app.utils.oauth_security import TokenEncryption
+
+                config_data = dict(connector.config)
+                token_encrypted = config_data.get("_token_encrypted", False)
+                if token_encrypted and config.SECRET_KEY:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+                    if config_data.get("token"):
+                        config_data["token"] = token_encryption.decrypt_token(
+                            config_data["token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+                    if config_data.get("client_secret"):
+                        config_data["client_secret"] = token_encryption.decrypt_token(
+                            config_data["client_secret"]
+                        )
+
+                exp = config_data.get("expiry", "")
+                if exp:
+                    exp = exp.replace("Z", "")
+
+                creds = Credentials(
+                    token=config_data.get("token"),
+                    refresh_token=config_data.get("refresh_token"),
+                    token_uri=config_data.get("token_uri"),
+                    client_id=config_data.get("client_id"),
+                    client_secret=config_data.get("client_secret"),
+                    scopes=config_data.get("scopes", []),
+                    expiry=datetime.fromisoformat(exp) if exp else None,
+                )
+
+            from googleapiclient.discovery import build
+
+            gmail_service = build("gmail", "v1", credentials=creds)
+
+            message = MIMEText(final_body)
+            message["to"] = final_to
+            message["subject"] = final_subject
+            if final_cc:
+                message["cc"] = final_cc
+            if final_bcc:
+                message["bcc"] = final_bcc
+            raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
+
+            try:
+                created = await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    lambda: (
+                        gmail_service.users()
+                        .drafts()
+                        .create(userId="me", body={"message": {"raw": raw}})
+                        .execute()
+                    ),
+                )
+            except Exception as api_err:
+                from googleapiclient.errors import HttpError
+
+                if isinstance(api_err, HttpError) and api_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {actual_connector_id}: {api_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        _res = await db_session.execute(
+                            select(SearchSourceConnector).where(
+                                SearchSourceConnector.id == actual_connector_id
+                            )
+                        )
+                        _conn = _res.scalar_one_or_none()
+                        if _conn and not _conn.config.get("auth_expired"):
+                            _conn.config = {**_conn.config, "auth_expired": True}
+                            flag_modified(_conn, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            actual_connector_id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": actual_connector_id,
+                        "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            logger.info(f"Gmail draft created: id={created.get('id')}")
+
+            kb_message_suffix = ""
+            try:
+                from app.services.gmail import GmailKBSyncService
+
+                kb_service = GmailKBSyncService(db_session)
+                draft_message = created.get("message", {})
+                kb_result = await kb_service.sync_after_create(
+                    message_id=draft_message.get("id", ""),
+                    thread_id=draft_message.get("threadId", ""),
+                    subject=final_subject,
+                    sender="me",
+                    date_str=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                    body_text=final_body,
+                    connector_id=actual_connector_id,
+                    search_space_id=search_space_id,
+                    user_id=user_id,
+                    draft_id=created.get("id"),
+                )
+                if kb_result["status"] == "success":
+                    kb_message_suffix = " Your knowledge base has also been updated."
+                else:
+                    kb_message_suffix = " This draft will be added to your knowledge base in the next scheduled sync."
+            except Exception as kb_err:
+                logger.warning(f"KB sync after create failed: {kb_err}")
+                kb_message_suffix = " This draft will be added to your knowledge base in the next scheduled sync."
+
+            return {
+                "status": "success",
+                "draft_id": created.get("id"),
+                "message": f"Successfully created Gmail draft with subject '{final_subject}'.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error creating Gmail draft: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while creating the draft. Please try again.",
+            }
+
+    return create_gmail_draft

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/index.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+)
+
+from .create_draft import create_create_gmail_draft_tool
+from .read_email import create_read_gmail_email_tool
+from .search_emails import create_search_gmail_tool
+from .send_email import create_send_gmail_email_tool
+from .trash_email import create_trash_gmail_email_tool
+from .update_draft import create_update_gmail_draft_tool
+
+
+def load_tools(*, dependencies: dict[str, Any] | None = None, **kwargs: Any) -> ToolsPermissions:
+    d = {**(dependencies or {}), **kwargs}
+    common = {
+        "db_session": d["db_session"],
+        "search_space_id": d["search_space_id"],
+        "user_id": d["user_id"],
+    }
+    search = create_search_gmail_tool(**common)
+    read = create_read_gmail_email_tool(**common)
+    draft = create_create_gmail_draft_tool(**common)
+    send = create_send_gmail_email_tool(**common)
+    trash = create_trash_gmail_email_tool(**common)
+    updraft = create_update_gmail_draft_tool(**common)
+    return {
+        "allow": [
+            {"name": getattr(search, "name", "") or "", "tool": search},
+            {"name": getattr(read, "name", "") or "", "tool": read},
+        ],
+        "ask": [
+            {"name": getattr(draft, "name", "") or "", "tool": draft},
+            {"name": getattr(send, "name", "") or "", "tool": send},
+            {"name": getattr(trash, "name", "") or "", "tool": trash},
+            {"name": getattr(updraft, "name", "") or "", "tool": updraft},
+        ],
+    }

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/read_email.py

@@ -0,0 +1,100 @@
+import logging
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+
+from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+logger = logging.getLogger(__name__)
+
+_GMAIL_TYPES = [
+    SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+    SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+]
+
+
+def create_read_gmail_email_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def read_gmail_email(message_id: str) -> dict[str, Any]:
+        """Read the full content of a specific Gmail email by its message ID.
+
+        Use after search_gmail to get the complete body of an email.
+
+        Args:
+            message_id: The Gmail message ID (from search_gmail results).
+
+        Returns:
+            Dictionary with status and the full email content formatted as markdown.
+        """
+        if db_session is None or search_space_id is None or user_id is None:
+            return {"status": "error", "message": "Gmail tool not properly configured."}
+
+        try:
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type.in_(_GMAIL_TYPES),
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "No Gmail connector found. Please connect Gmail in your workspace settings.",
+                }
+
+            from app.agents.new_chat.tools.gmail.search_emails import _build_credentials
+
+            creds = _build_credentials(connector)
+
+            from app.connectors.google_gmail_connector import GoogleGmailConnector
+
+            gmail = GoogleGmailConnector(
+                credentials=creds,
+                session=db_session,
+                user_id=user_id,
+                connector_id=connector.id,
+            )
+
+            detail, error = await gmail.get_message_details(message_id)
+            if error:
+                if (
+                    "re-authenticate" in error.lower()
+                    or "authentication failed" in error.lower()
+                ):
+                    return {
+                        "status": "auth_error",
+                        "message": error,
+                        "connector_type": "gmail",
+                    }
+                return {"status": "error", "message": error}
+
+            if not detail:
+                return {
+                    "status": "not_found",
+                    "message": f"Email with ID '{message_id}' not found.",
+                }
+
+            content = gmail.format_message_to_markdown(detail)
+
+            return {"status": "success", "message_id": message_id, "content": content}
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+            logger.error("Error reading Gmail email: %s", e, exc_info=True)
+            return {
+                "status": "error",
+                "message": "Failed to read email. Please try again.",
+            }
+
+    return read_gmail_email

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/search_emails.py

@@ -0,0 +1,182 @@
+import logging
+from datetime import datetime
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+
+from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+logger = logging.getLogger(__name__)
+
+_GMAIL_TYPES = [
+    SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+    SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+]
+
+_token_encryption_cache: object | None = None
+
+
+def _get_token_encryption():
+    global _token_encryption_cache
+    if _token_encryption_cache is None:
+        from app.config import config
+        from app.utils.oauth_security import TokenEncryption
+
+        if not config.SECRET_KEY:
+            raise RuntimeError("SECRET_KEY not configured for token decryption.")
+        _token_encryption_cache = TokenEncryption(config.SECRET_KEY)
+    return _token_encryption_cache
+
+
+def _build_credentials(connector: SearchSourceConnector):
+    """Build Google OAuth Credentials from a connector's stored config.
+
+    Handles both native OAuth connectors (with encrypted tokens) and
+    Composio-backed connectors. Shared by Gmail and Calendar tools.
+    """
+    from app.utils.google_credentials import COMPOSIO_GOOGLE_CONNECTOR_TYPES
+
+    if connector.connector_type in COMPOSIO_GOOGLE_CONNECTOR_TYPES:
+        from app.utils.google_credentials import build_composio_credentials
+
+        cca_id = connector.config.get("composio_connected_account_id")
+        if not cca_id:
+            raise ValueError("Composio connected account ID not found.")
+        return build_composio_credentials(cca_id)
+
+    from google.oauth2.credentials import Credentials
+
+    cfg = dict(connector.config)
+    if cfg.get("_token_encrypted"):
+        enc = _get_token_encryption()
+        for key in ("token", "refresh_token", "client_secret"):
+            if cfg.get(key):
+                cfg[key] = enc.decrypt_token(cfg[key])
+
+    exp = (cfg.get("expiry") or "").replace("Z", "")
+    return Credentials(
+        token=cfg.get("token"),
+        refresh_token=cfg.get("refresh_token"),
+        token_uri=cfg.get("token_uri"),
+        client_id=cfg.get("client_id"),
+        client_secret=cfg.get("client_secret"),
+        scopes=cfg.get("scopes", []),
+        expiry=datetime.fromisoformat(exp) if exp else None,
+    )
+
+
+def create_search_gmail_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def search_gmail(
+        query: str,
+        max_results: int = 10,
+    ) -> dict[str, Any]:
+        """Search emails in the user's Gmail inbox using Gmail search syntax.
+
+        Args:
+            query: Gmail search query, same syntax as the Gmail search bar.
+                Examples: "from:alice@example.com", "subject:meeting",
+                "is:unread", "after:2024/01/01 before:2024/02/01",
+                "has:attachment", "in:sent".
+            max_results: Number of emails to return (default 10, max 20).
+
+        Returns:
+            Dictionary with status and a list of email summaries including
+            message_id, subject, from, date, snippet.
+        """
+        if db_session is None or search_space_id is None or user_id is None:
+            return {"status": "error", "message": "Gmail tool not properly configured."}
+
+        max_results = min(max_results, 20)
+
+        try:
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type.in_(_GMAIL_TYPES),
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "No Gmail connector found. Please connect Gmail in your workspace settings.",
+                }
+
+            creds = _build_credentials(connector)
+
+            from app.connectors.google_gmail_connector import GoogleGmailConnector
+
+            gmail = GoogleGmailConnector(
+                credentials=creds,
+                session=db_session,
+                user_id=user_id,
+                connector_id=connector.id,
+            )
+
+            messages_list, error = await gmail.get_messages_list(
+                max_results=max_results, query=query
+            )
+            if error:
+                if (
+                    "re-authenticate" in error.lower()
+                    or "authentication failed" in error.lower()
+                ):
+                    return {
+                        "status": "auth_error",
+                        "message": error,
+                        "connector_type": "gmail",
+                    }
+                return {"status": "error", "message": error}
+
+            if not messages_list:
+                return {
+                    "status": "success",
+                    "emails": [],
+                    "total": 0,
+                    "message": "No emails found.",
+                }
+
+            emails = []
+            for msg in messages_list:
+                detail, err = await gmail.get_message_details(msg["id"])
+                if err:
+                    continue
+                headers = {
+                    h["name"].lower(): h["value"]
+                    for h in detail.get("payload", {}).get("headers", [])
+                }
+                emails.append(
+                    {
+                        "message_id": detail.get("id"),
+                        "thread_id": detail.get("threadId"),
+                        "subject": headers.get("subject", "No Subject"),
+                        "from": headers.get("from", "Unknown"),
+                        "to": headers.get("to", ""),
+                        "date": headers.get("date", ""),
+                        "snippet": detail.get("snippet", ""),
+                        "labels": detail.get("labelIds", []),
+                    }
+                )
+
+            return {"status": "success", "emails": emails, "total": len(emails)}
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+            logger.error("Error searching Gmail: %s", e, exc_info=True)
+            return {
+                "status": "error",
+                "message": "Failed to search Gmail. Please try again.",
+            }
+
+    return search_gmail

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/send_email.py

@@ -0,0 +1,315 @@
+import asyncio
+import base64
+import logging
+from datetime import datetime
+from email.mime.text import MIMEText
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.services.gmail import GmailToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_send_gmail_email_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def send_gmail_email(
+        to: str,
+        subject: str,
+        body: str,
+        cc: str | None = None,
+        bcc: str | None = None,
+    ) -> dict[str, Any]:
+        """Send an email via Gmail.
+
+        Use when the user explicitly asks to send an email. This sends the
+        email immediately - it cannot be unsent.
+
+        Args:
+            to: Recipient email address.
+            subject: Email subject line.
+            body: Email body content.
+            cc: Optional CC recipient(s), comma-separated.
+            bcc: Optional BCC recipient(s), comma-separated.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", or "error"
+            - message_id: Gmail message ID (if success)
+            - thread_id: Gmail thread ID (if success)
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined the action.
+              Respond with a brief acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry the action.
+
+        Examples:
+            - "Send an email to alice@example.com about the meeting"
+            - "Email Bob the project update"
+        """
+        logger.info(f"send_gmail_email called: to='{to}', subject='{subject}'")
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Gmail tool not properly configured. Please contact support.",
+            }
+
+        try:
+            metadata_service = GmailToolMetadataService(db_session)
+            context = await metadata_service.get_creation_context(
+                search_space_id, user_id
+            )
+
+            if "error" in context:
+                logger.error(f"Failed to fetch creation context: {context['error']}")
+                return {"status": "error", "message": context["error"]}
+
+            accounts = context.get("accounts", [])
+            if accounts and all(a.get("auth_expired") for a in accounts):
+                logger.warning("All Gmail accounts have expired authentication")
+                return {
+                    "status": "auth_error",
+                    "message": "All connected Gmail accounts need re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "gmail",
+                }
+
+            logger.info(
+                f"Requesting approval for sending Gmail email: to='{to}', subject='{subject}'"
+            )
+            result = request_approval(
+                action_type="gmail_email_send",
+                tool_name="send_gmail_email",
+                params={
+                    "to": to,
+                    "subject": subject,
+                    "body": body,
+                    "cc": cc,
+                    "bcc": bcc,
+                    "connector_id": None,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The email was not sent. Do not ask again or suggest alternatives.",
+                }
+
+            final_to = result.params.get("to", to)
+            final_subject = result.params.get("subject", subject)
+            final_body = result.params.get("body", body)
+            final_cc = result.params.get("cc", cc)
+            final_bcc = result.params.get("bcc", bcc)
+            final_connector_id = result.params.get("connector_id")
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _gmail_types = [
+                SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+            ]
+
+            if final_connector_id is not None:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.id == final_connector_id,
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_gmail_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "Selected Gmail connector is invalid or has been disconnected.",
+                    }
+                actual_connector_id = connector.id
+            else:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_gmail_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "No Gmail connector found. Please connect Gmail in your workspace settings.",
+                    }
+                actual_connector_id = connector.id
+
+            logger.info(
+                f"Sending Gmail email: to='{final_to}', subject='{final_subject}', connector={actual_connector_id}"
+            )
+
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    creds = build_composio_credentials(cca_id)
+                else:
+                    return {
+                        "status": "error",
+                        "message": "Composio connected account ID not found for this Gmail connector.",
+                    }
+            else:
+                from google.oauth2.credentials import Credentials
+
+                from app.config import config
+                from app.utils.oauth_security import TokenEncryption
+
+                config_data = dict(connector.config)
+                token_encrypted = config_data.get("_token_encrypted", False)
+                if token_encrypted and config.SECRET_KEY:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+                    if config_data.get("token"):
+                        config_data["token"] = token_encryption.decrypt_token(
+                            config_data["token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+                    if config_data.get("client_secret"):
+                        config_data["client_secret"] = token_encryption.decrypt_token(
+                            config_data["client_secret"]
+                        )
+
+                exp = config_data.get("expiry", "")
+                if exp:
+                    exp = exp.replace("Z", "")
+
+                creds = Credentials(
+                    token=config_data.get("token"),
+                    refresh_token=config_data.get("refresh_token"),
+                    token_uri=config_data.get("token_uri"),
+                    client_id=config_data.get("client_id"),
+                    client_secret=config_data.get("client_secret"),
+                    scopes=config_data.get("scopes", []),
+                    expiry=datetime.fromisoformat(exp) if exp else None,
+                )
+
+            from googleapiclient.discovery import build
+
+            gmail_service = build("gmail", "v1", credentials=creds)
+
+            message = MIMEText(final_body)
+            message["to"] = final_to
+            message["subject"] = final_subject
+            if final_cc:
+                message["cc"] = final_cc
+            if final_bcc:
+                message["bcc"] = final_bcc
+            raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
+
+            try:
+                sent = await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    lambda: (
+                        gmail_service.users()
+                        .messages()
+                        .send(userId="me", body={"raw": raw})
+                        .execute()
+                    ),
+                )
+            except Exception as api_err:
+                from googleapiclient.errors import HttpError
+
+                if isinstance(api_err, HttpError) and api_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {actual_connector_id}: {api_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        _res = await db_session.execute(
+                            select(SearchSourceConnector).where(
+                                SearchSourceConnector.id == actual_connector_id
+                            )
+                        )
+                        _conn = _res.scalar_one_or_none()
+                        if _conn and not _conn.config.get("auth_expired"):
+                            _conn.config = {**_conn.config, "auth_expired": True}
+                            flag_modified(_conn, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            actual_connector_id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": actual_connector_id,
+                        "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            logger.info(
+                f"Gmail email sent: id={sent.get('id')}, threadId={sent.get('threadId')}"
+            )
+
+            kb_message_suffix = ""
+            try:
+                from app.services.gmail import GmailKBSyncService
+
+                kb_service = GmailKBSyncService(db_session)
+                kb_result = await kb_service.sync_after_create(
+                    message_id=sent.get("id", ""),
+                    thread_id=sent.get("threadId", ""),
+                    subject=final_subject,
+                    sender="me",
+                    date_str=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                    body_text=final_body,
+                    connector_id=actual_connector_id,
+                    search_space_id=search_space_id,
+                    user_id=user_id,
+                )
+                if kb_result["status"] == "success":
+                    kb_message_suffix = " Your knowledge base has also been updated."
+                else:
+                    kb_message_suffix = " This email will be added to your knowledge base in the next scheduled sync."
+            except Exception as kb_err:
+                logger.warning(f"KB sync after send failed: {kb_err}")
+                kb_message_suffix = " This email will be added to your knowledge base in the next scheduled sync."
+
+            return {
+                "status": "success",
+                "message_id": sent.get("id"),
+                "thread_id": sent.get("threadId"),
+                "message": f"Successfully sent email to '{final_to}' with subject '{final_subject}'.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error sending Gmail email: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while sending the email. Please try again.",
+            }
+
+    return send_gmail_email

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/trash_email.py

@@ -0,0 +1,309 @@
+import asyncio
+import logging
+from datetime import datetime
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.services.gmail import GmailToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_trash_gmail_email_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def trash_gmail_email(
+        email_subject_or_id: str,
+        delete_from_kb: bool = False,
+    ) -> dict[str, Any]:
+        """Move an email or draft to trash in Gmail.
+
+        Use when the user asks to delete, remove, or trash an email or draft.
+
+        Args:
+            email_subject_or_id: The exact subject line or message ID of the
+                email to trash (as it appears in the inbox).
+            delete_from_kb: Whether to also remove the email from the knowledge base.
+                           Default is False.
+                           Set to True to remove from both Gmail and knowledge base.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", "not_found", or "error"
+            - message_id: Gmail message ID (if success)
+            - deleted_from_kb: whether the document was removed from the knowledge base
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined. Respond with a brief
+              acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "not_found", relay the exact message to the user and ask them
+              to verify the email subject or check if it has been indexed.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry this tool.
+        Examples:
+            - "Delete the email about 'Meeting Cancelled'"
+            - "Trash the email from Bob about the project"
+        """
+        logger.info(
+            f"trash_gmail_email called: email_subject_or_id='{email_subject_or_id}', delete_from_kb={delete_from_kb}"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Gmail tool not properly configured. Please contact support.",
+            }
+
+        try:
+            metadata_service = GmailToolMetadataService(db_session)
+            context = await metadata_service.get_trash_context(
+                search_space_id, user_id, email_subject_or_id
+            )
+
+            if "error" in context:
+                error_msg = context["error"]
+                if "not found" in error_msg.lower():
+                    logger.warning(f"Email not found: {error_msg}")
+                    return {"status": "not_found", "message": error_msg}
+                logger.error(f"Failed to fetch trash context: {error_msg}")
+                return {"status": "error", "message": error_msg}
+
+            account = context.get("account", {})
+            if account.get("auth_expired"):
+                logger.warning(
+                    "Gmail account %s has expired authentication",
+                    account.get("id"),
+                )
+                return {
+                    "status": "auth_error",
+                    "message": "The Gmail account for this email needs re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "gmail",
+                }
+
+            email = context["email"]
+            message_id = email["message_id"]
+            document_id = email.get("document_id")
+            connector_id_from_context = context["account"]["id"]
+
+            if not message_id:
+                return {
+                    "status": "error",
+                    "message": "Message ID is missing from the indexed document. Please re-index the email and try again.",
+                }
+
+            logger.info(
+                f"Requesting approval for trashing Gmail email: '{email_subject_or_id}' (message_id={message_id}, delete_from_kb={delete_from_kb})"
+            )
+            result = request_approval(
+                action_type="gmail_email_trash",
+                tool_name="trash_gmail_email",
+                params={
+                    "message_id": message_id,
+                    "connector_id": connector_id_from_context,
+                    "delete_from_kb": delete_from_kb,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The email was not trashed. Do not ask again or suggest alternatives.",
+                }
+
+            final_message_id = result.params.get("message_id", message_id)
+            final_connector_id = result.params.get(
+                "connector_id", connector_id_from_context
+            )
+            final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
+
+            if not final_connector_id:
+                return {
+                    "status": "error",
+                    "message": "No connector found for this email.",
+                }
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _gmail_types = [
+                SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+            ]
+
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == final_connector_id,
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type.in_(_gmail_types),
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "Selected Gmail connector is invalid or has been disconnected.",
+                }
+
+            logger.info(
+                f"Trashing Gmail email: message_id='{final_message_id}', connector={final_connector_id}"
+            )
+
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    creds = build_composio_credentials(cca_id)
+                else:
+                    return {
+                        "status": "error",
+                        "message": "Composio connected account ID not found for this Gmail connector.",
+                    }
+            else:
+                from google.oauth2.credentials import Credentials
+
+                from app.config import config
+                from app.utils.oauth_security import TokenEncryption
+
+                config_data = dict(connector.config)
+                token_encrypted = config_data.get("_token_encrypted", False)
+                if token_encrypted and config.SECRET_KEY:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+                    if config_data.get("token"):
+                        config_data["token"] = token_encryption.decrypt_token(
+                            config_data["token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+                    if config_data.get("client_secret"):
+                        config_data["client_secret"] = token_encryption.decrypt_token(
+                            config_data["client_secret"]
+                        )
+
+                exp = config_data.get("expiry", "")
+                if exp:
+                    exp = exp.replace("Z", "")
+
+                creds = Credentials(
+                    token=config_data.get("token"),
+                    refresh_token=config_data.get("refresh_token"),
+                    token_uri=config_data.get("token_uri"),
+                    client_id=config_data.get("client_id"),
+                    client_secret=config_data.get("client_secret"),
+                    scopes=config_data.get("scopes", []),
+                    expiry=datetime.fromisoformat(exp) if exp else None,
+                )
+
+            from googleapiclient.discovery import build
+
+            gmail_service = build("gmail", "v1", credentials=creds)
+
+            try:
+                await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    lambda: (
+                        gmail_service.users()
+                        .messages()
+                        .trash(userId="me", id=final_message_id)
+                        .execute()
+                    ),
+                )
+            except Exception as api_err:
+                from googleapiclient.errors import HttpError
+
+                if isinstance(api_err, HttpError) and api_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {connector.id}: {api_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        if not connector.config.get("auth_expired"):
+                            connector.config = {
+                                **connector.config,
+                                "auth_expired": True,
+                            }
+                            flag_modified(connector, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            connector.id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": connector.id,
+                        "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            logger.info(f"Gmail email trashed: message_id={final_message_id}")
+
+            trash_result: dict[str, Any] = {
+                "status": "success",
+                "message_id": final_message_id,
+                "message": f"Successfully moved email '{email.get('subject', email_subject_or_id)}' to trash.",
+            }
+
+            deleted_from_kb = False
+            if final_delete_from_kb and document_id:
+                try:
+                    from app.db import Document
+
+                    doc_result = await db_session.execute(
+                        select(Document).filter(Document.id == document_id)
+                    )
+                    document = doc_result.scalars().first()
+                    if document:
+                        await db_session.delete(document)
+                        await db_session.commit()
+                        deleted_from_kb = True
+                        logger.info(
+                            f"Deleted document {document_id} from knowledge base"
+                        )
+                    else:
+                        logger.warning(f"Document {document_id} not found in KB")
+                except Exception as e:
+                    logger.error(f"Failed to delete document from KB: {e}")
+                    await db_session.rollback()
+                    trash_result["warning"] = (
+                        f"Email trashed, but failed to remove from knowledge base: {e!s}"
+                    )
+
+            trash_result["deleted_from_kb"] = deleted_from_kb
+            if deleted_from_kb:
+                trash_result["message"] = (
+                    f"{trash_result.get('message', '')} (also removed from knowledge base)"
+                )
+
+            return trash_result
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error trashing Gmail email: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while trashing the email. Please try again.",
+            }
+
+    return trash_gmail_email

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/gmail/tools/update_draft.py

@@ -0,0 +1,410 @@
+import asyncio
+import base64
+import logging
+from datetime import datetime
+from email.mime.text import MIMEText
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.services.gmail import GmailToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_update_gmail_draft_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def update_gmail_draft(
+        draft_subject_or_id: str,
+        body: str,
+        to: str | None = None,
+        subject: str | None = None,
+        cc: str | None = None,
+        bcc: str | None = None,
+    ) -> dict[str, Any]:
+        """Update an existing Gmail draft.
+
+        Use when the user asks to modify, edit, or add content to an existing
+        email draft. This replaces the draft content with the new version.
+        The user will be able to review and edit the content before it is applied.
+
+        If the user simply wants to "edit" a draft without specifying exact changes,
+        generate the body yourself using your best understanding of the conversation
+        context. The user will review and can freely edit the content in the approval
+        card before confirming.
+
+        IMPORTANT: This tool is ONLY for modifying Gmail draft content, NOT for
+        deleting/trashing drafts (use trash_gmail_email instead), Notion pages,
+        calendar events, or any other content type.
+
+        Args:
+            draft_subject_or_id: The exact subject line of the draft to update
+                (as it appears in Gmail drafts).
+            body: The full updated body content for the draft. Generate this
+                  yourself based on the user's request and conversation context.
+            to: Optional new recipient email address (keeps original if omitted).
+            subject: Optional new subject line (keeps original if omitted).
+            cc: Optional CC recipient(s), comma-separated.
+            bcc: Optional BCC recipient(s), comma-separated.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", "not_found", or "error"
+            - draft_id: Gmail draft ID (if success)
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined the action.
+              Respond with a brief acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "not_found", relay the exact message to the user and ask them
+              to verify the draft subject or check if it has been indexed.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry the action.
+
+        Examples:
+            - "Update the Kurseong Plan draft with the new itinerary details"
+            - "Edit my draft about the project proposal and change the recipient"
+            - "Let me edit the meeting notes draft" (call with current body content so user can edit in the approval card)
+        """
+        logger.info(
+            f"update_gmail_draft called: draft_subject_or_id='{draft_subject_or_id}'"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Gmail tool not properly configured. Please contact support.",
+            }
+
+        try:
+            metadata_service = GmailToolMetadataService(db_session)
+            context = await metadata_service.get_update_context(
+                search_space_id, user_id, draft_subject_or_id
+            )
+
+            if "error" in context:
+                error_msg = context["error"]
+                if "not found" in error_msg.lower():
+                    logger.warning(f"Draft not found: {error_msg}")
+                    return {"status": "not_found", "message": error_msg}
+                logger.error(f"Failed to fetch update context: {error_msg}")
+                return {"status": "error", "message": error_msg}
+
+            account = context.get("account", {})
+            if account.get("auth_expired"):
+                logger.warning(
+                    "Gmail account %s has expired authentication",
+                    account.get("id"),
+                )
+                return {
+                    "status": "auth_error",
+                    "message": "The Gmail account for this draft needs re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "gmail",
+                }
+
+            email = context["email"]
+            message_id = email["message_id"]
+            document_id = email.get("document_id")
+            connector_id_from_context = account["id"]
+            draft_id_from_context = context.get("draft_id")
+
+            original_subject = email.get("subject", draft_subject_or_id)
+            final_subject_default = subject if subject else original_subject
+            final_to_default = to if to else ""
+
+            logger.info(
+                f"Requesting approval for updating Gmail draft: '{original_subject}' "
+                f"(message_id={message_id}, draft_id={draft_id_from_context})"
+            )
+            result = request_approval(
+                action_type="gmail_draft_update",
+                tool_name="update_gmail_draft",
+                params={
+                    "message_id": message_id,
+                    "draft_id": draft_id_from_context,
+                    "to": final_to_default,
+                    "subject": final_subject_default,
+                    "body": body,
+                    "cc": cc,
+                    "bcc": bcc,
+                    "connector_id": connector_id_from_context,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The draft was not updated. Do not ask again or suggest alternatives.",
+                }
+
+            final_to = result.params.get("to", final_to_default)
+            final_subject = result.params.get("subject", final_subject_default)
+            final_body = result.params.get("body", body)
+            final_cc = result.params.get("cc", cc)
+            final_bcc = result.params.get("bcc", bcc)
+            final_connector_id = result.params.get(
+                "connector_id", connector_id_from_context
+            )
+            final_draft_id = result.params.get("draft_id", draft_id_from_context)
+
+            if not final_connector_id:
+                return {
+                    "status": "error",
+                    "message": "No connector found for this draft.",
+                }
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _gmail_types = [
+                SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
+            ]
+
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == final_connector_id,
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type.in_(_gmail_types),
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "Selected Gmail connector is invalid or has been disconnected.",
+                }
+
+            logger.info(
+                f"Updating Gmail draft: subject='{final_subject}', connector={final_connector_id}"
+            )
+
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    creds = build_composio_credentials(cca_id)
+                else:
+                    return {
+                        "status": "error",
+                        "message": "Composio connected account ID not found for this Gmail connector.",
+                    }
+            else:
+                from google.oauth2.credentials import Credentials
+
+                from app.config import config
+                from app.utils.oauth_security import TokenEncryption
+
+                config_data = dict(connector.config)
+                token_encrypted = config_data.get("_token_encrypted", False)
+                if token_encrypted and config.SECRET_KEY:
+                    token_encryption = TokenEncryption(config.SECRET_KEY)
+                    if config_data.get("token"):
+                        config_data["token"] = token_encryption.decrypt_token(
+                            config_data["token"]
+                        )
+                    if config_data.get("refresh_token"):
+                        config_data["refresh_token"] = token_encryption.decrypt_token(
+                            config_data["refresh_token"]
+                        )
+                    if config_data.get("client_secret"):
+                        config_data["client_secret"] = token_encryption.decrypt_token(
+                            config_data["client_secret"]
+                        )
+
+                exp = config_data.get("expiry", "")
+                if exp:
+                    exp = exp.replace("Z", "")
+
+                creds = Credentials(
+                    token=config_data.get("token"),
+                    refresh_token=config_data.get("refresh_token"),
+                    token_uri=config_data.get("token_uri"),
+                    client_id=config_data.get("client_id"),
+                    client_secret=config_data.get("client_secret"),
+                    scopes=config_data.get("scopes", []),
+                    expiry=datetime.fromisoformat(exp) if exp else None,
+                )
+
+            from googleapiclient.discovery import build
+
+            gmail_service = build("gmail", "v1", credentials=creds)
+
+            # Resolve draft_id if not already available
+            if not final_draft_id:
+                logger.info(
+                    f"draft_id not in metadata, looking up via drafts.list for message_id={message_id}"
+                )
+                final_draft_id = await _find_draft_id_by_message(
+                    gmail_service, message_id
+                )
+
+            if not final_draft_id:
+                return {
+                    "status": "error",
+                    "message": (
+                        "Could not find this draft in Gmail. "
+                        "It may have already been sent or deleted."
+                    ),
+                }
+
+            message = MIMEText(final_body)
+            if final_to:
+                message["to"] = final_to
+            message["subject"] = final_subject
+            if final_cc:
+                message["cc"] = final_cc
+            if final_bcc:
+                message["bcc"] = final_bcc
+            raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
+
+            try:
+                updated = await asyncio.get_event_loop().run_in_executor(
+                    None,
+                    lambda: (
+                        gmail_service.users()
+                        .drafts()
+                        .update(
+                            userId="me",
+                            id=final_draft_id,
+                            body={"message": {"raw": raw}},
+                        )
+                        .execute()
+                    ),
+                )
+            except Exception as api_err:
+                from googleapiclient.errors import HttpError
+
+                if isinstance(api_err, HttpError) and api_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {connector.id}: {api_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        if not connector.config.get("auth_expired"):
+                            connector.config = {
+                                **connector.config,
+                                "auth_expired": True,
+                            }
+                            flag_modified(connector, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            connector.id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": connector.id,
+                        "message": "This Gmail account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                if isinstance(api_err, HttpError) and api_err.resp.status == 404:
+                    return {
+                        "status": "error",
+                        "message": "Draft no longer exists in Gmail. It may have been sent or deleted.",
+                    }
+                raise
+
+            logger.info(f"Gmail draft updated: id={updated.get('id')}")
+
+            kb_message_suffix = ""
+            if document_id:
+                try:
+                    from sqlalchemy.future import select as sa_select
+                    from sqlalchemy.orm.attributes import flag_modified
+
+                    from app.db import Document
+
+                    doc_result = await db_session.execute(
+                        sa_select(Document).filter(Document.id == document_id)
+                    )
+                    document = doc_result.scalars().first()
+                    if document:
+                        document.source_markdown = final_body
+                        document.title = final_subject
+                        meta = dict(document.document_metadata or {})
+                        meta["subject"] = final_subject
+                        meta["draft_id"] = updated.get("id", final_draft_id)
+                        updated_msg = updated.get("message", {})
+                        if updated_msg.get("id"):
+                            meta["message_id"] = updated_msg["id"]
+                        document.document_metadata = meta
+                        flag_modified(document, "document_metadata")
+                        await db_session.commit()
+                        kb_message_suffix = (
+                            " Your knowledge base has also been updated."
+                        )
+                        logger.info(
+                            f"KB document {document_id} updated for draft {final_draft_id}"
+                        )
+                    else:
+                        kb_message_suffix = " This draft will be fully updated in your knowledge base in the next scheduled sync."
+                except Exception as kb_err:
+                    logger.warning(f"KB update after draft edit failed: {kb_err}")
+                    await db_session.rollback()
+                    kb_message_suffix = " This draft will be fully updated in your knowledge base in the next scheduled sync."
+
+            return {
+                "status": "success",
+                "draft_id": updated.get("id"),
+                "message": f"Successfully updated Gmail draft with subject '{final_subject}'.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error updating Gmail draft: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while updating the draft. Please try again.",
+            }
+
+    return update_gmail_draft
+
+
+async def _find_draft_id_by_message(gmail_service: Any, message_id: str) -> str | None:
+    """Look up a draft's ID by its message ID via the Gmail API."""
+    try:
+        page_token = None
+        while True:
+            kwargs: dict[str, Any] = {"userId": "me", "maxResults": 100}
+            if page_token:
+                kwargs["pageToken"] = page_token
+
+            response = await asyncio.get_event_loop().run_in_executor(
+                None,
+                lambda kwargs=kwargs: (
+                    gmail_service.users().drafts().list(**kwargs).execute()
+                ),
+            )
+
+            for draft in response.get("drafts", []):
+                if draft.get("message", {}).get("id") == message_id:
+                    return draft["id"]
+
+            page_token = response.get("nextPageToken")
+            if not page_token:
+                break
+
+        return None
+    except Exception as e:
+        logger.warning(f"Failed to look up draft by message_id: {e}")
+        return None

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/agent.py

@@ -0,0 +1,54 @@
+"""`google_drive` route: ``SubAgent`` spec for deepagents."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from deepagents import SubAgent
+from langchain_core.language_models import BaseChatModel
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.md_file_reader import (
+    read_md_file,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+    merge_tools_permissions,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.subagent_builder import (
+    pack_subagent,
+)
+
+from .tools.index import load_tools
+
+NAME = "google_drive"
+
+
+def build_subagent(
+    *,
+    dependencies: dict[str, Any],
+    model: BaseChatModel | None = None,
+    extra_middleware: Sequence[Any] | None = None,
+    extra_tools_bucket: ToolsPermissions | None = None,
+) -> SubAgent:
+    buckets = load_tools(dependencies=dependencies)
+    merged_tools_bucket = merge_tools_permissions(buckets, extra_tools_bucket)
+    tools = [
+        row["tool"]
+        for row in (*merged_tools_bucket["allow"], *merged_tools_bucket["ask"])
+        if row.get("tool") is not None
+    ]
+    interrupt_on = {r["name"]: True for r in merged_tools_bucket["ask"] if r.get("name")}
+    description = read_md_file(__package__, "description").strip()
+    if not description:
+        description = "Handles google drive tasks for this workspace."
+    system_prompt = read_md_file(__package__, "system_prompt").strip()
+    return pack_subagent(
+        name=NAME,
+        description=description,
+        system_prompt=system_prompt,
+        tools=tools,
+        interrupt_on=interrupt_on,
+        model=model,
+        extra_middleware=extra_middleware,
+    )

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/description.md

@@ -0,0 +1 @@
+Use for Google Drive document/file tasks: locate files, inspect content, and manage Drive files or folders.

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/system_prompt.md

@@ -0,0 +1,54 @@
+You are the Google Drive operations sub-agent.
+You receive delegated instructions from a supervisor agent and return structured results for supervisor synthesis.
+
+<goal>
+Execute Google Drive file operations accurately in the connected account.
+</goal>
+
+<available_tools>
+- `create_google_drive_file`
+- `delete_google_drive_file`
+</available_tools>
+
+<tool_policy>
+- Use only tools in `<available_tools>`.
+- Ensure target file identity/path is explicit before mutate actions.
+- If target is ambiguous, return `status=blocked` with candidate files.
+- Never invent file IDs/names or mutation outcomes.
+</tool_policy>
+
+<out_of_scope>
+- Do not perform non-Google-Drive tasks.
+</out_of_scope>
+
+<safety>
+- Never claim file mutation success without tool confirmation.
+</safety>
+
+<failure_policy>
+- On tool failure, return `status=error` with concise recovery `next_step`.
+- On target ambiguity, return `status=blocked` with candidate files.
+</failure_policy>
+
+<output_contract>
+Return **only** one JSON object (no markdown/prose):
+{
+  "status": "success" | "partial" | "blocked" | "error",
+  "action_summary": string,
+  "evidence": {
+    "file_id": string | null,
+    "file_name": string | null,
+    "operation": "create" | "delete" | null,
+    "matched_candidates": [
+      { "file_id": string, "file_name": string | null }
+    ] | null
+  },
+  "next_step": string | null,
+  "missing_fields": string[] | null,
+  "assumptions": string[] | null
+}
+Rules:
+- `status=success` -> `next_step=null`, `missing_fields=null`.
+- `status=partial|blocked|error` -> `next_step` must be non-null.
+- `status=blocked` due to missing required inputs -> `missing_fields` must be non-null.
+</output_contract>

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/tools/__init__.py

@@ -0,0 +1,11 @@
+from app.agents.new_chat.tools.google_drive.create_file import (
+    create_create_google_drive_file_tool,
+)
+from app.agents.new_chat.tools.google_drive.trash_file import (
+    create_delete_google_drive_file_tool,
+)
+
+__all__ = [
+    "create_create_google_drive_file_tool",
+    "create_delete_google_drive_file_tool",
+]

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/tools/create_file.py

@@ -0,0 +1,283 @@
+import logging
+from typing import Any, Literal
+
+from googleapiclient.errors import HttpError
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.connectors.google_drive.client import GoogleDriveClient
+from app.connectors.google_drive.file_types import GOOGLE_DOC, GOOGLE_SHEET
+from app.services.google_drive import GoogleDriveToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+_MIME_MAP: dict[str, str] = {
+    "google_doc": GOOGLE_DOC,
+    "google_sheet": GOOGLE_SHEET,
+}
+
+
+def create_create_google_drive_file_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def create_google_drive_file(
+        name: str,
+        file_type: Literal["google_doc", "google_sheet"],
+        content: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a new Google Doc or Google Sheet in Google Drive.
+
+        Use this tool when the user explicitly asks to create a new document
+        or spreadsheet in Google Drive. The user MUST specify a topic before
+        you call this tool. If the request does not contain a topic (e.g.
+        "create a drive doc" or "make a Google Sheet"), ask what the file
+        should be about. Never call this tool without a clear topic from the user.
+
+        Args:
+            name: The file name (without extension).
+            file_type: Either "google_doc" or "google_sheet".
+            content: Optional initial content. Generate from the user's topic.
+                     For google_doc, provide markdown text. For google_sheet, provide CSV-formatted text.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", or "error"
+            - file_id: Google Drive file ID (if success)
+            - name: File name (if success)
+            - web_view_link: URL to open the file (if success)
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined the action.
+              Respond with a brief acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry the action.
+
+        Examples:
+            - "Create a Google Doc with today's meeting notes"
+            - "Create a spreadsheet for the 2026 budget"
+        """
+        logger.info(
+            f"create_google_drive_file called: name='{name}', type='{file_type}'"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Google Drive tool not properly configured. Please contact support.",
+            }
+
+        if file_type not in _MIME_MAP:
+            return {
+                "status": "error",
+                "message": f"Unsupported file type '{file_type}'. Use 'google_doc' or 'google_sheet'.",
+            }
+
+        try:
+            metadata_service = GoogleDriveToolMetadataService(db_session)
+            context = await metadata_service.get_creation_context(
+                search_space_id, user_id
+            )
+
+            if "error" in context:
+                logger.error(f"Failed to fetch creation context: {context['error']}")
+                return {"status": "error", "message": context["error"]}
+
+            accounts = context.get("accounts", [])
+            if accounts and all(a.get("auth_expired") for a in accounts):
+                logger.warning("All Google Drive accounts have expired authentication")
+                return {
+                    "status": "auth_error",
+                    "message": "All connected Google Drive accounts need re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "google_drive",
+                }
+
+            logger.info(
+                f"Requesting approval for creating Google Drive file: name='{name}', type='{file_type}'"
+            )
+            result = request_approval(
+                action_type="google_drive_file_creation",
+                tool_name="create_google_drive_file",
+                params={
+                    "name": name,
+                    "file_type": file_type,
+                    "content": content,
+                    "connector_id": None,
+                    "parent_folder_id": None,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The file was not created. Do not ask again or suggest alternatives.",
+                }
+
+            final_name = result.params.get("name", name)
+            final_file_type = result.params.get("file_type", file_type)
+            final_content = result.params.get("content", content)
+            final_connector_id = result.params.get("connector_id")
+            final_parent_folder_id = result.params.get("parent_folder_id")
+
+            if not final_name or not final_name.strip():
+                return {"status": "error", "message": "File name cannot be empty."}
+
+            mime_type = _MIME_MAP.get(final_file_type)
+            if not mime_type:
+                return {
+                    "status": "error",
+                    "message": f"Unsupported file type '{final_file_type}'.",
+                }
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _drive_types = [
+                SearchSourceConnectorType.GOOGLE_DRIVE_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR,
+            ]
+
+            if final_connector_id is not None:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.id == final_connector_id,
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_drive_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "Selected Google Drive connector is invalid or has been disconnected.",
+                    }
+                actual_connector_id = connector.id
+            else:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type.in_(_drive_types),
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "No Google Drive connector found. Please connect Google Drive in your workspace settings.",
+                    }
+                actual_connector_id = connector.id
+
+            logger.info(
+                f"Creating Google Drive file: name='{final_name}', type='{final_file_type}', connector={actual_connector_id}"
+            )
+
+            pre_built_creds = None
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    pre_built_creds = build_composio_credentials(cca_id)
+
+            client = GoogleDriveClient(
+                session=db_session,
+                connector_id=actual_connector_id,
+                credentials=pre_built_creds,
+            )
+            try:
+                created = await client.create_file(
+                    name=final_name,
+                    mime_type=mime_type,
+                    parent_folder_id=final_parent_folder_id,
+                    content=final_content,
+                )
+            except HttpError as http_err:
+                if http_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {actual_connector_id}: {http_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        _res = await db_session.execute(
+                            select(SearchSourceConnector).where(
+                                SearchSourceConnector.id == actual_connector_id
+                            )
+                        )
+                        _conn = _res.scalar_one_or_none()
+                        if _conn and not _conn.config.get("auth_expired"):
+                            _conn.config = {**_conn.config, "auth_expired": True}
+                            flag_modified(_conn, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            actual_connector_id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": actual_connector_id,
+                        "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            logger.info(
+                f"Google Drive file created: id={created.get('id')}, name={created.get('name')}"
+            )
+
+            kb_message_suffix = ""
+            try:
+                from app.services.google_drive import GoogleDriveKBSyncService
+
+                kb_service = GoogleDriveKBSyncService(db_session)
+                kb_result = await kb_service.sync_after_create(
+                    file_id=created.get("id"),
+                    file_name=created.get("name", final_name),
+                    mime_type=mime_type,
+                    web_view_link=created.get("webViewLink"),
+                    content=final_content,
+                    connector_id=actual_connector_id,
+                    search_space_id=search_space_id,
+                    user_id=user_id,
+                )
+                if kb_result["status"] == "success":
+                    kb_message_suffix = " Your knowledge base has also been updated."
+                else:
+                    kb_message_suffix = " This file will be added to your knowledge base in the next scheduled sync."
+            except Exception as kb_err:
+                logger.warning(f"KB sync after create failed: {kb_err}")
+                kb_message_suffix = " This file will be added to your knowledge base in the next scheduled sync."
+
+            return {
+                "status": "success",
+                "file_id": created.get("id"),
+                "name": created.get("name"),
+                "web_view_link": created.get("webViewLink"),
+                "message": f"Successfully created '{created.get('name')}' in Google Drive.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error creating Google Drive file: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while creating the file. Please try again.",
+            }
+
+    return create_google_drive_file

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/tools/index.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+)
+
+from .create_file import create_create_google_drive_file_tool
+from .trash_file import create_delete_google_drive_file_tool
+
+
+def load_tools(*, dependencies: dict[str, Any] | None = None, **kwargs: Any) -> ToolsPermissions:
+    d = {**(dependencies or {}), **kwargs}
+    common = {
+        "db_session": d["db_session"],
+        "search_space_id": d["search_space_id"],
+        "user_id": d["user_id"],
+    }
+    create = create_create_google_drive_file_tool(**common)
+    delete = create_delete_google_drive_file_tool(**common)
+    return {
+        "allow": [],
+        "ask": [
+            {"name": getattr(create, "name", "") or "", "tool": create},
+            {"name": getattr(delete, "name", "") or "", "tool": delete},
+        ],
+    }

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/google_drive/tools/trash_file.py

@@ -0,0 +1,262 @@
+import logging
+from typing import Any
+
+from googleapiclient.errors import HttpError
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.connectors.google_drive.client import GoogleDriveClient
+from app.services.google_drive import GoogleDriveToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_delete_google_drive_file_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+):
+    @tool
+    async def delete_google_drive_file(
+        file_name: str,
+        delete_from_kb: bool = False,
+    ) -> dict[str, Any]:
+        """Move a Google Drive file to trash.
+
+        Use this tool when the user explicitly asks to delete, remove, or trash
+        a file in Google Drive.
+
+        Args:
+            file_name: The exact name of the file to trash (as it appears in Drive).
+            delete_from_kb: Whether to also remove the file from the knowledge base.
+                          Default is False.
+                          Set to True to remove from both Google Drive and knowledge base.
+
+        Returns:
+            Dictionary with:
+            - status: "success", "rejected", "not_found", or "error"
+            - file_id: Google Drive file ID (if success)
+            - deleted_from_kb: whether the document was removed from the knowledge base
+            - message: Result message
+
+            IMPORTANT:
+            - If status is "rejected", the user explicitly declined. Respond with a brief
+              acknowledgment and do NOT retry or suggest alternatives.
+            - If status is "not_found", relay the exact message to the user and ask them
+              to verify the file name or check if it has been indexed.
+            - If status is "insufficient_permissions", the connector lacks the required OAuth scope.
+              Inform the user they need to re-authenticate and do NOT retry this tool.
+        Examples:
+            - "Delete the 'Meeting Notes' file from Google Drive"
+            - "Trash the 'Old Budget' spreadsheet"
+        """
+        logger.info(
+            f"delete_google_drive_file called: file_name='{file_name}', delete_from_kb={delete_from_kb}"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {
+                "status": "error",
+                "message": "Google Drive tool not properly configured. Please contact support.",
+            }
+
+        try:
+            metadata_service = GoogleDriveToolMetadataService(db_session)
+            context = await metadata_service.get_trash_context(
+                search_space_id, user_id, file_name
+            )
+
+            if "error" in context:
+                error_msg = context["error"]
+                if "not found" in error_msg.lower():
+                    logger.warning(f"File not found: {error_msg}")
+                    return {"status": "not_found", "message": error_msg}
+                logger.error(f"Failed to fetch trash context: {error_msg}")
+                return {"status": "error", "message": error_msg}
+
+            account = context.get("account", {})
+            if account.get("auth_expired"):
+                logger.warning(
+                    "Google Drive account %s has expired authentication",
+                    account.get("id"),
+                )
+                return {
+                    "status": "auth_error",
+                    "message": "The Google Drive account for this file needs re-authentication. Please re-authenticate in your connector settings.",
+                    "connector_type": "google_drive",
+                }
+
+            file = context["file"]
+            file_id = file["file_id"]
+            document_id = file.get("document_id")
+            connector_id_from_context = context["account"]["id"]
+
+            if not file_id:
+                return {
+                    "status": "error",
+                    "message": "File ID is missing from the indexed document. Please re-index the file and try again.",
+                }
+
+            logger.info(
+                f"Requesting approval for deleting Google Drive file: '{file_name}' (file_id={file_id}, delete_from_kb={delete_from_kb})"
+            )
+            result = request_approval(
+                action_type="google_drive_file_trash",
+                tool_name="delete_google_drive_file",
+                params={
+                    "file_id": file_id,
+                    "connector_id": connector_id_from_context,
+                    "delete_from_kb": delete_from_kb,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. The file was not trashed. Do not ask again or suggest alternatives.",
+                }
+
+            final_file_id = result.params.get("file_id", file_id)
+            final_connector_id = result.params.get(
+                "connector_id", connector_id_from_context
+            )
+            final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
+
+            if not final_connector_id:
+                return {
+                    "status": "error",
+                    "message": "No connector found for this file.",
+                }
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            _drive_types = [
+                SearchSourceConnectorType.GOOGLE_DRIVE_CONNECTOR,
+                SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR,
+            ]
+
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == final_connector_id,
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type.in_(_drive_types),
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "Selected Google Drive connector is invalid or has been disconnected.",
+                }
+
+            logger.info(
+                f"Deleting Google Drive file: file_id='{final_file_id}', connector={final_connector_id}"
+            )
+
+            pre_built_creds = None
+            if (
+                connector.connector_type
+                == SearchSourceConnectorType.COMPOSIO_GOOGLE_DRIVE_CONNECTOR
+            ):
+                from app.utils.google_credentials import build_composio_credentials
+
+                cca_id = connector.config.get("composio_connected_account_id")
+                if cca_id:
+                    pre_built_creds = build_composio_credentials(cca_id)
+
+            client = GoogleDriveClient(
+                session=db_session,
+                connector_id=connector.id,
+                credentials=pre_built_creds,
+            )
+            try:
+                await client.trash_file(file_id=final_file_id)
+            except HttpError as http_err:
+                if http_err.resp.status == 403:
+                    logger.warning(
+                        f"Insufficient permissions for connector {connector.id}: {http_err}"
+                    )
+                    try:
+                        from sqlalchemy.orm.attributes import flag_modified
+
+                        if not connector.config.get("auth_expired"):
+                            connector.config = {
+                                **connector.config,
+                                "auth_expired": True,
+                            }
+                            flag_modified(connector, "config")
+                            await db_session.commit()
+                    except Exception:
+                        logger.warning(
+                            "Failed to persist auth_expired for connector %s",
+                            connector.id,
+                            exc_info=True,
+                        )
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": connector.id,
+                        "message": "This Google Drive account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            logger.info(
+                f"Google Drive file deleted (moved to trash): file_id={final_file_id}"
+            )
+
+            trash_result: dict[str, Any] = {
+                "status": "success",
+                "file_id": final_file_id,
+                "message": f"Successfully moved '{file['name']}' to trash.",
+            }
+
+            deleted_from_kb = False
+            if final_delete_from_kb and document_id:
+                try:
+                    from app.db import Document
+
+                    doc_result = await db_session.execute(
+                        select(Document).filter(Document.id == document_id)
+                    )
+                    document = doc_result.scalars().first()
+                    if document:
+                        await db_session.delete(document)
+                        await db_session.commit()
+                        deleted_from_kb = True
+                        logger.info(
+                            f"Deleted document {document_id} from knowledge base"
+                        )
+                    else:
+                        logger.warning(f"Document {document_id} not found in KB")
+                except Exception as e:
+                    logger.error(f"Failed to delete document from KB: {e}")
+                    await db_session.rollback()
+                    trash_result["warning"] = (
+                        f"File moved to trash, but failed to remove from knowledge base: {e!s}"
+                    )
+
+            trash_result["deleted_from_kb"] = deleted_from_kb
+            if deleted_from_kb:
+                trash_result["message"] = (
+                    f"{trash_result.get('message', '')} (also removed from knowledge base)"
+                )
+
+            return trash_result
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+
+            logger.error(f"Error deleting Google Drive file: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while trashing the file. Please try again.",
+            }
+
+    return delete_google_drive_file

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/agent.py

@@ -0,0 +1,54 @@
+"""`jira` route: ``SubAgent`` spec for deepagents."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from deepagents import SubAgent
+from langchain_core.language_models import BaseChatModel
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.md_file_reader import (
+    read_md_file,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+    merge_tools_permissions,
+)
+from app.agents.multi_agent_with_deepagents.subagents.shared.subagent_builder import (
+    pack_subagent,
+)
+
+from .tools.index import load_tools
+
+NAME = "jira"
+
+
+def build_subagent(
+    *,
+    dependencies: dict[str, Any],
+    model: BaseChatModel | None = None,
+    extra_middleware: Sequence[Any] | None = None,
+    extra_tools_bucket: ToolsPermissions | None = None,
+) -> SubAgent:
+    buckets = load_tools(dependencies=dependencies)
+    merged_tools_bucket = merge_tools_permissions(buckets, extra_tools_bucket)
+    tools = [
+        row["tool"]
+        for row in (*merged_tools_bucket["allow"], *merged_tools_bucket["ask"])
+        if row.get("tool") is not None
+    ]
+    interrupt_on = {r["name"]: True for r in merged_tools_bucket["ask"] if r.get("name")}
+    description = read_md_file(__package__, "description").strip()
+    if not description:
+        description = "Handles jira tasks for this workspace."
+    system_prompt = read_md_file(__package__, "system_prompt").strip()
+    return pack_subagent(
+        name=NAME,
+        description=description,
+        system_prompt=system_prompt,
+        tools=tools,
+        interrupt_on=interrupt_on,
+        model=model,
+        extra_middleware=extra_middleware,
+    )

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/description.md

@@ -0,0 +1 @@
+Use for Jira issue/project workflows: search issues, inspect fields, update tickets, and move work through workflow states.

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/system_prompt.md

@@ -0,0 +1,46 @@
+You are the Jira MCP operations sub-agent.
+You receive delegated instructions from a supervisor agent and return structured results for supervisor synthesis.
+
+<goal>
+Execute Jira MCP operations accurately, including discovery and issue mutation flows.
+</goal>
+
+<available_tools>
+- Runtime-provided Jira MCP tools for site/project discovery, issue search, create, and update.
+</available_tools>
+
+<tool_policy>
+- Respect discovery dependencies (site/project/issue-type) before mutate calls.
+- If required fields are missing or targets are ambiguous, return `status=blocked` with `missing_fields`.
+- Do not guess keys/IDs.
+- Never claim create/update success without tool confirmation.
+</tool_policy>
+
+<out_of_scope>
+- Do not execute non-Jira tasks.
+</out_of_scope>
+
+<safety>
+- Never perform destructive/mutating actions without explicit target resolution.
+</safety>
+
+<failure_policy>
+- On tool failure, return `status=error` with concise recovery `next_step`.
+- On unresolved ambiguity, return `status=blocked` with candidates or missing fields.
+</failure_policy>
+
+<output_contract>
+Return **only** one JSON object (no markdown/prose):
+{
+  "status": "success" | "partial" | "blocked" | "error",
+  "action_summary": string,
+  "evidence": { "items": object | null },
+  "next_step": string | null,
+  "missing_fields": string[] | null,
+  "assumptions": string[] | null
+}
+Rules:
+- `status=success` -> `next_step=null`, `missing_fields=null`.
+- `status=partial|blocked|error` -> `next_step` must be non-null.
+- `status=blocked` due to missing required inputs -> `missing_fields` must be non-null.
+</output_contract>

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/tools/__init__.py

@@ -0,0 +1,11 @@
+"""Jira tools for creating, updating, and deleting issues."""
+
+from .create_issue import create_create_jira_issue_tool
+from .delete_issue import create_delete_jira_issue_tool
+from .update_issue import create_update_jira_issue_tool
+
+__all__ = [
+    "create_create_jira_issue_tool",
+    "create_delete_jira_issue_tool",
+    "create_update_jira_issue_tool",
+]

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/tools/create_issue.py

@@ -0,0 +1,216 @@
+import asyncio
+import logging
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.connectors.jira_history import JiraHistoryConnector
+from app.services.jira import JiraToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_create_jira_issue_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+    connector_id: int | None = None,
+):
+    @tool
+    async def create_jira_issue(
+        project_key: str,
+        summary: str,
+        issue_type: str = "Task",
+        description: str | None = None,
+        priority: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a new issue in Jira.
+
+        Use this tool when the user explicitly asks to create a new Jira issue/ticket.
+
+        Args:
+            project_key: The Jira project key (e.g. "PROJ", "ENG").
+            summary: Short, descriptive issue title.
+            issue_type: Issue type (default "Task"). Others: "Bug", "Story", "Epic".
+            description: Optional description body for the issue.
+            priority: Optional priority name (e.g. "High", "Medium", "Low").
+
+        Returns:
+            Dictionary with status, issue_key, and message.
+
+            IMPORTANT:
+            - If status is "rejected", the user declined. Do NOT retry.
+            - If status is "insufficient_permissions", inform user to re-authenticate.
+        """
+        logger.info(
+            f"create_jira_issue called: project_key='{project_key}', summary='{summary}'"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {"status": "error", "message": "Jira tool not properly configured."}
+
+        try:
+            metadata_service = JiraToolMetadataService(db_session)
+            context = await metadata_service.get_creation_context(
+                search_space_id, user_id
+            )
+
+            if "error" in context:
+                return {"status": "error", "message": context["error"]}
+
+            accounts = context.get("accounts", [])
+            if accounts and all(a.get("auth_expired") for a in accounts):
+                return {
+                    "status": "auth_error",
+                    "message": "All connected Jira accounts need re-authentication.",
+                    "connector_type": "jira",
+                }
+
+            result = request_approval(
+                action_type="jira_issue_creation",
+                tool_name="create_jira_issue",
+                params={
+                    "project_key": project_key,
+                    "summary": summary,
+                    "issue_type": issue_type,
+                    "description": description,
+                    "priority": priority,
+                    "connector_id": connector_id,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. Do not retry or suggest alternatives.",
+                }
+
+            final_project_key = result.params.get("project_key", project_key)
+            final_summary = result.params.get("summary", summary)
+            final_issue_type = result.params.get("issue_type", issue_type)
+            final_description = result.params.get("description", description)
+            final_priority = result.params.get("priority", priority)
+            final_connector_id = result.params.get("connector_id", connector_id)
+
+            if not final_summary or not final_summary.strip():
+                return {"status": "error", "message": "Issue summary cannot be empty."}
+            if not final_project_key:
+                return {"status": "error", "message": "A project must be selected."}
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            actual_connector_id = final_connector_id
+            if actual_connector_id is None:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type
+                        == SearchSourceConnectorType.JIRA_CONNECTOR,
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {"status": "error", "message": "No Jira connector found."}
+                actual_connector_id = connector.id
+            else:
+                result = await db_session.execute(
+                    select(SearchSourceConnector).filter(
+                        SearchSourceConnector.id == actual_connector_id,
+                        SearchSourceConnector.search_space_id == search_space_id,
+                        SearchSourceConnector.user_id == user_id,
+                        SearchSourceConnector.connector_type
+                        == SearchSourceConnectorType.JIRA_CONNECTOR,
+                    )
+                )
+                connector = result.scalars().first()
+                if not connector:
+                    return {
+                        "status": "error",
+                        "message": "Selected Jira connector is invalid.",
+                    }
+
+            try:
+                jira_history = JiraHistoryConnector(
+                    session=db_session, connector_id=actual_connector_id
+                )
+                jira_client = await jira_history._get_jira_client()
+                api_result = await asyncio.to_thread(
+                    jira_client.create_issue,
+                    project_key=final_project_key,
+                    summary=final_summary,
+                    issue_type=final_issue_type,
+                    description=final_description,
+                    priority=final_priority,
+                )
+            except Exception as api_err:
+                if "status code 403" in str(api_err).lower():
+                    try:
+                        _conn = connector
+                        _conn.config = {**_conn.config, "auth_expired": True}
+                        flag_modified(_conn, "config")
+                        await db_session.commit()
+                    except Exception:
+                        pass
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": actual_connector_id,
+                        "message": "This Jira account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            issue_key = api_result.get("key", "")
+            issue_url = (
+                f"{jira_history._base_url}/browse/{issue_key}"
+                if jira_history._base_url and issue_key
+                else ""
+            )
+
+            kb_message_suffix = ""
+            try:
+                from app.services.jira import JiraKBSyncService
+
+                kb_service = JiraKBSyncService(db_session)
+                kb_result = await kb_service.sync_after_create(
+                    issue_id=issue_key,
+                    issue_identifier=issue_key,
+                    issue_title=final_summary,
+                    description=final_description,
+                    state="To Do",
+                    connector_id=actual_connector_id,
+                    search_space_id=search_space_id,
+                    user_id=user_id,
+                )
+                if kb_result["status"] == "success":
+                    kb_message_suffix = " Your knowledge base has also been updated."
+                else:
+                    kb_message_suffix = " This issue will be added to your knowledge base in the next scheduled sync."
+            except Exception as kb_err:
+                logger.warning(f"KB sync after create failed: {kb_err}")
+                kb_message_suffix = " This issue will be added to your knowledge base in the next scheduled sync."
+
+            return {
+                "status": "success",
+                "issue_key": issue_key,
+                "issue_url": issue_url,
+                "message": f"Jira issue {issue_key} created successfully.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+            logger.error(f"Error creating Jira issue: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while creating the issue.",
+            }
+
+    return create_jira_issue

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/tools/delete_issue.py

@@ -0,0 +1,183 @@
+import asyncio
+import logging
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.connectors.jira_history import JiraHistoryConnector
+from app.services.jira import JiraToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_delete_jira_issue_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+    connector_id: int | None = None,
+):
+    @tool
+    async def delete_jira_issue(
+        issue_title_or_key: str,
+        delete_from_kb: bool = False,
+    ) -> dict[str, Any]:
+        """Delete a Jira issue.
+
+        Use this tool when the user asks to delete or remove a Jira issue.
+
+        Args:
+            issue_title_or_key: The issue key (e.g. "PROJ-42") or title.
+            delete_from_kb: Whether to also remove from the knowledge base.
+
+        Returns:
+            Dictionary with status, message, and deleted_from_kb.
+
+            IMPORTANT:
+            - If status is "rejected", do NOT retry.
+            - If status is "not_found", relay the message to the user.
+            - If status is "insufficient_permissions", inform user to re-authenticate.
+        """
+        logger.info(
+            f"delete_jira_issue called: issue_title_or_key='{issue_title_or_key}'"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {"status": "error", "message": "Jira tool not properly configured."}
+
+        try:
+            metadata_service = JiraToolMetadataService(db_session)
+            context = await metadata_service.get_deletion_context(
+                search_space_id, user_id, issue_title_or_key
+            )
+
+            if "error" in context:
+                error_msg = context["error"]
+                if context.get("auth_expired"):
+                    return {
+                        "status": "auth_error",
+                        "message": error_msg,
+                        "connector_id": context.get("connector_id"),
+                        "connector_type": "jira",
+                    }
+                if "not found" in error_msg.lower():
+                    return {"status": "not_found", "message": error_msg}
+                return {"status": "error", "message": error_msg}
+
+            issue_data = context["issue"]
+            issue_key = issue_data["issue_id"]
+            document_id = issue_data["document_id"]
+            connector_id_from_context = context.get("account", {}).get("id")
+
+            result = request_approval(
+                action_type="jira_issue_deletion",
+                tool_name="delete_jira_issue",
+                params={
+                    "issue_key": issue_key,
+                    "connector_id": connector_id_from_context,
+                    "delete_from_kb": delete_from_kb,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. Do not retry or suggest alternatives.",
+                }
+
+            final_issue_key = result.params.get("issue_key", issue_key)
+            final_connector_id = result.params.get(
+                "connector_id", connector_id_from_context
+            )
+            final_delete_from_kb = result.params.get("delete_from_kb", delete_from_kb)
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            if not final_connector_id:
+                return {
+                    "status": "error",
+                    "message": "No connector found for this issue.",
+                }
+
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == final_connector_id,
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type
+                    == SearchSourceConnectorType.JIRA_CONNECTOR,
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "Selected Jira connector is invalid.",
+                }
+
+            try:
+                jira_history = JiraHistoryConnector(
+                    session=db_session, connector_id=final_connector_id
+                )
+                jira_client = await jira_history._get_jira_client()
+                await asyncio.to_thread(jira_client.delete_issue, final_issue_key)
+            except Exception as api_err:
+                if "status code 403" in str(api_err).lower():
+                    try:
+                        connector.config = {**connector.config, "auth_expired": True}
+                        flag_modified(connector, "config")
+                        await db_session.commit()
+                    except Exception:
+                        pass
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": final_connector_id,
+                        "message": "This Jira account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            deleted_from_kb = False
+            if final_delete_from_kb and document_id:
+                try:
+                    from app.db import Document
+
+                    doc_result = await db_session.execute(
+                        select(Document).filter(Document.id == document_id)
+                    )
+                    document = doc_result.scalars().first()
+                    if document:
+                        await db_session.delete(document)
+                        await db_session.commit()
+                        deleted_from_kb = True
+                except Exception as e:
+                    logger.error(f"Failed to delete document from KB: {e}")
+                    await db_session.rollback()
+
+            message = f"Jira issue {final_issue_key} deleted successfully."
+            if deleted_from_kb:
+                message += " Also removed from the knowledge base."
+
+            return {
+                "status": "success",
+                "issue_key": final_issue_key,
+                "deleted_from_kb": deleted_from_kb,
+                "message": message,
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+            logger.error(f"Error deleting Jira issue: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while deleting the issue.",
+            }
+
+    return delete_jira_issue

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/tools/index.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.agents.multi_agent_with_deepagents.subagents.shared.permissions import (
+    ToolsPermissions,
+)
+
+from .create_issue import create_create_jira_issue_tool
+from .delete_issue import create_delete_jira_issue_tool
+from .update_issue import create_update_jira_issue_tool
+
+
+def load_tools(*, dependencies: dict[str, Any] | None = None, **kwargs: Any) -> ToolsPermissions:
+    d = {**(dependencies or {}), **kwargs}
+    common = {
+        "db_session": d["db_session"],
+        "search_space_id": d["search_space_id"],
+        "user_id": d["user_id"],
+        "connector_id": d.get("connector_id"),
+    }
+    create = create_create_jira_issue_tool(**common)
+    update = create_update_jira_issue_tool(**common)
+    delete = create_delete_jira_issue_tool(**common)
+    return {
+        "allow": [],
+        "ask": [
+            {"name": getattr(create, "name", "") or "", "tool": create},
+            {"name": getattr(update, "name", "") or "", "tool": update},
+            {"name": getattr(delete, "name", "") or "", "tool": delete},
+        ],
+    }

surfsense_backend/app/agents/multi_agent_with_deepagents/subagents/connectors/jira/tools/update_issue.py

@@ -0,0 +1,226 @@
+import asyncio
+import logging
+from typing import Any
+
+from langchain_core.tools import tool
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.agents.new_chat.tools.hitl import request_approval
+from app.connectors.jira_history import JiraHistoryConnector
+from app.services.jira import JiraToolMetadataService
+
+logger = logging.getLogger(__name__)
+
+
+def create_update_jira_issue_tool(
+    db_session: AsyncSession | None = None,
+    search_space_id: int | None = None,
+    user_id: str | None = None,
+    connector_id: int | None = None,
+):
+    @tool
+    async def update_jira_issue(
+        issue_title_or_key: str,
+        new_summary: str | None = None,
+        new_description: str | None = None,
+        new_priority: str | None = None,
+    ) -> dict[str, Any]:
+        """Update an existing Jira issue.
+
+        Use this tool when the user asks to modify, edit, or update a Jira issue.
+
+        Args:
+            issue_title_or_key: The issue key (e.g. "PROJ-42") or title to identify the issue.
+            new_summary: Optional new title/summary for the issue.
+            new_description: Optional new description.
+            new_priority: Optional new priority name.
+
+        Returns:
+            Dictionary with status and message.
+
+            IMPORTANT:
+            - If status is "rejected", do NOT retry.
+            - If status is "not_found", relay the message and ask user to verify.
+            - If status is "insufficient_permissions", inform user to re-authenticate.
+        """
+        logger.info(
+            f"update_jira_issue called: issue_title_or_key='{issue_title_or_key}'"
+        )
+
+        if db_session is None or search_space_id is None or user_id is None:
+            return {"status": "error", "message": "Jira tool not properly configured."}
+
+        try:
+            metadata_service = JiraToolMetadataService(db_session)
+            context = await metadata_service.get_update_context(
+                search_space_id, user_id, issue_title_or_key
+            )
+
+            if "error" in context:
+                error_msg = context["error"]
+                if context.get("auth_expired"):
+                    return {
+                        "status": "auth_error",
+                        "message": error_msg,
+                        "connector_id": context.get("connector_id"),
+                        "connector_type": "jira",
+                    }
+                if "not found" in error_msg.lower():
+                    return {"status": "not_found", "message": error_msg}
+                return {"status": "error", "message": error_msg}
+
+            issue_data = context["issue"]
+            issue_key = issue_data["issue_id"]
+            document_id = issue_data.get("document_id")
+            connector_id_from_context = context.get("account", {}).get("id")
+
+            result = request_approval(
+                action_type="jira_issue_update",
+                tool_name="update_jira_issue",
+                params={
+                    "issue_key": issue_key,
+                    "document_id": document_id,
+                    "new_summary": new_summary,
+                    "new_description": new_description,
+                    "new_priority": new_priority,
+                    "connector_id": connector_id_from_context,
+                },
+                context=context,
+            )
+
+            if result.rejected:
+                return {
+                    "status": "rejected",
+                    "message": "User declined. Do not retry or suggest alternatives.",
+                }
+
+            final_issue_key = result.params.get("issue_key", issue_key)
+            final_summary = result.params.get("new_summary", new_summary)
+            final_description = result.params.get("new_description", new_description)
+            final_priority = result.params.get("new_priority", new_priority)
+            final_connector_id = result.params.get(
+                "connector_id", connector_id_from_context
+            )
+            final_document_id = result.params.get("document_id", document_id)
+
+            from sqlalchemy.future import select
+
+            from app.db import SearchSourceConnector, SearchSourceConnectorType
+
+            if not final_connector_id:
+                return {
+                    "status": "error",
+                    "message": "No connector found for this issue.",
+                }
+
+            result = await db_session.execute(
+                select(SearchSourceConnector).filter(
+                    SearchSourceConnector.id == final_connector_id,
+                    SearchSourceConnector.search_space_id == search_space_id,
+                    SearchSourceConnector.user_id == user_id,
+                    SearchSourceConnector.connector_type
+                    == SearchSourceConnectorType.JIRA_CONNECTOR,
+                )
+            )
+            connector = result.scalars().first()
+            if not connector:
+                return {
+                    "status": "error",
+                    "message": "Selected Jira connector is invalid.",
+                }
+
+            fields: dict[str, Any] = {}
+            if final_summary:
+                fields["summary"] = final_summary
+            if final_description is not None:
+                fields["description"] = {
+                    "type": "doc",
+                    "version": 1,
+                    "content": [
+                        {
+                            "type": "paragraph",
+                            "content": [{"type": "text", "text": final_description}],
+                        }
+                    ],
+                }
+            if final_priority:
+                fields["priority"] = {"name": final_priority}
+
+            if not fields:
+                return {"status": "error", "message": "No changes specified."}
+
+            try:
+                jira_history = JiraHistoryConnector(
+                    session=db_session, connector_id=final_connector_id
+                )
+                jira_client = await jira_history._get_jira_client()
+                await asyncio.to_thread(
+                    jira_client.update_issue, final_issue_key, fields
+                )
+            except Exception as api_err:
+                if "status code 403" in str(api_err).lower():
+                    try:
+                        connector.config = {**connector.config, "auth_expired": True}
+                        flag_modified(connector, "config")
+                        await db_session.commit()
+                    except Exception:
+                        pass
+                    return {
+                        "status": "insufficient_permissions",
+                        "connector_id": final_connector_id,
+                        "message": "This Jira account needs additional permissions. Please re-authenticate in connector settings.",
+                    }
+                raise
+
+            issue_url = (
+                f"{jira_history._base_url}/browse/{final_issue_key}"
+                if jira_history._base_url and final_issue_key
+                else ""
+            )
+
+            kb_message_suffix = ""
+            if final_document_id:
+                try:
+                    from app.services.jira import JiraKBSyncService
+
+                    kb_service = JiraKBSyncService(db_session)
+                    kb_result = await kb_service.sync_after_update(
+                        document_id=final_document_id,
+                        issue_id=final_issue_key,
+                        user_id=user_id,
+                        search_space_id=search_space_id,
+                    )
+                    if kb_result["status"] == "success":
+                        kb_message_suffix = (
+                            " Your knowledge base has also been updated."
+                        )
+                    else:
+                        kb_message_suffix = (
+                            " The knowledge base will be updated in the next sync."
+                        )
+                except Exception as kb_err:
+                    logger.warning(f"KB sync after update failed: {kb_err}")
+                    kb_message_suffix = (
+                        " The knowledge base will be updated in the next sync."
+                    )
+
+            return {
+                "status": "success",
+                "issue_key": final_issue_key,
+                "issue_url": issue_url,
+                "message": f"Jira issue {final_issue_key} updated successfully.{kb_message_suffix}",
+            }
+
+        except Exception as e:
+            from langgraph.errors import GraphInterrupt
+
+            if isinstance(e, GraphInterrupt):
+                raise
+            logger.error(f"Error updating Jira issue: {e}", exc_info=True)
+            return {
+                "status": "error",
+                "message": "Something went wrong while updating the issue.",
+            }
+
+    return update_jira_issue