Add index-page link discovery and simplify licence/priority fields

Adds link_discovery.py: sitemap-first, same-domain-crawl-fallback
  discovery of sub-pages from a single index URL, so a source with many
  sub-pages no longer needs each page entered by hand. Never pulls page
  titles or body text into a ticket field, only URLs, to avoid a
  prompt-injection path into tickets Claude Code later reads. Adds a
  review-and-approve screen so nothing is queued without a human
  checking the actual discovered list first.

  Replaces the free-text licence field with a required typed permission
  question, since licence itself is better researched from the source
  during actual processing rather than typed in at ticket time. Replaces
  the priority dropdown with a single urgent checkbox.

  Fixes a few real bugs found along the way: colliding ticket filenames
  when writing many tickets in one batch, a couple of layout issues
  where long labels or URLs were getting clipped instead of wrapping or
  scrolling, and dead links slipping through discovery from malformed
  relative link resolution

ambrosiana_intake.py

@@ -36,10 +36,13 @@ import tomllib
 from datetime import datetime
 from pathlib import Path
 
+from textual import work
 from textual.app import App, ComposeResult
 from textual.containers import Horizontal, VerticalScroll
+from textual.screen import Screen
 from textual.widgets import (
     Button,
+    Checkbox,
     Footer,
     Header,
     Input,
@@ -49,6 +52,8 @@ from textual.widgets import (
     TextArea,
 )
 
+from link_discovery import DiscoveredPage, discover_links
+
 # ---------------------------------------------------------------------------
 # Configuration
 # ---------------------------------------------------------------------------
@@ -93,12 +98,9 @@ SOURCE_TIER_OPTIONS = [
     ("forum/unmoderated", "forum-unmoderated"),
 ]
 
-PRIORITY_OPTIONS = [
+PERMISSION_QUESTION = "Do you have permission to access this document and add it to the RAG?"
-    ("Immediate need", "immediate"),
-    ("Add to queue", "queue"),
-]
 
-REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "licence", "priority")
+REQUIRED_FIELDS = ("source_location", "source_type", "source_tier", "permission")
 
 
 def slugify(text: str, fallback: str) -> str:
@@ -110,6 +112,81 @@ def slugify(text: str, fallback: str) -> str:
     return text[:60] or fallback
 
 
+# ---------------------------------------------------------------------------
+# Discovery review screen
+# ---------------------------------------------------------------------------
+
+
+class LinkReviewScreen(Screen[list[DiscoveredPage]]):
+    """
+    Shows discovered sub-pages for approval before any ticket is written.
+    Titles shown here are derived from the URL only (see link_discovery.py)
+    — never from fetched page content, deliberately, to avoid a prompt-
+    injection vector into the ticket queue Claude Code later reads.
+    Dismisses with the list of approved DiscoveredPage entries (empty list
+    if cancelled).
+    """
+
+    CSS = """
+    #review-scroll { width: 100%; padding: 1 2; overflow-x: auto; }
+    #review-header { width: 100%; padding-bottom: 1; color: $text-muted; }
+    .review-row { width: auto; min-width: 100%; }
+    .not-in-sitemap { color: $warning; }
+    #review-button-row { width: 100%; height: auto; padding-top: 1; align-horizontal: center; }
+    #review-button-row Button { margin: 0 1; }
+    """
+
+    BINDINGS = [("escape", "cancel", "Cancel")]
+
+    def __init__(self, pages: list[DiscoveredPage], method: str) -> None:
+        super().__init__()
+        self.pages = pages
+        self.method = method
+
+    def compose(self) -> ComposeResult:
+        yield Header()
+        with VerticalScroll(id="review-scroll"):
+            confirmed_note = (
+                "sitemap-verified" if self.method == "sitemap-verified" else "HTML crawl (no sitemap found)"
+            )
+            yield Static(
+                f"Found {len(self.pages)} page(s) via {confirmed_note}. "
+                f"Review and deselect anything you don't want queued.",
+                id="review-header",
+            )
+            for index, page in enumerate(self.pages):
+                warn = page.sitemap_confirmed is False
+                label = f"{page.title}  [dim]{page.url}[/dim]"
+                if warn:
+                    label += "  [warning](not in sitemap)[/warning]"
+                yield Checkbox(label, value=True, id=f"pick-{index}", classes="review-row")
+            with Horizontal(id="review-button-row"):
+                yield Button("Select all", id="select-all")
+                yield Button("Select none", id="select-none")
+                yield Button("Approve selected", id="approve", variant="success")
+                yield Button("Cancel", id="cancel", variant="error")
+        yield Footer()
+
+    def _checkboxes(self) -> list[Checkbox]:
+        return list(self.query(Checkbox))
+
+    def on_button_pressed(self, event: Button.Pressed) -> None:
+        if event.button.id == "select-all":
+            for box in self._checkboxes():
+                box.value = True
+        elif event.button.id == "select-none":
+            for box in self._checkboxes():
+                box.value = False
+        elif event.button.id == "approve":
+            approved = [page for page, box in zip(self.pages, self._checkboxes()) if box.value]
+            self.dismiss(approved)
+        elif event.button.id == "cancel":
+            self.dismiss([])
+
+    def action_cancel(self) -> None:
+        self.dismiss([])
+
+
 # ---------------------------------------------------------------------------
 # App
 # ---------------------------------------------------------------------------
@@ -123,8 +200,7 @@ class AmbrosianaIntakeApp(App):
     }
 
     #form-scroll {
-        width: 90;
+        width: 100%;
-        max-width: 100%;
         padding: 1 2;
     }
 
@@ -136,6 +212,7 @@ class AmbrosianaIntakeApp(App):
     }
 
     .field-label {
+        width: 100%;
         padding-top: 1;
         color: $text;
         text-style: bold;
@@ -203,17 +280,24 @@ class AmbrosianaIntakeApp(App):
             yield Label("Source location  [required]", classes="field-label")
             yield Input(placeholder="filepath or URL", id="source_location")
 
+            yield Label(
+                "Discovery depth  (only used by \"Discover sub-pages\", below)",
+                classes="field-label optional-marker",
+            )
+            yield Input(placeholder="e.g. 4", id="discovery_depth", value="1")
+
             yield Label("Source type  [required]", classes="field-label")
             yield Select(SOURCE_TYPE_OPTIONS, id="source_type", prompt="Choose source type")
 
             yield Label("Source tier  [required]", classes="field-label")
             yield Select(SOURCE_TIER_OPTIONS, id="source_tier", prompt="Choose source tier")
 
-            yield Label("Licence  [required]", classes="field-label")
+            yield Label(
-            yield Input(placeholder="e.g. The Unlicense, CC-BY-4.0, proprietary...", id="licence")
+                f"{PERMISSION_QUESTION}  [required — type yes/no]", classes="field-label"
+            )
+            yield Input(placeholder="yes / no", id="permission")
 
-            yield Label("Priority  [required]", classes="field-label")
+            yield Checkbox("Urgent (immediate need — otherwise added to queue)", value=False, id="urgent")
-            yield Select(PRIORITY_OPTIONS, id="priority", prompt="Choose priority")
 
             yield Label("Title / description  (optional)", classes="field-label optional-marker")
             yield Input(placeholder="short human-readable title", id="title")
@@ -223,6 +307,7 @@ class AmbrosianaIntakeApp(App):
 
             with Horizontal(id="button-row"):
                 yield Button("Save ticket", id="submit", variant="success")
+                yield Button("Discover sub-pages", id="discover", variant="primary")
                 yield Button("Clear", id="reset", variant="warning")
                 yield Button("Quit", id="quit", variant="error")
 
@@ -236,8 +321,8 @@ class AmbrosianaIntakeApp(App):
             "source_location": self.query_one("#source_location", Input).value.strip(),
             "source_type": self.query_one("#source_type", Select).value,
             "source_tier": self.query_one("#source_tier", Select).value,
-            "licence": self.query_one("#licence", Input).value.strip(),
+            "permission": self.query_one("#permission", Input).value.strip(),
-            "priority": self.query_one("#priority", Select).value,
+            "priority": "immediate" if self.query_one("#urgent", Checkbox).value else "queue",
             "title": self.query_one("#title", Input).value.strip(),
             "notes": self.query_one("#notes-area", TextArea).text.strip(),
         }
@@ -253,8 +338,8 @@ class AmbrosianaIntakeApp(App):
         self.query_one("#source_location", Input).value = ""
         self.query_one("#source_type", Select).clear()
         self.query_one("#source_tier", Select).clear()
-        self.query_one("#licence", Input).value = ""
+        self.query_one("#permission", Input).value = ""
-        self.query_one("#priority", Select).clear()
+        self.query_one("#urgent", Checkbox).value = False
         self.query_one("#title", Input).value = ""
         self.query_one("#notes-area", TextArea).text = ""
         self.query_one("#source_location", Input).focus()
@@ -264,14 +349,25 @@ class AmbrosianaIntakeApp(App):
         if missing:
             self._set_status(f"Missing required field(s): {', '.join(missing)}", error=True)
             return None
+        if data["permission"].strip().lower() not in ("yes", "y"):
+            self._set_status(
+                "Permission must be confirmed by typing 'yes' — ticket not saved.", error=True
+            )
+            return None
 
         now = datetime.now()
 
         fallback_slug = Path(data["source_location"]).stem or "untitled"
         slug = slugify(data["title"] or fallback_slug, fallback_slug)
         priority_tag = "immediate_" if data["priority"] == "immediate" else ""
-        filename = f"{priority_tag}{now.strftime('%Y%m%d-%H%M%S')}_{slug}.md"
+        base_filename = f"{priority_tag}{now.strftime('%Y%m%d-%H%M%S')}_{slug}"
-        ticket_path = TICKETS_ROOT / filename
+        ticket_path = TICKETS_ROOT / f"{base_filename}.md"
+        # Second-precision timestamps collide when writing many tickets in a
+        # tight loop (batch discovery) — disambiguate rather than overwrite.
+        suffix = 2
+        while ticket_path.exists():
+            ticket_path = TICKETS_ROOT / f"{base_filename}-{suffix}.md"
+            suffix += 1
 
         front_matter_lines = [
             "---",
@@ -281,7 +377,7 @@ class AmbrosianaIntakeApp(App):
             f"source_location: \"{data['source_location']}\"",
             f"source_type: {data['source_type']}",
             f"source_tier: {data['source_tier']}",
-            f"licence: \"{data['licence']}\"",
+            f"permission_confirmed: \"{data['permission']}\"",
             f"priority: {data['priority']}",
             f"title: \"{data['title']}\"" if data["title"] else "title: \"\"",
             "---",
@@ -295,9 +391,15 @@ class AmbrosianaIntakeApp(App):
         body_lines.append(f"**Source:** {data['source_location']}")
         body_lines.append(f"**Source-type:** `{data['source_type']}`")
         body_lines.append(f"**Source-tier:** `{data['source_tier']}`")
-        body_lines.append(f"**Licence:** {data['licence']}")
+        body_lines.append(f"**Permission confirmed:** {data['permission']}")
         body_lines.append(f"**Priority:** {data['priority']}")
         body_lines.append("")
+        body_lines.append(
+            "*Licence is not captured here — research it from the source itself "
+            "during drafting; default to \"restricted, in-house use only\" if it "
+            "can't be determined.*"
+        )
+        body_lines.append("")
         if data["notes"]:
             body_lines.append("## Notes / context")
             body_lines.append("")
@@ -328,12 +430,72 @@ class AmbrosianaIntakeApp(App):
     def on_button_pressed(self, event: Button.Pressed) -> None:
         if event.button.id == "submit":
             self._do_submit()
+        elif event.button.id == "discover":
+            self._do_discover()
         elif event.button.id == "reset":
             self._clear_form()
             self._set_status("")
         elif event.button.id == "quit":
             self.exit()
 
+    # -- discovery flow ---------------------------------------------------
+
+    def _do_discover(self) -> None:
+        data = self._collect()
+        shared_missing = [f for f in REQUIRED_FIELDS if f != "source_location"
+                           and (not data.get(f) or data.get(f) is Select.BLANK)]
+        if shared_missing:
+            self._set_status(
+                f"Fill in {', '.join(shared_missing)} first — every discovered "
+                f"page's ticket will reuse these.", error=True,
+            )
+            return
+        if data["permission"].strip().lower() not in ("yes", "y"):
+            self._set_status(
+                "Permission must be confirmed by typing 'yes' before discovering.", error=True
+            )
+            return
+        if not re.match(r"^https?://", data["source_location"]):
+            self._set_status("Discovery needs an http(s) URL in Source location.", error=True)
+            return
+
+        try:
+            depth = int(self.query_one("#discovery_depth", Input).value.strip() or "1")
+        except ValueError:
+            self._set_status("Discovery depth must be a whole number.", error=True)
+            return
+
+        self._set_status(f"Discovering (depth {depth})... this may take a moment.")
+        self._run_discovery(data["source_location"], depth, data)
+
+    @work(thread=True)
+    def _run_discovery(self, start_url: str, depth: int, shared_data: dict[str, str]) -> None:
+        pages, method = discover_links(start_url, max_depth=depth)
+        self.call_from_thread(self._on_discovery_done, pages, method, shared_data)
+
+    def _on_discovery_done(
+        self, pages: list[DiscoveredPage], method: str, shared_data: dict[str, str]
+    ) -> None:
+        if not pages:
+            self._set_status(f"No pages discovered via {method} — nothing to review.", error=True)
+            return
+
+        def handle_approved(approved: list[DiscoveredPage]) -> None:
+            if not approved:
+                self._set_status("Discovery cancelled — no tickets written.")
+                return
+            written = 0
+            for page in approved:
+                page_data = dict(shared_data)
+                page_data["source_location"] = page.url
+                page_data["title"] = page.title
+                if self._write_ticket(page_data) is not None:
+                    written += 1
+            self._set_status(f"Wrote {written} ticket(s) from discovery.")
+            self._clear_form()
+
+        self.push_screen(LinkReviewScreen(pages, method), handle_approved)
+
 
 if __name__ == "__main__":
     AmbrosianaIntakeApp().run()

link_discovery.py

@@ -0,0 +1,243 @@
+#!/usr/bin/env python3
+# Created by John A. Hoeven with the ethical assistance of Claude AI
+# ---------------------------------------------------------------------------
+# link_discovery.py
+# ~/projects/ambrosiana-document-intake/link_discovery.py
+# Version: v0.2.0 | Status: BETA
+# ---------------------------------------------------------------------------
+# Purpose:  URL discovery from an index page, for the "add index page,
+#           review discovered sub-pages" intake flow. Fetches pages to find
+#           URLs only — never writes a ticket itself; that stays a separate,
+#           human-approved step in the App.
+# Target:   Any Ambrosiana device.
+# Entry:    import as a module.
+# Depends:  requests, beautifulsoup4 (see requirements.txt)
+# ---------------------------------------------------------------------------
+"""
+Security note (deliberate, not an oversight): this module never extracts
+free-text content from a fetched page (titles, headings, link text) into
+anything that ends up in a ticket. A ticket is later read by Claude Code
+as a pipeline instruction, so pulling untrusted third-party prose into a
+ticket field would be a real prompt-injection vector. Only URLs — which
+this tool itself discovered under a scope the human chose — are ever
+returned. Display titles shown in the review UI are derived from the URL
+path alone, never from fetched page content.
+
+Two discovery strategies, tried in this order:
+
+1. Sitemap (preferred) — sitemap.xml is pure structured data (<loc>,
+   <lastmod>), nothing to inject into. Checked at the domain root and via
+   robots.txt's Sitemap: directive.
+2. HTML link crawl (fallback) — same-domain breadth-first crawl for sites
+   without a sitemap. Extracts hrefs only, never page titles or body text.
+"""
+
+from __future__ import annotations
+
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass
+from urllib.parse import urljoin, urlparse, urlunparse
+
+import requests
+from bs4 import BeautifulSoup
+
+USER_AGENT = "ambrosiana-intake-discovery/0.2 (personal RAG intake tool)"
+REQUEST_TIMEOUT = 10
+DEFAULT_MAX_PAGES = 300
+SKIP_EXTENSIONS = (
+    ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico",
+    ".woff", ".woff2", ".ttf", ".xml", ".json", ".pdf", ".zip", ".md",
+)
+SITEMAP_XML_NS = "{http://www.sitemaps.org/schemas/sitemap/0.9}"
+
+
+@dataclass
+class DiscoveredPage:
+    url: str
+    title: str  # derived from the URL path only — never from fetched content
+    depth: int
+    sitemap_confirmed: bool | None = None  # None = no sitemap available to check against
+
+
+def _display_title(url: str) -> str:
+    """Human-readable label derived purely from the URL — no page content."""
+    parsed = urlparse(url)
+    path = parsed.path.strip("/")
+    if not path:
+        return parsed.netloc or url  # site root — use the domain, not the raw URL
+    last_segment = path.rsplit("/", 1)[-1]
+    return last_segment.replace("-", " ").replace("_", " ").strip() or url
+
+
+def _normalize(url: str) -> str:
+    """Strip fragment/query for de-dup purposes; keep the path as-is."""
+    parsed = urlparse(url)
+    return urlunparse((parsed.scheme, parsed.netloc, parsed.path.rstrip("/") or "/", "", "", ""))
+
+
+def _is_same_domain(url: str, domain: str) -> bool:
+    return urlparse(url).netloc == domain
+
+
+def _is_content_link(url: str) -> bool:
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        return False
+    return not parsed.path.lower().endswith(SKIP_EXTENSIONS)
+
+
+def _fetch(url: str) -> requests.Response | None:
+    try:
+        response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT)
+        response.raise_for_status()
+        return response
+    except requests.RequestException:
+        return None
+
+
+# ── Strategy 1: sitemap ──────────────────────────────────────────────────
+
+def find_sitemap_url(start_url: str) -> str | None:
+    """Check robots.txt's Sitemap: directive first, then the conventional path."""
+    parsed = urlparse(start_url)
+    root = f"{parsed.scheme}://{parsed.netloc}"
+
+    robots = _fetch(f"{root}/robots.txt")
+    if robots is not None:
+        for line in robots.text.splitlines():
+            if line.strip().lower().startswith("sitemap:"):
+                return line.split(":", 1)[1].strip()
+
+    candidate = f"{root}/sitemap.xml"
+    response = _fetch(candidate)
+    if response is not None and "xml" in response.headers.get("content-type", ""):
+        return candidate
+    return None
+
+
+def fetch_sitemap_urls(sitemap_url: str, _depth: int = 0) -> set[str]:
+    """
+    Parse a sitemap.xml (or a sitemap index listing nested sitemaps, handled
+    up to 2 levels deep) and return the full set of normalized <loc> URLs —
+    used only as a cross-reference trust signal, never as the primary
+    discovery source (see discover_links). ElementTree's default parser
+    does not resolve external entities/DTDs, so this is safe against
+    classic XXE without needing an extra dependency.
+    """
+    if _depth > 2:
+        return set()
+
+    response = _fetch(sitemap_url)
+    if response is None:
+        return set()
+
+    try:
+        root = ET.fromstring(response.content)
+    except ET.ParseError:
+        return set()
+
+    locs = [el.text.strip() for el in root.iter(f"{SITEMAP_XML_NS}loc") if el.text]
+
+    if root.tag == f"{SITEMAP_XML_NS}sitemapindex":
+        urls: set[str] = set()
+        for nested_url in locs[:20]:  # bounded — a sitemap index fan-out cap
+            urls |= fetch_sitemap_urls(nested_url, _depth=_depth + 1)
+        return urls
+
+    return {_normalize(loc) for loc in locs}
+
+
+# ── Strategy 2: HTML crawl fallback ──────────────────────────────────────
+
+def discover_from_crawl(
+    start_url: str,
+    max_depth: int,
+    max_pages: int = DEFAULT_MAX_PAGES,
+    sitemap_urls: set[str] | None = None,
+) -> list[DiscoveredPage]:
+    """
+    Breadth-first same-domain crawl from start_url, following the actual
+    links on each page — this is what "N links deep from the index"
+    literally means, including for hub/TOC-style index pages whose real
+    sub-content lives at sibling paths rather than nested under the index's
+    own URL. Extracts hrefs only — never page titles or body text (see
+    module docstring). The start page itself is not included in the
+    returned list, only pages found from it.
+
+    If sitemap_urls is given, each discovered page is annotated with
+    whether it's also present in the sitemap — a trust signal shown in the
+    review UI, not a filter (a sitemap can be incomplete or stale).
+    """
+    domain = urlparse(start_url).netloc
+    visited = {_normalize(start_url)}
+    queue: list[tuple[str, int]] = [(start_url, 0)]
+    discovered: list[DiscoveredPage] = []
+    fetched_count = 0
+
+    while queue and fetched_count < max_pages:
+        url, depth = queue.pop(0)
+        if depth >= max_depth:
+            continue
+
+        response = _fetch(url)
+        fetched_count += 1
+        if response is None:
+            continue
+
+        soup = BeautifulSoup(response.text, "html.parser")
+        for anchor in soup.find_all("a", href=True):
+            link = urljoin(url, anchor["href"])
+            if not _is_content_link(link) or not _is_same_domain(link, domain):
+                continue
+            link_norm = _normalize(link)
+            if link_norm in visited:
+                continue
+            visited.add(link_norm)
+
+            confirmed = (link_norm in sitemap_urls) if sitemap_urls is not None else None
+            discovered.append(
+                DiscoveredPage(url=link, title=_display_title(link), depth=depth + 1, sitemap_confirmed=confirmed)
+            )
+            if len(discovered) + fetched_count >= max_pages:
+                break
+            queue.append((link, depth + 1))
+
+    return [p for p in discovered if p.sitemap_confirmed or _is_reachable(p.url)]
+
+
+def _is_reachable(url: str) -> bool:
+    """
+    Cheap liveness check for links not already confirmed via the sitemap.
+    Relative-link resolution against an already-nested page can produce
+    plausible-looking but nonexistent URLs (e.g. a same-name breadcrumb
+    href resolving into a doubled path segment) — verified live 2026-08-07
+    against textual.textualize.io, where this caught real 404s a sitemap-
+    confirmed link would never need checking for.
+    """
+    try:
+        response = requests.head(
+            url, headers={"User-Agent": USER_AGENT}, timeout=REQUEST_TIMEOUT, allow_redirects=True
+        )
+        return response.status_code < 400
+    except requests.RequestException:
+        return False
+
+
+# ── Entry point ───────────────────────────────────────────────────────────
+
+def discover_links(
+    start_url: str,
+    max_depth: int,
+    max_pages: int = DEFAULT_MAX_PAGES,
+) -> tuple[list[DiscoveredPage], str]:
+    """
+    Returns (discovered_pages, method) — method is "sitemap-verified" or
+    "crawl" so the UI can tell the human whether a sitemap was available to
+    cross-check against.
+    """
+    sitemap_url = find_sitemap_url(start_url)
+    sitemap_urls = fetch_sitemap_urls(sitemap_url) if sitemap_url else None
+
+    pages = discover_from_crawl(start_url, max_depth=max_depth, max_pages=max_pages, sitemap_urls=sitemap_urls)
+    method = "sitemap-verified" if sitemap_urls else "crawl"
+    return pages, method

requirements.txt

@@ -1 +1,3 @@
 textual>=8.2.0
+requests>=2.34.0
+beautifulsoup4>=4.15.0