feat(colab): CTI Colab Gemini mapping + ac_refresh_colab_contexts upkeep skill (PyAutoHands#324)

.claude/commands/ac_refresh_colab_contexts.md

@@ -0,0 +1 @@
+../../skills/ac_refresh_colab_contexts.md

.claude/skills/ac_refresh_colab_contexts.md

@@ -0,0 +1 @@
+../../skills/ac_refresh_colab_contexts.md

.codex/skills/autocti-assistant-ac-refresh-colab-contexts/SKILL.md

@@ -0,0 +1,7 @@
+---
+name: autocti-assistant-ac-refresh-colab-contexts
+description: "Maintainer upkeep for the link between this assistant's skills and the Colab notebooks of autocti_workspace. Runs the deterministic coverage check (`python -m autoassistant.colab_contexts check` plus PyAutoHands' `autohands gemini_context report`), then AI-scans what it flags — notebooks reached only by `default`, mapping entries that match nothing, missing or renamed skills, entries near the size cap — against the workspace script headers and the skill descriptions, proposes minimal `colab/contexts.yaml` edits, re-checks, re-stamps the workspace notebooks (the release does not regenerate them), and opens the matching PRs. Use after a skill is added, renamed or substantially resized, after the workspace gains or moves folders, and before a release. Not for user CTI questions, and never for editing the skills or the workspace scripts themselves."
+---
+<!-- Generated by PyAutoBrain/bin/project_discovery.py; do not edit. -->
+
+Read and follow the canonical skill: [SKILL](../../../skills/ac_refresh_colab_contexts.md).

.github/workflows/colab-contexts.yml

@@ -0,0 +1,36 @@
+# Colab-contexts check — colab/contexts.yaml stays loadable by PyAutoHands.
+#
+# PyAutoHands' `autohands gemini_context apply` stamps every autocti_workspace notebook
+# with a Colab Gemini instruction assembled from colab/contexts.yaml (preface + core
+# rules + the mapped skills), and fails if a mapped skill is missing or an instruction
+# exceeds the mapping's cap (`cap:`, default 20,000). This job runs the same checks on
+# every PR, so renaming a skill or growing one past the cap fails here, beside the
+# change, rather than when the notebooks are next stamped. See
+# autoassistant/colab_contexts.py.
+name: colab-contexts
+
+on:
+  pull_request:
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  colab-contexts:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install PyYAML + pytest
+        run: python -m pip install PyYAML pytest
+
+      - name: Every mapped skill exists and every entry fits the cap
+        run: python -m autoassistant.colab_contexts check
+
+      - name: Mapping tests
+        run: python -m pytest -q autoassistant/tests/test_colab_contexts.py

Makefile

@@ -1,4 +1,4 @@
-.PHONY: validate-literature-citations audit test
+.PHONY: validate-literature-citations audit test colab-contexts
 
 validate-literature-citations:
 	python -m autoassistant.literature validate-citations
@@ -10,3 +10,7 @@ audit:
 # Assistant tooling test suite (slow: the gate tests import autocti per case).
 test:
 	python -m pytest autoassistant/tests -q
+
+# colab/contexts.yaml: mapped skills exist and every entry fits the Gemini cap (`cap:` in the file).
+colab-contexts:
+	python -m autoassistant.colab_contexts check

autoassistant/colab_contexts.py

@@ -0,0 +1,190 @@
+"""Check `colab/contexts.yaml` and stamp this repository's own Colab notebooks with it.
+
+`colab/contexts.yaml` maps workspace notebook paths to skills. PyAutoHands
+(`autohands/build_util.py`, `inject_gemini_instructions`, run on autocti_workspace by
+`autohands gemini_context apply`) assembles, for each notebook, the preface + core
+rules + mapped skills and writes them into the notebook's metadata as its Colab Gemini instruction (`metadata.colab.aiContexts` +
+`selectedAiContext`). Colab's Gemini cannot read this repository; the instruction is how
+the skills reach it.
+
+This module is the assistant's side of that contract, self-contained (PyYAML + stdlib)
+because the assistant does not depend on PyAutoHands:
+
+- ``check`` — every mapped skill exists, none is on the file's ``never_map`` list, and
+  every entry assembles to at most the mapping's cap (`cap:`, default 20,000). CI runs it on every PR
+  (``.github/workflows/colab-contexts.yml``), so a renamed skill or an over-long entry
+  fails here before it can fail a stamping run.
+- ``apply`` — stamps one notebook, for notebooks this repository ships itself, keeping the file's JSON layout so the diff touches metadata only.
+
+The assembly and the instruction id mirror PyAutoHands' ``assemble_gemini_context`` and
+``gemini_context_id``; change both or neither.
+
+Usage:
+
+    python -m autoassistant.colab_contexts check
+    python -m autoassistant.colab_contexts apply path/to/notebook.ipynb \\
+        --name autocti_workspace/notebooks/dataset_1d/modeling/start_here.ipynb
+"""
+
+from __future__ import annotations
+
+import argparse
+import fnmatch
+import itertools
+import json
+import sys
+import uuid
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+CONTEXTS_FILE = Path("colab") / "contexts.yaml"
+CAP = 20_000
+# A mapping may set its own `cap:` (its skills may be longer) up to this ceiling, the
+# size verified in Colab. Mirrors PyAutoHands' GEMINI_CONTEXT_CAP / GEMINI_CONTEXT_CAP_MAX.
+CAP_MAX = 50_000
+_ID_URL = "https://github.com/PyAutoLabs/colab-gemini/{project}"
+
+# ``check`` sizes each entry with this stand-in notebook name, not the entry's glob: the
+# rules text embeds the real path, and a glob can be shorter than the paths it matches.
+# 240 characters is longer than any workspace notebook path (the longest today is ~110),
+# so an entry that passes here cannot fail the release's per-notebook cap on path length.
+_PROBE_NAME = "repository/" + "p" * (240 - len("repository/") - len(".ipynb")) + ".ipynb"
+
+
+def load(root: Path = ROOT) -> dict:
+    import yaml
+
+    return yaml.safe_load((Path(root) / CONTEXTS_FILE).read_text(encoding="utf-8"))
+
+
+def context_id(project: str) -> str:
+    return str(uuid.uuid5(uuid.NAMESPACE_URL, _ID_URL.format(project=project)))
+
+
+def match(contexts: dict, notebook_name: str) -> tuple[str, list[str]]:
+    """First matching entry wins (most specific first); else ``default``."""
+    for entry in contexts["contexts"]:
+        if fnmatch.fnmatchcase(notebook_name, entry["match"]):
+            return entry["match"], list(entry["skills"])
+    return "default", list(contexts["default"]["skills"])
+
+
+def assemble(contexts: dict, notebook_name: str, root: Path = ROOT) -> tuple[str, str]:
+    """``(entry, text)`` for ``notebook_name`` ("<repo>/<path from the repo root>")."""
+    label, skills = match(contexts, notebook_name)
+    repo, _, relpath = notebook_name.partition("/")
+    sections = [
+        contexts["preface"].strip(),
+        contexts["rules"].replace("{notebook}", relpath).replace("{repo}", repo).strip(),
+        "# Reference material: assistant skills",
+    ]
+    for skill in skills:
+        text = (Path(root) / "skills" / f"{skill}.md").read_text(encoding="utf-8").strip()
+        sections.append(f"## Skill: {skill}\n\n{text}")
+    return label, "\n\n".join(sections) + "\n"
+
+
+def cap_of(contexts: dict) -> int:
+    """The cap in force: the mapping's `cap:`, else ``CAP``. Raises outside (0, CAP_MAX]."""
+    cap = contexts.get("cap", CAP)
+    if isinstance(cap, bool) or not isinstance(cap, int) or not 0 < cap <= CAP_MAX:
+        raise ValueError(f"cap {cap!r} must be a positive integer no larger than {CAP_MAX}")
+    return cap
+
+
+def check(root: Path = ROOT) -> list[str]:
+    """Problems with the mapping; empty when it is sound."""
+    contexts = load(root)
+    problems = []
+    for key in ("name", "project", "preface", "rules", "default", "contexts"):
+        if key not in contexts:
+            problems.append(f"missing top-level '{key}'")
+    if problems:
+        return problems
+    try:
+        cap = cap_of(contexts)
+    except ValueError as exc:
+        return [str(exc)]
+    never_map = set(contexts.get("never_map") or [])
+    entries = [("default", contexts["default"].get("skills") or [])] + [
+        (e.get("match"), e.get("skills") or []) for e in contexts["contexts"]
+    ]
+    seen = set()
+    for label, skills in entries:
+        if not label or not skills:
+            problems.append(f"entry {label!r} needs both 'match' and 'skills'")
+            continue
+        if label in seen:
+            problems.append(f"entry {label!r} appears twice (the second can never match)")
+        seen.add(label)
+        missing = [s for s in skills if not (Path(root) / "skills" / f"{s}.md").is_file()]
+        if missing:
+            problems.append(f"entry {label!r} maps missing skill(s): {', '.join(missing)}")
+            continue
+        banned = sorted(never_map.intersection(skills))
+        if banned:
+            problems.append(f"entry {label!r} maps never_map skill(s): {', '.join(banned)}")
+        size = len(_assemble_skills(contexts, _PROBE_NAME, skills, root))
+        if size > cap:
+            problems.append(f"entry {label!r} assembles to {size} characters, over the {cap} cap")
+    return problems
+
+
+def _assemble_skills(contexts: dict, name: str, skills: list[str], root: Path) -> str:
+    probe = dict(contexts, contexts=[{"match": "*", "skills": skills}])
+    return assemble(probe, name, root)[1]
+
+
+def dump_like(original_text: str, notebook: dict) -> str:
+    """Serialise ``notebook`` in the JSON layout ``original_text`` uses (metadata-only diffs)."""
+    original = json.loads(original_text)
+    newline = "\n" if original_text.endswith("\n") else ""
+    for indent, ensure_ascii, sort_keys in itertools.product((1, 2, 4), (False, True), (False, True)):
+        kwargs = dict(indent=indent, ensure_ascii=ensure_ascii, sort_keys=sort_keys)
+        if json.dumps(original, **kwargs) + newline == original_text:
+            return json.dumps(notebook, **kwargs) + newline
+    return json.dumps(notebook, indent=1, ensure_ascii=False, sort_keys=True) + "\n"
+
+
+def apply(notebook_path: Path, notebook_name: str, root: Path = ROOT) -> str:
+    """Write the instruction for ``notebook_name`` into ``notebook_path``; returns the entry."""
+    contexts = load(root)
+    label, text = assemble(contexts, notebook_name, root)
+    cap = cap_of(contexts)
+    if len(text) > cap:
+        raise ValueError(f"{notebook_name}: entry {label!r} assembles to {len(text)} characters, over {cap}")
+    original_text = Path(notebook_path).read_text(encoding="utf-8")
+    notebook = json.loads(original_text)
+    cid = context_id(contexts["project"])
+    colab = notebook.setdefault("metadata", {}).setdefault("colab", {})
+    colab.setdefault("aiContexts", {})[cid] = {"id": cid, "name": contexts["name"], "context": text}
+    colab["selectedAiContext"] = cid
+    new_text = dump_like(original_text, notebook)
+    if new_text != original_text:
+        Path(notebook_path).write_text(new_text, encoding="utf-8")
+    return label
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="python -m autoassistant.colab_contexts")
+    sub = parser.add_subparsers(dest="command", required=True)
+    sub.add_parser("check", help="Validate colab/contexts.yaml")
+    p_apply = sub.add_parser("apply", help="Stamp one notebook")
+    p_apply.add_argument("notebook")
+    p_apply.add_argument("--name", required=True, help='"<repo>/<path from the repo root>"')
+    args = parser.parse_args(argv)
+
+    if args.command == "check":
+        problems = check()
+        for problem in problems:
+            print(f"colab/contexts.yaml: {problem}", file=sys.stderr)
+        if not problems:
+            print(f"colab/contexts.yaml: OK ({len(load()['contexts'])} entries + default, cap {cap_of(load())})")
+        return 1 if problems else 0
+
+    print(f"{args.notebook}: {apply(Path(args.notebook), args.name)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

autoassistant/tests/test_colab_contexts.py

@@ -0,0 +1,103 @@
+"""Pin `colab/contexts.yaml`: sound mapping, PyAutoHands-identical ids, metadata-only stamping.
+
+Stdlib + PyYAML, fast: nothing is executed.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from autoassistant import colab_contexts
+
+NOTEBOOK_NAME = "autocti_workspace/notebooks/dataset_1d/modeling/start_here.ipynb"
+
+
+def test_mapping_is_sound():
+    assert colab_contexts.check() == []
+
+
+def test_project_and_name():
+    contexts = colab_contexts.load()
+    assert contexts["project"] == "autocti"
+    assert contexts["default"]["skills"]
+
+
+def test_context_id_matches_pyautohands():
+    # PyAutoHands' build_util.gemini_context_id("autocti") — the id every stamped
+    # autocti_workspace notebook carries. Literal, so a drift on either side fails here.
+    assert colab_contexts.context_id("autocti") == "a3d7c160-b952-56fb-b83a-321b7b858913"
+
+
+def test_check_flags_never_map_and_over_cap(tmp_path):
+    root = tmp_path / "assistant"
+    (root / "colab").mkdir(parents=True)
+    (root / "skills").mkdir()
+    (root / "skills" / "small.md").write_text("small")
+    (root / "skills" / "maint.md").write_text("maint")
+    (root / "skills" / "huge.md").write_text("x" * colab_contexts.CAP)
+    (root / "colab" / "contexts.yaml").write_text(
+        "name: T\nproject: autocti\npreface: P\nrules: R {notebook} {repo}\n"
+        "never_map: [maint]\ndefault: {skills: [small]}\ncontexts:\n"
+        "  - {match: a/*, skills: [maint]}\n  - {match: b/*, skills: [huge]}\n"
+        "  - {match: c/*, skills: [missing]}\n"
+    )
+    problems = "\n".join(colab_contexts.check(root))
+    assert "never_map" in problems and "over the" in problems and "missing" in problems
+
+
+def test_apply_changes_metadata_only(tmp_path):
+    notebook = {
+        "cells": [{"cell_type": "markdown", "id": "cell-0", "metadata": {}, "source": ["Hi é"]}],
+        "metadata": {"colab": {"provenance": []}, "kernelspec": {"name": "python3"}},
+        "nbformat": 4,
+        "nbformat_minor": 5,
+    }
+    path = tmp_path / "x.ipynb"
+    path.write_text(json.dumps(notebook, indent=1, ensure_ascii=False, sort_keys=True) + "\n")
+    colab_contexts.apply(path, NOTEBOOK_NAME)
+    first = path.read_text()
+    colab_contexts.apply(path, NOTEBOOK_NAME)  # idempotent
+    assert path.read_text() == first
+    stamped = json.loads(first)
+    assert stamped["cells"] == notebook["cells"]
+    assert stamped["metadata"]["kernelspec"] == {"name": "python3"}
+    colab = stamped["metadata"]["colab"]
+    assert colab["provenance"] == []
+    assert colab["selectedAiContext"] == "a3d7c160-b952-56fb-b83a-321b7b858913"
+    (entry,) = colab["aiContexts"].values()
+    assert NOTEBOOK_NAME.split("/", 1)[1] in entry["context"]
+    assert first.endswith("}\n") and "Hi é" in first
+
+
+def _mini(tmp_path, cap_line=""):
+    root = tmp_path / "assistant"
+    (root / "colab").mkdir(parents=True)
+    (root / "skills").mkdir()
+    (root / "skills" / "mid.md").write_text("y" * 25_000)
+    (root / "colab" / "contexts.yaml").write_text(
+        f"{cap_line}\nname: T\nproject: p\npreface: P\nrules: R {{notebook}} {{repo}}\n"
+        "default: {skills: [mid]}\ncontexts: []\n"
+    )
+    return root
+
+
+def test_cap_defaults_to_20k(tmp_path):
+    root = _mini(tmp_path)
+    assert colab_contexts.cap_of(colab_contexts.load(root)) == colab_contexts.CAP == 20_000
+    assert "over the 20000 cap" in "\n".join(colab_contexts.check(root))
+
+
+def test_custom_cap_honoured(tmp_path):
+    root = _mini(tmp_path, "cap: 30000")
+    assert colab_contexts.check(root) == []
+
+
+@pytest.mark.parametrize("cap", ["50001", "0", "-1", "lots", "true"])
+def test_cap_out_of_range_rejected(tmp_path, cap):
+    root = _mini(tmp_path, f"cap: {cap}")
+    assert "no larger than 50000" in "\n".join(colab_contexts.check(root))
+    with pytest.raises(ValueError, match="no larger than 50000"):
+        colab_contexts.cap_of(colab_contexts.load(root))