test: A-12 inline-LLM-POST guard + OAuth scope-escalation coverage (Fix #10)

- tests/test_a12_invariant.py: fails if a new inline `.post(...chat/completions
  ...)` appears outside codec_llm. Allowlists the documented vision sites
  (dashboard, watcher, both bridges, screenshot_text) + codec_core's generated
  session-script string. Runs in CI via the full pytest suite (no separate
  workflow step needed). Includes a synthetic-violation test proving the guard
  is not a no-op.
- tests/test_oauth_provider.py: regression tests for the refresh-token
  scope-escalation defense (requested scopes must be a subset of the original
  grant) — previously untested.

permission_gate mutation cases are already covered by test_agent_runner.py
(8 tests incl. D-5 traversal/symlink); not duplicated. The flaky time.sleep ->
threading.Event refactor is deferred — no specific flaky test identified, and
rewriting passing tests without evidence risks regressions.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: Mikarina13

tests/test_a12_invariant.py

@@ -0,0 +1,75 @@
+"""Fix #10: CI guard for the A-12 invariant.
+
+A-12 (see AGENTS.md §2) routed every chat/completions *text* call site onto
+codec_llm. The only inline `requests.post(.../chat/completions)` calls left
+are vision sites (pending A-11 cleanup) and codec_core's generated session
+script. This guard fails if a NEW inline chat/completions POST appears
+anywhere else — i.e. someone bypassed codec_llm.
+
+The detector matches the precise anti-pattern: a `.post(` whose first argument
+literally contains `chat/completions`. URL-in-a-variable callers (codec_llm,
+codec_vision) don't match and don't need allowlisting; that's intended — the
+guard targets the literal inline-POST shape that bypasses the canonical caller.
+"""
+import re
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+
+# Files permitted to contain an inline `.post(...chat/completions...)`.
+# Every entry is a vision site (pending A-11 migration onto codec_vision) or
+# codec_core's build_session_script, which EMITS the call as a string into the
+# generated session script (not a live POST in codec_core itself).
+_ALLOWLIST = {
+    "codec_dashboard.py",        # screen-vision POSTs (A-11 pending)
+    "codec_watcher.py",          # screen-vision POST (A-11 pending)
+    "codec_imessage.py",         # bridge vision POST (A-11 pending)
+    "codec_telegram.py",         # bridge vision POST (A-11 pending)
+    "skills/screenshot_text.py",  # OCR vision POST (A-11 pending)
+    "codec_core.py",             # generated session-script string, not a live POST
+}
+
+_INLINE_POST_RE = re.compile(r"\.post\s*\([^)]*chat/completions")
+_SKIP_PREFIXES = ("tests/", ".claude/", "scripts/")
+
+
+def _scan(root: Path) -> set:
+    """Return the set of repo-relative .py paths containing an inline
+    chat/completions POST."""
+    found = set()
+    for p in root.rglob("*.py"):
+        rel = p.relative_to(root).as_posix()
+        if rel.startswith(_SKIP_PREFIXES) or "__pycache__" in rel:
+            continue
+        try:
+            text = p.read_text(encoding="utf-8")
+        except OSError:
+            continue
+        if _INLINE_POST_RE.search(text):
+            found.add(rel)
+    return found
+
+
+def test_no_new_inline_llm_post_outside_codec_llm():
+    found = _scan(REPO)
+    offenders = found - _ALLOWLIST
+    assert not offenders, (
+        "New inline chat/completions POST(s) outside codec_llm "
+        f"(A-12 invariant violated): {sorted(offenders)}.\n"
+        "Route LLM text calls through codec_llm.call/stream/acall/astream. "
+        "If this is a legitimate vision site pending A-11, add it to the "
+        "documented _ALLOWLIST in this test with a reason."
+    )
+
+
+def test_a12_guard_actually_detects_a_violation(tmp_path):
+    # Proves the detector is not a no-op: a synthetic rogue inline POST is found.
+    (tmp_path / "rogue_skill.py").write_text(
+        "import requests\n"
+        "def run(t):\n"
+        '    return requests.post("http://127.0.0.1:8090/v1/chat/completions", json={}).text\n'
+    )
+    (tmp_path / "innocent.py").write_text("x = 1\n")
+    found = _scan(tmp_path)
+    assert "rogue_skill.py" in found, "guard failed to detect an inline chat/completions POST"
+    assert "innocent.py" not in found, "guard false-positived on an unrelated file"

tests/test_oauth_provider.py

@@ -154,6 +154,60 @@ def _counting_fsync(fd):
     assert mode == 0o600, f"fallback file must be 0600; got {oct(mode)}"
 
 
+# ── Fix #10: OAuth scope-escalation regression coverage ──────────────────────
+def test_refresh_rejects_scope_escalation(tmp_path):
+    """A refresh token issued for ['read'] must NOT be exchangeable for
+    ['read','write'] — exchange_refresh_token enforces requested ⊆ original.
+    Pins the scope-escalation defense (codec_oauth_provider.py:244-249) so a
+    future refactor that drops the subset check fails CI."""
+    import time
+
+    from mcp.server.auth.provider import RefreshToken, TokenError
+
+    p = PersistentOAuthProvider(
+        base_url="https://test.example.com",
+        client_registration_options=ClientRegistrationOptions(enabled=True),
+        state_path=tmp_path / "state.json",
+    )
+    client = _make_client("escal")
+    asyncio.run(p.register_client(client))
+    rt = RefreshToken(
+        token="codec_rt_orig",
+        client_id="escal",
+        scopes=["read"],
+        expires_at=int(time.time() + 3600),
+    )
+    with pytest.raises(TokenError) as exc:
+        asyncio.run(p.exchange_refresh_token(client, rt, ["read", "write"]))
+    assert "scope" in str(exc.value).lower()
+
+
+def test_refresh_allows_subset_scopes(tmp_path):
+    """A refresh for a SUBSET of the original scopes succeeds, and the new
+    access token carries only the narrowed scopes (no silent widening)."""
+    import time
+
+    from mcp.server.auth.provider import RefreshToken
+
+    p = PersistentOAuthProvider(
+        base_url="https://test.example.com",
+        client_registration_options=ClientRegistrationOptions(enabled=True),
+        state_path=tmp_path / "state.json",
+    )
+    client = _make_client("narrow")
+    asyncio.run(p.register_client(client))
+    rt = RefreshToken(
+        token="codec_rt_orig2",
+        client_id="narrow",
+        scopes=["read", "write"],
+        expires_at=int(time.time() + 3600),
+    )
+    p.refresh_tokens[rt.token] = rt  # register so the internal revoke is clean
+    token = asyncio.run(p.exchange_refresh_token(client, rt, ["read"]))
+    assert "read" in (token.scope or ""), "narrowed scope must include the requested scope"
+    assert "write" not in (token.scope or ""), "new token must not carry the dropped scope"
+
+
 if __name__ == "__main__":
     import tempfile
     with tempfile.TemporaryDirectory() as d: