diff --git a/.agents/skills/email-to-asana/SKILL.md b/.agents/skills/email-to-asana/SKILL.md new file mode 100644 index 0000000..7153419 --- /dev/null +++ b/.agents/skills/email-to-asana/SKILL.md @@ -0,0 +1,279 @@ +--- +name: email-to-asana +description: Filter important RIB emails and create Asana action items classified into the 4 Stanier management-action categories (Information Gathering, Nudging, Being the Example, Taking Decision). +--- + +# Email-to-Asana Triage Skill + +## Purpose + +**email-to-asana** automatically triages incoming emails into one of four management-action categories, creates idempotent Asana tasks, and extracts target dates from email context. The triage engine combines deterministic importance scoring (sender tiers + customer status + content signals) with a single LLM call to categorize each email and identify deadlines. + +The skill: +1. Scores each email's importance via sender membership, customer domain checks, and escalation-signal keywords +2. Fetches email bodies for qualifying messages (recall-biased: always fetch for listed senders or customers, over-fetch the rest) +3. Passes qualifying emails to an LLM for category classification + deadline extraction +4. Creates one Asana task per email (idempotent, dedup-scoped to the target project) +5. Reports preview decisions (dry-run) or committed task creation (with full error handling) + +All per-email errors (LLM timeouts, Asana API failures, date parsing) are caught and logged; they do not propagate and do not prevent processing of subsequent emails. + +## The 4 Categories & Their Asana Projects + +| Category | Project Key | Definition (Stanier) | +|----------|-------------|----------------------| +| **Information Gathering** | `information_gathering` | Observe & collect a signal. No direction change yet; the data informs future action. | +| **Nudging** | `nudging` | Subtly steer someone toward a better outcome while preserving their autonomy. | +| **Being the Example** | `being_the_example` | Model a habit, standard, or boundary visibly so others learn by observing. | +| **Taking Decision** | `taking_decision` | A decision is required of the reader; deadlock or clarity gap blocks progress. | + +## Category Templates + +Each category has a **standard text template** (task description structure) and a **Done-when definition** (acceptance criteria). These are verbatim as stored in the triage engine: + +### Information Gathering + +**Standard text:** +``` +Observe & collect: {what data/signal}. Source: {email link}. +Watching for: {pattern}. No direction change yet. +``` + +**Done when:** +``` +Data reviewed and a note recorded — either 'warrants action → +follow-up spawned' or 'no action needed' — then closed. +``` + +### Nudging + +**Standard text:** +``` +Nudge {who} toward {better outcome} via {subtle mechanism — link / +open question / framing}. Preserve their autonomy. +``` + +**Done when:** +``` +Nudge delivered (message/question/resource sent) and you noted +whether it landed. +``` + +### Being the Example + +**Standard text:** +``` +Model {habit / standard / boundary} in {context}. Demonstrate, +don't instruct. +``` + +**Done when:** +``` +A visible artifact exists (doc written, PR comment left, boundary +set) that others can see. +``` + +### Taking Decision + +**Standard text:** +``` +Decide: {question}. Options: {A / B}. Constraint/deadline: {date}. +Communicate to: {stakeholders}. +``` + +**Done when:** +``` +Decision made, communicated to stakeholders, recorded; deadlock resolved. +``` + +## Importance & Eligibility Rubric + +### Sender Tiers + +The triage engine evaluates sender membership against three configured tiers, each with a weight: + +- **extremely_important** (weight 1.0) — C-suite, founders, or equivalent decision-making authority +- **very_important** (weight 0.8) — Directors, heads of major functions, or strategic partners; triggers a soft bias toward `taking_decision` +- **also_important** (weight 0.5) — Team leads, senior individual contributors, or important collaborators + +Tier members are matched by case-insensitive substring (name or email address). + +### Boosts + +- **Customer boost:** Any sender from an external domain (not `internal_domain`) is treated as a customer and always qualifies +- **Escalation boost:** Content containing 2+ signal keywords (urgency, decision, approval, blocker, deadline, etc.) elevates the email for review + +### Eligibility Gate + +An email qualifies for triage when **any** of the following are true: + +1. Sender is in extremely/very tier **AND** content has actionable signal (score > 0) +2. Sender is in also tier **AND** content is strong (score >= content_high_threshold, default 0.6) +3. Sender is a customer (external domain) +4. Content has high-signal escalation keywords (score >= content_high_threshold) + +### Body Fetch (Recall-Biased) + +- **Always fetch** for: listed senders (any tier), customers +- **Over-fetch** the rest: broad subject pre-scan for signal keywords; declined fetches are logged at DEBUG level + +### Decisive-Sender Bias + +Senders in the `very_important` tier receive a soft bias (instruction to the LLM, not a hard rule) toward the `taking_decision` category when the email genuinely contains a decision point. + +## Configuration + +Triage is configured in `config.yaml` under two sections: + +### `asana:` section (required when triage is enabled) + +```yaml +asana: + pat: "REPLACE_WITH_ASANA_PAT" # Personal Access Token + workspace_gid: "REPLACE_WITH_WORKSPACE_GID" # Workspace GUID + project_gids: + information_gathering: "REPLACE_WITH_PROJECT_GID" + nudging: "REPLACE_WITH_PROJECT_GID" + being_the_example: "REPLACE_WITH_PROJECT_GID" + taking_decision: "REPLACE_WITH_PROJECT_GID" +``` + +**Notes:** +- The PAT is never committed; it lives in the gitignored `config.yaml` +- All four projects must already exist in Asana (validate-only; never auto-created) +- One task is created in exactly one project (the LLM-selected category) + +### `triage:` section (optional, disabled by default) + +```yaml +triage: + enabled: false # Enable/disable triage + scan_rebuild: false # Opt-in rebuild-phase triage + internal_domain: "rib-software.com" # Internal domain (for customer check) + content_high_threshold: 0.6 # Escalation signal threshold (0..1) + sender_tiers: + extremely_important: + - "Rolf Helmes" + - "René Wolf" + # ... more names or addresses ... + very_important: + - "Arthur Berganski" + - "Sanket Khandare" + # ... more names or addresses ... + also_important: + - "Helen Wiersma" + - "Jaan Tasane" + # ... more names or addresses ... +``` + +**Notes:** +- `scan_rebuild`: defaults to `false`; set to `true` to enable triage during the rebuild phase (scanning archive + old inbox) +- `sender_tiers`: each tier is a list of name or email substrings (case-insensitive match) +- `content_high_threshold`: tuned empirically; 0.6 is a reasonable starting point + +## Operating Procedure + +### Preview Decisions (Dry-Run) + +Always preview before committing: + +```bash +kontor-cli triage --dry-run +``` + +This outputs the LLM category decisions, target dates, and task names **without writing to Asana**. No email bodies are fetched or moved. No credentials are validated beyond syntax checks. + +### Commit Decisions (Live) + +When you are confident in the preview output: + +```bash +kontor-cli process +``` + +With triage enabled in the config, the `process` command will: +- Triage realtime inbox messages (INBOX-only by default) +- Create Asana tasks for qualifying emails +- Log a summary: tasks created, deduplicated, or skipped due to error + +### Rebuild-Phase Triage (Optional) + +To triage archive and old inbox during the rebuild phase: + +1. Set `triage.scan_rebuild: true` in config.yaml +2. Run: `kontor-cli process --phase rebuild` + +This is opt-in because it can be slow (full mailbox scan) and may create many tasks retroactively. + +## Failure Boundary + +### Fast-Fail Errors (Prevent All Triage) + +These errors block the entire triage run before processing any emails: + +- Asana PAT is missing or invalid +- Workspace GID is missing +- One or more project GIDs are missing or invalid +- LLM API key or base URL is missing + +### Per-Email Errors (Skip & Log) + +These errors are caught inside `maybe_create_task`, logged at WARNING level, and converted to `outcome="skipped_error"`. Processing continues: + +- Body fetch failure (himalaya read-failure) +- LLM request timeout or HTTP error +- LLM response parsing failure (invalid JSON, missing category) +- Date parsing failure (unparseable deadline string) +- Asana API error (network, quota, 403, 5xx) + +The run summary reports: +- `triage_tasks_created`: count of tasks successfully written to Asana +- `triage_skipped_dedup`: count of emails already in the target project (marker found) +- `triage_skipped_errors`: count of per-email errors (body/LLM/date/Asana failures) + +### Deduplication Marker + +Tasks are idempotent via a stable marker stored in task notes: + +``` + +``` + +The marker is scoped to the **target project only** (the category's project GID). This allows the same email to appear in multiple projects if recategorized, but prevents duplicate tasks within a single category. + +- **Message-ID** is preferred (stable across moves) +- **UID fallback** if Message-ID extraction fails + +### Mailbox Immutability + +The triage engine **never mutates** the mailbox: +- Emails are read via `himalaya --preview` (preview mode only) +- No moves, no deletes, no flag changes +- The mailbox is read-only from triage's perspective + +## LLM Categorization + +The LLM receives: + +- The 4 category definitions (information_gathering, nudging, being_the_example, taking_decision) +- A soft bias toward `taking_decision` if the sender is in the `very_important` tier (does not override genuine content judgment) +- The email's From, Subject, Date, and (conditionally) Body + +The LLM responds with strict JSON: + +```json +{ + "category": "", + "deadline": "", + "rationale": "" +} +``` + +- **deadline** is parsed (if present) relative to the email's date; if parsing fails, falls back to the email's date as target date +- **rationale** is included in task notes for transparency +- Any invalid category or parse failure is logged and the email is skipped + +--- + +**Version:** 1.0 +**Last updated:** 2026-06-28 diff --git a/config.example.yaml b/config.example.yaml index dcf13ba..dcd6cdf 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -66,4 +66,65 @@ pipeline: logging: level: "INFO" - format: "json" \ No newline at end of file + format: "json" + +# --------------------------------------------------------------------------- +# Asana integration (required when triage.enabled is true) +# --------------------------------------------------------------------------- +# asana: +# pat: "REPLACE_WITH_ASANA_PAT" +# workspace_gid: "REPLACE_WITH_WORKSPACE_GID" +# project_gids: +# information_gathering: "REPLACE_WITH_PROJECT_GID" +# nudging: "REPLACE_WITH_PROJECT_GID" +# being_the_example: "REPLACE_WITH_PROJECT_GID" +# taking_decision: "REPLACE_WITH_PROJECT_GID" + +# --------------------------------------------------------------------------- +# Triage pipeline (optional — disabled by default) +# --------------------------------------------------------------------------- +# triage: +# enabled: false +# scan_rebuild: false +# internal_domain: "rib-software.com" +# content_high_threshold: 0.6 +# sender_tiers: +# extremely_important: +# - "Rolf Helmes" +# - "René Wolf" +# - "Martin Biesinger" +# - "Martin Muth" +# - "Georg Reitschmidt" +# - "Joe de Klerk" +# - "Kevin Thompson" +# very_important: +# - "Arthur Berganski" +# - "Sanket Khandare" +# - "Guille Majchrzak" +# - "Julien Seroi" +# - "Stefan Stelzer" +# - "Beate Kasper" +# - "Reinhardt Fraunhoffer" +# - "Christopher Leineweber" +# - "Roman Trottner" +# - "Han Che" +# - "Tim Laine" +# - "Florian Haag" +# - "Frank Bädeker" +# - "Silvio Brendel" +# - "Patrick Janas" +# - "Jignasa Purohit" +# - "Ashwini Bhujabalaiah" +# - "Gautam Makker" +# - "Jeff Ruan" +# also_important: +# - "Helen Wiersma" +# - "Jaan Tasane" +# - "Estela Grana Perez" +# - "Carmen Fernández" +# - "Thomas Rixner" +# - "Kim Fischer" +# - "Sinda Bouzir" +# - "Jinlin Shen" +# - "Simon Welß" +# - "Benjamin Balkanci" \ No newline at end of file diff --git a/src/kontor_cli/asana_client.py b/src/kontor_cli/asana_client.py new file mode 100644 index 0000000..cc89f27 --- /dev/null +++ b/src/kontor_cli/asana_client.py @@ -0,0 +1,151 @@ +"""Asana API client for task lookup and creation.""" + +from __future__ import annotations + +import logging +from typing import Any + +import httpx + +logger = logging.getLogger(__name__) + + +class AsanaError(Exception): + """Raised on Asana API errors (HTTP or network).""" + + +class AsanaClient: + BASE = "https://app.asana.com/api/1.0" + + def __init__( + self, + pat: str, + workspace_gid: str, + project_gids: dict[str, str], + timeout: int = 30, + ) -> None: + self._pat = pat + self.workspace_gid = workspace_gid + self.project_gids = project_gids + self.timeout = timeout + + def _auth_headers(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self._pat}"} + + def validate_projects(self) -> None: + """GET BASE/projects/ for each project_gid value. + + Raises AsanaError naming any missing or inaccessible project. + Never POSTs — never creates a project. + """ + for label, gid in self.project_gids.items(): + url = f"{self.BASE}/projects/{gid}" + try: + response = httpx.get( + url, + headers=self._auth_headers(), + timeout=self.timeout, + ) + response.raise_for_status() + except httpx.HTTPStatusError as exc: + raise AsanaError( + f"Project {gid!r} ({label}) is missing or inaccessible: " + f"HTTP {exc.response.status_code}" + ) from exc + except httpx.RequestError as exc: + raise AsanaError( + f"Network error validating project {gid!r} ({label}): {exc}" + ) from exc + + def find_task_by_marker(self, project_gid: str, marker: str) -> bool: + """Search tasks in project for marker substring in notes, with pagination. + + Returns True if `marker` appears in any task's notes field. + """ + url = f"{self.BASE}/projects/{project_gid}/tasks" + params: dict[str, str] = {"opt_fields": "notes"} + + while True: + try: + response = httpx.get( + url, + headers=self._auth_headers(), + params=params, + timeout=self.timeout, + ) + response.raise_for_status() + except httpx.HTTPStatusError as exc: + raise AsanaError( + f"Failed to fetch tasks for project {project_gid!r}: " + f"HTTP {exc.response.status_code}" + ) from exc + except httpx.RequestError as exc: + raise AsanaError( + f"Network error fetching tasks for project {project_gid!r}: {exc}" + ) from exc + + data = response.json() + for task in data.get("data", []): + notes: str = task.get("notes", "") + if marker in notes: + return True + + next_page = data.get("next_page") + if not next_page: + break + # Follow pagination offset + params = {"opt_fields": "notes", "offset": next_page["offset"]} + + return False + + def create_task( + self, + project_gid: str, + name: str, + notes: str, + due_on: str, + ) -> dict[str, Any]: + """POST a new task to Asana. + + Args: + project_gid: GID of the project to add the task to. + name: Task title. + notes: Task body / description. + due_on: ISO date string 'YYYY-MM-DD'. + + Returns: + The created task dict (contents of response['data']). + + Raises: + AsanaError: On HTTP or network errors. + """ + url = f"{self.BASE}/tasks" + payload: dict[str, Any] = { + "data": { + "name": name, + "notes": notes, + "due_on": due_on, + "projects": [project_gid], + "workspace": self.workspace_gid, + } + } + try: + response = httpx.post( + url, + headers=self._auth_headers(), + json=payload, + timeout=self.timeout, + ) + response.raise_for_status() + except httpx.HTTPStatusError as exc: + raise AsanaError( + f"Failed to create task in project {project_gid!r}: " + f"HTTP {exc.response.status_code}" + ) from exc + except httpx.RequestError as exc: + raise AsanaError( + f"Network error creating task in project {project_gid!r}: {exc}" + ) from exc + + result: dict[str, Any] = response.json()["data"] + return result diff --git a/src/kontor_cli/cli.py b/src/kontor_cli/cli.py index 8e4732a..9c6b2df 100644 --- a/src/kontor_cli/cli.py +++ b/src/kontor_cli/cli.py @@ -16,9 +16,11 @@ DavMailNotReachableError, HimalayaNotFoundError, ) +from kontor_cli.himalaya import list_emails, read_message_body from kontor_cli.logging_config import configure_logging from kontor_cli.mailbox_cleanup import restore_archive_projects from kontor_cli.pipeline import HealPipeline, RealtimePipeline, RebuildPipeline +from kontor_cli.triage import Triage logger = logging.getLogger("kontor_cli") @@ -259,6 +261,55 @@ def rules_freeze_cmd(config_path: Path | None) -> None: _rules_freeze(cfg, root) +@cli.command("triage") +@click.option("--folder", default="INBOX", help="Source folder (default: INBOX)") +@click.option( + "--config", + "config_path", + type=click.Path(exists=False, path_type=Path), + default=None, +) +@click.option( + "--dry-run", + is_flag=True, + default=True, + help="Preview mode — no Asana writes (always on; this command is preview-only)", +) +def triage_cmd(folder: str, config_path: Path | None, dry_run: bool) -> None: + """Preview which emails qualify for Asana task creation (preview-only). + + Always runs in preview mode — no Asana writes. Task creation is the job of + ``kontor-cli process``. + """ + try: + cfg = Config.load(config_path) + except ConfigError as exc: + click.echo(f"Config error: {exc}", err=True) + sys.exit(1) + + root = (config_path or Path.cwd() / "config.yaml").parent + triage = Triage(cfg, cwd=root) + emails = list_emails(folder, cwd=root) + + for email in emails: + + def _body_fetcher(e: Any, _root: Path = root, _folder: str = folder) -> str: + return read_message_body(e.id, _folder, cwd=_root) + + # Preview-only: always dry-run. Writes are the job of `process`. + decision = triage.maybe_create_task( + email, body_fetcher=_body_fetcher, dry_run=True + ) + qualify_flag = "y" if decision.qualifies else "n" + category = decision.category or "-" + target_date = decision.target_date or "-" + task_name = decision.task_name or "-" + click.echo( + f"{email.id} qualify={qualify_flag} reason={decision.reason}" + f" category={category} due={target_date} task={task_name}" + ) + + def _rules_freeze(cfg: Config, root: Path) -> None: """Write a frozen snapshot of the evolved rules directory.""" import json diff --git a/src/kontor_cli/config.py b/src/kontor_cli/config.py index 0129007..f362463 100644 --- a/src/kontor_cli/config.py +++ b/src/kontor_cli/config.py @@ -58,6 +58,22 @@ def __init__(self, data: dict[str, Any], config_dir: Path | None = None) -> None ] self.log_level: str = data["logging"]["level"] self.log_format: str = data["logging"]["format"] + # Optional asana section + asana = data.get("asana", {}) + self.asana_pat: str | None = asana.get("pat") or None + self.asana_workspace_gid: str | None = asana.get("workspace_gid") or None + self.asana_project_gids: dict[str, str] = asana.get("project_gids") or {} + # Optional triage section + triage = data.get("triage", {}) + self.triage_enabled: bool = bool(triage.get("enabled", False)) + self.triage_scan_rebuild: bool = bool(triage.get("scan_rebuild", False)) + self.triage_internal_domain: str = triage.get( + "internal_domain", "rib-software.com" + ) + self.triage_sender_tiers: dict[str, list[str]] = triage.get("sender_tiers", {}) + self.triage_content_high_threshold: float = float( + triage.get("content_high_threshold", 0.6) + ) @classmethod def load(cls, path: str | Path | None = None) -> Config: @@ -101,6 +117,30 @@ def _validate_required(cls, data: dict[str, Any], path: Path) -> None: if not isinstance(data["davmail"].get("smtp_port"), int): raise ConfigError("davmail.smtp_port must be an integer") + # Conditional asana validation: only required when triage.enabled is true + if data.get("triage", {}).get("enabled"): + asana = data.get("asana") + if asana is None: + raise ConfigError("triage.enabled is true but asana section is missing") + if not asana.get("pat"): + raise ConfigError("triage.enabled is true but asana.pat is missing") + if not asana.get("workspace_gid"): + raise ConfigError( + "triage.enabled is true but asana.workspace_gid is missing" + ) + _required_project_gids = [ + "information_gathering", + "nudging", + "being_the_example", + "taking_decision", + ] + project_gids = asana.get("project_gids") or {} + for key in _required_project_gids: + if key not in project_gids: + raise ConfigError( + f"triage.enabled is true but asana.project_gids.{key} is missing" + ) + def check_prerequisites(self) -> None: """Run startup checks: himalaya, himalaya version, DavMail connectivity.""" self._check_himalaya() diff --git a/src/kontor_cli/himalaya.py b/src/kontor_cli/himalaya.py index 04e3e11..4ff646b 100644 --- a/src/kontor_cli/himalaya.py +++ b/src/kontor_cli/himalaya.py @@ -34,18 +34,22 @@ class Email: date: datetime flags: dict[str, bool] folder: str + from_name: str = "" @classmethod def from_json(cls, obj: dict[str, Any], folder: str) -> Email: """Parse from a himalaya envelope JSON dict.""" from_field = obj.get("from", {}) - addr = ( - from_field.get("addr", from_field.get("address", "")) - if isinstance(from_field, dict) - else str(from_field) - ) + if isinstance(from_field, dict): + addr = from_field.get("addr", from_field.get("address", "")) + name = from_field.get("name", "") + else: + addr = str(from_field) + name = "" if not isinstance(addr, str): addr = str(addr) + if not isinstance(name, str): + name = str(name) date_str = obj.get("date", "") try: date = datetime.fromisoformat(date_str.replace("Z", "+00:00")) @@ -54,6 +58,7 @@ def from_json(cls, obj: dict[str, Any], folder: str) -> Email: return cls( id=str(obj.get("id", "")), from_addr=addr, + from_name=name, subject=obj.get("subject", ""), date=date, flags=obj.get("flags", {}), @@ -165,6 +170,35 @@ def delete_folder(folder_name: str, cwd: str | Path | None = None) -> None: _run(["folder", "delete", folder_name], cwd=cwd) +def read_message_body( + email_id: str, folder: str = "INBOX", cwd: str | Path | None = None +) -> str: + """Fetch plaintext body read-only (no seen flag) via himalaya v1.2.0.""" + return _run( + ["message", "read", email_id, "-f", folder, "--no-headers", "--preview"], + cwd=cwd, + ) + + +def read_message_id( + email_id: str, folder: str = "INBOX", cwd: str | Path | None = None +) -> str | None: + """Fetch the Message-ID header (stable dedup key); None if absent. + + Issues a separate himalaya call with -H (mutually exclusive with --no-headers). + Parses the 'Message-Id:' line case-insensitively and strips surrounding <...>. + """ + out = _run( + ["message", "read", email_id, "-f", folder, "-H", "Message-Id", "--preview"], + cwd=cwd, + ) + for line in out.splitlines(): + if line.lower().startswith("message-id:"): + value = line.split(":", 1)[1].strip() + return value.strip("<>") + return None + + def delete_email(email_id: str, folder: str, cwd: str | Path | None = None) -> None: """Deletion is not supported — raises DeleteNotSupportedError.""" raise DeleteNotSupportedError( diff --git a/src/kontor_cli/pipeline.py b/src/kontor_cli/pipeline.py index 222b08f..ee108ea 100644 --- a/src/kontor_cli/pipeline.py +++ b/src/kontor_cli/pipeline.py @@ -20,8 +20,10 @@ create_folder, list_emails, move_email, + read_message_body, ) from kontor_cli.rules_engine import RulesEngine +from kontor_cli.triage import Triage logger = logging.getLogger("kontor_cli.pipeline") @@ -95,6 +97,27 @@ def __init__(self, config: Config, cwd: Path | None = None) -> None: self.llm_failures = 0 self._created_folders: set[str] = set() # cache of folders already created + # Email → Asana triage (Step 8). Only constructed when enabled. + self.triage = Triage(config, cwd) if config.triage_enabled else None + self.triage_tasks_created = 0 + self.triage_skipped_dedup = 0 + self.triage_skipped_errors = 0 + self._triage_validated = False # validate_projects() runs at most once + + def _validate_triage_projects(self, dry_run: bool, triage_scope: bool) -> None: + """Validate Asana projects once, up-front, before processing emails. + + Loud fail-fast: an AsanaError here aborts the whole run. Skipped in + dry-run, when triage is disabled/out-of-scope, or when already run. + """ + if self.triage is None or not triage_scope or dry_run: + return + if self._triage_validated: + return + self._triage_validated = True + if self.triage.asana is not None: + self.triage.asana.validate_projects() + def _ensure_folder(self, folder: str) -> None: """Ensure a folder exists. Creates it if valid and missing.""" if folder in self._created_folders: @@ -118,13 +141,45 @@ def _classify(self, email: Email) -> str | None: result = self._llm_classify(email) return result.folder if result else None - def _process_email(self, email: Email, dry_run: bool = False) -> str | None: - """Process a single email: classify → decide target folder → move.""" + def _process_email( + self, email: Email, dry_run: bool = False, triage_scope: bool = False + ) -> str | None: + """Process a single email: classify → decide target folder → move. + + ``triage_scope`` enables content-driven email → Asana triage for this + phase (realtime always; rebuild only when configured; heal never). + """ current_folder = email.folder # Step 1: classify (rules, then LLM fallback) and decide the target target = self.folder_policy.target_for(email.date, self._classify(email)) + # Step 1b: content-driven triage — fires on the classified email + # regardless of the move outcome below (loop-skip / already-correct / + # move failure all still triage). A triage bug must NEVER break the + # move loop, so everything here is defensively contained. + if self.triage is not None and triage_scope: + try: + decision = self.triage.maybe_create_task( + email, + body_fetcher=lambda e: read_message_body( + e.id, e.folder, cwd=self.cwd + ), + dry_run=dry_run, + ) + if decision.outcome == "created": + self.triage_tasks_created += 1 + elif decision.outcome == "skipped_dedup": + self.triage_skipped_dedup += 1 + elif decision.outcome == "skipped_error": + self.triage_skipped_errors += 1 + except Exception: + logger.exception( + f"Triage failed for email {email.id}", + extra={"email_id": email.id}, + ) + self.triage_skipped_errors += 1 + # Step 2: Loop prevention if (email.id, target) in self.move_history: self.skipped_loop += 1 @@ -243,6 +298,9 @@ def _summary(self, phase: str, total: int) -> dict[str, int | str]: "skipped_already_correct": self.skipped_already_correct, "skipped_loop": self.skipped_loop, "llm_failures": self.llm_failures, + "triage_tasks_created": self.triage_tasks_created, + "triage_skipped_dedup": self.triage_skipped_dedup, + "triage_skipped_errors": self.triage_skipped_errors, } logger.info(f"Phase {phase} complete", extra={**s, "phase": phase}) return s @@ -253,6 +311,8 @@ class RebuildPipeline(Pipeline): def run(self, dry_run: bool = False) -> dict[str, Any]: logger.info("Starting Historical Rebuild", extra={"phase": "rebuild"}) + triage_scope = self.config.triage_scan_rebuild + self._validate_triage_projects(dry_run, triage_scope) total_processed = 0 for folder in SCAN_FOLDERS: try: @@ -262,7 +322,7 @@ def run(self, dry_run: bool = False) -> dict[str, Any]: continue for email in emails: - self._process_email(email, dry_run=dry_run) + self._process_email(email, dry_run=dry_run, triage_scope=triage_scope) total_processed += 1 return self._summary("rebuild", total_processed) @@ -273,6 +333,7 @@ class RealtimePipeline(Pipeline): def run(self, dry_run: bool = False) -> dict[str, Any]: logger.info("Starting Real-Time Processing", extra={"phase": "realtime"}) + self._validate_triage_projects(dry_run, triage_scope=True) try: emails = list_emails("INBOX", cwd=self.cwd) except HimalayaError as exc: @@ -281,7 +342,7 @@ def run(self, dry_run: bool = False) -> dict[str, Any]: total = 0 for email in emails: - self._process_email(email, dry_run=dry_run) + self._process_email(email, dry_run=dry_run, triage_scope=True) total += 1 return self._summary("realtime", total) diff --git a/src/kontor_cli/triage.py b/src/kontor_cli/triage.py new file mode 100644 index 0000000..1d90e22 --- /dev/null +++ b/src/kontor_cli/triage.py @@ -0,0 +1,576 @@ +"""Email → Asana management-action triage. + +Deterministic importance scoring, recall-biased body-fetch selection, one +mocked LLM boundary for category + deadline judgment, and idempotent task +creation orchestration. All per-email LLM/Asana/date/body errors are caught +inside ``maybe_create_task`` and turned into ``outcome="skipped_error"`` — they +never propagate. +""" + +from __future__ import annotations + +import json +import logging +from collections.abc import Callable +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import httpx +from dateutil import parser as date_parser + +from . import himalaya +from .asana_client import AsanaClient, AsanaError + +if TYPE_CHECKING: + from .config import Config + from .himalaya import Email + +logger = logging.getLogger(__name__) + +# Sender-tier weights for the deterministic importance component. +TIER_WEIGHTS: dict[str, float] = { + "extremely_important": 1.0, + "very_important": 0.8, + "also_important": 0.5, +} + +# Broad subject-keyword pre-scan for the unlisted/non-customer remainder. +# Over-fetch biased (recall over precision); the LLM is the precision filter. +_CONTENT_KEYWORDS: tuple[str, ...] = ( + "escalation", + "escalate", + "urgent", + "asap", + "decision", + "decide", + "approve", + "approval", + "blocker", + "blocked", + "can you", + "could you", + "please advise", + "please review", + "deadline", + "go-live", + "go live", + "hypercare", + "critical", + "priority", + "action required", +) + +# Category rubric → standard text + definition-of-done, encoded VERBATIM. +CATEGORY_TEMPLATES: dict[str, dict[str, str]] = { + "information_gathering": { + "standard_text": ( + "Observe & collect: {what data/signal}. Source: {email link}. " + "Watching for: {pattern}. No direction change yet." + ), + "done_when": ( + "Data reviewed and a note recorded — either 'warrants action → " + "follow-up spawned' or 'no action needed' — then closed." + ), + }, + "nudging": { + "standard_text": ( + "Nudge {who} toward {better outcome} via {subtle mechanism — link / " + "open question / framing}. Preserve their autonomy." + ), + "done_when": ( + "Nudge delivered (message/question/resource sent) and you noted " + "whether it landed." + ), + }, + "being_the_example": { + "standard_text": ( + "Model {habit / standard / boundary} in {context}. Demonstrate, " + "don't instruct." + ), + "done_when": ( + "A visible artifact exists (doc written, PR comment left, boundary " + "set) that others can see." + ), + }, + "taking_decision": { + "standard_text": ( + "Decide: {question}. Options: {A / B}. Constraint/deadline: {date}. " + "Communicate to: {stakeholders}." + ), + "done_when": ( + "Decision made, communicated to stakeholders, recorded; deadlock resolved." + ), + }, +} + +_VALID_CATEGORIES: frozenset[str] = frozenset(CATEGORY_TEMPLATES) + + +@dataclass +class ImportanceScore: + """Result of the deterministic importance evaluation.""" + + sender_component: float + content_component: float + customer_boost: bool + escalation_boost: bool + decisive_prior: bool + qualifies: bool + reason: str + + +@dataclass +class CategoryDecision: + """LLM judgment of category + (optional) deadline for a qualifying email.""" + + category: str + deadline: date | None + rationale: str + + +@dataclass +class TriageDecision: + """Per-email triage outcome (preview-safe; no side effects implied).""" + + email_id: str + qualifies: bool + reason: str + category: str | None + target_date: str | None + task_name: str | None + task_notes: str | None + outcome: str # preview | created | skipped_dedup | skipped_error | not_qualified + + +class Triage: + """Composed, deterministic-core email → Asana triage engine.""" + + def __init__(self, config: Config, cwd: Path | None = None) -> None: + self.config = config + self.cwd = cwd + # Lazy Asana client: only when a PAT is configured. + self.asana: AsanaClient | None = None + if config.asana_pat and config.asana_workspace_gid: + self.asana = AsanaClient( + config.asana_pat, + config.asana_workspace_gid, + config.asana_project_gids, + ) + self.tiers = config.triage_sender_tiers + self.internal_domain = config.triage_internal_domain + self.content_high_threshold = config.triage_content_high_threshold + # Decisive senders bias the LLM toward taking_decision (soft, not forced). + self.decisive: list[str] = list(self.tiers.get("very_important", [])) + + # ------------------------------------------------------------------ # + # Step 3 — deterministic scoring + eligibility gate + # ------------------------------------------------------------------ # + @staticmethod + def _matches_member(from_addr: str, from_name: str, member: str) -> bool: + """Match an email's from-field against a tier member string. + + Matches by case-insensitive substring on either the display name or + the address — seeded tier entries may be a display name (e.g. "Rolf + Helmes") or an address (e.g. "rolf.helmes@rib-software.com"). Both the + envelope ``from_name`` and ``from_addr`` are searched. + """ + needle = member.lower().strip() + if not needle: + return False + haystack = f"{from_name}\n{from_addr}".lower() + return needle in haystack + + def score_sender(self, email: Email) -> tuple[float, bool, bool]: + """Return ``(weight, decisive_prior, in_any_tier)`` for the sender. + + Highest matching tier wins. ``decisive_prior`` is True iff the sender + is in the ``very_important`` tier. + """ + weight = 0.0 + in_any_tier = False + decisive_prior = False + for tier, members in self.tiers.items(): + for member in members: + if self._matches_member(email.from_addr, email.from_name, member): + in_any_tier = True + tier_weight = TIER_WEIGHTS.get(tier, 0.0) + if tier_weight > weight: + weight = tier_weight + if tier == "very_important": + decisive_prior = True + return weight, decisive_prior, in_any_tier + + def is_customer(self, from_addr: str) -> bool: + """True when the sender's domain is external (not the internal domain).""" + if "@" not in from_addr: + return False + domain = from_addr.rsplit("@", 1)[-1].strip().lower().strip("<>") + if not domain: + return False + return domain != self.internal_domain.lower() + + def score_content(self, subject: str, body: str) -> float: + """Deterministic 0..1 content-signal heuristic. + + Scales with the number of distinct signal keywords present across the + subject + body. Transparent and testable. + """ + haystack = f"{subject}\n{body}".lower() + hits = sum(1 for kw in _CONTENT_KEYWORDS if kw in haystack) + if hits == 0: + return 0.0 + # Two distinct signals saturates to HIGH; one signal lands mid-band. + return min(1.0, 0.35 + 0.35 * hits) + + def evaluate(self, email: Email, body: str) -> ImportanceScore: + """Deterministic eligibility gate. + + Qualifies when ANY of: + - sender in extremely/very tier AND content is actionable (>0) + - sender in also tier AND content is strong (>= threshold) + - sender is a customer (external) + - content is a HIGH escalation signal (>= threshold) + """ + weight, decisive_prior, in_any_tier = self.score_sender(email) + content = self.score_content(email.subject, body) + customer = self.is_customer(email.from_addr) + escalation = content >= self.content_high_threshold + + high_tier = weight >= TIER_WEIGHTS["very_important"] + also_tier = in_any_tier and not high_tier + + qualifies = ( + (high_tier and content > 0.0) + or (also_tier and content >= self.content_high_threshold) + or customer + or escalation + ) + + reasons: list[str] = [] + if high_tier and content > 0.0: + reasons.append("high-tier sender with actionable content") + if also_tier and content >= self.content_high_threshold: + reasons.append("also-important sender with strong content") + if customer: + reasons.append("external customer sender") + if escalation: + reasons.append("high-signal escalation/decision content") + if not reasons: + reasons.append("no qualifying signal (low importance)") + + return ImportanceScore( + sender_component=weight, + content_component=content, + customer_boost=customer, + escalation_boost=escalation, + decisive_prior=decisive_prior, + qualifies=qualifies, + reason="; ".join(reasons), + ) + + # ------------------------------------------------------------------ # + # Step 4 — recall-biased body-fetch selector + # ------------------------------------------------------------------ # + def should_fetch_body(self, email: Email) -> bool: + """Decide whether to fetch the body for an email. + + Always fetch for any listed sender or any customer (no subject gate). + Otherwise apply a broad over-fetch-biased subject pre-scan. A declined + fetch emits a DEBUG log naming the email id. + """ + _, _, in_any_tier = self.score_sender(email) + if in_any_tier or self.is_customer(email.from_addr): + return True + if self.score_content(email.subject, "") > 0.0: + return True + logger.debug("triage pre-filter declined %s: subject had no signal", email.id) + return False + + # ------------------------------------------------------------------ # + # Step 6 — LLM category + deadline judgment + deterministic date + # ------------------------------------------------------------------ # + def judge_category( + self, email: Email, body: str, decisive_prior: bool + ) -> CategoryDecision | None: + """One LLM call returning ``{category, deadline, rationale}``. + + Returns None on any HTTP/parse/validation error (e.g. category not in + the 4 slugs). + """ + bias = "" + if decisive_prior: + bias = ( + " The sender is a decisive stakeholder; softly favor " + "'taking_decision' when the content is genuinely a decision, " + "but do not force it." + ) + rubric = ( + "Categorize this email into exactly ONE management-action slug:\n" + "- information_gathering: observe & collect a signal, no action yet.\n" + "- nudging: subtly steer someone toward a better outcome.\n" + "- being_the_example: model a habit/standard/boundary visibly.\n" + "- taking_decision: a decision is required of the reader." + bias + ) + date_str = email.date.isoformat() if email.date is not None else "unknown" + user_prompt = ( + f"{rubric}\n\n" + f"From: {email.from_addr}\n" + f"Subject: {email.subject}\n" + f"Date: {date_str}\n\n" + f"Body:\n{body}\n\n" + "Respond with STRICT JSON only: " + '{"category": "", ' + '"deadline": "", ' + '"rationale": ""}' + ) + + try: + response = httpx.post( + f"{self.config.llm_base_url.rstrip('/')}/chat/completions", + headers={ + "Authorization": f"Bearer {self.config.llm_api_key}", + "Content-Type": "application/json", + }, + json={ + "model": self.config.llm_model, + "messages": [ + {"role": "user", "content": user_prompt}, + ], + "temperature": self.config.llm_temperature, + }, + timeout=self.config.llm_timeout, + ) + response.raise_for_status() + except httpx.HTTPStatusError as exc: + logger.error( + "triage LLM returned %s for %s", + exc.response.status_code, + email.id, + ) + return None + except httpx.RequestError as exc: + logger.error("triage LLM request failed for %s: %s", email.id, exc) + return None + + try: + data = response.json() + content = data["choices"][0]["message"]["content"] + if content.strip().startswith("```"): + content = content.strip()[content.strip().find("\n") + 1 :] + if content.endswith("```"): + content = content[:-3].strip() + parsed: dict[str, Any] = json.loads(content) + except (KeyError, IndexError, json.JSONDecodeError) as exc: + logger.error("triage LLM parse failed for %s: %r", email.id, exc) + return None + + category = parsed.get("category") + if category not in _VALID_CATEGORIES: + logger.error( + "triage LLM returned invalid category %r for %s", category, email.id + ) + return None + + deadline_raw = parsed.get("deadline") + deadline: date | None = None + if deadline_raw: + try: + anchor = email.date if email.date is not None else None + deadline = ( + date_parser.parse(str(deadline_raw), default=anchor).date() + if anchor is not None + else date_parser.parse(str(deadline_raw)).date() + ) + except (ValueError, OverflowError, TypeError): + # Keep the raw phrase out; resolve_target_date falls back. + deadline = None + + return CategoryDecision( + category=str(category), + deadline=deadline, + rationale=str(parsed.get("rationale", "")), + ) + + def resolve_target_date(self, decision: CategoryDecision, email: Email) -> str: + """Resolve a concrete 'YYYY-MM-DD' due date. + + Uses the decision's deadline when present; else falls back to the + email's date. Raises ValueError when neither is usable. + """ + if decision.deadline is not None: + return decision.deadline.isoformat() + if email.date is not None: + return email.date.date().isoformat() + raise ValueError("no usable deadline and email.date is missing") + + # ------------------------------------------------------------------ # + # Step 7 — task assembly + orchestration + # ------------------------------------------------------------------ # + def _resolve_marker(self, email: Email) -> str: + """Stable dedup marker: Message-ID preferred, UID fallback.""" + message_id: str | None = None + try: + message_id = himalaya.read_message_id(email.id, email.folder, self.cwd) + except Exception as exc: # noqa: BLE001 — fall back to UID on any error + logger.debug("read_message_id failed for %s: %s", email.id, exc) + key = message_id or email.id + return f"kontor-id:{key}" + + def _build_notes( + self, + email: Email, + decision: CategoryDecision, + marker: str, + ) -> str: + template = CATEGORY_TEMPLATES[decision.category] + return ( + f"Sender: {email.from_addr}\n" + f"Date: {email.date.isoformat()}\n" + f"Email reference: {email.folder}/{email.id}\n" + f"Rationale: {decision.rationale}\n\n" + f"{template['standard_text']}\n\n" + f"Done when: {template['done_when']}\n\n" + f"" + ) + + def maybe_create_task( + self, + email: Email, + body_fetcher: Callable[[Email], str], + dry_run: bool, + ) -> TriageDecision: + """Orchestrate one email → at-most-one idempotent Asana task. + + All LLM/Asana/date/body errors are caught here and converted into + ``outcome="skipped_error"``; they never propagate. + """ + # 1. Body (recall-biased selector). + if self.should_fetch_body(email): + try: + body = body_fetcher(email) + except Exception as exc: # noqa: BLE001 — per-email skip-and-log + logger.warning("triage body fetch failed for %s: %s", email.id, exc) + return TriageDecision( + email_id=email.id, + qualifies=False, + reason="body fetch failed", + category=None, + target_date=None, + task_name=None, + task_notes=None, + outcome="skipped_error", + ) + else: + body = "" + + # 2. Deterministic gate. + score = self.evaluate(email, body) + if not score.qualifies: + return TriageDecision( + email_id=email.id, + qualifies=False, + reason=score.reason, + category=None, + target_date=None, + task_name=None, + task_notes=None, + outcome="not_qualified", + ) + + # 3. LLM category judgment. + decision = self.judge_category(email, body, score.decisive_prior) + if decision is None: + logger.warning("triage LLM produced no decision for %s", email.id) + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=None, + target_date=None, + task_name=None, + task_notes=None, + outcome="skipped_error", + ) + + # 4. Deterministic date. + try: + target_date = self.resolve_target_date(decision, email) + except ValueError as exc: + logger.warning("triage date resolution failed for %s: %s", email.id, exc) + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=decision.category, + target_date=None, + task_name=None, + task_notes=None, + outcome="skipped_error", + ) + + # 5. Stable marker. + marker = self._resolve_marker(email) + + # 6. Assemble name + notes. + category_title = decision.category.replace("_", " ").title() + task_name = f"[{category_title}] {email.subject}" + task_notes = self._build_notes(email, decision, marker) + + # 7. Dry-run preview — no Asana calls. + if dry_run: + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=decision.category, + target_date=target_date, + task_name=task_name, + task_notes=task_notes, + outcome="preview", + ) + + # 8. Dedup scoped to the ONE target project. + target_project_gid = self.config.asana_project_gids[decision.category] + try: + if self.asana is None: + raise AsanaError("Asana client not configured") + if self.asana.find_task_by_marker(target_project_gid, marker): + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=decision.category, + target_date=target_date, + task_name=task_name, + task_notes=task_notes, + outcome="skipped_dedup", + ) + # 9. Create. + self.asana.create_task( + target_project_gid, task_name, task_notes, target_date + ) + except AsanaError as exc: + logger.warning("triage Asana call failed for %s: %s", email.id, exc) + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=decision.category, + target_date=target_date, + task_name=task_name, + task_notes=task_notes, + outcome="skipped_error", + ) + + return TriageDecision( + email_id=email.id, + qualifies=True, + reason=score.reason, + category=decision.category, + target_date=target_date, + task_name=task_name, + task_notes=task_notes, + outcome="created", + ) diff --git a/tests/unit/asana_client_test.py b/tests/unit/asana_client_test.py new file mode 100644 index 0000000..d9b5199 --- /dev/null +++ b/tests/unit/asana_client_test.py @@ -0,0 +1,202 @@ +"""Unit tests for AsanaClient.""" + +from __future__ import annotations + +from unittest import mock + +import httpx +import pytest + +from kontor_cli.asana_client import AsanaClient, AsanaError + +PAT = "test-pat-token" +WORKSPACE_GID = "ws-123" +PROJECT_GIDS = {"bugs": "proj-bugs-456", "features": "proj-features-789"} + + +def _make_client() -> AsanaClient: + return AsanaClient( + pat=PAT, + workspace_gid=WORKSPACE_GID, + project_gids=PROJECT_GIDS, + timeout=10, + ) + + +def _mock_response(json_data: object, status_code: int = 200) -> mock.MagicMock: + resp = mock.MagicMock() + resp.json.return_value = json_data + resp.raise_for_status = mock.MagicMock() + resp.status_code = status_code + return resp + + +# --------------------------------------------------------------------------- +# validate_projects +# --------------------------------------------------------------------------- + + +def test_validate_projects_all_exist_passes() -> None: + client = _make_client() + ok = _mock_response({"data": {"gid": "proj-bugs-456", "name": "Bugs"}}) + + with mock.patch("httpx.get", return_value=ok) as mock_get: + client.validate_projects() # should not raise + + # Called once per project gid + assert mock_get.call_count == len(PROJECT_GIDS) + + +def test_validate_projects_missing_gid_raises_asana_error() -> None: + client = _make_client() + + status_err = httpx.HTTPStatusError( + "404", + request=mock.MagicMock(), + response=mock.MagicMock(status_code=404), + ) + err_resp = _mock_response({}, status_code=404) + err_resp.raise_for_status.side_effect = status_err + + with mock.patch("httpx.get", return_value=err_resp): + with pytest.raises(AsanaError, match="proj-"): + client.validate_projects() + + +def test_validate_never_posts_to_projects_endpoint() -> None: + """validate_projects must never POST (i.e., never create a project).""" + client = _make_client() + ok = _mock_response({"data": {"gid": "proj-bugs-456", "name": "Bugs"}}) + + with mock.patch("httpx.get", return_value=ok): + with mock.patch("httpx.post") as mock_post: + client.validate_projects() + + mock_post.assert_not_called() + + +# --------------------------------------------------------------------------- +# find_task_by_marker +# --------------------------------------------------------------------------- + + +def test_find_task_by_marker_found_returns_true() -> None: + client = _make_client() + marker = "TICKET-42" + page = _mock_response( + { + "data": [ + {"gid": "t1", "notes": f"Some notes with {marker} embedded"}, + ], + "next_page": None, + } + ) + + with mock.patch("httpx.get", return_value=page): + found = client.find_task_by_marker("proj-bugs-456", marker) + + assert found is True + + +def test_find_task_by_marker_not_found_returns_false() -> None: + client = _make_client() + page = _mock_response( + { + "data": [ + {"gid": "t1", "notes": "Unrelated task"}, + ], + "next_page": None, + } + ) + + with mock.patch("httpx.get", return_value=page): + found = client.find_task_by_marker("proj-bugs-456", "TICKET-99") + + assert found is False + + +def test_find_task_by_marker_paginates() -> None: + """First page has a next_page offset; marker appears only on second page.""" + client = _make_client() + marker = "TICKET-77" + + page1 = _mock_response( + { + "data": [{"gid": "t1", "notes": "nothing here"}], + "next_page": {"offset": "eyJsaW1pdCI6MjV9"}, + } + ) + page2 = _mock_response( + { + "data": [{"gid": "t2", "notes": f"contains {marker} here"}], + "next_page": None, + } + ) + + with mock.patch("httpx.get", side_effect=[page1, page2]): + found = client.find_task_by_marker("proj-bugs-456", marker) + + assert found is True + + +# --------------------------------------------------------------------------- +# create_task +# --------------------------------------------------------------------------- + + +def test_create_task_posts_expected_payload() -> None: + client = _make_client() + project_gid = "proj-bugs-456" + name = "Bug: something broke" + notes = "Details about the bug" + due_on = "2026-07-15" + + created = {"gid": "task-new-999", "name": name} + resp = _mock_response({"data": created}) + + with mock.patch("httpx.post", return_value=resp) as mock_post: + result = client.create_task(project_gid, name, notes, due_on) + + assert result == created + + mock_post.assert_called_once() + call_kwargs = mock_post.call_args + payload = call_kwargs.kwargs.get("json") or call_kwargs.args[1] + data = payload["data"] + assert data["name"] == name + assert data["notes"] == notes + assert data["due_on"] == due_on + assert project_gid in data["projects"] + assert data["workspace"] == WORKSPACE_GID + + +# --------------------------------------------------------------------------- +# Error handling +# --------------------------------------------------------------------------- + + +def test_asana_http_error_raises_asana_error() -> None: + client = _make_client() + + status_err = httpx.HTTPStatusError( + "403 Forbidden", + request=mock.MagicMock(), + response=mock.MagicMock(status_code=403), + ) + err_resp = _mock_response({}, status_code=403) + err_resp.raise_for_status.side_effect = status_err + + with mock.patch("httpx.get", return_value=err_resp): + with pytest.raises(AsanaError): + client.find_task_by_marker("proj-bugs-456", "TICKET-1") + + +def test_asana_request_error_raises_asana_error() -> None: + client = _make_client() + + with mock.patch( + "httpx.get", + side_effect=httpx.RequestError("connection refused", request=mock.MagicMock()), + ): + with pytest.raises(AsanaError): + client.find_task_by_marker("proj-bugs-456", "TICKET-1") diff --git a/tests/unit/cli_test.py b/tests/unit/cli_test.py index 928a456..5075563 100644 --- a/tests/unit/cli_test.py +++ b/tests/unit/cli_test.py @@ -340,3 +340,154 @@ def test_classify_recommend_requires_no_llm_api_key(self, tmp_path: Path) -> Non assert data["email"]["from"] == "boss@example.com" assert "rules_based_target" in data assert "taxonomy" in data + + +class TestTriage: + """Tests for the `triage` CLI command.""" + + def _make_email(self) -> object: + from datetime import datetime + + from kontor_cli.himalaya import Email + + return Email( + id="99", + from_addr="customer@external.com", + subject="Urgent: blocker on go-live", + date=datetime(2026, 6, 28, 10, 0, 0, tzinfo=UTC), + flags={}, + folder="INBOX", + ) + + def _make_decision(self) -> object: + from kontor_cli.triage import TriageDecision + + return TriageDecision( + email_id="99", + qualifies=True, + reason="external customer sender", + category="taking_decision", + target_date="2026-06-30", + task_name="[Taking Decision] Urgent: blocker on go-live", + task_notes="some notes", + outcome="preview", + ) + + def test_triage_dry_run_lists_per_email_qualify_reason_category_targetdate_task( + self, + ) -> None: + from click.testing import CliRunner + + from kontor_cli.cli import cli + from kontor_cli.config import Config + + mock_cfg = mock.MagicMock(spec=Config) + email = self._make_email() + decision = self._make_decision() + + with mock.patch("kontor_cli.cli.Config.load", return_value=mock_cfg): + with mock.patch("kontor_cli.cli.list_emails", return_value=[email]): + with mock.patch("kontor_cli.cli.Triage") as mock_triage_cls: + instance = mock_triage_cls.return_value + instance.maybe_create_task.return_value = decision + + runner = CliRunner() + result = runner.invoke( + cli, ["triage", "--dry-run"], catch_exceptions=False + ) + + assert result.exit_code == 0, result.output + assert "99" in result.output + assert "taking_decision" in result.output + assert "2026-06-30" in result.output + assert "[Taking Decision] Urgent: blocker on go-live" in result.output + # qualify indicator and reason + assert "y" in result.output + assert "external customer sender" in result.output + + def test_triage_dry_run_never_calls_asana_write(self) -> None: + from click.testing import CliRunner + + from kontor_cli.cli import cli + from kontor_cli.config import Config + + mock_cfg = mock.MagicMock(spec=Config) + email = self._make_email() + decision = self._make_decision() + + with mock.patch("kontor_cli.cli.Config.load", return_value=mock_cfg): + with mock.patch("kontor_cli.cli.list_emails", return_value=[email]): + with mock.patch("kontor_cli.cli.Triage") as mock_triage_cls: + instance = mock_triage_cls.return_value + instance.maybe_create_task.return_value = decision + + runner = CliRunner() + result = runner.invoke( + cli, ["triage", "--dry-run"], catch_exceptions=False + ) + + assert result.exit_code == 0, result.output + # maybe_create_task must be called with dry_run=True + instance.maybe_create_task.assert_called_once() + call_kwargs = instance.maybe_create_task.call_args + assert call_kwargs.kwargs.get("dry_run") is True or call_kwargs.args[2] is True + + # AsanaClient.create_task must never be called + with mock.patch( + "kontor_cli.asana_client.AsanaClient.create_task" + ) as mock_create: + mock_create.assert_not_called() + + def test_triage_no_dry_run_flag_is_rejected(self) -> None: + """`--no-dry-run` must NOT exist — the command is preview-only.""" + from click.testing import CliRunner + + from kontor_cli.cli import cli + + runner = CliRunner() + result = runner.invoke(cli, ["triage", "--no-dry-run"]) + # Unknown option → click usage error, exit code 2, no real run. + assert result.exit_code == 2 + assert ( + "no-dry-run" in result.output or "no such option" in result.output.lower() + ) + + def test_triage_always_calls_maybe_create_task_with_dry_run_true(self) -> None: + """Even invoked plainly, maybe_create_task is always dry_run=True.""" + from click.testing import CliRunner + + from kontor_cli.cli import cli + from kontor_cli.config import Config + + mock_cfg = mock.MagicMock(spec=Config) + email = self._make_email() + decision = self._make_decision() + + with mock.patch("kontor_cli.cli.Config.load", return_value=mock_cfg): + with mock.patch("kontor_cli.cli.list_emails", return_value=[email]): + with mock.patch("kontor_cli.cli.Triage") as mock_triage_cls: + instance = mock_triage_cls.return_value + instance.maybe_create_task.return_value = decision + + runner = CliRunner() + result = runner.invoke(cli, ["triage"], catch_exceptions=False) + + assert result.exit_code == 0, result.output + instance.maybe_create_task.assert_called_once() + call = instance.maybe_create_task.call_args + assert call.kwargs.get("dry_run") is True or call.args[2] is True + + def test_triage_config_error_exits_1(self) -> None: + from click.testing import CliRunner + + from kontor_cli.cli import cli + from kontor_cli.config import ConfigError + + with mock.patch( + "kontor_cli.cli.Config.load", side_effect=ConfigError("bad config") + ): + runner = CliRunner() + result = runner.invoke(cli, ["triage"], catch_exceptions=False) + + assert result.exit_code == 1 + assert "Config error" in result.output diff --git a/tests/unit/config_test.py b/tests/unit/config_test.py index 794a9d8..8d5c2c7 100644 --- a/tests/unit/config_test.py +++ b/tests/unit/config_test.py @@ -188,3 +188,143 @@ def test_check_davmail_ok(self, tmp_path: Path) -> None: mock_sock = mock.MagicMock() with mock.patch("socket.create_connection", return_value=mock_sock): cfg._check_davmail() # should not raise + + +class TestTriageAndAsanaConfig: + def test_loads_asana_and_triage_sections(self, tmp_path: Path) -> None: + import yaml + + cfg_file = tmp_path / "config.yaml" + data = _minimal_config() + data["asana"] = { + "pat": "REPLACE_WITH_ASANA_PAT", + "workspace_gid": "12345678", + "project_gids": { + "information_gathering": "111", + "nudging": "222", + "being_the_example": "333", + "taking_decision": "444", + }, + } + data["triage"] = { + "enabled": True, + "scan_rebuild": False, + "internal_domain": "rib-software.com", + "sender_tiers": { + "extremely_important": ["Rolf Helmes"], + "very_important": ["Arthur Berganski"], + "also_important": ["Helen Wiersma"], + }, + "content_high_threshold": 0.6, + } + yaml.safe_dump(data, open(cfg_file, "w")) + cfg = Config.load(cfg_file) + assert cfg.asana_pat == "REPLACE_WITH_ASANA_PAT" + assert cfg.asana_workspace_gid == "12345678" + assert cfg.asana_project_gids["information_gathering"] == "111" + assert cfg.asana_project_gids["nudging"] == "222" + assert cfg.asana_project_gids["being_the_example"] == "333" + assert cfg.asana_project_gids["taking_decision"] == "444" + assert cfg.triage_enabled is True + assert cfg.triage_scan_rebuild is False + assert cfg.triage_internal_domain == "rib-software.com" + assert cfg.triage_sender_tiers["extremely_important"] == ["Rolf Helmes"] + assert cfg.triage_content_high_threshold == 0.6 + + def test_triage_disabled_no_asana_section_loads_fine(self, tmp_path: Path) -> None: + import yaml + + cfg_file = tmp_path / "config.yaml" + data = _minimal_config() + # No asana section, triage.enabled is false (default) + yaml.safe_dump(data, open(cfg_file, "w")) + cfg = Config.load(cfg_file) + assert cfg.triage_enabled is False + assert cfg.asana_pat is None + assert cfg.asana_workspace_gid is None + assert cfg.asana_project_gids == {} + + def test_triage_enabled_missing_asana_section_raises_config_error_naming_key( + self, tmp_path: Path + ) -> None: + import yaml + + cfg_file = tmp_path / "config.yaml" + data = _minimal_config() + data["triage"] = {"enabled": True} + yaml.safe_dump(data, open(cfg_file, "w")) + with pytest.raises(ConfigError, match="asana"): + Config.load(cfg_file) + + def test_triage_enabled_missing_workspace_gid_or_pat_raises( + self, tmp_path: Path + ) -> None: + import yaml + + cfg_file = tmp_path / "config.yaml" + data = _minimal_config() + data["asana"] = { + "pat": "REPLACE_WITH_ASANA_PAT", + # workspace_gid is missing + "project_gids": { + "information_gathering": "111", + "nudging": "222", + "being_the_example": "333", + "taking_decision": "444", + }, + } + data["triage"] = {"enabled": True} + yaml.safe_dump(data, open(cfg_file, "w")) + with pytest.raises(ConfigError, match="asana.workspace_gid"): + Config.load(cfg_file) + + def test_triage_enabled_missing_pat_raises(self, tmp_path: Path) -> None: + import yaml + + cfg_file = tmp_path / "config.yaml" + data = _minimal_config() + data["asana"] = { + # pat is missing + "workspace_gid": "12345678", + "project_gids": { + "information_gathering": "111", + "nudging": "222", + "being_the_example": "333", + "taking_decision": "444", + }, + } + data["triage"] = {"enabled": True} + yaml.safe_dump(data, open(cfg_file, "w")) + with pytest.raises(ConfigError, match="asana.pat"): + Config.load(cfg_file) + + def test_triage_enabled_missing_any_of_4_project_gids_raises_naming_key( + self, tmp_path: Path + ) -> None: + import yaml + + missing_keys = [ + "information_gathering", + "nudging", + "being_the_example", + "taking_decision", + ] + all_gids = { + "information_gathering": "111", + "nudging": "222", + "being_the_example": "333", + "taking_decision": "444", + } + for missing_key in missing_keys: + cfg_file = tmp_path / f"config_{missing_key}.yaml" + data = _minimal_config() + gids = {k: v for k, v in all_gids.items() if k != missing_key} + data["asana"] = { + "pat": "REPLACE_WITH_ASANA_PAT", + "workspace_gid": "12345678", + "project_gids": gids, + } + data["triage"] = {"enabled": True} + yaml.safe_dump(data, open(cfg_file, "w")) + with pytest.raises(ConfigError, match=missing_key): + Config.load(cfg_file) diff --git a/tests/unit/coverage_gaps_test.py b/tests/unit/coverage_gaps_test.py index ab5af4b..33774e5 100644 --- a/tests/unit/coverage_gaps_test.py +++ b/tests/unit/coverage_gaps_test.py @@ -503,6 +503,9 @@ class MockConfig: llm_temperature = 0.0 llm_timeout = 30 pipeline_confidence_threshold = 0.7 + # Triage disabled: Pipeline.triage stays None for these gap tests. + triage_enabled = False + triage_scan_rebuild = False class TestPipelineGaps: diff --git a/tests/unit/himalaya_test.py b/tests/unit/himalaya_test.py index 0cfb438..307c291 100644 --- a/tests/unit/himalaya_test.py +++ b/tests/unit/himalaya_test.py @@ -19,6 +19,8 @@ list_emails, list_folders, move_email, + read_message_body, + read_message_id, ) SAMPLE_ENVELOPES = [ @@ -55,6 +57,30 @@ def test_email_from_json_basic(self) -> None: assert email.flags == {"seen": True} assert email.folder == "INBOX" + def test_email_from_json_captures_name_and_address(self) -> None: + # Real himalaya envelope shape carries both address AND name. + env = { + "id": "50", + "from": {"address": "rolf.helmes@rib-software.com", "name": "Rolf Helmes"}, + "subject": "Decision needed", + "date": "2024-06-15T09:00:00Z", + "flags": {}, + } + email = Email.from_json(env, "INBOX") + assert email.from_addr == "rolf.helmes@rib-software.com" + assert email.from_name == "Rolf Helmes" + + def test_email_from_json_name_defaults_empty_when_absent(self) -> None: + env = { + "id": "51", + "from": {"address": "x@y.com"}, + "subject": "No name", + "date": "2024-06-15T09:00:00Z", + "flags": {}, + } + email = Email.from_json(env, "INBOX") + assert email.from_name == "" + def test_email_from_json_supports_addr_field(self) -> None: env = { "id": "43", @@ -308,6 +334,64 @@ def test_delete_folder_command(self) -> None: ] +class TestReadMessageBody: + def test_read_message_body_invokes_exact_arg_vector(self) -> None: + with mock.patch("kontor_cli.himalaya._run", return_value="body text") as p: + read_message_body("42", folder="INBOX") + p.assert_called_once_with( + ["message", "read", "42", "-f", "INBOX", "--no-headers", "--preview"], + cwd=None, + ) + + def test_read_message_body_returns_plaintext(self) -> None: + expected = "Hello,\n\nThis is the body.\n" + with mock.patch("kontor_cli.himalaya._run", return_value=expected): + result = read_message_body("42") + assert result == expected + + def test_read_message_body_himalaya_error_propagates(self) -> None: + with mock.patch( + "kontor_cli.himalaya._run", + side_effect=HimalayaError("command failed: boom"), + ): + with pytest.raises(HimalayaError, match="command failed"): + read_message_body("42") + + +class TestReadMessageId: + def test_read_message_id_header_parsed(self) -> None: + # Case-insensitive; strips angle brackets + output = "Message-Id: \n\nBody text here.\n" + with mock.patch("kontor_cli.himalaya._run", return_value=output) as p: + result = read_message_id("42", folder="INBOX") + p.assert_called_once_with( + [ + "message", + "read", + "42", + "-f", + "INBOX", + "-H", + "Message-Id", + "--preview", + ], + cwd=None, + ) + assert result == "abc123@mail.example.com" + + def test_read_message_id_case_insensitive(self) -> None: + output = "message-id: \n\nBody.\n" + with mock.patch("kontor_cli.himalaya._run", return_value=output): + result = read_message_id("7") + assert result == "lower-case@example.com" + + def test_read_message_id_absent_returns_none(self) -> None: + output = "Subject: hello\n\nBody only, no message-id header.\n" + with mock.patch("kontor_cli.himalaya._run", return_value=output): + result = read_message_id("99") + assert result is None + + class TestRunErrors: def test_himalaya_command_failure(self) -> None: exc = subprocess.CalledProcessError( diff --git a/tests/unit/pipeline_test.py b/tests/unit/pipeline_test.py index 56cb654..e433653 100644 --- a/tests/unit/pipeline_test.py +++ b/tests/unit/pipeline_test.py @@ -35,6 +35,9 @@ class MockConfig: llm_temperature = 0.0 llm_timeout = 30 pipeline_confidence_threshold = 0.7 + # Triage defaults: disabled, so Pipeline.triage stays None for existing tests. + triage_enabled = False + triage_scan_rebuild = False class TestRebuildPipeline: @@ -347,6 +350,175 @@ def test_realtime_himalaya_error_returns_error_dict(self, tmp_path: Path) -> Non assert "error" in result +class TriageConfig(MockConfig): + """MockConfig variant with triage enabled.""" + + triage_enabled = True + triage_scan_rebuild = False + asana_pat = "pat-test" + asana_workspace_gid = "ws-test" + asana_project_gids = {"taking_decision": "proj-1"} + triage_sender_tiers: dict[str, list[str]] = {} + triage_internal_domain = "example.com" + triage_content_high_threshold = 0.7 + + +def _decision(outcome: str): + """Build a minimal TriageDecision-like stub with the given outcome.""" + return mock.MagicMock(outcome=outcome) + + +class TestTriageIntegration: + def _pipeline(self, tmp_path: Path, *, scan_rebuild: bool = False, cls=None): + """Construct a pipeline with a fake triage engine and stubbed classify.""" + from kontor_cli.pipeline import RealtimePipeline + + cls = cls or RealtimePipeline + cfg = TriageConfig() + cfg.triage_scan_rebuild = scan_rebuild + with mock.patch("kontor_cli.pipeline.Triage") as mock_triage_cls: + fake_triage = mock_triage_cls.return_value + fake_triage.asana = mock.MagicMock() + fake_triage.maybe_create_task.return_value = _decision("created") + p = cls(cfg, cwd=tmp_path) + p.rules_engine.classify = lambda e: "2_Projects/PRJ_Test" + p.rules_engine.get_nl_context = lambda: "" + return p + + def test_triage_fires_from_classification_even_when_move_email_raises( + self, tmp_path: Path + ) -> None: + p = self._pipeline(tmp_path) + email = _email("1", "INBOX") + err = HimalayaError("connection timeout") + with mock.patch("kontor_cli.pipeline.move_email", side_effect=err): + with mock.patch("kontor_cli.pipeline.create_folder"): + with mock.patch( + "kontor_cli.pipeline.read_message_body", return_value="b" + ): + p._process_email(email, dry_run=False, triage_scope=True) + + p.triage.maybe_create_task.assert_called_once() + assert p.triage_tasks_created == 1 + + def test_triage_not_invoked_when_triage_enabled_false(self, tmp_path: Path) -> None: + from kontor_cli.pipeline import RealtimePipeline + + p = RealtimePipeline(MockConfig(), cwd=tmp_path) + assert p.triage is None + p.rules_engine.classify = lambda e: "2_Projects/PRJ_Test" + p.rules_engine.get_nl_context = lambda: "" + email = _email("1", "INBOX") + with mock.patch("kontor_cli.pipeline.move_email"): + with mock.patch("kontor_cli.pipeline.create_folder"): + p._process_email(email, dry_run=False, triage_scope=True) + # No crash, counters stay zero. + assert p.triage_tasks_created == 0 + + def test_triage_dry_run_propagates(self, tmp_path: Path) -> None: + p = self._pipeline(tmp_path) + p.triage.maybe_create_task.return_value = _decision("preview") + email = _email("1", "INBOX") + with mock.patch("kontor_cli.pipeline.read_message_body", return_value="b"): + p._process_email(email, dry_run=True, triage_scope=True) + _, kwargs = p.triage.maybe_create_task.call_args + assert kwargs["dry_run"] is True + + def test_triage_skipped_in_rebuild_when_scan_rebuild_false( + self, tmp_path: Path + ) -> None: + from kontor_cli.pipeline import RebuildPipeline + + emails = [_email("1", "INBOX")] + p = self._pipeline(tmp_path, scan_rebuild=False, cls=RebuildPipeline) + with mock.patch("kontor_cli.pipeline.list_emails", return_value=emails): + with mock.patch("kontor_cli.pipeline.move_email"): + with mock.patch("kontor_cli.pipeline.create_folder"): + p.run(dry_run=False) + p.triage.maybe_create_task.assert_not_called() + + def test_triage_runs_in_rebuild_when_scan_rebuild_true( + self, tmp_path: Path + ) -> None: + from kontor_cli.pipeline import RebuildPipeline + + emails = [_email("1", "INBOX")] + p = self._pipeline(tmp_path, scan_rebuild=True, cls=RebuildPipeline) + with mock.patch("kontor_cli.pipeline.list_emails", return_value=emails): + with mock.patch("kontor_cli.pipeline.move_email"): + with mock.patch("kontor_cli.pipeline.create_folder"): + with mock.patch( + "kontor_cli.pipeline.read_message_body", return_value="b" + ): + p.run(dry_run=False) + p.triage.maybe_create_task.assert_called() + + def test_triage_exception_does_not_break_move_loop(self, tmp_path: Path) -> None: + emails = [_email("1", "INBOX"), _email("2", "INBOX")] + p = self._pipeline(tmp_path) + p.triage.maybe_create_task.side_effect = RuntimeError("triage bug") + with mock.patch("kontor_cli.pipeline.list_emails", return_value=emails): + with mock.patch("kontor_cli.pipeline.move_email") as mock_move: + with mock.patch("kontor_cli.pipeline.create_folder"): + with mock.patch( + "kontor_cli.pipeline.read_message_body", return_value="b" + ): + result = p.run(dry_run=False) + # Both emails still moved despite triage raising on each. + assert mock_move.call_count == 2 + assert p.triage_skipped_errors == 2 + assert result["phase"] == "realtime" + + def test_validate_projects_called_up_front_when_enabled_and_not_dry_run( + self, tmp_path: Path + ) -> None: + p = self._pipeline(tmp_path) + with mock.patch("kontor_cli.pipeline.list_emails", return_value=[]): + p.run(dry_run=False) + p.triage.asana.validate_projects.assert_called_once() + + def test_validate_projects_not_called_in_dry_run(self, tmp_path: Path) -> None: + p = self._pipeline(tmp_path) + with mock.patch("kontor_cli.pipeline.list_emails", return_value=[]): + p.run(dry_run=True) + p.triage.asana.validate_projects.assert_not_called() + + def test_validate_projects_error_aborts_run(self, tmp_path: Path) -> None: + from kontor_cli.asana_client import AsanaError + + p = self._pipeline(tmp_path) + p.triage.asana.validate_projects.side_effect = AsanaError("missing project") + with mock.patch("kontor_cli.pipeline.list_emails", return_value=[]): + try: + p.run(dry_run=False) + except AsanaError: + pass + else: + raise AssertionError("expected AsanaError to propagate") + + def test_triage_tally_by_outcome(self, tmp_path: Path) -> None: + p = self._pipeline(tmp_path) + email = _email("1", "INBOX") + with mock.patch("kontor_cli.pipeline.read_message_body", return_value="b"): + with mock.patch("kontor_cli.pipeline.move_email"): + with mock.patch("kontor_cli.pipeline.create_folder"): + p.triage.maybe_create_task.return_value = _decision("skipped_dedup") + p._process_email(email, dry_run=False, triage_scope=True) + p.triage.maybe_create_task.return_value = _decision("skipped_error") + p._process_email( + _email("2", "INBOX"), dry_run=False, triage_scope=True + ) + assert p.triage_skipped_dedup == 1 + assert p.triage_skipped_errors == 1 + + def test_summary_includes_triage_counters(self, tmp_path: Path) -> None: + p = self._pipeline(tmp_path) + s = p._summary("realtime", 0) + assert "triage_tasks_created" in s + assert "triage_skipped_dedup" in s + assert "triage_skipped_errors" in s + + class TestHealPipelineViolationPaths: def test_heal_pipeline_archive_violation_fixed(self, tmp_path: Path) -> None: from kontor_cli.pipeline import HealPipeline diff --git a/tests/unit/triage_test.py b/tests/unit/triage_test.py new file mode 100644 index 0000000..d437fe3 --- /dev/null +++ b/tests/unit/triage_test.py @@ -0,0 +1,492 @@ +"""Unit tests for kontor_cli.triage.""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime +from unittest import mock + +import pytest + +from kontor_cli.asana_client import AsanaError +from kontor_cli.config import Config +from kontor_cli.himalaya import Email +from kontor_cli.triage import ( + CATEGORY_TEMPLATES, + CategoryDecision, + Triage, + TriageDecision, +) + +INTERNAL = "rib-software.com" + + +def _make_config() -> Config: + cfg = mock.MagicMock(spec=Config) + cfg.asana_pat = "pat-test" + cfg.asana_workspace_gid = "ws-1" + cfg.asana_project_gids = { + "information_gathering": "gid-info", + "nudging": "gid-nudge", + "being_the_example": "gid-example", + "taking_decision": "gid-decide", + } + cfg.triage_internal_domain = INTERNAL + cfg.triage_content_high_threshold = 0.6 + cfg.triage_sender_tiers = { + "extremely_important": ["ceo@rib-software.com", "Big Boss"], + "very_important": ["vp@rib-software.com", "Decisive Dan"], + "also_important": ["lead@rib-software.com"], + } + cfg.llm_base_url = "https://llm.test/v1" + cfg.llm_api_key = "sk-test" + cfg.llm_model = "test-model" + cfg.llm_temperature = 0.0 + cfg.llm_timeout = 30 + return cfg + + +def _make_email( + email_id: str = "42", + from_addr: str = "someone@example.com", + subject: str = "Hello there", + date: datetime | None = None, + folder: str = "INBOX", + from_name: str = "", +) -> Email: + return Email( + id=email_id, + from_addr=from_addr, + from_name=from_name, + subject=subject, + date=date or datetime(2026, 6, 28, 9, 0, 0), + flags={}, + folder=folder, + ) + + +def _make_triage(cfg: Config | None = None) -> Triage: + return Triage(cfg or _make_config()) + + +def _llm_response(category: str, deadline: object, rationale: str = "r") -> mock.Mock: + payload = {"category": category, "deadline": deadline, "rationale": rationale} + result = mock.MagicMock() + result.json.return_value = { + "choices": [{"message": {"content": json.dumps(payload)}}] + } + result.raise_for_status = mock.MagicMock() + return result + + +# --------------------------------------------------------------------------- # +# Step 3 — scoring + gate +# --------------------------------------------------------------------------- # +class TestScoringGate: + def test_extremely_important_qualifies_on_actionable_content(self) -> None: + t = _make_triage() + email = _make_email(from_addr="ceo@rib-software.com", subject="urgent thing") + score = t.evaluate(email, "please decide soon") + assert score.qualifies + assert score.sender_component == 1.0 + + def test_very_important_sets_cat4_decisive_prior(self) -> None: + t = _make_triage() + email = _make_email(from_addr="vp@rib-software.com") + weight, decisive, in_tier = t.score_sender(email) + assert weight == 0.8 + assert decisive is True + assert in_tier is True + + def test_also_important_needs_strong_content(self) -> None: + t = _make_triage() + email = _make_email(from_addr="lead@rib-software.com", subject="fyi update") + weak = t.evaluate(email, "just a note") + assert not weak.qualifies + strong = t.evaluate(email, "urgent decision needed, please approve") + assert strong.qualifies + + def test_unlisted_lowsignal_does_not_qualify(self) -> None: + cfg = _make_config() + cfg.triage_internal_domain = "example.com" # make sender internal + t = _make_triage(cfg) + email = _make_email(from_addr="random@example.com", subject="lunch?") + score = t.evaluate(email, "want to grab lunch tomorrow") + assert not score.qualifies + + def test_external_domain_boosted(self) -> None: + t = _make_triage() + email = _make_email(from_addr="client@customer.com") + score = t.evaluate(email, "hi") + assert score.qualifies + assert score.customer_boost is True + + def test_internal_domain_not_auto_boosted(self) -> None: + t = _make_triage() + assert t.is_customer("colleague@rib-software.com") is False + assert t.is_customer("client@customer.com") is True + + def test_internal_colleague_asking_of_me_boosts(self) -> None: + cfg = _make_config() + t = _make_triage(cfg) + email = _make_email(from_addr="peer@rib-software.com") + # Internal sender, not in tiers — only content escalation can qualify. + score = t.evaluate(email, "Can you please advise on this blocker? urgent") + assert score.qualifies + assert score.escalation_boost is True + + def test_escalation_keyword_qualifies_regardless_of_tier(self) -> None: + cfg = _make_config() + t = _make_triage(cfg) + email = _make_email(from_addr="nobody@rib-software.com") + score = t.evaluate(email, "ESCALATION: hypercare go-live blocker") + assert score.qualifies + assert score.escalation_boost is True + + def test_qualifies_when_listed_OR_customer_OR_high_content(self) -> None: # noqa: N802 + t = _make_triage() + # customer alone + assert t.evaluate(_make_email(from_addr="a@customer.com"), "").qualifies + # listed + content + assert t.evaluate( + _make_email(from_addr="ceo@rib-software.com"), "please approve" + ).qualifies + + def test_name_and_address_matching(self) -> None: + t = _make_triage() + # Production shape: bare address in from_addr, display name in from_name. + # Match by display name (address does not contain the name). + by_name = _make_email( + from_addr="unknown.person@rib-software.com", from_name="Big Boss" + ) + _, _, in_tier_name = t.score_sender(by_name) + assert in_tier_name is True + # Match by address (no display name present at all). + by_addr = _make_email(from_addr="ceo@rib-software.com", from_name="") + weight, _, _ = t.score_sender(by_addr) + assert weight == 1.0 + + def test_named_sender_qualifies_extremely_important_tier(self) -> None: + # Regression: a listed sender configured by DISPLAY NAME must match the + # real envelope shape (bare address + separate display name) and land in + # the extremely_important tier (weight 1.0) end-to-end. + cfg = _make_config() + cfg.triage_sender_tiers = { + "extremely_important": ["Rolf Helmes"], + "very_important": [], + "also_important": [], + } + t = _make_triage(cfg) + email = _make_email( + from_addr="rolf.helmes@rib-software.com", + from_name="Rolf Helmes", + subject="please decide", + ) + weight, decisive, in_tier = t.score_sender(email) + assert in_tier is True + assert weight == 1.0 + assert decisive is False + score = t.evaluate(email, "please approve this decision") + assert score.qualifies + assert score.sender_component == 1.0 + + +# --------------------------------------------------------------------------- # +# Step 4 — fetch selector +# --------------------------------------------------------------------------- # +class TestFetchSelector: + def test_listed_sender_bypasses_subject_gate(self) -> None: + t = _make_triage() + email = _make_email(from_addr="ceo@rib-software.com", subject="nothing here") + assert t.should_fetch_body(email) is True + + def test_customer_bypasses_subject_gate(self) -> None: + t = _make_triage() + email = _make_email(from_addr="x@customer.com", subject="nothing here") + assert t.should_fetch_body(email) is True + + def test_unlisted_noncustomer_with_signal_subject_fetches(self) -> None: + cfg = _make_config() + t = _make_triage(cfg) + email = _make_email( + from_addr="peer@rib-software.com", subject="URGENT decision needed" + ) + assert t.should_fetch_body(email) is True + + def test_unlisted_noncustomer_lowsignal_declines_and_logs_debug( + self, caplog: pytest.LogCaptureFixture + ) -> None: + cfg = _make_config() + t = _make_triage(cfg) + email = _make_email( + email_id="99", from_addr="peer@rib-software.com", subject="lunch plans" + ) + # The "kontor_cli" parent logger may have propagate=False set by another + # test's configure_logging(); attach caplog's handler to the triage + # logger directly so capture is independent of parent state. + triage_logger = logging.getLogger("kontor_cli.triage") + triage_logger.addHandler(caplog.handler) + prev_level = triage_logger.level + triage_logger.setLevel(logging.DEBUG) + try: + with caplog.at_level(logging.DEBUG, logger="kontor_cli.triage"): + assert t.should_fetch_body(email) is False + finally: + triage_logger.removeHandler(caplog.handler) + triage_logger.setLevel(prev_level) + assert "declined 99" in caplog.text + + +# --------------------------------------------------------------------------- # +# Step 6 — LLM judgment + date +# --------------------------------------------------------------------------- # +class TestLLMJudgment: + def test_llm_assigns_exactly_one_of_4_categories(self) -> None: + t = _make_triage() + email = _make_email() + with mock.patch("httpx.post", return_value=_llm_response("nudging", None)): + decision = t.judge_category(email, "body", decisive_prior=False) + assert decision is not None + assert decision.category == "nudging" + assert decision.category in CATEGORY_TEMPLATES + + def test_decisive_prior_biases_toward_taking_decision(self) -> None: + t = _make_triage() + email = _make_email() + with mock.patch( + "httpx.post", return_value=_llm_response("taking_decision", None) + ) as post: + t.judge_category(email, "body", decisive_prior=True) + sent = post.call_args.kwargs["json"]["messages"][0]["content"] + assert "taking_decision" in sent + assert "decisive" in sent.lower() + + def test_llm_failure_returns_none(self) -> None: + import httpx + + t = _make_triage() + with mock.patch("httpx.post", side_effect=httpx.RequestError("boom")): + assert t.judge_category(_make_email(), "b", False) is None + + def test_llm_invalid_json_returns_none(self) -> None: + t = _make_triage() + bad = mock.MagicMock() + bad.json.return_value = {"choices": [{"message": {"content": "not json"}}]} + bad.raise_for_status = mock.MagicMock() + with mock.patch("httpx.post", return_value=bad): + assert t.judge_category(_make_email(), "b", False) is None + + def test_llm_category_not_in_4_returns_none(self) -> None: + t = _make_triage() + with mock.patch("httpx.post", return_value=_llm_response("nonsense", None)): + assert t.judge_category(_make_email(), "b", False) is None + + def test_date_llm_absolute_deadline_used(self) -> None: + t = _make_triage() + email = _make_email(date=datetime(2026, 6, 28)) + with mock.patch( + "httpx.post", return_value=_llm_response("nudging", "2026-07-15") + ): + decision = t.judge_category(email, "b", False) + assert decision is not None + assert t.resolve_target_date(decision, email) == "2026-07-15" + + def test_date_no_deadline_falls_back_to_email_date(self) -> None: + t = _make_triage() + email = _make_email(date=datetime(2026, 6, 28, 14, 0)) + decision = CategoryDecision(category="nudging", deadline=None, rationale="r") + assert t.resolve_target_date(decision, email) == "2026-06-28" + + def test_date_relative_phrase_parsed_via_dateutil_anchored_on_email_date( + self, + ) -> None: + t = _make_triage() + # email dated Sunday 2026-06-28; "Friday" anchored on that date. + email = _make_email(date=datetime(2026, 6, 28)) + with mock.patch( + "httpx.post", return_value=_llm_response("taking_decision", "Friday") + ): + decision = t.judge_category(email, "b", False) + assert decision is not None + assert decision.deadline is not None + # dateutil resolves "Friday" anchored on 2026-06-28 → 2026-07-03. + resolved = t.resolve_target_date(decision, email) + assert resolved == "2026-07-03" + + def test_date_unparseable_or_missing_email_date_raises(self) -> None: + t = _make_triage() + email = _make_email() + email.date = None # type: ignore[assignment] + decision = CategoryDecision(category="nudging", deadline=None, rationale="r") + with pytest.raises(ValueError): + t.resolve_target_date(decision, email) + + +# --------------------------------------------------------------------------- # +# Step 7 — orchestration +# --------------------------------------------------------------------------- # +class TestOrchestration: + def test_maybe_create_task_dry_run_returns_preview_no_write(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + email = _make_email(from_addr="ceo@rib-software.com", subject="please approve") + with mock.patch("httpx.post", return_value=_llm_response("nudging", None)): + result = t.maybe_create_task(email, lambda e: "body decide", dry_run=True) + assert result.outcome == "preview" + t.asana.create_task.assert_not_called() + assert result.task_name is not None + + def test_maybe_create_task_skips_when_not_qualified(self) -> None: + cfg = _make_config() + cfg.triage_internal_domain = "example.com" + t = _make_triage(cfg) + email = _make_email(from_addr="rand@example.com", subject="lunch") + result = t.maybe_create_task(email, lambda e: "let's eat", dry_run=False) + assert result.outcome == "not_qualified" + + def test_dedup_hit_returns_skipped_dedup(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + t.asana.find_task_by_marker.return_value = True + email = _make_email(from_addr="ceo@rib-software.com", subject="approve please") + with ( + mock.patch("httpx.post", return_value=_llm_response("nudging", None)), + mock.patch( + "kontor_cli.triage.himalaya.read_message_id", return_value="mid-1" + ), + ): + result = t.maybe_create_task(email, lambda e: "body", dry_run=False) + assert result.outcome == "skipped_dedup" + # scoped to the one target project gid + t.asana.find_task_by_marker.assert_called_once_with( + "gid-nudge", "kontor-id:mid-1" + ) + t.asana.create_task.assert_not_called() + + def test_marker_prefers_message_id_falls_back_to_uid(self) -> None: + t = _make_triage() + email = _make_email(email_id="uid-7") + with mock.patch( + "kontor_cli.triage.himalaya.read_message_id", return_value="msg-abc" + ): + assert t._resolve_marker(email) == "kontor-id:msg-abc" + with mock.patch( + "kontor_cli.triage.himalaya.read_message_id", return_value=None + ): + assert t._resolve_marker(email) == "kontor-id:uid-7" + with mock.patch( + "kontor_cli.triage.himalaya.read_message_id", + side_effect=RuntimeError("x"), + ): + assert t._resolve_marker(email) == "kontor-id:uid-7" + + def test_task_notes_contain_sender_date_rationale_standardtext_dod(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + t.asana.find_task_by_marker.return_value = False + email = _make_email( + from_addr="ceo@rib-software.com", + subject="approve", + date=datetime(2026, 6, 28), + ) + with ( + mock.patch( + "httpx.post", + return_value=_llm_response("nudging", None, rationale="needs a nudge"), + ), + mock.patch("kontor_cli.triage.himalaya.read_message_id", return_value="m1"), + ): + t.maybe_create_task(email, lambda e: "please approve", dry_run=False) + notes = t.asana.create_task.call_args.args[2] + assert "ceo@rib-software.com" in notes + assert "2026-06-28" in notes + assert "needs a nudge" in notes + assert CATEGORY_TEMPLATES["nudging"]["standard_text"] in notes + assert CATEGORY_TEMPLATES["nudging"]["done_when"] in notes + assert "kontor-id:m1" in notes + + def test_task_name_format_bracket_category(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + t.asana.find_task_by_marker.return_value = False + email = _make_email(from_addr="ceo@rib-software.com", subject="Quarterly sync") + with ( + mock.patch( + "httpx.post", return_value=_llm_response("taking_decision", None) + ), + mock.patch("kontor_cli.triage.himalaya.read_message_id", return_value="m1"), + ): + result = t.maybe_create_task(email, lambda e: "decide", dry_run=False) + assert result.task_name == "[Taking Decision] Quarterly sync" + assert result.outcome == "created" + + def test_asana_error_returns_skipped_error(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + t.asana.find_task_by_marker.return_value = False + t.asana.create_task.side_effect = AsanaError("500") + email = _make_email(from_addr="ceo@rib-software.com", subject="approve") + with ( + mock.patch("httpx.post", return_value=_llm_response("nudging", None)), + mock.patch("kontor_cli.triage.himalaya.read_message_id", return_value="m1"), + ): + result = t.maybe_create_task(email, lambda e: "b", dry_run=False) + assert result.outcome == "skipped_error" + + def test_llm_none_returns_skipped_error(self) -> None: + import httpx + + t = _make_triage() + email = _make_email(from_addr="ceo@rib-software.com", subject="approve") + with mock.patch("httpx.post", side_effect=httpx.RequestError("x")): + result = t.maybe_create_task(email, lambda e: "b", dry_run=False) + assert result.outcome == "skipped_error" + assert result.category is None + + def test_body_fetch_fail_returns_skipped_error(self) -> None: + t = _make_triage() + email = _make_email(from_addr="ceo@rib-software.com") + + def boom(_e: Email) -> str: + raise RuntimeError("fetch failed") + + result = t.maybe_create_task(email, boom, dry_run=False) + assert result.outcome == "skipped_error" + + def test_partial_run_resume_skips_already_created_via_marker(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + t.asana.find_task_by_marker.return_value = True + email = _make_email(from_addr="ceo@rib-software.com", subject="approve") + with ( + mock.patch("httpx.post", return_value=_llm_response("nudging", None)), + mock.patch("kontor_cli.triage.himalaya.read_message_id", return_value="m1"), + ): + result = t.maybe_create_task(email, lambda e: "b", dry_run=False) + assert result.outcome == "skipped_dedup" + t.asana.create_task.assert_not_called() + + def test_date_resolution_failure_returns_skipped_error(self) -> None: + t = _make_triage() + t.asana = mock.MagicMock() + email = _make_email(from_addr="ceo@rib-software.com", subject="approve") + email.date = None # type: ignore[assignment] + with mock.patch("httpx.post", return_value=_llm_response("nudging", None)): + result = t.maybe_create_task(email, lambda e: "b", dry_run=False) + assert result.outcome == "skipped_error" + assert result.target_date is None + + +def test_triage_decision_dataclass_shape() -> None: + d = TriageDecision( + email_id="1", + qualifies=True, + reason="r", + category="nudging", + target_date="2026-06-28", + task_name="[Nudging] x", + task_notes="notes", + outcome="created", + ) + assert d.outcome == "created"