Add provider-neutral task external-resource actions

src/backend/integrations/google/gmail_service.py

@@ -893,6 +893,86 @@ def get_message(
     return request.execute() or {}
 
 
+def resolve_message_thread_navigation_target(
+    profile_id: str,
+    message_id: str,
+    *,
+    profiles: Optional[Dict[str, GmailProfile]] = None,
+    audit_context: Optional[Mapping[str, object]] = None,
+) -> dict[str, str]:
+    """Read the smallest Gmail state needed to open a message's thread.
+
+    Gmail's REST message identifier is a stable API handle, not a browser
+    permalink.  Resolve its parent thread with a ``minimal`` partial response,
+    then return the authenticated mailbox address needed by a browser caller to
+    select the right Gmail account.  This helper performs no label or message
+    mutation and never returns message content, headers, or credentials.
+    """
+
+    cleaned_message_id = message_id.strip() if isinstance(message_id, str) else ""
+    if not cleaned_message_id:
+        raise ValueError("message_id is required to resolve a Gmail thread link")
+
+    profile = get_profile(profile_id, profiles)
+    _log_gmail_audit(
+        "resolve_message_thread_navigation_target",
+        profile_id=profile.profile_id,
+        audit_context=audit_context,
+        message_id=cleaned_message_id,
+    )
+    service = get_service(profile.profile_id, profiles)
+    message = (
+        service.users()
+        .messages()
+        .get(
+            userId=profile.user_id,
+            id=cleaned_message_id,
+            format="minimal",
+            fields="threadId",
+        )
+        .execute()
+        or {}
+    )
+    thread_id = message.get("threadId") if isinstance(message, Mapping) else None
+    if not isinstance(thread_id, str) or not thread_id.strip():
+        raise RuntimeError("Gmail did not return a thread ID for the message")
+
+    authorised_email: str | None = None
+    if callable(_get_agent_gmail_token_status):
+        try:
+            status = _get_agent_gmail_token_status(profile.profile_id)
+            candidate = getattr(status, "authorised_email", None)
+            if isinstance(candidate, str) and candidate.strip():
+                authorised_email = candidate.strip()
+        except Exception as exc:  # provider identity fallback is bounded
+            logger.debug(
+                "[gmail_service] token-status identity lookup failed for %s: %s",
+                profile.profile_id,
+                type(exc).__name__,
+            )
+
+    if authorised_email is None:
+        profile_payload = (
+            service.users().getProfile(userId=profile.user_id).execute() or {}
+        )
+        candidate = (
+            profile_payload.get("emailAddress")
+            if isinstance(profile_payload, Mapping)
+            else None
+        )
+        if isinstance(candidate, str) and candidate.strip():
+            authorised_email = candidate.strip()
+    if authorised_email is None:
+        raise RuntimeError("Gmail did not return an authorised mailbox address")
+
+    return {
+        "profile_id": profile.profile_id,
+        "message_id": cleaned_message_id,
+        "thread_id": thread_id.strip(),
+        "authorised_email": authorised_email,
+    }
+
+
 def get_attachment(
     profile_id: str,
     message_id: str,

src/backend/integrations/google/gmail_task_external_resource_actions.py

@@ -0,0 +1,136 @@
+"""Navigation-only external-resource actions for Gmail-backed Von tasks.
+
+The provider deliberately knows only the durable Gmail source evidence emitted
+by the mail-ingestion workflow.  It does not interpret task prose, perform
+Gmail mutations, or let a task-supplied profile bypass the normal actor-scoped
+Gmail authority boundary.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+from urllib.parse import quote
+
+from src.backend.services.task_external_resource_action_service import (
+    OPEN_RESOURCE_ACTION_KIND,
+    TaskExternalResourceActionAccessError,
+    TaskExternalResourceActionResolutionError,
+)
+
+GMAIL_OPEN_SOURCE_ACTION_ID = "gmail.open_source"
+_GMAIL_SOURCE_SYSTEM = "gmail"
+_MAX_SOURCE_FIELD_LENGTH = 512
+
+
+def _clean_source_field(value: Any) -> str | None:
+    if not isinstance(value, str):
+        return None
+    cleaned = value.strip()
+    if not cleaned or len(cleaned) > _MAX_SOURCE_FIELD_LENGTH:
+        return None
+    return cleaned
+
+
+def _parse_gmail_source_evidence(task: Mapping[str, Any]) -> dict[str, str] | None:
+    """Read the exact, delimited Gmail source fields from a task's evidence.
+
+    Evidence is a small legacy text field, not a general serialisation format.
+    Requiring the exact key names avoids accidentally promoting prose such as a
+    copied message subject into an external resource identifier.
+    """
+
+    evidence = task.get("evidence")
+    if not isinstance(evidence, str) or not evidence.strip():
+        return None
+
+    fields: dict[str, str] = {}
+    for part in evidence.split(";"):
+        key, separator, value = part.partition("=")
+        key = key.strip()
+        cleaned_value = _clean_source_field(value)
+        if not separator or not key or cleaned_value is None or key in fields:
+            return None
+        fields[key] = cleaned_value
+
+    if fields.get("source_system") != _GMAIL_SOURCE_SYSTEM:
+        return None
+    source_profile = fields.get("source_profile")
+    source_item_id = fields.get("source_item_id")
+    if source_profile is None or source_item_id is None:
+        return None
+    return {
+        "source_system": _GMAIL_SOURCE_SYSTEM,
+        "source_profile": source_profile,
+        "source_item_id": source_item_id,
+    }
+
+
+def list_actions(task: Mapping[str, Any]) -> list[dict[str, str]]:
+    """List the one safe action available for a Gmail-ingestion review task."""
+
+    source = _parse_gmail_source_evidence(task)
+    if source is None:
+        return []
+    return [
+        {
+            "action_id": GMAIL_OPEN_SOURCE_ACTION_ID,
+            "kind": OPEN_RESOURCE_ACTION_KIND,
+            "label": "Open source in Gmail",
+            "source_system": _GMAIL_SOURCE_SYSTEM,
+            "resource_type": "gmail_thread",
+        }
+    ]
+
+
+def resolve_action(task: Mapping[str, Any], action_id: str) -> str | None:
+    """Resolve the advertised Gmail action through trusted profile authority."""
+
+    if action_id != GMAIL_OPEN_SOURCE_ACTION_ID:
+        return None
+    source = _parse_gmail_source_evidence(task)
+    if source is None:
+        return None
+
+    from src.backend.integrations.google import gmail_service
+    from src.backend.services.gmail_profile_invocation_authority_service import (
+        GmailInvocationAuthorityError,
+        authorise_gmail_profile_for_invocation,
+    )
+
+    try:
+        authority = authorise_gmail_profile_for_invocation(source["source_profile"])
+    except GmailInvocationAuthorityError as exc:
+        raise TaskExternalResourceActionAccessError(
+            reason_code=exc.reason_code,
+            safe_message=exc.safe_message,
+        ) from exc
+
+    try:
+        target = gmail_service.resolve_message_thread_navigation_target(
+            profile_id=authority.profile_id,
+            message_id=source["source_item_id"],
+            audit_context={
+                "namespace": authority.audit_namespace,
+                "source": "task_external_resource_action",
+                "action": GMAIL_OPEN_SOURCE_ACTION_ID,
+            },
+        )
+    except Exception as exc:  # safe provider error boundary
+        raise TaskExternalResourceActionResolutionError(
+            reason_code="gmail_source_navigation_unavailable",
+            safe_message="The Gmail source could not be opened right now.",
+        ) from exc
+
+    return (
+        "https://mail.google.com/mail/?authuser="
+        f"{quote(target['authorised_email'], safe='')}"
+        f"#all/{quote(target['thread_id'], safe='')}"
+    )
+
+
+__all__ = [
+    "GMAIL_OPEN_SOURCE_ACTION_ID",
+    "list_actions",
+    "resolve_action",
+]

src/backend/server/routes/task_routes.py

@@ -4,35 +4,42 @@
 Tasks are stored as Vontology concepts.
 """
 
-from datetime import datetime, timezone
 import logging
 import time
+from datetime import datetime, timezone
 from typing import Any
 
-from flask import Blueprint, request, jsonify, session
+from flask import Blueprint, jsonify, redirect, request, session, url_for
 from flask.typing import ResponseReturnValue
 
+from ...services.task_external_resource_action_service import (
+    TaskExternalResourceActionAccessError,
+    TaskExternalResourceActionNotFoundError,
+    TaskExternalResourceActionResolutionError,
+    list_task_external_resource_actions,
+    resolve_task_external_resource_action,
+)
 from ...services.task_management_service import (
+    InvalidTaskDataError,
+    TaskManagementError,
+    TaskNotFoundError,
+    add_task_attachment,
+    add_task_comment,
+    apply_bulk_task_visibility,
+    backfill_jira_migration_bulk_task_collections,
     create_task,
+    delete_task,
     get_task,
+    get_task_history,
     get_task_taxonomy,
-    update_task_fields,
     get_tasks_for_conversation,
+    link_tasks,
+    list_task_attachments,
+    list_task_comments,
     list_tasks_with_visibility,
     search_tasks,
-    apply_bulk_task_visibility,
-    backfill_jira_migration_bulk_task_collections,
-    delete_task,
-    add_task_comment,
-    list_task_comments,
-    add_task_attachment,
-    list_task_attachments,
-    get_task_history,
-    link_tasks,
     unlink_tasks,
-    TaskNotFoundError,
-    InvalidTaskDataError,
-    TaskManagementError,
+    update_task_fields,
 )
 
 logger = logging.getLogger(__name__)
@@ -262,7 +269,19 @@ def get_task_taxonomy_route() -> ResponseReturnValue:
 def get_task_route(task_concept_id: str) -> ResponseReturnValue:
     """Get a task by concept_id."""
     try:
-        result = get_task(task_concept_id)
+        result = dict(get_task(task_concept_id))
+        actions = list_task_external_resource_actions(result)
+        result["external_resource_actions"] = [
+            {
+                **action,
+                "href": url_for(
+                    "tasks.open_task_external_resource_action_route",
+                    task_concept_id=task_concept_id,
+                    action_id=action["action_id"],
+                ),
+            }
+            for action in actions
+        ]
         return jsonify(result), 200
 
     except TaskNotFoundError as e:
@@ -272,6 +291,43 @@ def get_task_route(task_concept_id: str) -> ResponseReturnValue:
         return jsonify({"error": "Internal server error"}), 500
 
 
+@task_bp.route(
+    "/<task_concept_id>/external-resource-actions/<action_id>",
+    methods=["GET"],
+)
+def open_task_external_resource_action_route(
+    task_concept_id: str,
+    action_id: str,
+) -> ResponseReturnValue:
+    """Resolve one provider-neutral task navigation action."""
+
+    try:
+        task = get_task(task_concept_id)
+        target = resolve_task_external_resource_action(task, action_id)
+        return redirect(target, code=302)
+    except TaskNotFoundError as exc:
+        return jsonify({"error": str(exc)}), 404
+    except TaskExternalResourceActionNotFoundError as exc:
+        return jsonify(
+            {"error": exc.safe_message, "reason_code": exc.reason_code}
+        ), 404
+    except TaskExternalResourceActionAccessError as exc:
+        return jsonify(
+            {"error": exc.safe_message, "reason_code": exc.reason_code}
+        ), 403
+    except TaskExternalResourceActionResolutionError as exc:
+        return jsonify(
+            {"error": exc.safe_message, "reason_code": exc.reason_code}
+        ), 502
+    except Exception as exc:  # preserve the existing route error boundary
+        logger.error(
+            "Unexpected error resolving task external-resource action %s: %s",
+            action_id,
+            exc,
+        )
+        return jsonify({"error": "Internal server error"}), 500
+
+
 @task_bp.route("/<task_concept_id>", methods=["PATCH"])
 def update_task_route(task_concept_id: str) -> ResponseReturnValue:
     """Update a task.