feat(triggers): first real SKILL_OBSERVATION_TRIGGER — clipboard URL → web_fetch

codec_dashboard.py

@@ -2140,6 +2140,9 @@ async def web_search_endpoint(request: Request):
     "create_skill", "skill_forge", "ask_codec_to_build", "delegate",
     # Phase 2 Step 7 — end-of-day shift report (read-only, no destructive side effects)
     "shift_report",
+    # Phase 2 Step 6 — first declarative trigger (clipboard URL → web_fetch).
+    # Read-only network fetch, gated by codec_ask_user.ask consent on auto-fire.
+    "clipboard_url_fetch",
 }
 
 # ---------------------------------------------------------------------------

skills/clipboard_url_fetch.py

@@ -0,0 +1,144 @@
+"""CODEC Skill: Clipboard URL Auto-Fetch.
+
+First real declarative `SKILL_OBSERVATION_TRIGGER` shipped after Phase 2
+Step 6 (Trigger System). Demonstrates the full pipeline:
+
+  codec_observer.poll() → snapshot["clipboard"]["preview"]
+        → codec_triggers.evaluate() matches `clipboard_pattern`
+        → require_confirmation=True → codec_ask_user.ask("Fetch?")
+        → user yes → codec_dispatch.run_skill(...) → this skill's run()
+        → audit emit: trigger_fired
+
+Runtime behavior:
+  - Trigger fires when an HTTP/HTTPS URL appears in clipboard.
+  - 600 s cooldown per (skill, trigger) prevents repeated fires
+    when the user keeps the same URL on clipboard.
+  - Consent gate via codec_ask_user.ask is non-destructive
+    (read-only fetch, no side effects beyond the network request).
+  - On approval, this skill reads pbpaste at runtime to get the
+    current URL (handles the case where clipboard changed between
+    trigger fire and skill execution), then delegates to the
+    existing `web_fetch` skill.
+  - Output truncated to 2000 chars to keep the notification readable.
+
+Manual paths (chat / voice / MCP) ALSO work — the skill reads the
+clipboard or extracts a URL from the user's task string. So you can
+say "fetch the link I just copied" or "summarize https://..." and the
+same code path handles it.
+"""
+SKILL_NAME = "clipboard_url_fetch"
+SKILL_DESCRIPTION = (
+    "Fetch and return the content of an HTTP/HTTPS URL on the clipboard. "
+    "Auto-triggers when a URL is copied (with consent prompt)."
+)
+SKILL_TRIGGERS = [
+    "fetch clipboard url", "fetch the link", "fetch this url",
+    "summarize clipboard", "summarize this link", "what's at this url",
+    "what's at this link", "open this url",
+]
+SKILL_MCP_EXPOSE = False  # local-only; no value over MCP since clipboard isn't shared
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Phase 2 Step 6 declarative trigger.
+# Fires when the clipboard preview contains an http(s) URL.
+# Pattern is conservative: bounded character class, no whitespace,
+# no quote chars (avoids accidentally matching JSON-encoded URLs as part
+# of a larger blob).
+# ──────────────────────────────────────────────────────────────────────────────
+SKILL_OBSERVATION_TRIGGER = {
+    "type": "clipboard_pattern",
+    "pattern": r"https?://[^\s<>'\"]+",
+    "cooldown_seconds": 600,        # 10-min per-trigger cooldown
+    "require_confirmation": True,   # ask user before fetch
+    "destructive": False,           # read-only operation
+}
+
+
+import re
+import subprocess
+
+
+_URL_RE = re.compile(r"https?://[^\s<>'\"]+")
+
+
+def _read_clipboard() -> str:
+    """Read the system clipboard via `pbpaste` (macOS).
+
+    Returns empty string on any failure (no pbpaste, timeout, non-zero
+    exit, or non-text content). The skill handles empty-clipboard
+    gracefully by falling back to URLs in the task string.
+    """
+    try:
+        result = subprocess.run(
+            ["pbpaste"], capture_output=True, text=True, timeout=2
+        )
+        if result.returncode == 0:
+            return result.stdout
+    except (FileNotFoundError, subprocess.TimeoutExpired, Exception):
+        pass
+    return ""
+
+
+def _extract_url(text: str) -> str | None:
+    """Extract the first HTTP/HTTPS URL from `text`. Returns None if none."""
+    if not text:
+        return None
+    m = _URL_RE.search(text)
+    if not m:
+        return None
+    # Strip common trailing punctuation that's almost never part of a URL.
+    return m.group(0).rstrip(").,;'\"")
+
+
+def run(task: str = "", context: str = "") -> str:
+    """Fetch the URL from clipboard (or task text), return truncated content.
+
+    Parameters
+    ----------
+    task: str
+        For trigger auto-fires this is the rendered context string from
+        codec_triggers._render_task. For chat / voice / MCP invocations
+        this is the user's natural-language request, possibly containing
+        a URL inline.
+    context: str
+        Unused; accepted for skill-API uniformity.
+
+    Returns
+    -------
+    str
+        Either an error message ("...: no URL...") or
+        f"Fetched {url}:\n\n{content...}" where content is truncated
+        to 2000 chars with a tail summary.
+    """
+    # Try clipboard first (the trigger path), then fall back to URL in task.
+    text = _read_clipboard()
+    url = _extract_url(text)
+    if not url:
+        url = _extract_url(task)
+    if not url:
+        return "clipboard_url_fetch: no URL in clipboard or task"
+
+    # Delegate to the existing web_fetch skill for the actual HTTP work.
+    # Lazy import keeps this skill loadable even if web_fetch is missing
+    # from the runtime skills directory.
+    try:
+        import sys
+        from pathlib import Path
+        skills_dir = Path(__file__).resolve().parent
+        if str(skills_dir) not in sys.path:
+            sys.path.insert(0, str(skills_dir))
+        import web_fetch  # type: ignore
+        content = web_fetch.run(url)
+    except ImportError:
+        return f"clipboard_url_fetch: web_fetch skill not available; URL was {url}"
+    except Exception as e:
+        return f"clipboard_url_fetch: web_fetch failed for {url}: {e}"
+
+    if not isinstance(content, str):
+        content = str(content)
+
+    full_len = len(content)
+    if full_len > 2000:
+        content = content[:2000] + f"\n... [truncated, full length: {full_len} chars]"
+
+    return f"Fetched {url}:\n\n{content}"

tests/test_clipboard_url_fetch.py

@@ -0,0 +1,188 @@
+"""Tests for skills/clipboard_url_fetch.py — first real Step 6 trigger.
+
+15 tests covering:
+  - Metadata correctness (3): SKILL_NAME / SKILL_OBSERVATION_TRIGGER / chat allowlist
+  - Trigger schema validation (1): codec_triggers._validate_trigger_dict
+  - Trigger pattern semantics (3): matches https/http, rejects ftp / plain text
+  - URL extraction (3): basic, trailing punct, none
+  - run() behavior (5): no URL, URL via clipboard, URL via task, truncation, web_fetch failure
+
+All tests mock subprocess + web_fetch — never makes a real network call,
+never reads the user's actual clipboard.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+_REPO = Path(__file__).resolve().parents[1]
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+_SKILLS = _REPO / "skills"
+if str(_SKILLS) not in sys.path:
+    sys.path.insert(0, str(_SKILLS))
+
+import clipboard_url_fetch as cuf  # the skill
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Metadata correctness (3)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_metadata_constants():
+    assert cuf.SKILL_NAME == "clipboard_url_fetch"
+    assert cuf.SKILL_MCP_EXPOSE is False
+    assert "fetch clipboard url" in cuf.SKILL_TRIGGERS
+    assert isinstance(cuf.SKILL_OBSERVATION_TRIGGER, dict)
+
+
+def test_observation_trigger_has_required_keys():
+    """All Phase 2 Step 6 required keys present."""
+    t = cuf.SKILL_OBSERVATION_TRIGGER
+    for key in ("type", "pattern", "cooldown_seconds",
+                "require_confirmation", "destructive"):
+        assert key in t, f"missing required key: {key}"
+    assert t["type"] == "clipboard_pattern"
+    assert t["destructive"] is False
+    assert t["require_confirmation"] is True
+    assert t["cooldown_seconds"] >= 60  # not too aggressive
+
+
+def test_skill_in_chat_allowlist():
+    """clipboard_url_fetch must be allowed for chat-path dispatch.
+
+    Without this, the chat-path will silently drop the match and fall
+    through to LLM (same bug pattern as PR #13 for shift_report).
+    """
+    import codec_dashboard
+    assert "clipboard_url_fetch" in codec_dashboard.CHAT_SKILL_ALLOWLIST, (
+        "clipboard_url_fetch missing from CHAT_SKILL_ALLOWLIST — chat-path "
+        "dispatch will silently drop the match. See PR #13 hotfix history."
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Trigger schema validation (1)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_trigger_metadata_passes_codec_triggers_validation():
+    """Validate the SKILL_OBSERVATION_TRIGGER dict against the
+    codec_triggers schema validator. This locks the contract so any
+    schema change in codec_triggers will fail this test (and signal
+    that this skill needs an update)."""
+    from codec_triggers import _validate_trigger_dict
+    ok, why = _validate_trigger_dict(cuf.SKILL_OBSERVATION_TRIGGER)
+    assert ok, f"trigger dict invalid: {why}"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Trigger pattern semantics (3)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_pattern_matches_https():
+    import re
+    pattern = cuf.SKILL_OBSERVATION_TRIGGER["pattern"]
+    assert re.search(pattern, "https://example.com") is not None
+    assert re.search(pattern, "before https://test.com after") is not None
+
+
+def test_pattern_matches_http():
+    import re
+    pattern = cuf.SKILL_OBSERVATION_TRIGGER["pattern"]
+    assert re.search(pattern, "http://example.com/path?q=1") is not None
+
+
+def test_pattern_rejects_non_http_schemes_and_plain_text():
+    import re
+    pattern = cuf.SKILL_OBSERVATION_TRIGGER["pattern"]
+    assert re.search(pattern, "just plain text no url") is None
+    assert re.search(pattern, "ftp://example.com") is None
+    assert re.search(pattern, "ssh://host") is None
+    assert re.search(pattern, "") is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# URL extraction (3)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_extract_url_basic():
+    assert cuf._extract_url("check https://example.com") == "https://example.com"
+    assert cuf._extract_url("https://github.com/x/y") == "https://github.com/x/y"
+
+
+def test_extract_url_strips_trailing_punctuation():
+    """Trailing `).,;'"` chars are almost never part of the URL."""
+    assert cuf._extract_url("(see https://example.com)") == "https://example.com"
+    assert cuf._extract_url("https://example.com.") == "https://example.com"
+    assert cuf._extract_url("link: https://example.com;") == "https://example.com"
+
+
+def test_extract_url_returns_none_when_no_url():
+    assert cuf._extract_url("just text") is None
+    assert cuf._extract_url("") is None
+    assert cuf._extract_url(None) is None  # type: ignore
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# run() behavior (5)
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_run_no_url_anywhere(monkeypatch):
+    monkeypatch.setattr(cuf, "_read_clipboard", lambda: "no url here")
+    result = cuf.run("also no url")
+    assert "no URL" in result
+
+
+def test_run_url_from_clipboard(monkeypatch):
+    monkeypatch.setattr(cuf, "_read_clipboard", lambda: "https://example.com")
+
+    fake_web_fetch = MagicMock()
+    fake_web_fetch.run.return_value = "<html>page content here</html>"
+    monkeypatch.setitem(sys.modules, "web_fetch", fake_web_fetch)
+
+    result = cuf.run("[trigger fired]")
+    assert "https://example.com" in result
+    assert "page content here" in result
+    fake_web_fetch.run.assert_called_once_with("https://example.com")
+
+
+def test_run_url_from_task_when_clipboard_empty(monkeypatch):
+    monkeypatch.setattr(cuf, "_read_clipboard", lambda: "")
+
+    fake_web_fetch = MagicMock()
+    fake_web_fetch.run.return_value = "manual fetch"
+    monkeypatch.setitem(sys.modules, "web_fetch", fake_web_fetch)
+
+    result = cuf.run("please fetch https://test.com for me")
+    assert "https://test.com" in result
+    assert "manual fetch" in result
+
+
+def test_run_truncates_large_content(monkeypatch):
+    monkeypatch.setattr(cuf, "_read_clipboard", lambda: "https://big.example.com")
+
+    long_content = "x" * 5000
+    fake_web_fetch = MagicMock()
+    fake_web_fetch.run.return_value = long_content
+    monkeypatch.setitem(sys.modules, "web_fetch", fake_web_fetch)
+
+    result = cuf.run("")
+    assert "[truncated" in result
+    assert "5000" in result   # full-length number is reported
+    assert len(result) < 2500   # truncated body + header + tail
+
+
+def test_run_handles_web_fetch_failure(monkeypatch):
+    monkeypatch.setattr(cuf, "_read_clipboard", lambda: "https://broken.example.com")
+
+    fake_web_fetch = MagicMock()
+    fake_web_fetch.run.side_effect = RuntimeError("boom")
+    monkeypatch.setitem(sys.modules, "web_fetch", fake_web_fetch)
+
+    result = cuf.run("")
+    assert "web_fetch failed" in result
+    assert "boom" in result
+    assert "https://broken.example.com" in result