omniroute_offload: free-llm, claude-free

free-llm sends big texts to the free-best OmniRoute chain behind a
fail-closed privacy gate (public repos, /tmp, /var/log; credential
scan). claude-free runs Claude Code with the full ~/.claude setup on
free-best. Skill offload-free in ~/.claude says when to use them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWYwGi2FJwfZ6QRB1im

.github/workflows/omniroute_offload-precommit.yml

@@ -0,0 +1,73 @@
+name: omniroute_offload pre-commit
+
+on:
+  push:
+    branches: [main]
+    # The workflow's own file counts as omniroute_offload: editing only this file would
+    # otherwise trigger nothing, so a fix to the job itself could never prove
+    # itself.
+    paths: ['omniroute_offload/**', '.github/workflows/omniroute_offload-precommit.yml']
+  pull_request:
+    branches: [main]
+    paths: ['omniroute_offload/**', '.github/workflows/omniroute_offload-precommit.yml']
+
+jobs:
+  pre-commit:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.x"
+      # Only pre-commit itself. Every linting hook runs in its own isolated
+      # env -- ruff, mypy, pylint, bandit and codespell each come from their
+      # own repo+rev, and the local hooks need nothing beyond bash and
+      # python3 -- so installing the full requirements.txt spent 9s a run on
+      # packages no pre-commit-stage hook imports. Verified by running all
+      # three configs green in a venv holding pre-commit and nothing else.
+      # The pin is read out of requirements.txt rather than repeated here, so
+      # it cannot drift from what the dependency-freshness gate governs.
+      - name: Install pre-commit
+        run: |
+          pip install "$(grep '^pre-commit==' omniroute_offload/requirements.txt)"
+
+      # Keyed on THIS package's config. pre-commit/action@v3.0.1 caches on
+      # hashFiles('.pre-commit-config.yaml') -- the repo ROOT's -- so all three
+      # package jobs shared one key, raced to overwrite it with hook sets built
+      # from four different configs, and editing a per-package config did not
+      # invalidate it. The cache was therefore silently wrong rather than
+      # missing, and the hooks reinstalled from scratch on every run (38-49s).
+      - uses: actions/cache@v4
+        with:
+          path: ~/.cache/pre-commit
+          key: >-
+            pre-commit-omniroute_offload-${{ env.pythonLocation }}-${{
+            hashFiles('omniroute_offload/.pre-commit-config.yaml') }}
+          # Prefix fallback, or this fix would evaporate on the very
+          # event that motivated it: bumping a hook rev changes the
+          # exact key, and a total miss means reinstalling every hook
+          # again. pre-commit's store is keyed internally by hook
+          # repo+rev, so a superseded entry is ignored disk rather
+          # than a wrong hook, and cache still re-saves under the new
+          # exact key after a prefix-only hit.
+          restore-keys: pre-commit-omniroute_offload-${{ env.pythonLocation }}-
+      # mypy re-analyses the stdlib and every third-party stub from scratch
+      # without this, which measured 10s of the hook step in CI against 0.25s
+      # warm locally -- a 19MB cache dir does the whole difference. Keyed on
+      # the config, not the sources: mypy's own cache metadata records the
+      # interpreter, the flag set and a hash per source file, so it re-checks
+      # what changed and cannot pass stale code. Deliberately NOT keyed on
+      # github.sha -- that re-saves 19MB x3 every commit and would evict the
+      # 68MB hook caches under the repo's 10GB LRU budget. A fixed key keeps
+      # the expensive part, the dependency analysis, which is what costs 10s.
+      - uses: actions/cache@v4
+        with:
+          path: .mypy_cache
+          key: >-
+            mypy-omniroute_offload-${{ env.pythonLocation }}-${{
+            hashFiles('omniroute_offload/.pre-commit-config.yaml') }}
+          restore-keys: mypy-omniroute_offload-${{ env.pythonLocation }}-
+      - name: Run the hooks
+        run: >-
+          pre-commit run --config omniroute_offload/.pre-commit-config.yaml
+          --all-files --show-diff-on-failure

omniroute_offload/.pre-commit-config.yaml

@@ -0,0 +1,143 @@
+# Pre-commit Configuration for omniroute_offload
+# Install:  pre-commit install && pre-commit install --hook-type pre-push
+# Run:      pre-commit run --all-files
+# Update:   pre-commit autoupdate
+
+# NOTE: this subproject lives inside the ~/src/utils monorepo, whose git root is
+# one level up. pre-commit always runs hooks from the git root, so every path
+# below is written relative to THAT, not to this directory, and the top-level
+# `files` key keeps these hooks off the sibling subprojects. CI invokes this
+# config exactly the same way:
+#   pre-commit run --config omniroute_offload/.pre-commit-config.yaml --all-files
+files: ^omniroute_offload/
+
+default_language_version:
+  python: python3
+default_stages: [pre-commit]
+fail_fast: false
+
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.6.0
+    hooks:
+      - id: trailing-whitespace
+        args: [--markdown-linebreak-ext=md]
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-json
+      - id: check-toml
+      - id: check-added-large-files
+        args: [--maxkb=2000]
+      - id: check-merge-conflict
+      - id: detect-private-key
+      - id: debug-statements
+      - id: name-tests-test
+        args: [--pytest-test-first]
+      - id: check-ast
+      - id: mixed-line-ending
+        args: [--fix=lf]
+      - id: requirements-txt-fixer
+
+  - repo: local
+    hooks:
+      - id: file-length-cap
+        name: file length <= 250 lines
+        entry: bash scripts/check_file_length.sh
+        language: system
+        pass_filenames: true
+      - id: subproject-scope
+        name: subproject scope regex still matches
+        entry: scripts/check_subproject_scope.sh omniroute_offload
+        language: script
+        # always_run so the top-level `files` filter cannot skip the one hook
+        # whose job is to notice that filter has stopped matching anything.
+        always_run: true
+        pass_filenames: false
+      - id: no-noqa
+        name: Block noqa comments
+        entry: '(?i)#\s*(noqa|type:\s*ignore)'
+        language: pygrep
+        types: [python]
+      - id: no-ruff-noqa
+        name: Block ruff noqa file-level comments
+        entry: '(?i)#\s*ruff:\s*noqa'
+        language: pygrep
+        types: [python]
+
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.17.2
+    hooks:
+      - id: ruff
+        args: [--fix, --unsafe-fixes, --exit-non-zero-on-fix, --show-fixes]
+        types_or: [python, pyi]
+      - id: ruff-format
+        types_or: [python, pyi]
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.13.0
+    hooks:
+      - id: mypy
+        args:
+          - --ignore-missing-imports
+          - --no-error-summary
+          - --disable-error-code=no-untyped-def
+          - --disable-error-code=no-untyped-call
+          - --disable-error-code=var-annotated
+          - --disable-error-code=no-any-unimported
+          - --disable-error-code=type-arg
+          - --disable-error-code=no-any-return
+          - --disable-error-code=misc
+          - --disable-error-code=unused-ignore
+          - --disable-error-code=unreachable
+          - --disable-error-code=assignment
+          - --disable-error-code=no-redef
+          - --disable-error-code=attr-defined
+          - --disable-error-code=arg-type
+          - --disable-error-code=union-attr
+          - --disable-error-code=call-overload
+          - --disable-error-code=return-value
+          - --disable-error-code=redundant-cast
+          - --disable-error-code=empty-body
+          - --disable-error-code=list-item
+
+  - repo: https://github.com/pylint-dev/pylint
+    rev: v4.0.8
+    hooks:
+      - id: pylint
+        args:
+          - --rcfile=omniroute_offload/pyproject.toml
+          - --fail-under=10
+          - --jobs=4
+        additional_dependencies:
+          - pytest
+
+  - repo: https://github.com/PyCQA/bandit
+    rev: 1.9.4
+    hooks:
+      - id: bandit
+        args:
+          - -c
+          - omniroute_offload/pyproject.toml
+          - --severity-level=high
+          - --confidence-level=medium
+          - --skip=B113
+        additional_dependencies: ["bandit[toml]"]
+        exclude: ^(omniroute_offload/omniroute_offload/tests/|.*test.*\.py$)
+
+  - repo: local
+    hooks:
+      - id: pytest-coverage
+        name: pytest with coverage enforcement
+        entry: scripts/run_subproject_tests.sh omniroute_offload
+        language: script
+        types: [python]
+        pass_filenames: false
+        require_serial: true
+        stages: [pre-push]
+
+  - repo: https://github.com/codespell-project/codespell
+    rev: v2.3.0
+    hooks:
+      - id: codespell
+        args:
+          - --skip=*.json,*.lock,.git,__pycache__,.venv

omniroute_offload/README.md

@@ -0,0 +1,3 @@
+# omniroute_offload
+
+`free-llm`, `claude-free` and a Read hook that offload easy work to the free `free-best` chain on local OmniRoute. Install: `./install.sh`. Usage: skill `offload-free`.

omniroute_offload/bin/claude-free

@@ -0,0 +1,25 @@
+#!/bin/bash
+# Claude Code with your full ~/.claude setup, but every model (main, subagents,
+# background) is the free-best chain on the local OmniRoute gateway. Free providers log
+# prompts, so it refuses to start outside public repos unless --force.
+set -euo pipefail
+
+readonly GW="http://127.0.0.1:20128"
+readonly MODEL="${CLAUDE_FREE_MODEL:-free-best}"
+KEY="$(cat "${HOME}/.config/omniroute/api_key")"
+readonly KEY
+
+if [[ "${1:-}" == --force ]]; then
+  shift
+else
+  free-llm-check-dir . || { echo "claude-free: only public repos, /tmp, /var/log (or --force)" >&2; exit 2; }
+fi
+
+curl -fsS -m 3 -o /dev/null "${GW}/v1/models" -H "Authorization: Bearer ${KEY}" ||
+  { echo "claude-free: gateway down; run ~/src/omniroute-tetris-test/start.sh" >&2; exit 3; }
+
+export ANTHROPIC_BASE_URL="${GW}" ANTHROPIC_AUTH_TOKEN="${KEY}" ANTHROPIC_MODEL="${MODEL}" \
+  ANTHROPIC_SMALL_FAST_MODEL="${MODEL}" ANTHROPIC_DEFAULT_OPUS_MODEL="${MODEL}" \
+  ANTHROPIC_DEFAULT_SONNET_MODEL="${MODEL}" ANTHROPIC_DEFAULT_HAIKU_MODEL="${MODEL}" \
+  CLAUDE_CODE_SUBAGENT_MODEL="${MODEL}"
+exec claude --model "${MODEL}" "$@"

omniroute_offload/install.sh

@@ -0,0 +1,13 @@
+#!/bin/bash
+# Install free-llm (stdlib-only, system python, editable) and link claude-free into ~/.local/bin.
+set -euo pipefail
+
+REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+readonly REPO_DIR
+readonly SYSTEM_PYTHON="/usr/bin/python3"
+readonly BIN_DIR="${HOME}/.local/bin"
+
+"${SYSTEM_PYTHON}" -m pip install --user --no-deps --break-system-packages -q -e "${REPO_DIR}"
+mkdir -p "${BIN_DIR}"
+ln -sf "${REPO_DIR}/bin/claude-free" "${BIN_DIR}/claude-free"
+command -v free-llm free-llm-read-hook free-llm-check-dir claude-free

omniroute_offload/omniroute_offload/__init__.py

@@ -0,0 +1 @@
+"""Offload easy LLM work to free models behind the local OmniRoute gateway."""

omniroute_offload/omniroute_offload/_cli.py

@@ -0,0 +1,182 @@
+"""free-llm: ask the free-model chain behind OmniRoute, return only the answer.
+
+Use it to keep bulky input (logs, big files, pages) out of Claude's context:
+the file goes to the free model, only its short answer comes back. Inputs over
+CHUNK_CHARS are split, answered per part, and merged by one final call.
+Exit: 0 answer printed | 2 refused (privacy) | 3 gateway/model failure.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import urllib.error
+import urllib.request
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+from omniroute_offload import _privacy as privacy
+
+ENDPOINT = "http://127.0.0.1:20128/v1/chat/completions"
+KEY_FILE = Path.home() / ".config/omniroute/api_key"
+DEFAULT_MODEL = "free-best"
+CHUNK_CHARS = 60_000
+TIMEOUT_S = 40
+EXIT_REFUSED = 2
+EXIT_GATEWAY = 3
+SYSTEM = (
+    "You are a careful assistant. Answer only from the given material, say so "
+    "when it does not contain the answer, and keep the answer short and exact."
+)
+
+
+class GatewayError(RuntimeError):
+    """The gateway or every model in the chain failed."""
+
+
+def parse(argv: list[str]) -> argparse.Namespace:
+    """Command-line options."""
+    parser = argparse.ArgumentParser(
+        prog="free-llm", description=__doc__.splitlines()[0]
+    )
+    parser.add_argument("prompt", help="question or instruction")
+    parser.add_argument(
+        "-f",
+        "--file",
+        action="append",
+        default=[],
+        type=Path,
+        help="attach a file (repeatable)",
+    )
+    parser.add_argument("-m", "--model", default=DEFAULT_MODEL)
+    parser.add_argument("--max-tokens", type=int, default=4000)
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="skip path/repo checks (never the credential scan)",
+    )
+    return parser.parse_args(argv)
+
+
+def gather(args: argparse.Namespace) -> str:
+    """Attached files plus piped stdin as one labelled text; raises PermissionError."""
+    parts = []
+    for path in args.file:
+        reason = None if args.force else privacy.file_refusal(path, home=Path.home())
+        if reason:
+            raise PermissionError(reason)
+        try:
+            parts.append(f"=== {path.name} ===\n{path.read_text(errors='replace')}")
+        except OSError as err:
+            msg = f"cannot read {path}: {err.strerror}"
+            raise PermissionError(msg) from err
+    if not sys.stdin.isatty():
+        piped = sys.stdin.read()
+        if piped:
+            parts.append(f"=== stdin ===\n{piped}")
+    material = "\n\n".join(parts)
+    reason = privacy.secret_refusal(f"{material}\n{args.prompt}")
+    if reason:
+        raise PermissionError(reason)
+    return material
+
+
+def split_chunks(text: str, size: int) -> list[str]:
+    """Split into pieces of at most size chars, cutting at line breaks when possible."""
+    chunks = []
+    while text:
+        cut = min(len(text), size)
+        newline = text.rfind("\n", 0, cut)
+        if 0 < newline < cut < len(text):
+            cut = newline + 1
+        chunks.append(text[:cut])
+        text = text[cut:]
+    return chunks
+
+
+def ask(content: str, args: argparse.Namespace) -> str:
+    """One chat call through the gateway; returns the answer text."""
+    body = {
+        "model": args.model,
+        "max_tokens": args.max_tokens,
+        "messages": [
+            {"role": "system", "content": SYSTEM},
+            {"role": "user", "content": content},
+        ],
+    }
+    req = urllib.request.Request(
+        ENDPOINT,
+        data=json.dumps(body).encode(),
+        headers={
+            "content-type": "application/json",
+            "authorization": f"Bearer {KEY_FILE.read_text().strip()}",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=TIMEOUT_S) as resp:
+            reply = json.load(resp)
+    except urllib.error.HTTPError as err:
+        with err:
+            detail = err.read().decode(errors="replace")[:200]
+        msg = f"gateway {err.code}: {detail}"
+        raise GatewayError(msg) from err
+    except urllib.error.URLError as err:
+        msg = (
+            f"gateway unreachable ({err.reason}); start it with "
+            "~/src/omniroute-tetris-test/start.sh"
+        )
+        raise GatewayError(msg) from err
+    text = str(reply["choices"][0]["message"].get("content") or "").strip()
+    if not text:
+        msg = f"empty answer from {reply.get('model')}"
+        raise GatewayError(msg)
+    usage = reply.get("usage", {})
+    print(
+        f"[free-llm {reply.get('model')} in={usage.get('prompt_tokens')} "
+        f"out={usage.get('completion_tokens')}]",
+        file=sys.stderr,
+    )
+    return text
+
+
+def answer(material: str, args: argparse.Namespace) -> str:
+    """Answer directly, or map over chunks and merge the partial answers."""
+    if len(material) <= CHUNK_CHARS:
+        return ask(f"{material}\n\n{args.prompt}" if material else args.prompt, args)
+    chunks = split_chunks(material, CHUNK_CHARS)
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        partial = list(
+            pool.map(
+                lambda c: ask(
+                    f"{c}\n\n(This is one part of a larger input.) {args.prompt}", args
+                ),
+                chunks,
+            )
+        )
+    merged = "\n\n".join(
+        f"--- part {i}/{len(partial)} ---\n{p}" for i, p in enumerate(partial, 1)
+    )
+    return ask(
+        f"Partial answers, one per part of the input:\n{merged}\n\n"
+        f"Merge them into one answer to: {args.prompt}",
+        args,
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Entry point for the free-llm command."""
+    args = parse(sys.argv[1:] if argv is None else argv)
+    try:
+        print(answer(gather(args), args))
+    except PermissionError as err:
+        print(f"free-llm refused: {err}", file=sys.stderr)
+        return EXIT_REFUSED
+    except GatewayError as err:
+        print(f"free-llm failed: {err}", file=sys.stderr)
+        return EXIT_GATEWAY
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

omniroute_offload/omniroute_offload/_hook.py

@@ -0,0 +1,67 @@
+"""PreToolUse hook on Read: nudge to free-llm before reading a big shareable file.
+
+Never blocks (editing a big file still needs a Read); only adds context when a
+whole-file Read of a file over MIN_BYTES would pass the free-llm privacy gate.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from omniroute_offload import _privacy as privacy
+
+MIN_BYTES = 60_000
+NUDGE = (
+    "{path} is {kb} KB (~{tokens}k tokens). If you only need to understand or "
+    'summarize it (not edit it), run `free-llm -f {path} "<narrow question>"` '
+    "(skill offload-free) and read only its answer; if a pattern suffices, grep it. "
+    "Read it whole only if you must edit it or need exact text."
+)
+
+
+def nudge(payload: object) -> str | None:
+    """Context to add for this tool call, or None to stay silent."""
+    if not isinstance(payload, dict) or payload.get("tool_name") != "Read":
+        return None
+    tool_input = payload.get("tool_input")
+    if (
+        not isinstance(tool_input, dict)
+        or "limit" in tool_input
+        or "offset" in tool_input
+    ):
+        return None
+    path = Path(str(tool_input.get("file_path", "")))
+    try:
+        size = path.stat().st_size
+    except OSError:
+        return None
+    if size <= MIN_BYTES or privacy.file_refusal(path, home=Path.home()):
+        return None
+    return NUDGE.format(path=path, kb=size // 1000, tokens=size // 4000)
+
+
+def main() -> int:
+    """Read the hook JSON from stdin, print optional context JSON, always exit 0."""
+    try:
+        payload = json.loads(sys.stdin.read())
+    except ValueError:
+        return 0
+    context = nudge(payload)
+    if context:
+        print(
+            json.dumps(
+                {
+                    "hookSpecificOutput": {
+                        "hookEventName": "PreToolUse",
+                        "additionalContext": context,
+                    }
+                }
+            )
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

omniroute_offload/omniroute_offload/_privacy.py

@@ -0,0 +1,148 @@
+"""Fail-closed checks on what may be sent to free (logging, maybe training) providers.
+
+A file may go out only if it lives in a PUBLIC GitHub repo, or under /tmp or
+/var/log, and is not a secret-shaped file. Every text is scanned for credential
+patterns; that scan has no override.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+import sys
+import time
+from fnmatch import fnmatch
+from pathlib import Path
+
+CACHE = Path.home() / ".cache/omniroute_offload/visibility.json"
+CACHE_TTL_S = 86400
+SCRATCH_ROOTS = (Path("/tmp"), Path("/var/log"))
+DENIED_HOME_DIRS = (".ssh", ".config", ".gnupg", ".aws", ".claude")
+DENIED_NAMES = (
+    ".env",
+    ".env.*",
+    "*.pem",
+    "*.key",
+    "id_rsa*",
+    "id_ed25519*",
+    "id_ecdsa*",
+    "*secret*",
+    "*credential*",
+    "*.p12",
+    "*.pfx",
+    "*.keystore",
+    "*.jks",
+    ".netrc",
+)
+SECRET_PATTERNS = re.compile(
+    r"sk-[A-Za-z0-9_-]{20,}|gh[pousr]_[A-Za-z0-9]{30,}|github_pat_[A-Za-z0-9_]{20,}"
+    r"|AKIA[0-9A-Z]{16}|-----BEGIN [A-Z ]*PRIVATE KEY-----|AIza[0-9A-Za-z_-]{35}"
+    r"|xox[abprs]-[A-Za-z0-9-]{10,}"
+)
+GITHUB_REMOTE = re.compile(r"github\.com[:/]([^/\s]+/[^/\s]+?)(?:\.git)?/?$")
+
+
+def secret_refusal(text: str) -> str | None:
+    """Reason to refuse when the text contains something credential-shaped."""
+    hit = SECRET_PATTERNS.search(text)
+    return f"looks like a credential ({hit.group()[:8]}...)" if hit else None
+
+
+def path_refusal(path: Path, *, home: Path) -> str | None:
+    """Reason to refuse by file name or location alone (repo visibility aside)."""
+    if any(fnmatch(path.name, pattern) for pattern in DENIED_NAMES):
+        return f"{path}: secret-shaped file name"
+    for sub in DENIED_HOME_DIRS:
+        if path.is_relative_to(home / sub):
+            return f"{path}: under ~/{sub}"
+    if _repo_root(path) or any(path.is_relative_to(root) for root in SCRATCH_ROOTS):
+        return None
+    return f"{path}: not in a public repo, /tmp or /var/log"
+
+
+def file_refusal(path: Path, *, home: Path) -> str | None:
+    """Check one file; inside any repo (even under /tmp) its visibility decides."""
+    path = path.resolve()
+    reason = path_refusal(path, home=home)
+    root = _repo_root(path)
+    if reason or root is None:
+        return reason
+    return repo_refusal(root)
+
+
+def github_slug(remote: str) -> str | None:
+    """owner/repo of a GitHub remote URL, else None."""
+    match = GITHUB_REMOTE.search(remote.strip())
+    return match.group(1) if match else None
+
+
+def repo_refusal(root: Path) -> str | None:
+    """Reason to refuse unless the repo's GitHub origin is PUBLIC (cached a day)."""
+    remote = _git(root, "remote", "get-url", "origin")
+    slug = github_slug(remote) if remote else None
+    if slug is None:
+        return f"{root}: repo has no GitHub origin"
+    visibility = _cached_visibility(slug)
+    if visibility == "PUBLIC":
+        return None
+    return f"{root}: {slug} is {visibility or 'unknown (gh failed)'}"
+
+
+def _cached_visibility(slug: str) -> str | None:
+    try:
+        cache = json.loads(CACHE.read_text())
+    except OSError, ValueError:
+        cache = {}
+    entry = cache.get(slug)
+    if entry and time.time() - entry["at"] < CACHE_TTL_S:
+        return str(entry["visibility"])
+    visibility = _gh_visibility(slug)
+    if visibility is None:
+        return None
+    cache[slug] = {"visibility": visibility, "at": time.time()}
+    CACHE.parent.mkdir(parents=True, exist_ok=True)
+    CACHE.write_text(json.dumps(cache))
+    return visibility
+
+
+def _gh_visibility(slug: str) -> str | None:
+    try:
+        done = subprocess.run(
+            ["gh", "repo", "view", slug, "--json", "visibility"],
+            capture_output=True,
+            text=True,
+            timeout=15,
+            check=False,
+        )
+    except OSError, subprocess.TimeoutExpired:
+        return None
+    if done.returncode:
+        return None
+    return str(json.loads(done.stdout)["visibility"])
+
+
+def _repo_root(path: Path) -> Path | None:
+    start = path if path.is_dir() else path.parent
+    while not start.exists():
+        start = start.parent
+    top = _git(start, "rev-parse", "--show-toplevel")
+    return Path(top) if top else None
+
+
+def _git(cwd: Path, *args: str) -> str | None:
+    done = subprocess.run(
+        ["git", "-C", str(cwd), *args], capture_output=True, text=True, check=False
+    )
+    return done.stdout.strip() if not done.returncode and done.stdout.strip() else None
+
+
+def check_dir_main(argv: list[str] | None = None) -> int:
+    """free-llm-check-dir <dir>: exit 0 if its files may go to free models, else 2."""
+    args = sys.argv[1:] if argv is None else argv
+    target = Path(args[0] if args else ".").resolve()
+    reason = file_refusal(target, home=Path.home())
+    if reason:
+        print(f"refused: {reason}", file=sys.stderr)
+        return 2
+    return 0

omniroute_offload/omniroute_offload/tests/test_cli.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+import io
+import json
+import sys
+import urllib.error
+import urllib.request
+from email.message import Message
+from pathlib import Path
+from typing import Any, Self
+
+import pytest
+
+from omniroute_offload import _cli as cli
+from omniroute_offload import _privacy as privacy
+
+
+class FakeResp(io.BytesIO):
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(self, *_exc: object) -> None:
+        self.close()
+
+
+def _reply(text: str, model: str = "agnes-2.5-flash") -> FakeResp:
+    body = {
+        "model": model,
+        "choices": [{"message": {"content": text}}],
+        "usage": {"prompt_tokens": 10, "completion_tokens": 2},
+    }
+    return FakeResp(json.dumps(body).encode())
+
+
+@pytest.fixture
+def sent(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> list[dict[str, Any]]:
+    bodies: list[dict[str, Any]] = []
+
+    def fake_urlopen(req: Any, timeout: float) -> FakeResp:
+        bodies.append(json.loads(req.data))
+        return _reply(f"answer{len(bodies)}")
+
+    monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen)
+    key = tmp_path / "api_key"
+    key.write_text("k\n")
+    monkeypatch.setattr(cli, "KEY_FILE", key)
+    monkeypatch.setattr(sys, "stdin", io.StringIO(""))
+    return bodies
+
+
+def test_prompt_only(
+    sent: list[dict[str, Any]], capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert cli.main(["say hi"]) == 0
+    out, err = capsys.readouterr()
+    assert out == "answer1\n"
+    assert "agnes-2.5-flash" in err
+    assert sent[0]["model"] == "free-best"
+    assert sent[0]["messages"][-1]["content"] == "say hi"
+
+
+def test_file_is_attached(sent: list[dict[str, Any]], tmp_path: Path) -> None:
+    log = Path("/tmp") / f"free_llm_test_{tmp_path.name}.log"
+    log.write_text("ERROR one\n")
+    try:
+        assert cli.main(["-f", str(log), "list errors"]) == 0
+    finally:
+        log.unlink()
+    content = sent[0]["messages"][-1]["content"]
+    assert "ERROR one" in content
+    assert content.endswith("list errors")
+
+
+def test_stdin_is_attached(
+    sent: list[dict[str, Any]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sys, "stdin", io.StringIO("piped text"))
+    assert cli.main(["--max-tokens", "50", "-m", "nvidia/x", "sum"]) == 0
+    assert "piped text" in sent[0]["messages"][-1]["content"]
+    assert sent[0]["max_tokens"] == 50
+    assert sent[0]["model"] == "nvidia/x"
+
+
+def test_refused_file(
+    sent: list[dict[str, Any]], capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert cli.main(["-f", str(Path.home() / ".ssh/id_rsa"), "x"]) == cli.EXIT_REFUSED
+    assert "refused" in capsys.readouterr().err
+    assert sent == []
+
+
+def test_force_skips_path_but_not_secrets(
+    sent: list[dict[str, Any]],
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture[str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    doc = tmp_path / "notes.txt"
+    doc.write_text("plain")
+    monkeypatch.setattr(privacy, "SCRATCH_ROOTS", ())
+    assert cli.main(["-f", str(doc), "x"]) == cli.EXIT_REFUSED
+    assert cli.main(["--force", "-f", str(doc), "x"]) == 0
+    doc.write_text("AKIA" + "A" * 16)
+    assert cli.main(["--force", "-f", str(doc), "x"]) == cli.EXIT_REFUSED
+    assert len(sent) == 1
+    capsys.readouterr()
+
+
+def test_missing_file(
+    sent: list[dict[str, Any]], capsys: pytest.CaptureFixture[str]
+) -> None:
+    assert cli.main(["--force", "-f", "/nonexistent/x.log", "x"]) == cli.EXIT_REFUSED
+    assert "cannot read" in capsys.readouterr().err
+
+
+def test_chunked_map_reduce(
+    sent: list[dict[str, Any]],
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr(cli, "CHUNK_CHARS", 10)
+    monkeypatch.setattr(sys, "stdin", io.StringIO("a" * 25))
+    assert cli.main(["find"]) == 0
+    final = sent[-1]["messages"][-1]["content"]
+    parts = int(final.split("part 1/")[1].split()[0])
+    assert parts > 1
+    assert len(sent) == parts + 1
+    assert final.endswith("find")
+    assert capsys.readouterr().out == f"answer{parts + 1}\n"
+
+
+def test_split_prefers_line_breaks() -> None:
+    assert cli.split_chunks("aaaa\nbbbb\ncc", 6) == ["aaaa\n", "bbbb\n", "cc"]
+    assert cli.split_chunks("abcdefgh", 3) == ["abc", "def", "gh"]
+    assert cli.split_chunks("", 3) == []
+
+
+def test_gateway_http_error(
+    monkeypatch: pytest.MonkeyPatch,
+    sent: list[dict[str, Any]],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    def fail(req: Any, timeout: float) -> FakeResp:
+        raise urllib.error.HTTPError(
+            cli.ENDPOINT, 429, "busy", Message(), io.BytesIO(b'{"error":"cooling"}')
+        )
+
+    monkeypatch.setattr(urllib.request, "urlopen", fail)
+    assert cli.main(["x"]) == cli.EXIT_GATEWAY
+    assert "429" in capsys.readouterr().err
+
+
+def test_gateway_down(
+    monkeypatch: pytest.MonkeyPatch,
+    sent: list[dict[str, Any]],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    def down(req: Any, timeout: float) -> FakeResp:
+        msg = "refused"
+        raise urllib.error.URLError(msg)
+
+    monkeypatch.setattr(urllib.request, "urlopen", down)
+    assert cli.main(["x"]) == cli.EXIT_GATEWAY
+    assert "start.sh" in capsys.readouterr().err
+
+
+def test_empty_answer_is_an_error(
+    monkeypatch: pytest.MonkeyPatch,
+    sent: list[dict[str, Any]],
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout: _reply(""))
+    assert cli.main(["x"]) == cli.EXIT_GATEWAY
+    assert "empty" in capsys.readouterr().err
+
+
+def test_privacy_module_used(
+    monkeypatch: pytest.MonkeyPatch, sent: list[dict[str, Any]]
+) -> None:
+    monkeypatch.setattr(privacy, "secret_refusal", lambda _t: "secret!")
+    assert cli.main(["x"]) == cli.EXIT_REFUSED
+
+
+class Tty(io.StringIO):
+    def isatty(self) -> bool:
+        return True
+
+
+def test_tty_stdin_is_not_read(
+    sent: list[dict[str, Any]], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(sys, "stdin", Tty("must not be sent"))
+    assert cli.main(["hi"]) == 0
+    assert "must not be sent" not in sent[0]["messages"][-1]["content"]

omniroute_offload/omniroute_offload/tests/test_hook.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import io
+import json
+import sys
+from typing import TYPE_CHECKING
+
+import pytest
+
+from omniroute_offload import _hook as hook
+from omniroute_offload import _privacy as privacy
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _run(monkeypatch: pytest.MonkeyPatch, payload: object) -> str:
+    monkeypatch.setattr(sys, "stdin", io.StringIO(json.dumps(payload)))
+    out = io.StringIO()
+    monkeypatch.setattr(sys, "stdout", out)
+    assert hook.main() == 0
+    return out.getvalue()
+
+
+def _read(path: Path, **extra: object) -> dict[str, object]:
+    return {"tool_name": "Read", "tool_input": {"file_path": str(path), **extra}}
+
+
+def test_big_public_file_gets_nudge(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    doc = tmp_path / "doc.md"
+    doc.write_text("x" * (hook.MIN_BYTES + 1))
+    out = json.loads(_run(monkeypatch, _read(doc)))
+    ctx = out["hookSpecificOutput"]["additionalContext"]
+    assert out["hookSpecificOutput"]["hookEventName"] == "PreToolUse"
+    assert "free-llm" in ctx
+    assert str(doc) in ctx
+
+
+def test_small_file_is_silent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    doc = tmp_path / "doc.md"
+    doc.write_text("small")
+    assert _run(monkeypatch, _read(doc)) == ""
+
+
+def test_ranged_read_is_silent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    doc = tmp_path / "doc.md"
+    doc.write_text("x" * (hook.MIN_BYTES + 1))
+    assert _run(monkeypatch, _read(doc, limit=50)) == ""
+
+
+def test_private_file_is_silent(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    doc = tmp_path / "doc.md"
+    doc.write_text("x" * (hook.MIN_BYTES + 1))
+    monkeypatch.setattr(privacy, "file_refusal", lambda _p, home: "private")
+    assert _run(monkeypatch, _read(doc)) == ""
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        "not a dict",
+        {"tool_name": "Bash", "tool_input": {}},
+        {"tool_name": "Read"},
+        {"tool_name": "Read", "tool_input": {"file_path": "/nonexistent/x"}},
+    ],
+)
+def test_odd_input_is_silent(monkeypatch: pytest.MonkeyPatch, payload: object) -> None:
+    assert _run(monkeypatch, payload) == ""
+
+
+def test_invalid_json_is_silent(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(sys, "stdin", io.StringIO("{oops"))
+    assert hook.main() == 0

omniroute_offload/omniroute_offload/tests/test_privacy.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+from omniroute_offload import _privacy as privacy
+
+
+def _git_repo(path: Path, remote: str | None) -> Path:
+    subprocess.run(["git", "init", "-q", str(path)], check=True)
+    if remote:
+        subprocess.run(
+            ["git", "-C", str(path), "remote", "add", "origin", remote], check=True
+        )
+    return path
+
+
+@pytest.fixture
+def cache(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    path = tmp_path / "cache.json"
+    monkeypatch.setattr(privacy, "CACHE", path)
+    return path
+
+
+@pytest.mark.parametrize(
+    "name",
+    [
+        ".env",
+        ".env.local",
+        "server.pem",
+        "id_rsa",
+        "id_ed25519.pub",
+        "api.key",
+        "client_secret.json",
+        "credentials.json",
+        "app.keystore",
+    ],
+)
+def test_denied_file_names(tmp_path: Path, name: str) -> None:
+    assert privacy.path_refusal(tmp_path / name, home=tmp_path / "home") is not None
+
+
+@pytest.mark.parametrize("sub", [".ssh", ".config", ".gnupg", ".aws", ".claude"])
+def test_denied_home_dirs(tmp_path: Path, sub: str) -> None:
+    home = tmp_path / "home"
+    reason = privacy.path_refusal(home / sub / "x.txt", home=home)
+    assert reason is not None
+    assert sub in reason
+
+
+def test_outside_repo_only_tmp_and_var_log() -> None:
+    home = Path("/nonexistent-home")
+    assert privacy.path_refusal(Path("/tmp/big.log"), home=home) is None
+    assert privacy.path_refusal(Path("/var/log/pacman.log"), home=home) is None
+    reason = privacy.path_refusal(home / "Documents/notes.txt", home=home)
+    assert reason is not None
+    assert "not in a public repo" in reason
+
+
+def test_repo_without_remote_is_refused(tmp_path: Path, cache: Path) -> None:
+    repo = _git_repo(tmp_path / "r", None)
+    assert privacy.repo_refusal(repo) == f"{repo}: repo has no GitHub origin"
+
+
+def test_non_github_remote_is_refused(tmp_path: Path, cache: Path) -> None:
+    repo = _git_repo(tmp_path / "r", "https://gitlab.com/a/b.git")
+    assert "no GitHub origin" in str(privacy.repo_refusal(repo))
+
+
+@pytest.mark.parametrize(
+    ("remote", "slug"),
+    [
+        ("https://github.com/kuhyx/utils.git", "kuhyx/utils"),
+        ("git@github.com:kuhyx/utils.git", "kuhyx/utils"),
+        ("https://github.com/kuhyx/utils", "kuhyx/utils"),
+    ],
+)
+def test_github_slug(remote: str, slug: str) -> None:
+    assert privacy.github_slug(remote) == slug
+
+
+def test_public_repo_allowed_and_cached(
+    tmp_path: Path, cache: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    repo = _git_repo(tmp_path / "r", "https://github.com/kuhyx/utils.git")
+    calls: list[str] = []
+
+    def fake(slug: str) -> str | None:
+        calls.append(slug)
+        return "PUBLIC"
+
+    monkeypatch.setattr(privacy, "_gh_visibility", fake)
+    assert privacy.repo_refusal(repo) is None
+    assert privacy.repo_refusal(repo) is None
+    assert calls == ["kuhyx/utils"]
+    assert json.loads(cache.read_text())["kuhyx/utils"]["visibility"] == "PUBLIC"
+
+
+def test_stale_cache_is_refreshed(
+    tmp_path: Path, cache: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    repo = _git_repo(tmp_path / "r", "https://github.com/kuhyx/x.git")
+    cache.write_text(
+        json.dumps({"kuhyx/x": {"visibility": "PUBLIC", "at": time.time() - 2 * 86400}})
+    )
+    monkeypatch.setattr(privacy, "_gh_visibility", lambda _slug: "PRIVATE")
+    assert "PRIVATE" in str(privacy.repo_refusal(repo))
+
+
+def test_unknown_visibility_fails_closed(
+    tmp_path: Path, cache: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    repo = _git_repo(tmp_path / "r", "https://github.com/kuhyx/x.git")
+    monkeypatch.setattr(privacy, "_gh_visibility", lambda _slug: None)
+    assert "unknown" in str(privacy.repo_refusal(repo))
+    assert not cache.exists()
+
+
+def test_corrupt_cache_is_ignored(
+    tmp_path: Path, cache: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    repo = _git_repo(tmp_path / "r", "https://github.com/kuhyx/utils.git")
+    cache.write_text("{not json")
+    monkeypatch.setattr(privacy, "_gh_visibility", lambda _slug: "PUBLIC")
+    assert privacy.repo_refusal(repo) is None
+
+
+def test_gh_visibility_parses_and_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    def ok(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]:
+        return subprocess.CompletedProcess([], 0, '{"visibility":"PUBLIC"}', "")
+
+    def bad(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]:
+        return subprocess.CompletedProcess([], 1, "", "not found")
+
+    def boom(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]:
+        raise FileNotFoundError
+
+    monkeypatch.setattr(subprocess, "run", ok)
+    assert privacy._gh_visibility("a/b") == "PUBLIC"
+    monkeypatch.setattr(subprocess, "run", bad)
+    assert privacy._gh_visibility("a/b") is None
+    monkeypatch.setattr(subprocess, "run", boom)
+    assert privacy._gh_visibility("a/b") is None
+
+
+def test_file_refusal_combines_path_and_repo(
+    tmp_path: Path, cache: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    home = tmp_path / "home"
+    repo = _git_repo(home / "src/pub", "https://github.com/kuhyx/utils.git")
+    (repo / "a.py").write_text("x")
+    monkeypatch.setattr(privacy, "_gh_visibility", lambda _slug: "PUBLIC")
+    assert privacy.file_refusal(repo / "a.py", home=home) is None
+    (repo / ".env").write_text("x")
+    assert privacy.file_refusal(repo / ".env", home=home) is not None
+    priv = _git_repo(home / "src/priv", "https://github.com/kuhyx/secret.git")
+    (priv / "b.py").write_text("x")
+    monkeypatch.setattr(privacy, "_gh_visibility", lambda _slug: "PRIVATE")
+    assert "PRIVATE" in str(privacy.file_refusal(priv / "b.py", home=home))
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        "key sk-abcdefghijklmnopqrstuvwx",
+        "ghp_" + "a" * 36,
+        "github_pat_" + "a" * 30,
+        "AKIA" + "A" * 16,
+        "-----BEGIN OPENSSH " + "PRIVATE KEY-----",
+        "AIza" + "a" * 35,
+        "xoxb-1234567890-abc",
+    ],
+)
+def test_secret_scan_hits(text: str) -> None:
+    assert privacy.secret_refusal(f"log line\n{text}\n") is not None
+
+
+def test_secret_scan_clean() -> None:
+    assert privacy.secret_refusal("ERROR: disk full at 12:00\nsk-short") is None
+
+
+def test_check_dir_main(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cache: Path
+) -> None:
+    assert privacy.check_dir_main([str(tmp_path)]) == 0
+    monkeypatch.setattr(privacy, "SCRATCH_ROOTS", ())
+    assert privacy.check_dir_main([str(tmp_path)]) == 2
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr(sys, "argv", ["free-llm-check-dir"])
+    assert privacy.check_dir_main() == 2

omniroute_offload/pyproject.toml

@@ -0,0 +1,105 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "omniroute_offload"
+version = "0.1.0"
+description = "Offload easy LLM work to free models behind the local OmniRoute gateway"
+requires-python = ">=3.14"
+dependencies = []
+
+[project.scripts]
+free-llm = "omniroute_offload._cli:main"
+free-llm-read-hook = "omniroute_offload._hook:main"
+free-llm-check-dir = "omniroute_offload._privacy:check_dir_main"
+
+[tool.setuptools.packages.find]
+include = ["omniroute_offload*"]
+
+[tool.setuptools.package-data]
+omniroute_offload = ["py.typed"]
+
+[tool.ruff]
+target-version = "py314"
+include = ["*.py", "**/*.py"]
+exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"]
+
+[tool.ruff.lint]
+select = ["ALL"]
+ignore = [
+    "D203", "D213", "COM812", "ISC001", "S603", "S607", "T201",
+    # No per-file copyright headers in this monorepo; /tmp is an allowed source, not a temp file.
+    "CPY001", "S108",
+    # urlopen only ever gets the fixed local gateway URL.
+    "S310",
+]
+fixable = ["ALL"]
+unfixable = []
+
+[tool.ruff.lint.per-file-ignores]
+"**/tests/**/*.py" = ["ANN401", "ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
+
+[tool.pytest.ini_options]
+testpaths = ["omniroute_offload/tests"]
+addopts = [
+    "-q",
+    "--strict-markers",
+    "--strict-config",
+    "-ra",
+    "--cov=omniroute_offload",
+    "--cov-branch",
+    "--cov-report=term-missing",
+    "--cov-report=lcov",
+]
+filterwarnings = ["error", "ignore::DeprecationWarning", "ignore::pytest.PytestRemovedIn10Warning"]
+
+[tool.coverage.run]
+source = ["omniroute_offload"]
+branch = true
+omit = ["*/__pycache__/*", "*/tests/*"]
+
+[tool.coverage.report]
+fail_under = 100
+show_missing = true
+exclude_lines = ["pragma: no cover", 'if __name__ == "__main__":']
+
+[tool.pylint.main]
+# The repo root on the path, so pylint resolves the package's own imports the
+# same way pytest does instead of reporting import-error for them.
+init-hook = "import sys; sys.path.insert(0, 'omniroute_offload')"
+analyse-fallback-blocks = true
+persistent = true
+jobs = 0
+py-version = "3.14"
+ignore = [".venv", "__pycache__"]
+# Test suites intentionally use patterns (protected-access, fixture-arg
+# shadowing) that don't apply to source code; ruff's per-file-ignores carry
+# the equivalent exemptions, so tests stay linted -- by ruff. This is the same
+# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has
+# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the
+# basename of what pylint is *given*, and pre-commit hands it explicit file
+# paths, so a directory name in `ignore` never fires under the hook.
+ignore-paths = ["^(.*/)?tests/.*$"]
+ignore-patterns = [".*\\.pyi$"]
+
+[tool.pylint.messages_control]
+enable = "all"
+disable = []
+
+[tool.pylint.design]
+min-public-methods = 0
+max-module-lines = 1000
+max-attributes = 10
+
+[tool.pylint.typecheck]
+generated-members = [
+    ".*\\.assert_called_once_with",
+    ".*\\.assert_called_once",
+    ".*\\.assert_called",
+    ".*\\.assert_not_called",
+    ".*\\.assert_any_call",
+    ".*\\.call_args",
+    ".*\\.call_args_list",
+    ".*\\.call_count",
+]

omniroute_offload/requirements.txt

@@ -0,0 +1,10 @@
+# Exact pins, enforced by the shared dependency-freshness gate: a `>=`
+# floor is satisfied by anything newer, so it tells a checker nothing
+# about whether this repo is actually current.
+bandit[toml]==1.9.4
+mypy==2.4.1
+pre-commit==4.7.0
+pylint==4.1.3
+pytest==9.2.0
+pytest-cov==7.1.0
+ruff==0.17.3