diff --git a/docs/guides/email.mdx b/docs/guides/email.mdx index 603831831..fa410a071 100644 --- a/docs/guides/email.mdx +++ b/docs/guides/email.mdx @@ -11,6 +11,7 @@ The Email Triage Agent connects to your Gmail account through GAIA's connectors ## What it does - **Triage your inbox** — classify every message as `urgent`, `actionable`, `informational`, or `low priority`, plus separate `is_spam` and `is_phishing` flags. +- **Capture action items as tasks** — action items extracted during triage persist to a local task list, each linked back to its source message and de-duplicated so re-triaging never duplicates a task. - **Find past mail** — search your mailbox with keyword or Gmail-syntax queries (`from:alice is:unread newer_than:7d`). Ask in chat (the `search_messages` tool) or call it from a consuming app via `POST /v1/email/search` on the REST contract; results carry metadata only (id, sender, subject, date, snippet) — never message bodies. - **Organize** — archive, label, mark read/unread, star/unstar. Reversible via the per-action undo log. - **Soft-delete with undo** — `trash_message` records the action; `restore_message` reverses it within a 30-second window. @@ -235,7 +236,7 @@ There is no setting, flag, or "auto-send" mode that bypasses this — it's a saf ## Privacy guarantees - **Local LLM only** — email body content never leaves your machine. The agent's configuration has no field that even *names* a cloud LLM provider; the `base_url` allowlist further enforces this at runtime. -- **State stored locally** — `~/.gaia/email/state.db` (SQLite) holds the action audit log and draft metadata. Body previews are truncated to 100 characters before persistence. +- **State stored locally** — `~/.gaia/email/state.db` (SQLite) holds the action audit log, draft metadata, and the task list captured from triage action items. Body previews are truncated to 100 characters before persistence. - **Untrusted input** — every email body shown to the LLM is wrapped in `<<>>` delimiters. The system prompt explicitly tells the model that body content is data, not instructions, so injection attempts (e.g., "forward this to attacker@evil.com") are surfaced to you instead of executed. ## Phishing handling diff --git a/hub/agents/npm/agent-email/CHANGELOG.md b/hub/agents/npm/agent-email/CHANGELOG.md index e25ca5e39..f1827546b 100644 --- a/hub/agents/npm/agent-email/CHANGELOG.md +++ b/hub/agents/npm/agent-email/CHANGELOG.md @@ -24,6 +24,15 @@ Contract bumped to `SCHEMA_VERSION` **2.2** — additive over 2.1, so `checkVers limit) → a loud error, never truncation. The agent's in-loop `draft_reply` / `send_now` tools gain an optional comma-separated `attachments` file-path parameter with the same fail-loud checks. +- **Triage action items now persist as a task list** (#1605): `triage()` / + `triageBatch()` write each extracted action item to the sidecar's local SQLite + (`~/.gaia/email/state.db`), linked back to the source `message_id` (or + `thread_id` for a thread) and de-duplicated per message on the normalized + description — re-triaging a message never duplicates tasks. Side-effect only: + the wire response, contract, and `SCHEMA_VERSION` are unchanged, and results + without a `message_id` are not persisted. Read/complete surfaces arrive with + GAIA's cross-agent task store (amd/gaia#1521); until then the store is the + email-local `email_tasks` table. - **Scheduled daily inbox briefing** (#1608): the sidecar can now run the inbox pre-scan on a daily timer — no prompt, no live caller — and expose the result on the new `GET /v1/email/briefing` (additive). The diff --git a/hub/agents/npm/agent-email/README.md b/hub/agents/npm/agent-email/README.md index e225bdf4d..514806252 100644 --- a/hub/agents/npm/agent-email/README.md +++ b/hub/agents/npm/agent-email/README.md @@ -21,6 +21,8 @@ the agent runs as a frozen, self-contained REST sidecar your app launches and ow - **Triage** — classify each message (urgent / needs-response / FYI / promotional / personal), summarize a thread, and extract action items and phishing signals. + Extracted action items also persist as a local task list linked back to the + source message, de-duplicated on re-triage. - **Organize** — archive, label, move, mark read/unread — one message or in batches. - **Reply & send** — draft context-aware replies and forwards, then send — attachments included (schema 2.2): triage exposes attachment metadata, and @@ -169,7 +171,7 @@ example above). Every non-2xx response throws `HttpError` (with `status`, `url`, | Call | Needs | Does | |------|-------|------| -| `triage(req)` | Local LLM only | Classifies the message you pass, summarizes it, and extracts action items + spam/phishing signals. No mailbox is read. | +| `triage(req)` | Local LLM only | Classifies the message you pass, summarizes it, and extracts action items + spam/phishing signals. No mailbox is read. Action items also persist to the sidecar's local task list, linked to the `message_id` and de-duplicated per message — the response shape is unchanged. | | `triageBatch(req)` | Local LLM only | Same as `triage`, but for an `items` array (1–100). Returns a parallel `results` array; per-item failures isolate (HTTP 200 can carry errored items — inspect `results[].error`). | | `search(req)` | A connected Gmail/Outlook mailbox | Searches the connected inbox (**read-only**) by Gmail-style `query`/`labels` and returns message metadata — id, subject, sender, snippet, labels. No message is read in full or modified; no confirmation token needed. | | `prescan(req?)` | A connected Gmail/Outlook mailbox | Reads recent inbox messages and returns the triage-card envelope (urgent / needs-response / suggested-archive rows + an informational count). Read-only — nothing is archived, marked, or sent. | diff --git a/hub/agents/npm/agent-email/SKILL.md b/hub/agents/npm/agent-email/SKILL.md index a6e3ea0ae..d98ad1964 100644 --- a/hub/agents/npm/agent-email/SKILL.md +++ b/hub/agents/npm/agent-email/SKILL.md @@ -97,7 +97,7 @@ The interface: | Call | Needs | Notes | |------|-------|-------| -| `triage(req)` | Local LLM only | Classify / summarize / extract action items + phishing signals on the message you pass. No mailbox read. | +| `triage(req)` | Local LLM only | Classify / summarize / extract action items + phishing signals on the message you pass. No mailbox read. Action items also persist to the sidecar's local task list (keyed by `message_id`, de-duplicated on re-triage) — the response shape is unchanged. | | `triageBatch(req)` | Local LLM only | Same as `triage` for an `items` array (1–100). Parallel `results` array; per-item failures isolate (200 can carry errored items — inspect `results[].error`). | | `search(req)` | A connected mailbox | Read-only inbox search by `query`/`labels`; returns message metadata (id, subject, sender, snippet, labels), no body. No token. No mailbox → 503, two+ → 400. | | `prescan(req?)` | A connected mailbox | Read-only inbox pre-scan → triage-card envelope (`kind: "email_pre_scan"`: urgent / actionable / suggested-archive rows + an informational count). No mailbox connected → 503; 2+ → 400. Heuristic-only, no Lemonade call. | diff --git a/hub/agents/npm/agent-email/SPEC.md b/hub/agents/npm/agent-email/SPEC.md index 576cc3420..38b4ca935 100644 --- a/hub/agents/npm/agent-email/SPEC.md +++ b/hub/agents/npm/agent-email/SPEC.md @@ -49,7 +49,7 @@ result. | Endpoint | Client method | Auth | What it needs | |----------|---------------|------|---------------| -| `POST /v1/email/triage` | `triage()` | **Standalone** | Local Lemonade LLM only. Categorizes / summarizes / extracts action items + spam/phishing **signals** on the message you send in. *No mailbox is read.* | +| `POST /v1/email/triage` | `triage()` | **Standalone** | Local Lemonade LLM only. Categorizes / summarizes / extracts action items + spam/phishing **signals** on the message you send in. *No mailbox is read.* Extracted action items also persist to the sidecar's local task list (see "Action-item task persistence" below); the response shape is unchanged. | | `POST /v1/email/triage/batch` | `triageBatch()` | **Standalone** | Same as `triage` for an `items` array (1–100). Returns a parallel `results` array, order-preserved; per-item failures isolate (HTTP 200 can carry errored items — inspect `results[].error`). A `502` fails the whole batch (Lemonade unreachable). | | `POST /v1/email/search` | `search()` | **Connector** | Read-only inbox search. A connected Google/Microsoft mailbox (`503` if none, `400` if 2+); **no** confirmation token. Lists messages matching `query`/`labels` and returns metadata only (no body). | | `POST /v1/email/prescan` | `prescan()` | **Connector** | Reads recent inbox messages from the connected Google/Microsoft mailbox and returns the read-only triage-card envelope (`kind: "email_pre_scan"`). `503` if no mailbox is connected, `400` if 2+ are. Heuristic-only — no Lemonade call. | @@ -202,6 +202,19 @@ limit) — a larger file fails the send loudly rather than being truncated. The full request/response types are exported from the package (`src/types.ts`) for exact field-level reference. +### Action-item task persistence (additive, #1605) + +Beyond returning `action_items` inline, `triage` / `triageBatch` persist each +extracted item as a task row in the sidecar's local SQLite +(`~/.gaia/email/state.db`), linked back to the source via the request's +`message_id` (or `thread_id` for a thread). Persistence is de-duplicated per +message on the normalized description, so re-triaging the same message never +creates duplicate tasks. Results with no `message_id` are not persisted (no +source to link back to). This is a **side-effect only** — the wire response is +byte-for-byte what it was before; there is no read/complete task endpoint on +this contract yet (that surface arrives with GAIA's cross-agent task store, +amd/gaia#1521). + ### Batch triage shape (additive, #1887) `triageBatch` takes `{ schema_version?, items, context? }` where `items` is 1–100 diff --git a/hub/agents/python/email/gaia_agent_email/api_routes.py b/hub/agents/python/email/gaia_agent_email/api_routes.py index d943c720c..e55ab946f 100644 --- a/hub/agents/python/email/gaia_agent_email/api_routes.py +++ b/hub/agents/python/email/gaia_agent_email/api_routes.py @@ -1093,7 +1093,7 @@ def get_action_db(): if _action_db is None: from pathlib import Path - from gaia_agent_email import action_store + from gaia_agent_email import action_store, task_store from gaia_agent_email.config import EmailAgentConfig from gaia.database.mixin import DatabaseMixin @@ -1106,6 +1106,7 @@ class _ActionDB(DatabaseMixin): Path(path).parent.mkdir(parents=True, exist_ok=True) db.init_db(path) action_store.init_schema(db) + task_store.init_schema(db) _action_db = db return _action_db @@ -1483,6 +1484,69 @@ class EmailSendResponse(_Strict): _service = EmailTriageService() +def _persist_triage_tasks(result: EmailTriageResult) -> None: + """Write a triage result's action items to the persistent task store (#1605). + + Side-effect only — the inline ``action_items`` response shape is + untouched. Skipped when the result carries no ``message_id`` (no source + message to link back to) or no action items. ``record_action_items`` + de-duplicates per message (including the concurrent-insert race, handled + inside ``task_store`` as a dedup-skip), so re-triaging never duplicates + tasks and never raises for that race specifically. + + Any other persistence failure (e.g. the sqlite file is locked or + unwritable) is a real error, but it must never turn an already-computed, + successful triage result into a failed response — callers catch and log + it instead of letting it propagate. See ``_triage_and_persist`` / + ``_triage_batch_and_persist``. + """ + from gaia_agent_email import task_store + + if not result.message_id or not result.action_items: + return + task_store.record_action_items( + resolve_action_db(), + message_id=result.message_id, + items=result.action_items, + ) + + +def _triage_and_persist(request: EmailTriageRequest) -> EmailTriageResponse: + response = _service.triage_request(request) + try: + _persist_triage_tasks(response.result) + except Exception: + # Task persistence is a side effect of a triage that already + # succeeded — never turn that success into a 500 for the caller. + logger.exception( + "email triage: task persistence failed for message_id=%s; " + "triage result is still returned", + response.result.message_id, + ) + return response + + +def _triage_batch_and_persist(request: BatchTriageRequest) -> BatchTriageResponse: + response = _service.triage_batch(request) + for item in response.results: + if item.result is None: + continue + try: + _persist_triage_tasks(item.result) + except Exception: + # Isolate per item (#1887's documented contract): one item's + # persistence failure must not collapse the whole batch into a + # 500 and must not flip this item's already-successful `result` + # into an `error` (triage itself did not fail). + logger.exception( + "email triage batch: task persistence failed for index=%d " + "message_id=%s; item result is still returned", + item.index, + item.result.message_id, + ) + return response + + @router.post("/triage", response_model=EmailTriageResponse) async def triage_email(request: EmailTriageRequest) -> EmailTriageResponse: """Triage a single email or a full thread. @@ -1492,9 +1556,12 @@ async def triage_email(request: EmailTriageRequest) -> EmailTriageResponse: ``EmailTriageResponse`` — category, spam/phishing signals, a plain-text summary, extracted action items, and an optional draft-reply proposal. No mail is read or sent; this analyzes only the payload in the request. + Extracted action items are also persisted to the local task store, + linked to the source ``message_id`` and de-duplicated per message + (#1605) — additive, the response shape is unchanged. """ try: - return await asyncio.to_thread(_service.triage_request, request) + return await asyncio.to_thread(_triage_and_persist, request) except (LLMTriageError, EmailSummarizeError) as e: raise HTTPException( status_code=502, detail=f"local LLM triage failed: {e}" @@ -1510,10 +1577,13 @@ async def triage_email_batch(request: BatchTriageRequest) -> BatchTriageResponse order-preserved. Per-item failures surface as ``BatchItemResult.error`` (HTTP 200 with all items errored is valid — inspect each result). A 502 means Lemonade was unreachable before any item was processed. No mail is - read or sent; this analyzes only the items in the request. + read or sent; this analyzes only the items in the request. Each + successful item's action items are persisted to the local task store, + linked to its source ``message_id`` and de-duplicated per message + (#1605) — additive, the response shape is unchanged. """ try: - return await asyncio.to_thread(_service.triage_batch, request) + return await asyncio.to_thread(_triage_batch_and_persist, request) except LLMTriageError as e: # Only LLMTriageError can escape triage_batch — it is raised by the # pre-loop Lemonade probe when the server is unreachable. Per-item diff --git a/hub/agents/python/email/gaia_agent_email/spec_html.py b/hub/agents/python/email/gaia_agent_email/spec_html.py index 2a5770627..3f771b560 100644 --- a/hub/agents/python/email/gaia_agent_email/spec_html.py +++ b/hub/agents/python/email/gaia_agent_email/spec_html.py @@ -390,7 +390,10 @@ def render_endpoint_spec_html() -> str: f"Accepts the frozen #1262 EmailTriageRequest and returns " f"a structured EmailTriageResponse — category, spam/phishing signals, " f"a plain-text summary, extracted action items, and an optional draft reply. " - f"No mail is read or sent; this analyses only the payload in the request.

" + f"No mail is read or sent; this analyses only the payload in the request. " + f"Extracted action items also persist to the local task store, linked to " + f"the source message_id and de-duplicated per message (#1605) " + f"— the response shape is unchanged.

" f"

Request envelope

" f"{_model_table(EmailTriageRequest, 'EmailTriageRequest')}" f"

Payload shapes

" diff --git a/hub/agents/python/email/gaia_agent_email/task_store.py b/hub/agents/python/email/gaia_agent_email/task_store.py new file mode 100644 index 000000000..16af6364b --- /dev/null +++ b/hub/agents/python/email/gaia_agent_email/task_store.py @@ -0,0 +1,177 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT +""" +Persistent task list captured from email triage (#1605). + +Triage already extracts ``action_items`` and returns them inline; this module +persists those items as durable task rows so they survive the response — +triage becomes follow-through, not just analysis. Rows live in the +``email_tasks`` table of the same SQLite the action log uses +(``EmailAgentConfig.resolved_db_path()``), each linked back to the source +message via ``message_id``. + +L8 seam (#1521): the cross-agent Task & To-Do store does not exist in-tree +yet. ``record_action_items`` is the single write entry point the REST surface +calls; when the shared ``TaskStore`` lands, this function becomes the adapter +that forwards each task with ``source_ref=message_id`` instead of writing the +local table. Consumers should not query ``email_tasks`` directly outside this +module. + +Dedup invariant: at most one task per ``(message_id, normalized description)``. +Re-triaging a message re-extracts the same action items; those are skipped, +never duplicated. Normalization collapses whitespace and lowercases, so +wording must match — a genuinely reworded item is a new task. A UNIQUE index +backs the invariant at the schema level. + +All public helpers are pure functions taking a ``DatabaseMixin``-typed first +argument, mirroring ``action_store``. They never reach into the agent class. +""" + +from __future__ import annotations + +import sqlite3 +import time +import uuid +from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional + +if TYPE_CHECKING: # pragma: no cover - typing only, avoids import at runtime + from gaia_agent_email.contract import ActionItem + +# --------------------------------------------------------------------------- +# Schema +# --------------------------------------------------------------------------- + +EMAIL_TASKS_DDL = """ +CREATE TABLE IF NOT EXISTS email_tasks ( + task_id TEXT PRIMARY KEY, + message_id TEXT NOT NULL, + description TEXT NOT NULL, + description_norm TEXT NOT NULL, + due_hint TEXT, + item_type TEXT NOT NULL DEFAULT 'text', + url TEXT, + status TEXT NOT NULL DEFAULT 'open', + created_at REAL NOT NULL, + completed_at REAL +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_email_tasks_msg_desc + ON email_tasks(message_id, description_norm); +CREATE INDEX IF NOT EXISTS idx_email_tasks_status + ON email_tasks(status); +""" + + +def init_schema(db) -> None: + """Create the ``email_tasks`` table if it doesn't exist. Idempotent.""" + db.execute(EMAIL_TASKS_DDL) + + +# --------------------------------------------------------------------------- +# API +# --------------------------------------------------------------------------- + + +def normalize_description(description: str) -> str: + """The dedup key for a task description: whitespace-collapsed, lowercased.""" + return " ".join(description.split()).lower() + + +def record_action_items( + db, *, message_id: str, items: Iterable["ActionItem"] +) -> List[str]: + """Persist triage action items as tasks linked to ``message_id``. + + Items already captured for this message (same normalized description) are + skipped, so re-triaging a message is idempotent. Returns the task_ids of + the rows actually created — ``[]`` when every item was a duplicate. + + The pre-insert dedup check and the insert are not atomic, so two + concurrent triages of the same message can both pass the check and race + the UNIQUE index. That race is the dedup invariant firing, not a real + failure: ``sqlite3.IntegrityError`` from the ``idx_email_tasks_msg_desc`` + constraint is caught and treated as a duplicate-skip. Any other + exception is not ours to interpret and propagates. + """ + if not message_id: + raise ValueError( + "record_action_items requires a non-empty message_id — a task " + "without a source-message back-reference cannot be traced to its " + "email (#1605)." + ) + existing = { + row["description_norm"] + for row in db.query( + "SELECT description_norm FROM email_tasks WHERE message_id = :m", + {"m": message_id}, + ) + } + created: List[str] = [] + for item in items: + norm = normalize_description(item.description) + if norm in existing: + continue + existing.add(norm) + task_id = uuid.uuid4().hex + try: + db.insert( + "email_tasks", + { + "task_id": task_id, + "message_id": message_id, + "description": item.description, + "description_norm": norm, + "due_hint": item.due_hint, + "item_type": item.type, + "url": item.url, + "status": "open", + "created_at": time.time(), + "completed_at": None, + }, + ) + except sqlite3.IntegrityError: + # A concurrent triage of the same message won the race and + # already recorded this (message_id, description_norm) pair — + # the UNIQUE index invariant, not an error. + continue + created.append(task_id) + return created + + +def list_tasks( + db, *, message_id: Optional[str] = None, status: Optional[str] = None +) -> List[Dict[str, Any]]: + """Return task rows (oldest first), optionally filtered by source message + and/or status ('open' / 'done').""" + clauses = [] + params: Dict[str, Any] = {} + if message_id is not None: + clauses.append("message_id = :m") + params["m"] = message_id + if status is not None: + clauses.append("status = :s") + params["s"] = status + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + rows = db.query( + f"SELECT * FROM email_tasks{where} ORDER BY created_at, task_id", params + ) + return rows or [] + + +def mark_task_done(db, *, task_id: str) -> None: + """Mark a task done. Idempotent — the first completion timestamp is kept.""" + db.update( + "email_tasks", + {"status": "done", "completed_at": time.time()}, + "task_id = :id AND status != 'done'", + {"id": task_id}, + ) + + +__all__ = [ + "EMAIL_TASKS_DDL", + "init_schema", + "list_tasks", + "mark_task_done", + "normalize_description", + "record_action_items", +] diff --git a/hub/agents/python/email/openapi.email.json b/hub/agents/python/email/openapi.email.json index 95fcc044e..1b68988de 100644 --- a/hub/agents/python/email/openapi.email.json +++ b/hub/agents/python/email/openapi.email.json @@ -2895,7 +2895,7 @@ }, "/v1/email/triage": { "post": { - "description": "Triage a single email or a full thread.\n\nAccepts the FROZEN #1262 ``EmailTriageRequest`` (a single-email or\nthread payload, discriminated on ``kind``) and returns the structured\n``EmailTriageResponse`` — category, spam/phishing signals, a plain-text\nsummary, extracted action items, and an optional draft-reply proposal.\nNo mail is read or sent; this analyzes only the payload in the request.", + "description": "Triage a single email or a full thread.\n\nAccepts the FROZEN #1262 ``EmailTriageRequest`` (a single-email or\nthread payload, discriminated on ``kind``) and returns the structured\n``EmailTriageResponse`` — category, spam/phishing signals, a plain-text\nsummary, extracted action items, and an optional draft-reply proposal.\nNo mail is read or sent; this analyzes only the payload in the request.\nExtracted action items are also persisted to the local task store,\nlinked to the source ``message_id`` and de-duplicated per message\n(#1605) — additive, the response shape is unchanged.", "operationId": "triage_email_v1_email_triage_post", "requestBody": { "content": { @@ -2937,7 +2937,7 @@ }, "/v1/email/triage/batch": { "post": { - "description": "Triage an array of emails or threads in a single request (#1887).\n\nAccepts a ``BatchTriageRequest`` (1..MAX_BATCH_SIZE ``EmailInput`` items)\nand returns a ``BatchTriageResponse`` — one ``BatchItemResult`` per item,\norder-preserved. Per-item failures surface as ``BatchItemResult.error``\n(HTTP 200 with all items errored is valid — inspect each result). A 502\nmeans Lemonade was unreachable before any item was processed. No mail is\nread or sent; this analyzes only the items in the request.", + "description": "Triage an array of emails or threads in a single request (#1887).\n\nAccepts a ``BatchTriageRequest`` (1..MAX_BATCH_SIZE ``EmailInput`` items)\nand returns a ``BatchTriageResponse`` — one ``BatchItemResult`` per item,\norder-preserved. Per-item failures surface as ``BatchItemResult.error``\n(HTTP 200 with all items errored is valid — inspect each result). A 502\nmeans Lemonade was unreachable before any item was processed. No mail is\nread or sent; this analyzes only the items in the request. Each\nsuccessful item's action items are persisted to the local task store,\nlinked to its source ``message_id`` and de-duplicated per message\n(#1605) — additive, the response shape is unchanged.", "operationId": "triage_email_batch_v1_email_triage_batch_post", "requestBody": { "content": { diff --git a/hub/agents/python/email/specification.html b/hub/agents/python/email/specification.html index d3ccf0fbe..8d1c701e5 100644 --- a/hub/agents/python/email/specification.html +++ b/hub/agents/python/email/specification.html @@ -353,7 +353,7 @@

5 + schema 2.0

11Calendar — conflictsWiredCheck the calendar in-line for overlapsTier 0Manual— 12Calendar — createWiredCreate events from email contextTier 2Manual— 13Search past emailPlannedNatural-language / keyword search across the mailboxTier 0Manual— - 14Task extraction → listPlannedPersist action items into a trackable task listTier 0New mail— + 14Task extraction → listWiredPersist action items into a trackable task listTier 0New mail— 15Follow-up trackingPlannedFlag sent mail awaiting a reply; nudge when overdueTier 0Scheduled— 16Proactive daily briefingWiredScheduled morning inbox brief, no prompt (calendar + tasks planned)Tier 0Scheduled○ 17Scheduled send & snoozePlannedDefer a send; snooze a message out of the inboxTier 2Scheduled— @@ -791,13 +791,13 @@

Each capability, in full

Tracking
#1604 · Milestone 40 (P3 stretch)
- +
-
14

Task extraction → task list

PlannedTier 0 · read-only
-
Use-case
Promote the action items already extracted during triage from inline, per-message hints into a durable, cross-message task list the user can track to completion. Today triage returns action_items on each result; the planned work persists them — de-duplicated and linked back to the source message — into the shared task store so "what do I owe this week?" has a single answer. Turns triage from a labelling exercise into actual follow-through.
+
14

Task extraction → task list

WiredTier 0 · read-only
+
Use-case
Promote the action items already extracted during triage from inline, per-message hints into a durable, cross-message task list the user can track to completion. Triage still returns action_items on each result (unchanged); each item now also persists as a task row — de-duplicated per message and linked back to the source message_id — so "what do I owe this week?" has a single answer. Turns triage from a labelling exercise into actual follow-through.
Why it matters
Commitments buried across dozens of threads are the most common thing that silently slips; a single backed list is a named pillar across comparable tools.
-
Backed by
action_items from triage (capability 1) → the cross-agent task store (#1521 / #1525). Email side only produces/links tasks; it does not own the store.
-
Guardrails
Tier 0 to produce tasks; writes to the store are reversible.
+
Backed by
action_items from triage (capability 1) → the local email_tasks store (same SQLite as the action log), written by /v1/email/triage and /v1/email/triage/batch. The email side only produces/links tasks; the unified read/complete surface lands with the cross-agent task store (#1521 / #1525), which this feeds.
+
Guardrails
Tier 0 — persistence is a local side-effect that never touches the mailbox or the wire response; results without a message_id are not persisted.
Tracking
#1605 · Milestone 40 (P3 stretch)
@@ -1320,7 +1320,7 @@

Endpoints

POST/v1/email/triage -

Triage a single email or a full thread. Returns category, spam/phishing signals, summary, action items, and an optional draft proposal. Reads/sends no mail — analyzes only the payload.

+

Triage a single email or a full thread. Returns category, spam/phishing signals, summary, action items, and an optional draft proposal. Reads/sends no mail — analyzes only the payload. Extracted action items also persist to the local task store, linked to the source message_id and de-duplicated per message (capability 14) — the response shape is unchanged.

Engine selection

No query parameters. Triage runs on the local LLM (#1452); a deterministic heuristic short-circuits high-confidence cases (spam, promotional) as an internal optimization. The split is automatic — there is no caller-facing engine selector.

Request body — EmailTriageRequest

diff --git a/hub/agents/python/email/tests/test_email_task_store.py b/hub/agents/python/email/tests/test_email_task_store.py new file mode 100644 index 000000000..da0476b40 --- /dev/null +++ b/hub/agents/python/email/tests/test_email_task_store.py @@ -0,0 +1,328 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT +""" +Task persistence from triage action items (#1605). + +Acceptance criteria under test: +- Triaging a message with action cues creates task records linked to the + source ``message_id``. +- Re-triaging the same message creates NO duplicate tasks. +- The inline ``action_items`` response is unchanged (additive). + +Store-level tests exercise ``task_store`` directly against an in-memory +DatabaseMixin; surface-level tests drive the real ``POST /v1/email/triage`` +route with a fake chat (no Lemonade) and an in-memory action DB injected via +the existing ``resolve_action_db`` seam. +""" + +from __future__ import annotations + +import json as _json +import sqlite3 +import types as types_mod + +import pytest + +# The email agent ships as the standalone gaia-agent-email wheel (#1102); +# skip cleanly when a framework-only env lacks it. +pytest.importorskip("gaia_agent_email") + +from fastapi.testclient import TestClient # noqa: E402 +from gaia_agent_email import export_openapi, task_store # noqa: E402 +from gaia_agent_email.contract import ActionItem # noqa: E402 + +from gaia.database.mixin import DatabaseMixin # noqa: E402 + + +class _DB(DatabaseMixin): + pass + + +@pytest.fixture +def db(): + d = _DB() + d.init_db(":memory:") + task_store.init_schema(d) + return d + + +# --------------------------------------------------------------------------- +# Store level +# --------------------------------------------------------------------------- + + +def test_record_creates_tasks_linked_to_message(db): + items = [ + ActionItem(description="Please review the Q3 budget.", due_hint="Friday"), + ActionItem( + description="Follow up at https://example.com/doc", + type="link", + url="https://example.com/doc", + ), + ] + created = task_store.record_action_items(db, message_id="m-1", items=items) + assert len(created) == 2 + + rows = task_store.list_tasks(db, message_id="m-1") + assert [r["description"] for r in rows] == [ + "Please review the Q3 budget.", + "Follow up at https://example.com/doc", + ] + assert all(r["message_id"] == "m-1" for r in rows) + assert rows[0]["due_hint"] == "Friday" + assert rows[1]["item_type"] == "link" + assert rows[1]["url"] == "https://example.com/doc" + assert all(r["status"] == "open" for r in rows) + + +def test_rerecord_same_message_creates_no_duplicates(db): + items = [ActionItem(description="Please review the Q3 budget.")] + first = task_store.record_action_items(db, message_id="m-1", items=items) + assert len(first) == 1 + + again = task_store.record_action_items(db, message_id="m-1", items=items) + assert again == [] + assert len(task_store.list_tasks(db, message_id="m-1")) == 1 + + +def test_concurrent_duplicate_insert_raises_integrity_error_is_skipped(db, monkeypatch): + """A concurrent triage of the same message can win the pre-insert dedup + check race and hit the UNIQUE index on the actual INSERT. That must be + treated as the dedup invariant firing (skip), not re-raised as a failure + that would turn a successful triage into a 500 (bot review on #1917).""" + real_insert = db.insert + calls = {"n": 0} + + def flaky_insert(table, data): + calls["n"] += 1 + if calls["n"] == 1: + # Simulate another connection having already inserted the same + # (message_id, description_norm) row between our SELECT and INSERT. + raise sqlite3.IntegrityError( + "UNIQUE constraint failed: email_tasks.message_id, " + "email_tasks.description_norm" + ) + return real_insert(table, data) + + monkeypatch.setattr(db, "insert", flaky_insert) + + created = task_store.record_action_items( + db, + message_id="m-race", + items=[ + ActionItem(description="Please review the Q3 budget."), + ActionItem(description="A second, distinct item."), + ], + ) + + # The first item lost the race (IntegrityError -> skipped); the second + # item still gets recorded normally. + assert len(created) == 1 + rows = task_store.list_tasks(db, message_id="m-race") + assert [r["description"] for r in rows] == ["A second, distinct item."] + + +def test_non_integrity_error_on_insert_still_propagates(db, monkeypatch): + """Only the dedup-race IntegrityError is swallowed. Any other exception + (e.g. the sqlite file is locked/unwritable) is a real failure and must + still propagate loudly — no silent fallback.""" + + def broken_insert(table, data): + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(db, "insert", broken_insert) + + with pytest.raises(sqlite3.OperationalError, match="locked"): + task_store.record_action_items( + db, message_id="m-1", items=[ActionItem(description="x")] + ) + + +def test_dedup_key_is_normalized_description(): + # Whitespace/case drift in a re-extracted item must not duplicate the task. + d = _DB() + d.init_db(":memory:") + task_store.init_schema(d) + task_store.record_action_items( + d, message_id="m-1", items=[ActionItem(description="Please review the doc")] + ) + created = task_store.record_action_items( + d, message_id="m-1", items=[ActionItem(description=" please REVIEW the doc ")] + ) + assert created == [] + assert len(task_store.list_tasks(d, message_id="m-1")) == 1 + + +def test_same_item_different_messages_are_distinct_tasks(db): + item = [ActionItem(description="Please review the doc")] + task_store.record_action_items(db, message_id="m-1", items=item) + task_store.record_action_items(db, message_id="m-2", items=item) + assert len(task_store.list_tasks(db)) == 2 + assert len(task_store.list_tasks(db, message_id="m-2")) == 1 + + +def test_record_without_message_id_fails_loudly(db): + with pytest.raises(ValueError, match="message_id"): + task_store.record_action_items( + db, message_id="", items=[ActionItem(description="x")] + ) + + +def test_mark_task_done_is_idempotent(db): + (task_id,) = task_store.record_action_items( + db, message_id="m-1", items=[ActionItem(description="Please reply")] + ) + task_store.mark_task_done(db, task_id=task_id) + row = task_store.list_tasks(db, message_id="m-1")[0] + assert row["status"] == "done" + first_done = row["completed_at"] + assert first_done is not None + + task_store.mark_task_done(db, task_id=task_id) + assert task_store.list_tasks(db, message_id="m-1")[0]["completed_at"] == first_done + assert task_store.list_tasks(db, status="open") == [] + + +# --------------------------------------------------------------------------- +# REST surface — POST /v1/email/triage persists tasks as a side-effect +# --------------------------------------------------------------------------- + + +class _FakeChat: + """Deterministic chat stub: classification JSON for classify calls, + a fixed summary otherwise. No Lemonade needed.""" + + def send_messages(self, messages, system_prompt="", **kwargs): + resp = types_mod.SimpleNamespace() + first = messages[0].get("content", "") if messages else "" + resp.text = ( + _json.dumps( + {"category": "NEEDS_RESPONSE", "confidence": 0.9, "reasoning": "t"} + ) + if "Classify" in first + else "summary" + ) + return resp + + +def _triage_payload(message_id: str = "msg-77") -> dict: + return { + "payload": { + "kind": "single", + "principal": {"email": "me@example.com"}, + "message": { + "message_id": message_id, + "from": {"name": "Alice", "email": "alice@example.com"}, + "to": [{"email": "me@example.com"}], + "subject": "Budget", + "body": "Please review the Q3 budget and reply by Friday.", + }, + } + } + + +@pytest.fixture +def triage_env(monkeypatch, db): + """Real /triage route, fake chat, in-memory task DB. Returns (client, db).""" + from gaia_agent_email import api_routes as email_routes + + monkeypatch.setattr( + email_routes.EmailTriageService, + "_build_llm_chat", + lambda self, **kw: _FakeChat(), + ) + monkeypatch.setattr(email_routes, "resolve_action_db", lambda: db) + return TestClient(export_openapi.build_app()), db + + +def test_triage_endpoint_persists_linked_tasks(triage_env): + client, db = triage_env + resp = client.post("/v1/email/triage", json=_triage_payload()) + assert resp.status_code == 200, resp.text + + # Inline response unchanged (additive): action items still returned. + inline = resp.json()["result"]["action_items"] + assert inline, "expected the cue-bearing body to yield inline action items" + + # ...and now also persisted, linked to the source message. + rows = task_store.list_tasks(db, message_id="msg-77") + assert [r["description"] for r in rows] == [i["description"] for i in inline] + + +def test_retriage_endpoint_creates_no_duplicate_tasks(triage_env): + client, db = triage_env + assert client.post("/v1/email/triage", json=_triage_payload()).status_code == 200 + count_after_first = len(task_store.list_tasks(db, message_id="msg-77")) + assert count_after_first > 0 + + assert client.post("/v1/email/triage", json=_triage_payload()).status_code == 200 + assert len(task_store.list_tasks(db, message_id="msg-77")) == count_after_first + + +def test_batch_triage_endpoint_persists_per_item_tasks(triage_env): + client, db = triage_env + item = _triage_payload("msg-batch-1")["payload"] + resp = client.post("/v1/email/triage/batch", json={"items": [item, item]}) + assert resp.status_code == 200, resp.text + + # Two batch items for the SAME message dedup into one set of tasks. + rows = task_store.list_tasks(db, message_id="msg-batch-1") + inline = resp.json()["results"][0]["result"]["action_items"] + assert [r["description"] for r in rows] == [i["description"] for i in inline] + + +# --------------------------------------------------------------------------- +# Persistence failures must not turn a successful triage into a 500 +# (bot review on #1917: sqlite3.IntegrityError from a concurrent duplicate +# should not collapse the response; and /triage/batch's documented per-item +# isolation must hold for persistence failures too, not just triage failures). +# --------------------------------------------------------------------------- + + +def test_single_triage_persistence_failure_does_not_500(triage_env, monkeypatch): + def boom(db, *, message_id, items): + raise sqlite3.OperationalError("database is locked") + + monkeypatch.setattr(task_store, "record_action_items", boom) + + client, _db = triage_env + resp = client.post("/v1/email/triage", json=_triage_payload("msg-boom")) + + # Triage itself succeeded; persistence failing must not surface as a 500. + assert resp.status_code == 200, resp.text + assert resp.json()["result"]["action_items"] + + +def test_batch_triage_persistence_failure_is_isolated_per_item(triage_env, monkeypatch): + """One item's persistence failure must not collapse the whole batch into + a 500, and must not flip that item's already-successful `result` into an + `error` — only the triage step, not the persistence side-effect, decides + result vs error for a batch item.""" + real_record = task_store.record_action_items + + def flaky_record(db, *, message_id, items): + if message_id == "msg-batch-boom": + raise sqlite3.OperationalError("database is locked") + return real_record(db, message_id=message_id, items=items) + + monkeypatch.setattr(task_store, "record_action_items", flaky_record) + + client, db = triage_env + item_ok = _triage_payload("msg-batch-ok")["payload"] + item_boom = _triage_payload("msg-batch-boom")["payload"] + resp = client.post( + "/v1/email/triage/batch", json={"items": [item_boom, item_ok]} + ) + + assert resp.status_code == 200, resp.text + results = resp.json()["results"] + # Both items still report a successful triage `result` (not `error`) — + # persistence failing for one is invisible to the triage contract. + assert results[0]["result"] is not None + assert results[0]["error"] is None + assert results[1]["result"] is not None + assert results[1]["error"] is None + + # The item whose persistence failed has no task rows; the other does. + assert task_store.list_tasks(db, message_id="msg-batch-boom") == [] + assert len(task_store.list_tasks(db, message_id="msg-batch-ok")) > 0 diff --git a/tests/test_api.py b/tests/test_api.py index 9a53b4897..4f5eb5043 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1357,6 +1357,22 @@ def send_messages(self, messages, system_prompt="", **kwargs): monkeypatch.setattr( EmailTriageService, "_build_llm_chat", lambda self, **kw: _FakeChat() ) + + # Triage persists action items to the task store (#1605); point it at + # an in-memory DB so tests never write to the real ~/.gaia. + from gaia_agent_email import api_routes as email_routes + from gaia_agent_email import task_store + + from gaia.database.mixin import DatabaseMixin + + class _TaskDB(DatabaseMixin): + pass + + task_db = _TaskDB() + task_db.init_db(":memory:") + task_store.init_schema(task_db) + monkeypatch.setattr(email_routes, "resolve_action_db", lambda: task_db) + self.client = TestClient(app) def test_single_email_in_structured_out(self):