From 2a7a200a27d567b132d783b828b24c21beddbf36 Mon Sep 17 00:00:00 2001
From: Andy Dalton <andalton@redhat.com>
Date: Thu, 16 Jul 2026 11:54:52 -0400
Subject: [PATCH 1/5] Add scan.py to offload triage Jira scan from MCP to REST
 API

Replace the AI-driven scan phase (10-20 MCP round-trips per run) with a
deterministic Python script that calls the Jira REST API directly. Saves
50-100K tokens per scan by eliminating mechanical pagination and
normalization work that required zero AI judgment.

The script uses key-based cursor pagination (Jira Cloud ignores startAt),
extracts plain text from ADF descriptions, normalizes issues to a flat
schema, and supports both API token (Basic) and PAT (Bearer) auth.
Includes retry with exponential backoff on 429/5xx.

74 unit and integration tests with fixture data, no live API calls.

Assisted-by: Claude Opus 4.6 (1M) <noreply@anthropic.com>
---
 .github/workflows/lint.yaml |   1 -
 .github/workflows/test.yaml |  24 +
 triage/README.md            |   6 +-
 triage/SKILL.md             |   2 +-
 triage/guidelines.md        |   2 +-
 triage/scripts/scan.py      | 437 +++++++++++++++++++
 triage/scripts/test_scan.py | 842 ++++++++++++++++++++++++++++++++++++
 triage/skills/scan.md       | 162 ++-----
 8 files changed, 1343 insertions(+), 133 deletions(-)
 create mode 100644 .github/workflows/test.yaml
 create mode 100644 triage/scripts/scan.py
 create mode 100644 triage/scripts/test_scan.py

diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml
index 13dbb86..94e97a4 100644
--- a/.github/workflows/lint.yaml
+++ b/.github/workflows/lint.yaml
@@ -21,7 +21,6 @@ jobs:
         with:
           python-version: "3.12"
       - run: python3 skill-reviewer/scripts/pre-review-checks.py --all --repo-root .
-      - run: python3 -m unittest discover -s _shared/scripts -p 'test_*.py' -v
 
   skillsaw:
     name: Skillsaw Lint
diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
new file mode 100644
index 0000000..2800af6
--- /dev/null
+++ b/.github/workflows/test.yaml
@@ -0,0 +1,24 @@
+name: Test
+
+on:
+  pull_request:
+    branches: [main]
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  unit-tests:
+    name: Unit Tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - run: python3 -m unittest discover -s _shared/scripts -p 'test_*.py' -v
+      - run: python3 -m unittest discover -s triage/scripts -p 'test_*.py' -v
diff --git a/triage/README.md b/triage/README.md
index d4d3ffc..7904df6 100644
--- a/triage/README.md
+++ b/triage/README.md
@@ -17,8 +17,12 @@ graph TD
 
 ## Prerequisites
 
-- **Jira MCP server** — the `user-mcp-jira` MCP server must be configured and authenticated
 - **Jira access** — the authenticated user must have read access to the target project
+- **Jira MCP server** — the `user-mcp-jira` MCP server must be configured and authenticated (used by `/start` and `/assess`)
+- **Environment variables for `/scan` (and `/run`, which includes scan)** — the scan phase calls the Jira REST API directly via `scripts/scan.py` instead of MCP:
+  - `JIRA_URL` — Jira instance base URL (e.g., `https://redhat.atlassian.net`)
+  - `JIRA_TOKEN` — API token or Personal Access Token
+  - `JIRA_EMAIL` — (optional) account email; set this when using an API token (Basic auth); omit for PATs (Bearer auth)
 
 ## Phases
 
diff --git a/triage/SKILL.md b/triage/SKILL.md
index c59552c..4c82e25 100644
--- a/triage/SKILL.md
+++ b/triage/SKILL.md
@@ -1,6 +1,6 @@
 ---
 name: triage
-version: 0.1.0
+version: 0.2.0
 description: >-
   Bulk-triage unresolved Jira bugs with AI-driven recommendations and an
   interactive HTML report. Scan also loads recently resolved bugs for regression
diff --git a/triage/guidelines.md b/triage/guidelines.md
index 4bad58b..a135ff6 100644
--- a/triage/guidelines.md
+++ b/triage/guidelines.md
@@ -37,7 +37,7 @@ Artifacts go in `.artifacts/triage/{project}/`.
 |---------|-------------------------------------|--------------------------------|
 | Run     | Per phase below                     | Per phase below                |
 | Start   | `jira_search`                       | `mkdir` (create artifact dir)  |
-| Scan    | `jira_search`                       | Write `issues.json` and `resolved.json` |
+| Scan    | none (script calls REST API directly) | Run `scripts/scan.py`; read stdout/stderr |
 | Analyze | none                                | Read `issues.json`, read `resolved.json` (if present), write `analyzed.json` |
 | Report  | none                                | Read `analyzed.json`, read `templates/report.html`, write `report.html` |
 | Assess (`/assess`) | `jira_search`              | Optionally read `issues.json`; no required artifact writes |
diff --git a/triage/scripts/scan.py b/triage/scripts/scan.py
new file mode 100644
index 0000000..6320846
--- /dev/null
+++ b/triage/scripts/scan.py
@@ -0,0 +1,437 @@
+#!/usr/bin/env python3
+"""Scan Jira for unresolved and recently resolved bugs.
+
+Fetches all unresolved bugs and recently resolved bugs from a Jira
+project using JQL with key-based cursor pagination, normalizes the
+data, and writes issues.json and resolved.json.
+
+Usage: scan.py PROJECT_KEY [--window-days 90] [--output-dir DIR]
+
+Environment:
+    JIRA_URL    Jira instance base URL (e.g., https://redhat.atlassian.net)
+    JIRA_TOKEN  API token or Personal Access Token
+    JIRA_EMAIL  (optional) account email — when set, uses Basic auth
+                (required for API tokens); when absent, uses Bearer auth
+
+Exit codes:
+    0 — scan completed successfully
+    1 — missing input or configuration error
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import ssl
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections import Counter
+from collections.abc import Callable
+from datetime import datetime, timezone
+from functools import partial
+from pathlib import Path
+from typing import Any
+
+PAGE_SIZE = 50
+MAX_RETRIES = 3
+RETRY_BACKOFF_BASE = 2
+
+UNRESOLVED_FIELDS = (
+    "summary,status,priority,assignee,reporter,"
+    "created,updated,labels,components,description"
+)
+RESOLVED_FIELDS = f"{UNRESOLVED_FIELDS},resolution,resolutiondate"
+
+_ADF_BLOCK_CONTAINERS = frozenset({
+    "doc", "bulletList", "orderedList", "listItem",
+    "blockquote", "table", "tableRow", "tableCell", "tableHeader",
+    "panel", "taskList", "decisionList",
+    "expand", "nestedExpand", "layoutSection", "layoutColumn",
+})
+
+SearchFn = Callable[[str, str, int], dict[str, Any]]
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+class ScanError(Exception):
+    """Base exception for scan errors."""
+
+
+class JiraAPIError(ScanError):
+    """Jira API returned an error after exhausting retries."""
+
+    def __init__(self, status_code: int, body: str) -> None:
+        self.status_code = status_code
+        self.body = body
+        super().__init__(f"Jira API {status_code}: {body[:500]}")
+
+
+# ---------------------------------------------------------------------------
+# HTTP layer
+# ---------------------------------------------------------------------------
+
+def _read_error_body(exc: urllib.error.HTTPError) -> str:
+    try:
+        return exc.read().decode("utf-8", errors="replace")[:500]
+    except Exception:
+        return ""
+
+
+def _retry_delay(headers: Any, attempt: int) -> float:
+    retry_after = headers.get("Retry-After")
+    if retry_after is not None:
+        try:
+            return max(1, int(retry_after))
+        except (ValueError, TypeError):
+            pass
+    return RETRY_BACKOFF_BASE ** attempt
+
+
+def _http_get(url: str, headers: dict[str, str]) -> bytes:
+    """HTTP GET with retry on transient failures (429, 5xx)."""
+    ctx = ssl.create_default_context()
+
+    for attempt in range(MAX_RETRIES + 1):
+        req = urllib.request.Request(url, headers=headers)
+        try:
+            with urllib.request.urlopen(req, context=ctx) as resp:
+                return resp.read()
+        except urllib.error.HTTPError as exc:
+            body = _read_error_body(exc)
+            if exc.code == 429 and attempt < MAX_RETRIES:
+                time.sleep(_retry_delay(exc.headers, attempt))
+                continue
+            if exc.code >= 500 and attempt < MAX_RETRIES:
+                time.sleep(RETRY_BACKOFF_BASE ** attempt)
+                continue
+            raise JiraAPIError(exc.code, body) from exc
+        except urllib.error.URLError as exc:
+            if attempt < MAX_RETRIES:
+                time.sleep(RETRY_BACKOFF_BASE ** attempt)
+                continue
+            raise ScanError(f"Cannot reach Jira: {exc.reason}") from exc
+
+    # defensive: every loop iteration returns or raises, but this
+    # satisfies the type checker and guards against future refactors
+    raise ScanError(f"Retries exhausted after {MAX_RETRIES + 1} attempts")
+
+
+def build_auth_header(token: str, email: str | None = None) -> str:
+    """Build the Authorization header value.
+
+    API tokens (JIRA_EMAIL set) use Basic auth; PATs use Bearer.
+    """
+    if email:
+        credentials = base64.b64encode(
+            f"{email}:{token}".encode("utf-8"),
+        ).decode("ascii")
+        return f"Basic {credentials}"
+    return f"Bearer {token}"
+
+
+def jira_search(
+    base_url: str,
+    auth_header: str,
+    jql: str,
+    fields: str,
+    max_results: int = PAGE_SIZE,
+) -> dict[str, Any]:
+    """Execute a single JQL search against the Jira REST API."""
+    params = urllib.parse.urlencode({
+        "jql": jql,
+        "fields": fields,
+        "maxResults": max_results,
+    })
+    url = f"{base_url.rstrip('/')}/rest/api/3/search/jql?{params}"
+    headers = {
+        "Authorization": auth_header,
+        "Accept": "application/json",
+    }
+    raw = _http_get(url, headers)
+    return json.loads(raw.decode("utf-8"))
+
+
+# ---------------------------------------------------------------------------
+# Pagination
+# ---------------------------------------------------------------------------
+
+def fetch_all_issues(
+    search_fn: SearchFn,
+    jql_base: str,
+    fields: str,
+) -> list[dict[str, Any]]:
+    """Fetch all issues matching a JQL filter using key-based cursor pagination.
+
+    ``search_fn(jql, fields, max_results)`` is called repeatedly with
+    ``AND key > '{last_key}' ORDER BY key ASC`` appended to the base JQL
+    until a page returns fewer than PAGE_SIZE results.
+    """
+    all_issues: list[dict[str, Any]] = []
+    last_key = ""
+
+    while True:
+        jql = jql_base
+        if last_key:
+            jql += f" AND key > '{last_key}'"
+        jql += " ORDER BY key ASC"
+
+        data = search_fn(jql, fields, PAGE_SIZE)
+        page = data.get("issues", [])
+        if not page:
+            break
+
+        all_issues.extend(page)
+        last_key = page[-1]["key"]
+
+        if len(page) < PAGE_SIZE:
+            break
+
+    seen: set[str] = set()
+    deduped: list[dict[str, Any]] = []
+    for issue in all_issues:
+        key = issue["key"]
+        if key not in seen:
+            seen.add(key)
+            deduped.append(issue)
+    return deduped
+
+
+# ---------------------------------------------------------------------------
+# ADF text extraction
+# ---------------------------------------------------------------------------
+
+def extract_text(value: Any) -> str:
+    """Extract plain text from a value that may be a string or ADF JSON.
+
+    Handles Jira's Atlassian Document Format by recursively walking
+    content nodes and concatenating text leaves, with newlines between
+    block-level elements.
+    """
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, dict):
+        node_type = value.get("type")
+        if node_type == "text":
+            return value.get("text", "")
+        parts = [extract_text(c) for c in value.get("content", [])]
+        if node_type in _ADF_BLOCK_CONTAINERS:
+            return "\n".join(p for p in parts if p)
+        return "".join(parts)
+    if isinstance(value, list):
+        return "\n".join(p for p in (extract_text(item) for item in value) if p)
+    return ""
+
+
+# ---------------------------------------------------------------------------
+# Normalization
+# ---------------------------------------------------------------------------
+
+def _name_or_default(field: Any, key: str = "name", default: str = "") -> str:
+    """Extract a named attribute from a Jira object field, or return default."""
+    if isinstance(field, dict):
+        return field.get(key, default)
+    return default
+
+
+def normalize_issue(
+    raw: dict[str, Any],
+    *,
+    include_resolution: bool = False,
+) -> dict[str, Any]:
+    """Normalize a raw Jira API issue into a flat dictionary."""
+    fields = raw.get("fields", {})
+
+    normalized: dict[str, Any] = {
+        "key": raw.get("key", ""),
+        "summary": fields.get("summary", ""),
+        "status": _name_or_default(fields.get("status")),
+        "priority": _name_or_default(fields.get("priority")),
+        "assignee": _name_or_default(
+            fields.get("assignee"), "displayName", "Unassigned",
+        ),
+        "reporter": _name_or_default(fields.get("reporter"), "displayName"),
+        "created": fields.get("created", ""),
+        "updated": fields.get("updated", ""),
+        "labels": fields.get("labels", []),
+        "components": [
+            _name_or_default(c)
+            for c in fields.get("components", [])
+            if isinstance(c, dict)
+        ],
+        "description": extract_text(fields.get("description")),
+    }
+
+    if include_resolution:
+        normalized["resolution"] = _name_or_default(fields.get("resolution"))
+        normalized["resolved"] = fields.get("resolutiondate", "")
+
+    return normalized
+
+
+# ---------------------------------------------------------------------------
+# Output
+# ---------------------------------------------------------------------------
+
+def build_summary(issues: list[dict[str, Any]]) -> str:
+    """Build a human-readable summary grouped by priority and status."""
+    priority_counts: Counter[str] = Counter()
+    status_counts: Counter[str] = Counter()
+
+    for issue in issues:
+        priority_counts[issue.get("priority") or "Unknown"] += 1
+        status_counts[issue.get("status") or "Unknown"] += 1
+
+    lines: list[str] = []
+    if priority_counts:
+        lines.append("By priority:")
+        for name, count in priority_counts.most_common():
+            lines.append(f"  {name}: {count}")
+    if status_counts:
+        lines.append("By status:")
+        for name, count in status_counts.most_common():
+            lines.append(f"  {name}: {count}")
+    return "\n".join(lines)
+
+
+def build_output(
+    project: str,
+    jira_url: str,
+    issues: list[dict[str, Any]],
+    scanned_at: str,
+    *,
+    window_days: int | None = None,
+) -> dict[str, Any]:
+    """Assemble the output JSON structure (pure — no I/O or clock)."""
+    data: dict[str, Any] = {
+        "project": project,
+        "jiraBaseUrl": jira_url.rstrip("/"),
+        "scannedAt": scanned_at,
+        "totalCount": len(issues),
+        "issues": issues,
+    }
+    if window_days is not None:
+        data["windowDays"] = window_days
+    return data
+
+
+def write_json_file(path: Path, data: dict[str, Any]) -> None:
+    """Write a JSON object to a file, creating parent directories."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Scan Jira for unresolved and recently resolved bugs.",
+    )
+    parser.add_argument(
+        "project",
+        help="Jira project key (e.g., EDM)",
+    )
+    parser.add_argument(
+        "--window-days",
+        type=int,
+        default=90,
+        help="Number of days to look back for resolved bugs (default: 90)",
+    )
+    parser.add_argument(
+        "--output-dir",
+        type=Path,
+        default=None,
+        help="Output directory (default: .artifacts/triage/{PROJECT})",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    project = args.project
+    window_days = args.window_days
+    output_dir = args.output_dir or Path(f".artifacts/triage/{project}")
+
+    jira_url = os.environ.get("JIRA_URL", "").strip()
+    jira_token = os.environ.get("JIRA_TOKEN", "").strip()
+    jira_email = os.environ.get("JIRA_EMAIL", "").strip() or None
+    if not jira_url:
+        print("Error: JIRA_URL environment variable is not set", file=sys.stderr)
+        return 1
+    if not jira_token:
+        print("Error: JIRA_TOKEN environment variable is not set", file=sys.stderr)
+        return 1
+
+    auth_header = build_auth_header(jira_token, jira_email)
+    search = partial(jira_search, jira_url, auth_header)
+
+    try:
+        unresolved_jql = (
+            f"project = {project} AND issuetype = Bug "
+            f"AND resolution = Unresolved"
+        )
+        raw_unresolved = fetch_all_issues(
+            search, unresolved_jql, UNRESOLVED_FIELDS,
+        )
+        unresolved = [normalize_issue(r) for r in raw_unresolved]
+
+        resolved_jql = (
+            f"project = {project} AND issuetype = Bug "
+            f"AND resolution != Unresolved "
+            f"AND resolved >= -{window_days}d"
+        )
+        raw_resolved = fetch_all_issues(
+            search, resolved_jql, RESOLVED_FIELDS,
+        )
+        resolved = [
+            normalize_issue(r, include_resolution=True) for r in raw_resolved
+        ]
+    except ScanError as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return 1
+
+    scanned_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+    issues_path = output_dir / "issues.json"
+    resolved_path = output_dir / "resolved.json"
+
+    write_json_file(
+        issues_path,
+        build_output(project, jira_url, unresolved, scanned_at),
+    )
+    write_json_file(
+        resolved_path,
+        build_output(
+            project, jira_url, resolved, scanned_at,
+            window_days=window_days,
+        ),
+    )
+
+    print(f"Scan complete: {len(unresolved)} unresolved bugs in {project}")
+    print(f"Resolved (last {window_days} days): {len(resolved)} bugs")
+    print()
+
+    if unresolved:
+        print(build_summary(unresolved))
+        print()
+
+    print("Data saved to:")
+    print(f"  {issues_path}")
+    print(f"  {resolved_path}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/triage/scripts/test_scan.py b/triage/scripts/test_scan.py
new file mode 100644
index 0000000..ad5e340
--- /dev/null
+++ b/triage/scripts/test_scan.py
@@ -0,0 +1,842 @@
+#!/usr/bin/env python3
+"""Tests for triage/scripts/scan.py."""
+
+from __future__ import annotations
+
+import base64
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+_SCRIPT = Path(__file__).resolve().parent / "scan.py"
+_spec = importlib.util.spec_from_file_location("scan", _SCRIPT)
+assert _spec and _spec.loader
+scan = importlib.util.module_from_spec(_spec)
+sys.modules["scan"] = scan
+_spec.loader.exec_module(scan)
+
+
+# ---------------------------------------------------------------------------
+# Jira API response fixtures
+# ---------------------------------------------------------------------------
+
+def _raw_issue(
+    key: str,
+    *,
+    summary: str = "Bug summary",
+    status: str = "Open",
+    priority: str = "High",
+    assignee: str | None = "Alice",
+    reporter: str = "Bob",
+    created: str = "2026-01-15T10:00:00.000+0000",
+    updated: str = "2026-01-16T10:00:00.000+0000",
+    labels: list[str] | None = None,
+    components: list[str] | None = None,
+    description: str | dict | None = "Description text",
+    resolution: str | None = None,
+    resolutiondate: str | None = None,
+) -> dict:
+    """Build a raw Jira API issue object."""
+    fields: dict = {
+        "summary": summary,
+        "status": {"name": status},
+        "priority": {"name": priority},
+        "assignee": {"displayName": assignee} if assignee else None,
+        "reporter": {"displayName": reporter},
+        "created": created,
+        "updated": updated,
+        "labels": labels or [],
+        "components": [{"name": c} for c in (components or [])],
+        "description": description,
+    }
+    if resolution is not None:
+        fields["resolution"] = {"name": resolution}
+    if resolutiondate is not None:
+        fields["resolutiondate"] = resolutiondate
+    return {"key": key, "fields": fields}
+
+
+def _search_response(issues: list[dict]) -> dict:
+    """Wrap issues in a Jira search response envelope."""
+    return {"issues": issues, "total": len(issues)}
+
+
+# ---------------------------------------------------------------------------
+# extract_text — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestExtractText(unittest.TestCase):
+
+    def test_none_returns_empty(self) -> None:
+        self.assertEqual(scan.extract_text(None), "")
+
+    def test_plain_string_passthrough(self) -> None:
+        self.assertEqual(scan.extract_text("hello world"), "hello world")
+
+    def test_empty_string(self) -> None:
+        self.assertEqual(scan.extract_text(""), "")
+
+    def test_adf_text_node(self) -> None:
+        node = {"type": "text", "text": "inline text"}
+        self.assertEqual(scan.extract_text(node), "inline text")
+
+    def test_adf_text_node_missing_text_key(self) -> None:
+        node = {"type": "text"}
+        self.assertEqual(scan.extract_text(node), "")
+
+    def test_adf_paragraph(self) -> None:
+        doc = {
+            "type": "paragraph",
+            "content": [
+                {"type": "text", "text": "first "},
+                {"type": "text", "text": "second"},
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "first second")
+
+    def test_adf_document_with_paragraphs(self) -> None:
+        doc = {
+            "type": "doc",
+            "content": [
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "Line one"}],
+                },
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "Line two"}],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "Line one\nLine two")
+
+    def test_adf_heading(self) -> None:
+        doc = {
+            "type": "doc",
+            "content": [
+                {
+                    "type": "heading",
+                    "attrs": {"level": 1},
+                    "content": [{"type": "text", "text": "Title"}],
+                },
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "Body"}],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "Title\nBody")
+
+    def test_adf_bullet_list(self) -> None:
+        doc = {
+            "type": "bulletList",
+            "content": [
+                {
+                    "type": "listItem",
+                    "content": [
+                        {
+                            "type": "paragraph",
+                            "content": [{"type": "text", "text": "item one"}],
+                        },
+                    ],
+                },
+                {
+                    "type": "listItem",
+                    "content": [
+                        {
+                            "type": "paragraph",
+                            "content": [{"type": "text", "text": "item two"}],
+                        },
+                    ],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "item one\nitem two")
+
+    def test_adf_inline_marks_ignored(self) -> None:
+        node = {
+            "type": "text",
+            "text": "bold text",
+            "marks": [{"type": "strong"}],
+        }
+        self.assertEqual(scan.extract_text(node), "bold text")
+
+    def test_adf_empty_content(self) -> None:
+        doc = {"type": "doc", "content": []}
+        self.assertEqual(scan.extract_text(doc), "")
+
+    def test_adf_unknown_inline_type(self) -> None:
+        node = {
+            "type": "emoji",
+            "attrs": {"shortName": ":smile:"},
+        }
+        self.assertEqual(scan.extract_text(node), "")
+
+    def test_list_of_values(self) -> None:
+        values = ["first", "second", "third"]
+        self.assertEqual(scan.extract_text(values), "first\nsecond\nthird")
+
+    def test_numeric_value(self) -> None:
+        self.assertEqual(scan.extract_text(42), "")
+
+    def test_adf_expand(self) -> None:
+        doc = {
+            "type": "expand",
+            "attrs": {"title": "Details"},
+            "content": [
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "Hidden content"}],
+                },
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "More hidden"}],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "Hidden content\nMore hidden")
+
+    def test_adf_nested_expand(self) -> None:
+        doc = {
+            "type": "nestedExpand",
+            "content": [
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "Nested"}],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "Nested")
+
+    def test_adf_layout_section(self) -> None:
+        doc = {
+            "type": "layoutSection",
+            "content": [
+                {
+                    "type": "layoutColumn",
+                    "content": [
+                        {
+                            "type": "paragraph",
+                            "content": [{"type": "text", "text": "Col 1"}],
+                        },
+                    ],
+                },
+                {
+                    "type": "layoutColumn",
+                    "content": [
+                        {
+                            "type": "paragraph",
+                            "content": [{"type": "text", "text": "Col 2"}],
+                        },
+                    ],
+                },
+            ],
+        }
+        self.assertEqual(scan.extract_text(doc), "Col 1\nCol 2")
+
+    def test_nested_codeblock(self) -> None:
+        doc = {
+            "type": "codeBlock",
+            "attrs": {"language": "python"},
+            "content": [{"type": "text", "text": "print('hi')"}],
+        }
+        self.assertEqual(scan.extract_text(doc), "print('hi')")
+
+
+# ---------------------------------------------------------------------------
+# _name_or_default — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestNameOrDefault(unittest.TestCase):
+
+    def test_dict_with_key(self) -> None:
+        self.assertEqual(scan._name_or_default({"name": "High"}), "High")
+
+    def test_dict_missing_key(self) -> None:
+        self.assertEqual(scan._name_or_default({"id": "1"}), "")
+
+    def test_dict_custom_key(self) -> None:
+        self.assertEqual(
+            scan._name_or_default({"displayName": "Alice"}, "displayName"),
+            "Alice",
+        )
+
+    def test_dict_custom_default(self) -> None:
+        self.assertEqual(
+            scan._name_or_default({}, "displayName", "Unassigned"),
+            "Unassigned",
+        )
+
+    def test_none_returns_default(self) -> None:
+        self.assertEqual(scan._name_or_default(None), "")
+
+    def test_none_returns_custom_default(self) -> None:
+        self.assertEqual(
+            scan._name_or_default(None, "displayName", "Unassigned"),
+            "Unassigned",
+        )
+
+    def test_string_returns_default(self) -> None:
+        self.assertEqual(scan._name_or_default("not a dict"), "")
+
+    def test_list_returns_default(self) -> None:
+        self.assertEqual(scan._name_or_default([1, 2]), "")
+
+
+# ---------------------------------------------------------------------------
+# normalize_issue — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestNormalizeIssue(unittest.TestCase):
+
+    def test_full_unresolved_issue(self) -> None:
+        raw = _raw_issue(
+            "EDM-101",
+            summary="Login fails",
+            status="Open",
+            priority="Critical",
+            assignee="Alice",
+            reporter="Bob",
+            labels=["regression", "auth"],
+            components=["Backend", "API"],
+            description="Login page returns 500",
+        )
+        result = scan.normalize_issue(raw)
+        self.assertEqual(result["key"], "EDM-101")
+        self.assertEqual(result["summary"], "Login fails")
+        self.assertEqual(result["status"], "Open")
+        self.assertEqual(result["priority"], "Critical")
+        self.assertEqual(result["assignee"], "Alice")
+        self.assertEqual(result["reporter"], "Bob")
+        self.assertEqual(result["labels"], ["regression", "auth"])
+        self.assertEqual(result["components"], ["Backend", "API"])
+        self.assertEqual(result["description"], "Login page returns 500")
+        self.assertNotIn("resolution", result)
+        self.assertNotIn("resolved", result)
+
+    def test_unassigned(self) -> None:
+        raw = _raw_issue("EDM-102", assignee=None)
+        result = scan.normalize_issue(raw)
+        self.assertEqual(result["assignee"], "Unassigned")
+
+    def test_resolved_issue_with_resolution_fields(self) -> None:
+        raw = _raw_issue(
+            "EDM-103",
+            resolution="Fixed",
+            resolutiondate="2026-01-20T14:00:00.000+0000",
+        )
+        result = scan.normalize_issue(raw, include_resolution=True)
+        self.assertEqual(result["resolution"], "Fixed")
+        self.assertEqual(result["resolved"], "2026-01-20T14:00:00.000+0000")
+
+    def test_resolved_issue_without_resolution_date(self) -> None:
+        raw = _raw_issue("EDM-104", resolution="Won't Fix")
+        result = scan.normalize_issue(raw, include_resolution=True)
+        self.assertEqual(result["resolution"], "Won't Fix")
+        self.assertEqual(result["resolved"], "")
+
+    def test_missing_fields_produce_safe_defaults(self) -> None:
+        raw = {"key": "EDM-105", "fields": {}}
+        result = scan.normalize_issue(raw)
+        self.assertEqual(result["key"], "EDM-105")
+        self.assertEqual(result["summary"], "")
+        self.assertEqual(result["status"], "")
+        self.assertEqual(result["priority"], "")
+        self.assertEqual(result["assignee"], "Unassigned")
+        self.assertEqual(result["reporter"], "")
+        self.assertEqual(result["labels"], [])
+        self.assertEqual(result["components"], [])
+        self.assertEqual(result["description"], "")
+
+    def test_completely_empty_raw(self) -> None:
+        result = scan.normalize_issue({})
+        self.assertEqual(result["key"], "")
+
+    def test_adf_description_extracted(self) -> None:
+        adf = {
+            "type": "doc",
+            "content": [
+                {
+                    "type": "paragraph",
+                    "content": [{"type": "text", "text": "ADF content"}],
+                },
+            ],
+        }
+        raw = _raw_issue("EDM-106", description=adf)
+        result = scan.normalize_issue(raw)
+        self.assertEqual(result["description"], "ADF content")
+
+    def test_components_with_non_dict_entries_filtered(self) -> None:
+        raw = {"key": "EDM-107", "fields": {
+            "components": [{"name": "UI"}, "stray string", {"name": "API"}],
+        }}
+        result = scan.normalize_issue(raw)
+        self.assertEqual(result["components"], ["UI", "API"])
+
+
+# ---------------------------------------------------------------------------
+# build_summary — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestBuildSummary(unittest.TestCase):
+
+    def test_grouped_counts(self) -> None:
+        issues = [
+            {"priority": "High", "status": "Open"},
+            {"priority": "High", "status": "Open"},
+            {"priority": "Low", "status": "In Progress"},
+        ]
+        summary = scan.build_summary(issues)
+        self.assertIn("High: 2", summary)
+        self.assertIn("Low: 1", summary)
+        self.assertIn("Open: 2", summary)
+        self.assertIn("In Progress: 1", summary)
+
+    def test_empty_list(self) -> None:
+        self.assertEqual(scan.build_summary([]), "")
+
+    def test_missing_fields_counted_as_unknown(self) -> None:
+        issues = [{"priority": "", "status": None}]
+        summary = scan.build_summary(issues)
+        self.assertIn("Unknown: 1", summary)
+
+    def test_none_priority_counted_as_unknown(self) -> None:
+        issues = [{"priority": None, "status": "Open"}]
+        summary = scan.build_summary(issues)
+        self.assertIn("Unknown: 1", summary)
+
+
+# ---------------------------------------------------------------------------
+# build_output — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestBuildOutput(unittest.TestCase):
+
+    def test_basic_structure(self) -> None:
+        issues = [{"key": "EDM-1"}, {"key": "EDM-2"}]
+        result = scan.build_output(
+            "EDM", "https://jira.example.com", issues,
+            "2026-03-19T12:00:00Z",
+        )
+        self.assertEqual(result["project"], "EDM")
+        self.assertEqual(result["jiraBaseUrl"], "https://jira.example.com")
+        self.assertEqual(result["scannedAt"], "2026-03-19T12:00:00Z")
+        self.assertEqual(result["totalCount"], 2)
+        self.assertEqual(result["issues"], issues)
+        self.assertNotIn("windowDays", result)
+
+    def test_trailing_slash_stripped(self) -> None:
+        result = scan.build_output(
+            "X", "https://jira.example.com/", [],
+            "2026-01-01T00:00:00Z",
+        )
+        self.assertEqual(result["jiraBaseUrl"], "https://jira.example.com")
+
+    def test_with_window_days(self) -> None:
+        result = scan.build_output(
+            "X", "https://jira.example.com", [],
+            "2026-01-01T00:00:00Z", window_days=90,
+        )
+        self.assertEqual(result["windowDays"], 90)
+
+    def test_empty_issues(self) -> None:
+        result = scan.build_output(
+            "X", "https://jira.example.com", [],
+            "2026-01-01T00:00:00Z",
+        )
+        self.assertEqual(result["totalCount"], 0)
+        self.assertEqual(result["issues"], [])
+
+
+# ---------------------------------------------------------------------------
+# write_json_file — I/O tests
+# ---------------------------------------------------------------------------
+
+class TestWriteJsonFile(unittest.TestCase):
+
+    def test_writes_valid_json(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "output.json"
+            data = {"key": "value", "list": [1, 2, 3]}
+            scan.write_json_file(path, data)
+
+            content = path.read_text(encoding="utf-8")
+            self.assertTrue(content.endswith("\n"))
+            loaded = json.loads(content)
+            self.assertEqual(loaded, data)
+
+    def test_creates_parent_directories(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "deep" / "nested" / "output.json"
+            scan.write_json_file(path, {"ok": True})
+            self.assertTrue(path.is_file())
+
+    def test_pretty_printed(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "output.json"
+            scan.write_json_file(path, {"a": 1})
+            content = path.read_text(encoding="utf-8")
+            self.assertIn("\n", content.rstrip("\n"))
+
+
+# ---------------------------------------------------------------------------
+# fetch_all_issues — pagination tests with fake search function
+# ---------------------------------------------------------------------------
+
+def _fake_search(pages: list[list[dict]]):
+    """Return a search function that yields pages in order.
+
+    The returned callable also records each JQL it received, enabling
+    assertions on the cursor progression.
+    """
+    call_log: list[str] = []
+    page_iter = iter(pages)
+
+    def search(jql: str, fields: str, max_results: int = scan.PAGE_SIZE) -> dict:
+        call_log.append(jql)
+        page = next(page_iter, [])
+        return _search_response(page)
+
+    search.call_log = call_log  # type: ignore[attr-defined]
+    return search
+
+
+class TestFetchAllIssues(unittest.TestCase):
+
+    def test_single_partial_page(self) -> None:
+        issues = [_raw_issue(f"EDM-{i}") for i in range(1, 4)]
+        search = _fake_search([issues])
+        result = scan.fetch_all_issues(search, "project = EDM", "summary")
+        self.assertEqual(len(result), 3)
+        self.assertEqual(result[0]["key"], "EDM-1")
+
+    def test_multiple_full_pages(self) -> None:
+        page1 = [_raw_issue(f"EDM-{i}") for i in range(1, scan.PAGE_SIZE + 1)]
+        page2 = [_raw_issue(f"EDM-{scan.PAGE_SIZE + i}") for i in range(1, 4)]
+        search = _fake_search([page1, page2])
+        result = scan.fetch_all_issues(search, "project = EDM", "summary")
+        self.assertEqual(len(result), scan.PAGE_SIZE + 3)
+
+    def test_cursor_appears_in_jql(self) -> None:
+        page1 = [_raw_issue(f"EDM-{i}") for i in range(1, scan.PAGE_SIZE + 1)]
+        page2 = [_raw_issue(f"EDM-{scan.PAGE_SIZE + 1}")]
+        search = _fake_search([page1, page2])
+        scan.fetch_all_issues(search, "project = EDM", "summary")
+
+        self.assertEqual(len(search.call_log), 2)
+        self.assertNotIn("key >", search.call_log[0])
+        self.assertIn(f"key > 'EDM-{scan.PAGE_SIZE}'", search.call_log[1])
+
+    def test_order_by_appended(self) -> None:
+        search = _fake_search([[_raw_issue("EDM-1")]])
+        scan.fetch_all_issues(search, "project = EDM", "summary")
+        self.assertTrue(search.call_log[0].endswith("ORDER BY key ASC"))
+
+    def test_empty_result(self) -> None:
+        search = _fake_search([[]])
+        result = scan.fetch_all_issues(search, "project = EDM", "summary")
+        self.assertEqual(result, [])
+
+    def test_exact_page_boundary(self) -> None:
+        page = [_raw_issue(f"EDM-{i}") for i in range(1, scan.PAGE_SIZE + 1)]
+        search = _fake_search([page, []])
+        result = scan.fetch_all_issues(search, "project = EDM", "summary")
+        self.assertEqual(len(result), scan.PAGE_SIZE)
+        self.assertEqual(len(search.call_log), 2)
+
+    def test_deduplication(self) -> None:
+        issues = [
+            _raw_issue("EDM-1"),
+            _raw_issue("EDM-2"),
+            _raw_issue("EDM-1"),
+        ]
+        search = _fake_search([issues])
+        result = scan.fetch_all_issues(search, "project = EDM", "summary")
+        keys = [r["key"] for r in result]
+        self.assertEqual(keys, ["EDM-1", "EDM-2"])
+
+
+# ---------------------------------------------------------------------------
+# main — integration tests
+# ---------------------------------------------------------------------------
+
+class TestMain(unittest.TestCase):
+
+    _ENV = {
+        "JIRA_URL": "https://jira.example.com",
+        "JIRA_TOKEN": "test-token",
+    }
+
+    def setUp(self) -> None:
+        self._tmpdir = tempfile.mkdtemp()
+        self._output_dir = Path(self._tmpdir) / "output"
+
+    def tearDown(self) -> None:
+        import shutil
+        shutil.rmtree(self._tmpdir, ignore_errors=True)
+
+    def _run_main(
+        self,
+        search_responses: list[list[dict]],
+        *,
+        extra_args: list[str] | None = None,
+        env: dict[str, str] | None = None,
+    ) -> int:
+        """Run main() with a mocked jira_search. Output goes to self._output_dir."""
+        page_iter = iter(search_responses)
+
+        def fake_jira_search(
+            base_url: str,
+            auth_header: str,
+            jql: str,
+            fields: str,
+            max_results: int = scan.PAGE_SIZE,
+        ) -> dict:
+            page = next(page_iter, [])
+            return _search_response(page)
+
+        argv = ["EDM", "--output-dir", str(self._output_dir)]
+        if extra_args:
+            argv.extend(extra_args)
+
+        with (
+            patch.dict(os.environ, env or self._ENV, clear=False),
+            patch.object(scan, "jira_search", fake_jira_search),
+        ):
+            return scan.main(argv)
+
+    def test_successful_scan(self) -> None:
+        unresolved = [_raw_issue("EDM-1"), _raw_issue("EDM-2")]
+        resolved = [
+            _raw_issue(
+                "EDM-3",
+                resolution="Fixed",
+                resolutiondate="2026-01-20T14:00:00.000+0000",
+            ),
+        ]
+        code = self._run_main([unresolved, resolved])
+
+        self.assertEqual(code, 0)
+        self.assertTrue((self._output_dir / "issues.json").is_file())
+        self.assertTrue((self._output_dir / "resolved.json").is_file())
+
+        issues_data = json.loads(
+            (self._output_dir / "issues.json").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(issues_data["project"], "EDM")
+        self.assertEqual(issues_data["totalCount"], 2)
+        self.assertEqual(len(issues_data["issues"]), 2)
+        self.assertEqual(
+            issues_data["jiraBaseUrl"], "https://jira.example.com",
+        )
+        self.assertIn("scannedAt", issues_data)
+
+        resolved_data = json.loads(
+            (self._output_dir / "resolved.json").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(resolved_data["totalCount"], 1)
+        self.assertEqual(resolved_data["windowDays"], 90)
+        self.assertEqual(resolved_data["issues"][0]["resolution"], "Fixed")
+
+    def test_empty_project(self) -> None:
+        code = self._run_main([[], []])
+
+        self.assertEqual(code, 0)
+        issues_data = json.loads(
+            (self._output_dir / "issues.json").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(issues_data["totalCount"], 0)
+        self.assertEqual(issues_data["issues"], [])
+
+    def test_custom_window_days(self) -> None:
+        code = self._run_main(
+            [[], []],
+            extra_args=["--window-days", "30"],
+        )
+        self.assertEqual(code, 0)
+        resolved_data = json.loads(
+            (self._output_dir / "resolved.json").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(resolved_data["windowDays"], 30)
+
+    def test_missing_jira_url(self) -> None:
+        code = self._run_main(
+            [],
+            env={"JIRA_TOKEN": "tok"},
+        )
+        self.assertEqual(code, 1)
+
+    def test_missing_jira_token(self) -> None:
+        code = self._run_main(
+            [],
+            env={"JIRA_URL": "https://jira.example.com"},
+        )
+        self.assertEqual(code, 1)
+
+    def test_jira_api_error_returns_1(self) -> None:
+        def failing_search(
+            base_url: str,
+            auth_header: str,
+            jql: str,
+            fields: str,
+            max_results: int = scan.PAGE_SIZE,
+        ) -> dict:
+            raise scan.JiraAPIError(403, "Forbidden")
+
+        argv = ["EDM", "--output-dir", str(self._output_dir)]
+        with (
+            patch.dict(os.environ, self._ENV, clear=False),
+            patch.object(scan, "jira_search", failing_search),
+        ):
+            code = scan.main(argv)
+        self.assertEqual(code, 1)
+
+    def test_normalized_fields_in_output(self) -> None:
+        raw = [_raw_issue(
+            "EDM-1",
+            summary="Test bug",
+            status="Open",
+            priority="High",
+            assignee="Alice",
+            reporter="Bob",
+            labels=["regression"],
+            components=["Backend"],
+        )]
+        code = self._run_main([raw, []])
+        self.assertEqual(code, 0)
+
+        issues_data = json.loads(
+            (self._output_dir / "issues.json").read_text(encoding="utf-8"),
+        )
+        issue = issues_data["issues"][0]
+        self.assertEqual(issue["key"], "EDM-1")
+        self.assertEqual(issue["summary"], "Test bug")
+        self.assertEqual(issue["status"], "Open")
+        self.assertEqual(issue["priority"], "High")
+        self.assertEqual(issue["assignee"], "Alice")
+        self.assertEqual(issue["reporter"], "Bob")
+        self.assertEqual(issue["labels"], ["regression"])
+        self.assertEqual(issue["components"], ["Backend"])
+
+    def test_multi_page_scan(self) -> None:
+        page1 = [_raw_issue(f"EDM-{i}") for i in range(1, scan.PAGE_SIZE + 1)]
+        page2 = [_raw_issue(f"EDM-{scan.PAGE_SIZE + 1}")]
+        code = self._run_main([page1, page2, []])
+
+        self.assertEqual(code, 0)
+        issues_data = json.loads(
+            (self._output_dir / "issues.json").read_text(encoding="utf-8"),
+        )
+        self.assertEqual(issues_data["totalCount"], scan.PAGE_SIZE + 1)
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+class TestExceptions(unittest.TestCase):
+
+    def test_jira_api_error_attributes(self) -> None:
+        exc = scan.JiraAPIError(404, "Not found")
+        self.assertEqual(exc.status_code, 404)
+        self.assertEqual(exc.body, "Not found")
+        self.assertIn("404", str(exc))
+
+    def test_jira_api_error_truncates_long_body(self) -> None:
+        exc = scan.JiraAPIError(500, "x" * 1000)
+        self.assertTrue(len(str(exc)) < 600)
+
+    def test_scan_error_is_base(self) -> None:
+        self.assertTrue(issubclass(scan.JiraAPIError, scan.ScanError))
+
+
+# ---------------------------------------------------------------------------
+# _retry_delay — pure function tests
+# ---------------------------------------------------------------------------
+
+class _FakeHeaders:
+    """Minimal headers object for testing _retry_delay."""
+
+    def __init__(self, mapping: dict[str, str] | None = None) -> None:
+        self._mapping = mapping or {}
+
+    def get(self, key: str, default: object = None) -> object:
+        return self._mapping.get(key, default)
+
+
+class TestRetryDelay(unittest.TestCase):
+
+    def test_valid_retry_after_header(self) -> None:
+        headers = _FakeHeaders({"Retry-After": "5"})
+        self.assertEqual(scan._retry_delay(headers, 0), 5)
+
+    def test_retry_after_zero_clamps_to_one(self) -> None:
+        headers = _FakeHeaders({"Retry-After": "0"})
+        self.assertEqual(scan._retry_delay(headers, 0), 1)
+
+    def test_retry_after_negative_clamps_to_one(self) -> None:
+        headers = _FakeHeaders({"Retry-After": "-3"})
+        self.assertEqual(scan._retry_delay(headers, 0), 1)
+
+    def test_no_retry_after_falls_back_to_backoff(self) -> None:
+        headers = _FakeHeaders()
+        self.assertEqual(scan._retry_delay(headers, 2), scan.RETRY_BACKOFF_BASE ** 2)
+
+    def test_non_integer_retry_after_falls_back_to_backoff(self) -> None:
+        headers = _FakeHeaders({"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"})
+        self.assertEqual(scan._retry_delay(headers, 1), scan.RETRY_BACKOFF_BASE ** 1)
+
+
+# ---------------------------------------------------------------------------
+# build_auth_header
+# ---------------------------------------------------------------------------
+
+class TestBuildAuthHeader(unittest.TestCase):
+
+    def test_bearer_without_email(self) -> None:
+        header = scan.build_auth_header("my-pat-token")
+        self.assertEqual(header, "Bearer my-pat-token")
+
+    def test_bearer_with_none_email(self) -> None:
+        header = scan.build_auth_header("my-pat-token", None)
+        self.assertEqual(header, "Bearer my-pat-token")
+
+    def test_basic_with_email(self) -> None:
+        header = scan.build_auth_header("my-api-token", "user@example.com")
+        self.assertTrue(header.startswith("Basic "))
+        decoded = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
+        self.assertEqual(decoded, "user@example.com:my-api-token")
+
+
+# ---------------------------------------------------------------------------
+# parse_args
+# ---------------------------------------------------------------------------
+
+class TestParseArgs(unittest.TestCase):
+
+    def test_minimal_args(self) -> None:
+        args = scan.parse_args(["EDM"])
+        self.assertEqual(args.project, "EDM")
+        self.assertEqual(args.window_days, 90)
+        self.assertIsNone(args.output_dir)
+
+    def test_all_args(self) -> None:
+        args = scan.parse_args([
+            "PROJ", "--window-days", "30", "--output-dir", "/tmp/out",
+        ])
+        self.assertEqual(args.project, "PROJ")
+        self.assertEqual(args.window_days, 30)
+        self.assertEqual(args.output_dir, Path("/tmp/out"))
+
+    def test_missing_project_exits(self) -> None:
+        with self.assertRaises(SystemExit):
+            scan.parse_args([])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/triage/skills/scan.md b/triage/skills/scan.md
index 9bf7f1b..7a12e11 100644
--- a/triage/skills/scan.md
+++ b/triage/skills/scan.md
@@ -9,164 +9,68 @@ You are fetching **every unresolved bug** and **recently resolved bugs** (for re
 
 ## Allowed Tools
 
-- **Jira MCP (read-only):** `jira_search` — fetch issues via JQL
-- **Local:** write `issues.json` and `resolved.json` artifacts
-- **Prohibited:** all Jira write tools (create, update, delete, comment, transition)
+- **Shell:** run `triage/scripts/scan.py` to fetch, normalize, and write artifacts
+- **Local:** read script output (stdout, stderr, exit code)
+- **Prohibited:** all Jira MCP tools — the script calls the Jira REST API directly
 
 ## Prerequisites
 
 Before scanning, ensure you have:
 
 - **Project key** (required) — from `/start` or the user's message
+- **`JIRA_URL`** (required) — Jira instance base URL (e.g., `https://redhat.atlassian.net`)
+- **`JIRA_TOKEN`** (required) — API token or Personal Access Token
+- **`JIRA_EMAIL`** (optional) — account email; required when using an API token (Basic auth), omit for PATs (Bearer auth)
 
-If the project key is missing, ask the user before proceeding.
+If the project key is missing, ask the user before proceeding. If the environment variables are not set, tell the user which ones to set and stop.
 
 ## Process
 
-### Step 1: Fetch Unresolved Bugs (JQL with Key-Based Cursor Pagination)
+### Step 1: Verify Environment
 
-Use the `jira_search` MCP tool (server: `user-mcp-jira`) to fetch all unresolved bugs. The tool returns a maximum of 50 results per call, so you must paginate.
+Confirm that `JIRA_URL` and `JIRA_TOKEN` environment variables are set. If either is missing, tell the user what to set and stop.
 
-**Important:** The `start_at` parameter of the MCP tool is non-functional (the response always returns `total: -1` and ignores `start_at`). Use **JQL key-based cursor pagination** instead: sort by `key ASC` and add `AND key > '{LAST_KEY}'` to advance through pages.
+### Step 2: Run the Scan Script
 
-**First call** — fetches the first 50 issues sorted by key:
+Run the scan script to fetch and normalize all bugs. Resolve
+`{AI_WORKFLOWS_ROOT}` as the git root of the ai-workflows install
+(run `git rev-parse --show-toplevel` from any file in the workflow
+directory, or use `~/.ai-workflows` when symlinked). The `--output-dir`
+path is relative to the project root (CWD).
 
-```json
-{
-  "jql": "project = EDM AND issuetype = Bug AND resolution = Unresolved ORDER BY key ASC",
-  "fields": "summary,status,priority,assignee,reporter,created,updated,labels,components,description",
-  "limit": 50
-}
+```bash
+python3 "{AI_WORKFLOWS_ROOT}/triage/scripts/scan.py" {PROJECT} --output-dir .artifacts/triage/{PROJECT}
 ```
 
-**Pagination loop:**
-
-```text
-last_key = ""
-all_issues = []
-
-loop:
-  1. Build JQL:
-     - First call: "project = {PROJECT} AND issuetype = Bug AND resolution = Unresolved ORDER BY key ASC"
-     - Subsequent calls: "project = {PROJECT} AND issuetype = Bug AND resolution = Unresolved AND key > '{last_key}' ORDER BY key ASC"
-  2. Call jira_search with the JQL and limit=50
-  3. Let page_issues = the returned issues array
-  4. If page_issues is empty → stop (all issues fetched)
-  5. Append page_issues to all_issues
-  6. last_key = the key of the last issue in page_issues
-  7. If len(page_issues) < 50 → stop (final partial page)
-  8. Go to step 1
-```
-
-Important:
-- Always set `ORDER BY key ASC` — this gives a deterministic sort for cursor pagination.
-- Never use `start_at` for pagination — it is ignored by the MCP tool.
-- Stop when a page returns fewer than 50 issues (final page) or zero issues.
-- Deduplicate by key as a safety net before saving.
-
-### Step 2: Fetch Recently Resolved Bugs (Regression Context)
-
-After the unresolved scan completes, fetch **bugs resolved in the last 90 days** using the same key-based cursor pagination pattern.
-
-**JQL (first call):**
-
-```text
-project = {PROJECT} AND issuetype = Bug AND resolution != Unresolved AND resolved >= -90d ORDER BY key ASC
-```
-
-**Pagination:** Same loop as Step 1, but substitute the unresolved JQL with the resolved JQL above, and use `AND key > '{last_key}'` on subsequent pages.
-
-**Fields:** Include at minimum: `summary,status,priority,assignee,reporter,created,updated,labels,components,description,resolution`. Request resolution date if your Jira API exposes it (e.g. `resolutiondate` or equivalent) so `/analyze` can match fix timing.
-
-If the resolved query returns **zero** issues (quiet project), still write `resolved.json` with an empty `issues` array — `/analyze` must be able to read the file.
-
-### Step 3: Normalize Issue Data
-
-For each **unresolved** issue, extract and normalize:
-
-- `key` — Jira issue key (e.g. `EDM-1234`)
-- `summary` — issue title
-- `status` — current status name
-- `priority` — priority name (Critical, High, Medium, Low, etc.)
-- `assignee` — display name or "Unassigned"
-- `reporter` — display name
-- `created` — creation date (ISO 8601)
-- `updated` — last update date (ISO 8601)
-- `labels` — array of labels
-- `components` — array of component names
-- `description` — full description text (may be long; preserve it for analysis)
+The script handles pagination, normalization, and file output. It writes:
 
-For each **resolved** issue, normalize the same fields plus when available:
+- `.artifacts/triage/{PROJECT}/issues.json` — all unresolved bugs
+- `.artifacts/triage/{PROJECT}/resolved.json` — bugs resolved in the last 90 days
 
-- `resolution` — resolution name (e.g. Fixed, Done)
-- `resolved` — resolution date (ISO 8601), if available from the API response
-
-### Step 4: Extract Jira Base URL
-
-Inspect the `jira_search` response for a `self` URL or similar field that contains the Jira instance domain (e.g. `https://mycompany.atlassian.net`). Save this so the `/report` phase can link issue keys without making additional Jira calls.
-
-### Step 5: Save Raw Data
-
-Write the normalized issues to the artifact file:
+To change the resolved-bug lookback window (default 90 days):
 
-```
-.artifacts/triage/{PROJECT}/issues.json
+```bash
+python3 "{AI_WORKFLOWS_ROOT}/triage/scripts/scan.py" {PROJECT} --window-days 30 --output-dir .artifacts/triage/{PROJECT}
 ```
 
-Format as a JSON object:
+### Step 3: Handle Errors
 
-```json
-{
-  "project": "EDM",
-  "jiraBaseUrl": "https://mycompany.atlassian.net",
-  "scannedAt": "2026-03-19T12:00:00Z",
-  "totalCount": 87,
-  "issues": [ ... ]
-}
-```
+If the script exits with code 1, report the error from stderr to the user:
 
-Write the resolved-bug dataset to:
+- Missing environment variables → tell the user which variable to set
+- Jira API errors → report the HTTP status and suggest checking the token or project key
 
-```
-.artifacts/triage/{PROJECT}/resolved.json
-```
-
-Format:
-
-```json
-{
-  "project": "EDM",
-  "jiraBaseUrl": "https://mycompany.atlassian.net",
-  "scannedAt": "2026-03-19T12:05:00Z",
-  "windowDays": 90,
-  "totalCount": 42,
-  "issues": [ ... ]
-}
-```
+### Step 4: Read the Summary
 
-Use the same `jiraBaseUrl` and a `scannedAt` timestamp when you finish the resolved pass. `windowDays` should be `90` when using the default `-90d` JQL window.
+Read the script's stdout for the scan summary (issue counts by priority and status, artifact file paths).
 
-### Step 6: Present Summary
+### Step 5: Present Results
 
-Display a summary of the scan results:
+Display the summary from Step 4 to the user.
 
-```text
-Scan complete: 87 unresolved bugs in EDM
-Resolved (last 90 days): 42 bugs — saved for regression context
+### Step 6: Edge Case — Zero Unresolved Issues
 
-By priority:
-  Critical:  3
-  High:     12
-  ...
-
-By status:
-  Open:        52
-  ...
-
-Data saved to:
-  .artifacts/triage/EDM/issues.json
-  .artifacts/triage/EDM/resolved.json
-```
+If the summary shows zero unresolved bugs, the workflow is done — there is nothing to triage. Suggest verifying the project key or issue type filter. The `resolved.json` file may still contain data.
 
 ## Output
 

From 23d72957e373b4d7b62a08752001cdc1ebdf5d12 Mon Sep 17 00:00:00 2001
From: Andy Dalton <andalton@redhat.com>
Date: Thu, 16 Jul 2026 13:07:07 -0400
Subject: [PATCH 2/5] Address CodeRabbit review feedback on scan.py
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Add HTTP_TIMEOUT (30s) to urlopen to prevent hangs on stalled responses
- Catch UnicodeDecodeError/JSONDecodeError in jira_search, convert to ScanError
- Guard against non-advancing pagination cursor (raises ScanError)
- Validate JIRA_URL scheme (require https) before attaching credentials
- Validate project key against Jira grammar before JQL interpolation
- Fix test env isolation: clear=True prevents host JIRA_URL/JIRA_TOKEN leaking
- Remove ~/.ai-workflows absolute path from scan.md (pre-merge check failure)
- Clarify "confirm" → "check" in scan.md to prevent agents echoing tokens
- Document exit code 2 (argparse) in script docstring and scan.md
- Add 10 new tests (84 total): project key validation, URL validation,
  cursor guard

Assisted-by: Claude Opus 4.6 (1M) <noreply@anthropic.com>
---
 triage/scripts/scan.py      | 45 ++++++++++++++++++++++++--
 triage/scripts/test_scan.py | 63 ++++++++++++++++++++++++++++++++++++-
 triage/skills/scan.md       | 15 +++++----
 3 files changed, 111 insertions(+), 12 deletions(-)

diff --git a/triage/scripts/scan.py b/triage/scripts/scan.py
index 6320846..ba63e41 100644
--- a/triage/scripts/scan.py
+++ b/triage/scripts/scan.py
@@ -16,6 +16,7 @@
 Exit codes:
     0 — scan completed successfully
     1 — missing input or configuration error
+    2 — invalid arguments (from argparse)
 """
 
 from __future__ import annotations
@@ -24,6 +25,7 @@
 import base64
 import json
 import os
+import re
 import ssl
 import sys
 import time
@@ -40,6 +42,7 @@
 PAGE_SIZE = 50
 MAX_RETRIES = 3
 RETRY_BACKOFF_BASE = 2
+HTTP_TIMEOUT = 30
 
 UNRESOLVED_FIELDS = (
     "summary,status,priority,assignee,reporter,"
@@ -102,7 +105,7 @@ def _http_get(url: str, headers: dict[str, str]) -> bytes:
     for attempt in range(MAX_RETRIES + 1):
         req = urllib.request.Request(url, headers=headers)
         try:
-            with urllib.request.urlopen(req, context=ctx) as resp:
+            with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT, context=ctx) as resp:
                 return resp.read()
         except urllib.error.HTTPError as exc:
             body = _read_error_body(exc)
@@ -124,6 +127,27 @@ def _http_get(url: str, headers: dict[str, str]) -> bytes:
     raise ScanError(f"Retries exhausted after {MAX_RETRIES + 1} attempts")
 
 
+_PROJECT_KEY_RE = re.compile(r"^[A-Z][A-Z0-9_]+$")
+
+
+def validate_project_key(key: str) -> None:
+    """Reject values that don't match Jira's project key grammar."""
+    if not _PROJECT_KEY_RE.match(key):
+        raise ScanError(
+            f"Invalid project key {key!r} — must be uppercase letters, "
+            f"digits, and underscores (e.g., EDM, FLIGHTCTL)"
+        )
+
+
+def validate_jira_url(url: str) -> None:
+    """Reject URLs that would send credentials over cleartext or to unexpected destinations."""
+    parsed = urllib.parse.urlparse(url)
+    if parsed.scheme != "https":
+        raise ScanError(f"JIRA_URL must use https (got {parsed.scheme!r})")
+    if not parsed.hostname:
+        raise ScanError("JIRA_URL has no hostname")
+
+
 def build_auth_header(token: str, email: str | None = None) -> str:
     """Build the Authorization header value.
 
@@ -156,7 +180,10 @@ def jira_search(
         "Accept": "application/json",
     }
     raw = _http_get(url, headers)
-    return json.loads(raw.decode("utf-8"))
+    try:
+        return json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ScanError("Jira returned an invalid JSON response") from exc
 
 
 # ---------------------------------------------------------------------------
@@ -189,7 +216,12 @@ def fetch_all_issues(
             break
 
         all_issues.extend(page)
-        last_key = page[-1]["key"]
+        new_key = page[-1]["key"]
+        if new_key == last_key:
+            raise ScanError(
+                f"Pagination cursor did not advance (stuck at {last_key})"
+            )
+        last_key = new_key
 
         if len(page) < PAGE_SIZE:
             break
@@ -373,6 +405,13 @@ def main(argv: list[str] | None = None) -> int:
         print("Error: JIRA_TOKEN environment variable is not set", file=sys.stderr)
         return 1
 
+    try:
+        validate_project_key(project)
+        validate_jira_url(jira_url)
+    except ScanError as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return 1
+
     auth_header = build_auth_header(jira_token, jira_email)
     search = partial(jira_search, jira_url, auth_header)
 
diff --git a/triage/scripts/test_scan.py b/triage/scripts/test_scan.py
index ad5e340..5f89274 100644
--- a/triage/scripts/test_scan.py
+++ b/triage/scripts/test_scan.py
@@ -560,6 +560,16 @@ def test_deduplication(self) -> None:
         keys = [r["key"] for r in result]
         self.assertEqual(keys, ["EDM-1", "EDM-2"])
 
+    def test_non_advancing_cursor_raises(self) -> None:
+        stuck_page = [_raw_issue(f"EDM-{i}") for i in range(1, scan.PAGE_SIZE + 1)]
+
+        def stuck_search(jql: str, fields: str, max_results: int = scan.PAGE_SIZE) -> dict:
+            return _search_response(stuck_page)
+
+        with self.assertRaises(scan.ScanError) as ctx:
+            scan.fetch_all_issues(stuck_search, "project = EDM", "summary")
+        self.assertIn("did not advance", str(ctx.exception))
+
 
 # ---------------------------------------------------------------------------
 # main — integration tests
@@ -604,8 +614,9 @@ def fake_jira_search(
         if extra_args:
             argv.extend(extra_args)
 
+        effective_env = self._ENV if env is None else env
         with (
-            patch.dict(os.environ, env or self._ENV, clear=False),
+            patch.dict(os.environ, effective_env, clear=True),
             patch.object(scan, "jira_search", fake_jira_search),
         ):
             return scan.main(argv)
@@ -755,6 +766,56 @@ def test_scan_error_is_base(self) -> None:
         self.assertTrue(issubclass(scan.JiraAPIError, scan.ScanError))
 
 
+# ---------------------------------------------------------------------------
+# validate_project_key — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestValidateProjectKey(unittest.TestCase):
+
+    def test_valid_keys(self) -> None:
+        for key in ("EDM", "FLIGHTCTL", "MY_PROJ", "A1"):
+            scan.validate_project_key(key)
+
+    def test_lowercase_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_project_key("edm")
+
+    def test_empty_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_project_key("")
+
+    def test_injection_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_project_key("EDM' OR 1=1 --")
+
+    def test_path_traversal_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_project_key("../../etc")
+
+
+# ---------------------------------------------------------------------------
+# validate_jira_url — pure function tests
+# ---------------------------------------------------------------------------
+
+class TestValidateJiraUrl(unittest.TestCase):
+
+    def test_valid_https_url(self) -> None:
+        scan.validate_jira_url("https://redhat.atlassian.net")
+
+    def test_http_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError) as ctx:
+            scan.validate_jira_url("http://jira.example.com")
+        self.assertIn("https", str(ctx.exception))
+
+    def test_no_scheme_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_jira_url("jira.example.com")
+
+    def test_file_scheme_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError):
+            scan.validate_jira_url("file:///etc/passwd")
+
+
 # ---------------------------------------------------------------------------
 # _retry_delay — pure function tests
 # ---------------------------------------------------------------------------
diff --git a/triage/skills/scan.md b/triage/skills/scan.md
index 7a12e11..3194fdc 100644
--- a/triage/skills/scan.md
+++ b/triage/skills/scan.md
@@ -28,15 +28,14 @@ If the project key is missing, ask the user before proceeding. If the environmen
 
 ### Step 1: Verify Environment
 
-Confirm that `JIRA_URL` and `JIRA_TOKEN` environment variables are set. If either is missing, tell the user what to set and stop.
+Check that `JIRA_URL` and `JIRA_TOKEN` environment variables are set (do not print or echo their values). If either is missing, tell the user which variable to set and stop.
 
 ### Step 2: Run the Scan Script
 
 Run the scan script to fetch and normalize all bugs. Resolve
-`{AI_WORKFLOWS_ROOT}` as the git root of the ai-workflows install
-(run `git rev-parse --show-toplevel` from any file in the workflow
-directory, or use `~/.ai-workflows` when symlinked). The `--output-dir`
-path is relative to the project root (CWD).
+`{AI_WORKFLOWS_ROOT}` by running `git rev-parse --show-toplevel` from
+any file in the workflow directory. The `--output-dir` path is relative
+to the project root (CWD).
 
 ```bash
 python3 "{AI_WORKFLOWS_ROOT}/triage/scripts/scan.py" {PROJECT} --output-dir .artifacts/triage/{PROJECT}
@@ -55,10 +54,10 @@ python3 "{AI_WORKFLOWS_ROOT}/triage/scripts/scan.py" {PROJECT} --window-days 30
 
 ### Step 3: Handle Errors
 
-If the script exits with code 1, report the error from stderr to the user:
+If the script exits with a non-zero code, report the error from stderr to the user:
 
-- Missing environment variables → tell the user which variable to set
-- Jira API errors → report the HTTP status and suggest checking the token or project key
+- Exit 1 — missing environment variables, invalid project key, or Jira API errors
+- Exit 2 — invalid command-line arguments (e.g., missing project key)
 
 ### Step 4: Read the Summary
 

From 8137a9899e01f1c00e470a5b657acbb304fbc6e1 Mon Sep 17 00:00:00 2001
From: Andy Dalton <andalton@redhat.com>
Date: Thu, 16 Jul 2026 13:11:23 -0400
Subject: [PATCH 3/5] Reject userinfo and fragments in JIRA_URL validation

validate_jira_url() now also rejects URLs containing embedded
credentials (user:pass@) or fragment identifiers (#section).

Assisted-by: Claude Opus 4.6 (1M) <noreply@anthropic.com>
---
 triage/scripts/scan.py      |  4 ++++
 triage/scripts/test_scan.py | 10 ++++++++++
 2 files changed, 14 insertions(+)

diff --git a/triage/scripts/scan.py b/triage/scripts/scan.py
index ba63e41..3627b93 100644
--- a/triage/scripts/scan.py
+++ b/triage/scripts/scan.py
@@ -146,6 +146,10 @@ def validate_jira_url(url: str) -> None:
         raise ScanError(f"JIRA_URL must use https (got {parsed.scheme!r})")
     if not parsed.hostname:
         raise ScanError("JIRA_URL has no hostname")
+    if parsed.username or parsed.password:
+        raise ScanError("JIRA_URL must not contain credentials")
+    if parsed.fragment:
+        raise ScanError("JIRA_URL must not contain a fragment")
 
 
 def build_auth_header(token: str, email: str | None = None) -> str:
diff --git a/triage/scripts/test_scan.py b/triage/scripts/test_scan.py
index 5f89274..4c739b9 100644
--- a/triage/scripts/test_scan.py
+++ b/triage/scripts/test_scan.py
@@ -815,6 +815,16 @@ def test_file_scheme_rejected(self) -> None:
         with self.assertRaises(scan.ScanError):
             scan.validate_jira_url("file:///etc/passwd")
 
+    def test_userinfo_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError) as ctx:
+            scan.validate_jira_url("https://user:pass@jira.example.com")
+        self.assertIn("credentials", str(ctx.exception))
+
+    def test_fragment_rejected(self) -> None:
+        with self.assertRaises(scan.ScanError) as ctx:
+            scan.validate_jira_url("https://jira.example.com#section")
+        self.assertIn("fragment", str(ctx.exception))
+
 
 # ---------------------------------------------------------------------------
 # _retry_delay — pure function tests

From 9054a392b8d66a60390b51e191c5c2902638eaf9 Mon Sep 17 00:00:00 2001
From: Andy Dalton <andalton@redhat.com>
Date: Thu, 16 Jul 2026 13:20:16 -0400
Subject: [PATCH 4/5] Clarify AI_WORKFLOWS_ROOT resolution wording in scan.md

Assisted-by: Claude Opus 4.6 (1M) <noreply@anthropic.com>
---
 triage/skills/scan.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/triage/skills/scan.md b/triage/skills/scan.md
index 3194fdc..4fbc559 100644
--- a/triage/skills/scan.md
+++ b/triage/skills/scan.md
@@ -34,8 +34,8 @@ Check that `JIRA_URL` and `JIRA_TOKEN` environment variables are set (do not pri
 
 Run the scan script to fetch and normalize all bugs. Resolve
 `{AI_WORKFLOWS_ROOT}` by running `git rev-parse --show-toplevel` from
-any file in the workflow directory. The `--output-dir` path is relative
-to the project root (CWD).
+within the ai-workflows checkout (e.g., this skill file's directory).
+The `--output-dir` path is relative to the project root (CWD).
 
 ```bash
 python3 "{AI_WORKFLOWS_ROOT}/triage/scripts/scan.py" {PROJECT} --output-dir .artifacts/triage/{PROJECT}

From 6197d73c79219c79846605fe0215ae52d836bfdf Mon Sep 17 00:00:00 2001
From: Andy Dalton <andalton@redhat.com>
Date: Thu, 16 Jul 2026 14:44:15 -0400
Subject: [PATCH 5/5] Remove duplicate zero-results guidance from scan.md

Step 6 duplicated the edge case already documented in "On Completion".
Removed Step 6; the On Completion section is the single authority.

Assisted-by: Claude Opus 4.6 (1M) <noreply@anthropic.com>
---
 triage/skills/scan.md | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/triage/skills/scan.md b/triage/skills/scan.md
index 4fbc559..fc79f7c 100644
--- a/triage/skills/scan.md
+++ b/triage/skills/scan.md
@@ -67,10 +67,6 @@ Read the script's stdout for the scan summary (issue counts by priority and stat
 
 Display the summary from Step 4 to the user.
 
-### Step 6: Edge Case — Zero Unresolved Issues
-
-If the summary shows zero unresolved bugs, the workflow is done — there is nothing to triage. Suggest verifying the project key or issue type filter. The `resolved.json` file may still contain data.
-
 ## Output
 
 - Scan summary displayed to the user