From 07615f4eb1861ac90c1e0209e35eb50e516e7a74 Mon Sep 17 00:00:00 2001 From: cheshirecode Date: Sun, 26 Jul 2026 00:16:26 -0400 Subject: [PATCH 1/2] feat: add portable loop engineering skill --- README.md | 7 +- bin/doctor.sh | 21 +- bin/install-skills.sh | 97 +++--- install.sh | 28 +- manifest/skills.yaml | 12 +- skills/loop-engineering/SKILL.md | 68 +++++ skills/loop-engineering/agents/openai.yaml | 4 + .../loop-engineering/references/examples.md | 70 +++++ skills/loop-engineering/references/hosts.md | 49 +++ .../loop-engineering/references/protocol.md | 68 +++++ skills/loop-engineering/scripts/loop_state.py | 289 ++++++++++++++++++ .../loop-engineering/tests/test_loop_state.py | 167 ++++++++++ tests/run.sh | 95 +++++- 13 files changed, 905 insertions(+), 70 deletions(-) create mode 100644 skills/loop-engineering/SKILL.md create mode 100644 skills/loop-engineering/agents/openai.yaml create mode 100644 skills/loop-engineering/references/examples.md create mode 100644 skills/loop-engineering/references/hosts.md create mode 100644 skills/loop-engineering/references/protocol.md create mode 100755 skills/loop-engineering/scripts/loop_state.py create mode 100755 skills/loop-engineering/tests/test_loop_state.py diff --git a/README.md b/README.md index 064e6e2..79c819b 100644 --- a/README.md +++ b/README.md @@ -147,9 +147,10 @@ The `.cursor/` directory contains Cursor-specific configurations that are mainta git clone https://github.com/cheshirecode/dotfiles.git ~/.dotfiles cd ~/.dotfiles && bin/install.sh -# bin/install.sh is the SUPPORTED install path: detects your OS, installs -# runtime deps, runs the agent-skill installer (with the rmtree-safety -# sentinel), wires hooks, runs bin/doctor.sh. Idempotent — re-run is safe. +# bin/install.sh is the SUPPORTED install path: detects your OS, installs +# runtime deps, installs every manifest skill for Claude Code, Codex, and +# Cursor (with the rmtree-safety sentinel), wires hooks, and runs +# bin/doctor.sh. Idempotent — re-run is safe. # # If you want manual symlinks instead, BACK UP FIRST. `ln -sf` will # silently overwrite an existing real ~/.bashrc / ~/.zshrc / ~/.gitconfig diff --git a/bin/doctor.sh b/bin/doctor.sh index 1fadc99..d3de10b 100755 --- a/bin/doctor.sh +++ b/bin/doctor.sh @@ -5,7 +5,7 @@ # 1. Runtime deps on PATH (python3, gh, git, rg, jq, direnv) # 2. Python ≥ 3.10 (worklog lint helpers depend on 3.10+ syntax) # 3. PyYAML importable (install-skills.sh uses it) -# 4. Each manifest skill present as a SKILL.md under ~/.claude/skills/ +# 4. Each manifest skill present under Claude, shared-agent, and Cursor roots # 5. _worklog repo present and on a clean HEAD # 6. Hooks wired (.claude/settings.json mentions autosave) # 7. gh auth (warn-only — works for unauth'd public-repo flows) @@ -21,7 +21,10 @@ warn() { say "WARN" "$1"; WARN=$((WARN+1)); } REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" MANIFEST="$REPO_ROOT/manifest/skills.yaml" -SKILLS_DIR="${CLAUDE_SKILLS_DIR:-$HOME/.claude/skills}" +CLAUDE_SKILLS_DIR="${CLAUDE_SKILLS_DIR:-$HOME/.claude/skills}" +AGENT_SKILLS_DIR="${AGENT_SKILLS_DIR:-$HOME/.agents/skills}" +CURSOR_SKILLS_DIR="${CURSOR_SKILLS_DIR:-$HOME/.cursor/skills}" +SKILL_ROOTS=("$CLAUDE_SKILLS_DIR" "$AGENT_SKILLS_DIR" "$CURSOR_SKILLS_DIR") PROJECTS_DIR="${PROJECTS_DIR:-$HOME/Documents/projects}" echo "doctor: runtime deps" @@ -55,12 +58,14 @@ print('\n'.join(s['name'] for s in yaml.safe_load(open('$MANIFEST'))['skills'])) rm -f "$skill_list_err" while IFS= read -r name; do [[ -z "$name" ]] && continue - skill_md="$SKILLS_DIR/$name/SKILL.md" - if [[ -f "$skill_md" ]]; then - ok "$name → $skill_md" - else - warn "$name SKILL.md missing (run bin/install-skills.sh)" - fi + for skill_root in "${SKILL_ROOTS[@]}"; do + skill_md="$skill_root/$name/SKILL.md" + if [[ -f "$skill_md" ]]; then + ok "$name → $skill_md" + else + warn "$name SKILL.md missing at $skill_root (run bin/install-skills.sh)" + fi + done done <<< "$skill_list" else fail "manifest/skills.yaml missing" diff --git a/bin/install-skills.sh b/bin/install-skills.sh index c08006e..48e801b 100755 --- a/bin/install-skills.sh +++ b/bin/install-skills.sh @@ -1,10 +1,12 @@ #!/usr/bin/env bash -# Install agent skills from manifest/skills.yaml into ~/.claude/skills/. +# Install agent skills from manifest/skills.yaml into all supported user roots: +# ~/.claude/skills/ for Claude Code, ~/.agents/skills/ for Codex and shared +# discovery, and ~/.cursor/skills/ for Cursor's native discovery surface. # # Two source types: # - subpath: copy a directory from this repo (vendored skill). # - git: clone a repo at a pinned SHA, then symlink (Mac/Linux) or -# copy (WSL fallback) into ~/.claude/skills//. +# copy (WSL fallback) into all three user skill roots. # # Mac/Linux uses symlinks (cheap, easy upgrade). WSL2 inherits Linux behavior. # Windows-native is unsupported — install.sh refuses earlier. @@ -42,12 +44,14 @@ MANIFEST="$REPO_ROOT/manifest/skills.yaml" [[ -f "$MANIFEST" ]] || { echo "install-skills: manifest not found at $MANIFEST" >&2; exit 1; } SKILLS_DIR="${CLAUDE_SKILLS_DIR:-$HOME/.claude/skills}" -CACHE_DIR="${CLAUDE_AGENT_CACHE:-$HOME/.agents/skills}" +SHARED_SKILLS_DIR="${AGENT_SKILLS_DIR:-$HOME/.agents/skills}" +CURSOR_SKILLS_DIR="${CURSOR_SKILLS_DIR:-$HOME/.cursor/skills}" +SOURCE_CACHE_DIR="${CLAUDE_AGENT_CACHE:-$HOME/.cache/dotfiles-agent-skills}" -mkdir -p "$SKILLS_DIR" "$CACHE_DIR" +mkdir -p "$SKILLS_DIR" "$SHARED_SKILLS_DIR" "$CURSOR_SKILLS_DIR" "$SOURCE_CACHE_DIR" # Parse manifest via Python (yaml is stdlib-adjacent; safer than awk on YAML). -python3 - "$MANIFEST" "$SINGLE" "$DRY_RUN" "$SKILLS_DIR" "$CACHE_DIR" "$REPO_ROOT" <<'PY' +python3 - "$MANIFEST" "$SINGLE" "$DRY_RUN" "$SKILLS_DIR" "$SHARED_SKILLS_DIR" "$CURSOR_SKILLS_DIR" "$SOURCE_CACHE_DIR" "$REPO_ROOT" <<'PY' import os, sys, shutil, subprocess, pathlib try: import yaml @@ -55,10 +59,12 @@ except ImportError: sys.stderr.write("install-skills: PyYAML not installed. Run: pip3 install --user pyyaml\n") sys.exit(1) -manifest_path, single, dry_run, skills_dir, cache_dir, repo_root = sys.argv[1:7] +manifest_path, single, dry_run, skills_dir, shared_skills_dir, cursor_skills_dir, source_cache_dir, repo_root = sys.argv[1:9] dry = dry_run == "1" skills_dir = pathlib.Path(skills_dir).expanduser() -cache_dir = pathlib.Path(cache_dir).expanduser() +shared_skills_dir = pathlib.Path(shared_skills_dir).expanduser() +cursor_skills_dir = pathlib.Path(cursor_skills_dir).expanduser() +source_cache_dir = pathlib.Path(source_cache_dir).expanduser() repo_root = pathlib.Path(repo_root) m = yaml.safe_load(open(manifest_path)) @@ -122,9 +128,34 @@ def write_sentinel(dst, source_info): """Write a sentinel so future install-skills runs recognize this dir.""" (dst / SENTINEL).write_text(source_info + "\n") +def install_destinations(entry): + """Preserve the manifest destination and add shared and Cursor roots.""" + destinations = [ + pathlib.Path(entry["install_to"]).expanduser(), + shared_skills_dir / entry["name"], + cursor_skills_dir / entry["name"], + ] + return list(dict.fromkeys(destinations)) + +def link_or_copy(src, dst, name, source_info): + if dst.is_symlink() or dst.exists(): + print(f" refresh {name}: {dst}") + if not dry: + if dst.is_symlink(): dst.unlink() + elif dst.is_dir(): shutil.rmtree(dst) + else: dst.unlink() + else: + print(f" install {name}: {dst}") + if not dry: + dst.parent.mkdir(parents=True, exist_ok=True) + try: + os.symlink(src.resolve(), dst) + except OSError: + shutil.copytree(src, dst) + write_sentinel(dst, source_info) + def install_subpath(entry): src = repo_root / entry["source"]["path"] - dst = pathlib.Path(entry["install_to"]).expanduser() if not src.exists(): if entry.get("optional") is True: print(f" SKIP optional {entry['name']}: source {src} not present") @@ -141,23 +172,16 @@ def install_subpath(entry): if err: sys.stderr.write(f"install-skills: refusing {entry['name']}: {err}\n") sys.exit(3) - refuse_if_unowned(dst, entry["name"]) - if dst.is_symlink() or dst.exists(): - print(f" refresh {entry['name']}: {dst}") - if not dry: - if dst.is_symlink(): dst.unlink() - elif dst.is_dir(): shutil.rmtree(dst) - else: dst.unlink() - else: - print(f" install {entry['name']}: {dst}") - if not dry: - dst.parent.mkdir(parents=True, exist_ok=True) - try: - os.symlink(src.resolve(), dst) - except OSError: - # WSL or filesystems without symlink support — fall back to copy. - shutil.copytree(src, dst) - write_sentinel(dst, f"subpath:{entry['source']['path']}") + destinations = install_destinations(entry) + for dst in destinations: + refuse_if_unowned(dst, entry["name"]) + for dst in destinations: + link_or_copy( + src, + dst, + entry["name"], + f"subpath:{entry['source']['path']}", + ) return True def install_git(entry): @@ -174,7 +198,7 @@ def install_git(entry): f" Override with INSTALL_SKILLS_ALLOW_MOVING_REF=1 if you really mean it.\n" ) sys.exit(3) - cache = cache_dir / name + cache = source_cache_dir / name # Council guardrail #10: atomic clone-or-swap on git operations. # Mid-fetch network failure must not leave the cache dir in a half-state. import tempfile as _tempfile @@ -183,7 +207,7 @@ def install_git(entry): # there's no half-populated cache dir to confuse the next run. print(f" clone {name}: {repo} → {cache}") if not dry: - staging = pathlib.Path(_tempfile.mkdtemp(prefix=f".{name}-staging-", dir=cache_dir)) + staging = pathlib.Path(_tempfile.mkdtemp(prefix=f".{name}-staging-", dir=source_cache_dir)) shutil.rmtree(staging) # mkdtemp made it; git clone wants it absent try: run(["git", "clone", "--quiet", f"https://github.com/{repo}.git", str(staging)]) @@ -201,20 +225,11 @@ def install_git(entry): print(f" upgrade {name}: {cache} → {ref[:12]}") run(["git", "-C", str(cache), "fetch", "--quiet", "origin"]) run(["git", "-C", str(cache), "checkout", "--quiet", ref]) - dst = pathlib.Path(entry["install_to"]).expanduser() - refuse_if_unowned(dst, name) - if dst.is_symlink() or dst.exists(): - if not dry: - if dst.is_symlink(): dst.unlink() - elif dst.is_dir(): shutil.rmtree(dst) - else: dst.unlink() - if not dry: - dst.parent.mkdir(parents=True, exist_ok=True) - try: - os.symlink(cache.resolve(), dst) - except OSError: - shutil.copytree(cache, dst) - write_sentinel(dst, f"git:{repo}@{ref}") + destinations = install_destinations(entry) + for dst in destinations: + refuse_if_unowned(dst, name) + for dst in destinations: + link_or_copy(cache, dst, name, f"git:{repo}@{ref}") return True installed = skipped = 0 diff --git a/install.sh b/install.sh index 9df1f96..6faeadf 100755 --- a/install.sh +++ b/install.sh @@ -61,20 +61,22 @@ if [ -d "$REPO_DIR/.cursor" ]; then cp -R "$REPO_DIR/.cursor/." "$DEST/.cursor/" fi -# Symlink Claude Code skills shipped in this repo into ~/.claude/skills/. -# Code (skill + bin) is version-controlled here; per-machine data/config -# (e.g. worklog's WORKLOG_REPO) lives outside the repo via ~/.shell_common.local. +# Symlink every skill shipped in this repo into the shared Agent Skills root, +# Claude Code's personal skill root, and Cursor's native personal skill root. +# Code stays version-controlled here; per-machine data/config lives outside. if [ -d "$REPO_DIR/skills" ]; then - mkdir -p "$DEST/.claude/skills" - for skill in "$REPO_DIR"/skills/*/; do - sname="$(basename "$skill")" - starget="$DEST/.claude/skills/$sname" - if [ -L "$starget" ] && [ "$(readlink "$starget")" = "${skill%/}" ]; then - continue - fi - backup "$starget" - echo "Symlinking skill $sname into $starget..." - ln -s "${skill%/}" "$starget" + for skills_root in "$DEST/.agents/skills" "$DEST/.claude/skills" "$DEST/.cursor/skills"; do + mkdir -p "$skills_root" + for skill in "$REPO_DIR"/skills/*/; do + sname="$(basename "$skill")" + starget="$skills_root/$sname" + if [ -L "$starget" ] && [ "$(readlink "$starget")" = "${skill%/}" ]; then + continue + fi + backup "$starget" + echo "Symlinking skill $sname into $starget..." + ln -s "${skill%/}" "$starget" + done done fi diff --git a/manifest/skills.yaml b/manifest/skills.yaml index 47b0349..182d65e 100644 --- a/manifest/skills.yaml +++ b/manifest/skills.yaml @@ -4,8 +4,8 @@ # addressed). Bump SHAs explicitly when upgrading; don't track moving branches. # # Targets: -# user-installed Claude Code skills under ~/.claude/skills// -# sourced from public GitHub repos. +# every manifest skill is installed under ~/.claude/skills//, +# ~/.agents/skills//, and ~/.cursor/skills//. # # Skills that ship with Claude Code itself (init, review, security-review, # claude-api, loop, schedule, statusline-setup, update-config, etc.) are @@ -37,6 +37,14 @@ skills: # installs the skill files; bin/install-worklog.sh clones the repo + # wires hooks. + - name: loop-engineering + description: Bounded, evidence-driven loops with worklog-backed context and handoffs. + source: + type: subpath + ref: HEAD + path: skills/loop-engineering + install_to: ~/.claude/skills/loop-engineering + - name: ship-hygiene description: Pre-handoff sweep across worklog + PRs + CI. source: diff --git a/skills/loop-engineering/SKILL.md b/skills/loop-engineering/SKILL.md new file mode 100644 index 0000000..d3ea55b --- /dev/null +++ b/skills/loop-engineering/SKILL.md @@ -0,0 +1,68 @@ +--- +name: loop-engineering +description: Design and run bounded, evidence-driven loops for repeated, resumable, delegated, or scheduled engineering and research work. Use when an agent must iterate toward a verifiable condition, recover across context boundaries, coordinate subagents, or decide whether work belongs in a manual loop, worklog-backed handoff, or host-native scheduler. Skip trivial one-shot tasks. +--- + +# Loop Engineering + +Use deterministic state transitions around agent judgment. Repeated prompting is +not a loop design. + +## Route + +1. Skip this skill when one action plus one check is sufficient. +2. For interactive, resumable, or delegated loops, use the state script below. +3. For scheduled loops, also read [references/hosts.md](references/hosts.md) and + require a real, authorized recurrence primitive. +4. For exact transition, effect, worklog, or handoff rules, read + [references/protocol.md](references/protocol.md). + +## Initialize through the script + +Resolve `` as this `SKILL.md` file's directory. Choose an explicit, +authorized state path; do not hand-edit its JSON. + +```bash +python3 /scripts/loop_state.py init \ + --state \ + --goal "" \ + --evidence "" \ + --budget-unit "" \ + --budget-limit \ + --next-action "" +``` + +Add `--allowed-effect` and `--approval-boundary` whenever writes or external +effects are possible. If `python3` is unavailable, preserve the same five fields +manually and label the run as a non-deterministic fallback. + +## Run one bounded cycle + +1. Observe from tools or durable evidence. +2. Choose the smallest action that advances or falsifies the approach. +3. Check effect scope; serialize writes unless isolation is proven. +4. Execute and verify. A model's prose claim is not evidence. +5. On verified success, run `finish --status complete --verification + "" --evidence ""`. +6. Otherwise run `advance --evidence "" --next-action ""`. + The script emits `budget_exhausted` when the declared ceiling is consumed. +7. Run `show`; continue only while `terminal_status` is `running`. + +If later evidence contradicts a recorded fact, use `annotate --evidence +""`. Preserve the audit trail; do not reopen or hand-edit terminal +state. + +Use `finish` for `blocked`, `needs_human`, `cancelled`, or +`continue_scheduled`. Never translate those states or `budget_exhausted` into +`complete`. + +## Preserve durable context + +When the installed `worklog` protocol is available, hydrate resume context +before initialization and checkpoint verified state at compaction, delegation, +retry exhaustion, scheduled handoff, or termination. Before cold delegation, +pass the returned `context --for=compact` pack directly; do not pass the +parent transcript or imply that `spawn` enriches the pack. + +For brittle state classification or handoff sequencing, read +[references/examples.md](references/examples.md). Otherwise stay zero-shot. diff --git a/skills/loop-engineering/agents/openai.yaml b/skills/loop-engineering/agents/openai.yaml new file mode 100644 index 0000000..7a4be6d --- /dev/null +++ b/skills/loop-engineering/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Loop Engineering" + short_description: "Design bounded, evidence-driven agent loops" + default_prompt: "Use $loop-engineering to structure this repeated or resumable task as a bounded, evidence-driven loop." diff --git a/skills/loop-engineering/references/examples.md b/skills/loop-engineering/references/examples.md new file mode 100644 index 0000000..30550b2 --- /dev/null +++ b/skills/loop-engineering/references/examples.md @@ -0,0 +1,70 @@ +# Contrastive loop fixtures + +Read only when state classification, effect ordering, or handoff routing is +ambiguous. + +## 1. Interactive diagnosis + +INPUT + +> Find and fix a flaky unit test. Stop after three unsuccessful hypotheses. + +OUTPUT + +```text +goal: targeted test passes three consecutive runs +progress_evidence: failure reproduced once; no source mutation yet +budget: 3 hypotheses +next_action: isolate timing-dependent assertions +terminal_status: running +``` + +Run one discriminating check per hypothesis. Serialize edits and test after each +one. End `complete` only with three passing runs; otherwise end +`budget_exhausted` with failures and the safest next action. + +## 2. Worklog-backed delegation + +INPUT + +> Resume `auth-flake`, delegate context lookup, then decide the next fix. + +OUTPUT + +Invoke the installed `worklog` skill in `context` mode with +`auth-flake --for=resume`, then immediately hydrate the host tracker from its +tracker-ready snippet. Before delegation, invoke `context` mode with +`auth-flake --for=compact` and give that returned pack directly to each +independent read-only delegate. Follow worklog's dedupe rule so compact lookup +does not recreate tracker items already hydrated by resume. Ask one delegate +for failure-history evidence and another for current call-site evidence. +Reconcile both returns before editing. Persist new task evidence through the +worklog protocol, invoke `sync` mode for `auth-flake`, and rehydrate the tracker. + +Do not pass the full parent transcript or let delegates write concurrently. + +## 3. Scheduled monitor without a portable scheduler + +INPUT + +> Check CI every 15 minutes until it passes. + +OUTPUT + +Read `references/hosts.md`. If the active host exposes an authorized recurrence +primitive, schedule bounded checks and end each run `continue_scheduled` until +CI evidence satisfies the goal. If no such primitive is available, end +`needs_human` and name the missing capability. + +Do not claim that a timer or stop hook exists merely because another host +supports one. + +## Example-led review contract + +```text +shot_count: few +format: INPUT/OUTPUT +examples_or_skip_reason: three distinct fixtures cover diagnosis, delegated durable context, and scheduled handoff +risk_check: keep fixtures about classification and sequencing so agents do not copy task-specific details +acceptance_test: an unseen resumable task hydrates context, bounds the run, and ends with evidence plus one valid terminal status +``` diff --git a/skills/loop-engineering/references/hosts.md b/skills/loop-engineering/references/hosts.md new file mode 100644 index 0000000..0d82fc2 --- /dev/null +++ b/skills/loop-engineering/references/hosts.md @@ -0,0 +1,49 @@ +# Host capability routing + +Read only for installation, delegation primitives, recurrence, or host-specific +tracking. + +## Shared contract + +- Keep `skills/loop-engineering/` as the single source. +- Use only standard `name` and `description` frontmatter in `SKILL.md`. +- Discover capabilities before invoking them. +- Preserve the same run contract and terminal statuses on every host. + +## Codex + +- Discover shared skills from `~/.agents/skills/`. +- Use the current task plan and available subagent tools for in-session tracking + and bounded delegation. +- Invoke the installed `worklog` skill for durable context and checkpoints. +- For recurrence, use an available Codex automation or `loop-orchestrator`; if + neither is callable, end `needs_human`. + +## Claude Code + +- Discover personal skills from `~/.claude/skills/`. +- Use the available task tracker and Agent tool for in-session tracking and + bounded delegation. +- Invoke `/worklog context --for=compact` before cold delegation and + pass the returned pack directly. Use `/worklog sync` for the protocol's + confirmation/checkpoint boundary. +- Use Claude's real `/loop`, scheduled task, or hook capability only when exposed + and authorized. Otherwise end `needs_human`. + +## Cursor + +- Discover user skills from `~/.agents/skills/` or Cursor's native + `~/.cursor/skills/`; project-local alternatives may use `.agents/skills/` or + `.cursor/skills/`. +- Use Cursor todos and subagents when exposed. +- Invoke the installed worklog skill or its documented helper commands for + durable context. If unavailable, use one durable project tracker and label the + fallback. +- Use a configured Cursor automation or hook only after verifying it exists and + has a bounded stop rule. Otherwise end `needs_human`. + +## Compatibility rule + +Never encode a host-only hook, command substitution, tool name, or permission +grant in the portable root skill. Keep host differences in this reference and +degrade explicitly when a capability is absent. diff --git a/skills/loop-engineering/references/protocol.md b/skills/loop-engineering/references/protocol.md new file mode 100644 index 0000000..bf0216e --- /dev/null +++ b/skills/loop-engineering/references/protocol.md @@ -0,0 +1,68 @@ +# Loop protocol + +Read only when the root router needs exact transition, effect, worklog, or +handoff rules. + +## State CLI + +Use `scripts/loop_state.py`; do not reimplement its state machine or hand-edit +the JSON. + +- `init` creates a `running` state and refuses overwrite unless `--force`. +- `advance` records a failed or nonterminal cycle, consumes positive budget, + and atomically changes the state to `budget_exhausted` at the ceiling. +- `finish` records exactly one terminal outcome. `complete` requires + `--verification` naming tool output or an artifact. +- `annotate` appends corrected evidence without reopening terminal state or + changing the consumed budget. +- `validate` checks the schema and transition invariants. +- `show` prints the five-field contract; `--json` returns the full history. + +Every write is atomic. A failed transition leaves the previous state unchanged. + +## Effect boundary + +- Declare allowed effects and the approval boundary at initialization when + files, external systems, or user data may change. +- Parallelize independent read-only observations. +- Serialize writes unless the runtime proves isolation. +- After a contradiction, stop affected mutations, revise the hypothesis, and + replay the original discriminating check. + +The script records state; it never grants permission, executes the action, +verifies external truth, delegates, or schedules a wakeup. + +## Worklog composition + +For work spanning sessions, compaction, retries, or agents: + +1. Existing task: invoke installed `worklog` in `context` mode with + ` --for=resume`; immediately hydrate the host tracker from the + tracker-ready snippet. +2. New durable task: invoke `plan`, then slugless `sync`; let sync survey, + report, and wait for confirmation before creation. +3. Cold delegation: invoke `context` with ` --for=compact`, follow tracker + dedupe, and pass the returned pack directly to the delegate. +4. Boundary or terminal state: persist arbitrary evidence or task-body changes + through worklog, invoke `sync` or its supported checkpoint helper, then + rehydrate the tracker. + +If worklog is unavailable, use the host tracker plus one authorized durable +project file. Label that fallback and do not claim a worklog checkpoint. + +## Delegation + +- Delegate a bounded lookup, research, or verification question. +- Include the objective, evidence, constraints, budget, and requested return. +- Require evidence, uncertainty, and one proposed next action. +- Reconcile returns into the parent state before any write. + +## Terminal outcomes + +- `complete`: external evidence proves the goal. +- `blocked`: an external dependency prevents safe progress. +- `needs_human`: ambiguity, authority, or risk requires a decision. +- `budget_exhausted`: the declared ceiling was consumed. +- `cancelled`: the user or host stopped the run. +- `continue_scheduled`: this bounded run ended and a real scheduler owns the + next wakeup. diff --git a/skills/loop-engineering/scripts/loop_state.py b/skills/loop-engineering/scripts/loop_state.py new file mode 100755 index 0000000..f6cf8f6 --- /dev/null +++ b/skills/loop-engineering/scripts/loop_state.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +"""Manage the deterministic state boundary of a bounded agent loop.""" + +from __future__ import annotations + +import argparse +import json +import os +import pathlib +import sys +import tempfile +from datetime import datetime, timezone +from typing import Any + + +SCHEMA_VERSION = 1 +TERMINAL_STATUSES = { + "complete", + "blocked", + "needs_human", + "budget_exhausted", + "cancelled", + "continue_scheduled", +} +ALL_STATUSES = {"running", *TERMINAL_STATUSES} + + +class StateError(ValueError): + """Raised when a requested transition violates the loop contract.""" + + +def now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def read_state(path: pathlib.Path) -> dict[str, Any]: + try: + state = json.loads(path.read_text()) + except FileNotFoundError as exc: + raise StateError(f"state file does not exist: {path}") from exc + except json.JSONDecodeError as exc: + raise StateError(f"state file is not valid JSON: {path}: {exc}") from exc + validate_state(state) + return state + + +def write_state(path: pathlib.Path, state: dict[str, Any]) -> None: + validate_state(state) + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + temporary_path = pathlib.Path(temporary_name) + try: + with os.fdopen(fd, "w") as handle: + json.dump(state, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + finally: + temporary_path.unlink(missing_ok=True) + + +def validate_state(state: dict[str, Any]) -> None: + if state.get("schema_version") != SCHEMA_VERSION: + raise StateError(f"schema_version must be {SCHEMA_VERSION}") + for field in ("goal", "next_action", "terminal_status"): + if not isinstance(state.get(field), str): + raise StateError(f"{field} must be a string") + if not state["goal"].strip(): + raise StateError("goal must not be empty") + if state["terminal_status"] not in ALL_STATUSES: + raise StateError(f"invalid terminal_status: {state['terminal_status']}") + + evidence = state.get("progress_evidence") + if not isinstance(evidence, list) or not evidence: + raise StateError("progress_evidence must be a non-empty list") + if not all(isinstance(item, str) and item.strip() for item in evidence): + raise StateError("progress_evidence entries must be non-empty strings") + + budget = state.get("budget") + if not isinstance(budget, dict): + raise StateError("budget must be an object") + if not isinstance(budget.get("unit"), str) or not budget["unit"].strip(): + raise StateError("budget.unit must be a non-empty string") + limit = budget.get("limit") + used = budget.get("used") + if not isinstance(limit, int) or limit < 1: + raise StateError("budget.limit must be a positive integer") + if not isinstance(used, int) or used < 0 or used > limit: + raise StateError("budget.used must be between zero and budget.limit") + + if state["terminal_status"] == "running" and used >= limit: + raise StateError("running state cannot have an exhausted budget") + verification = state.get("verification") + if state["terminal_status"] == "complete" and ( + not isinstance(verification, str) or not verification.strip() + ): + raise StateError("complete state requires non-empty verification") + if not isinstance(state.get("history"), list): + raise StateError("history must be a list") + + +def require_running(state: dict[str, Any]) -> None: + if state["terminal_status"] != "running": + raise StateError( + f"cannot transition terminal state: {state['terminal_status']}" + ) + + +def append_history( + state: dict[str, Any], + event: str, + evidence: list[str], + next_action: str, +) -> None: + state["history"].append( + { + "at": now(), + "event": event, + "evidence": evidence, + "next_action": next_action, + } + ) + + +def command_init(args: argparse.Namespace) -> dict[str, Any]: + if args.state.exists() and not args.force: + raise StateError(f"refusing to overwrite existing state: {args.state}") + state: dict[str, Any] = { + "schema_version": SCHEMA_VERSION, + "goal": args.goal, + "progress_evidence": args.evidence, + "budget": { + "unit": args.budget_unit, + "limit": args.budget_limit, + "used": 0, + }, + "next_action": args.next_action, + "terminal_status": "running", + "allowed_effects": args.allowed_effect, + "approval_boundary": args.approval_boundary, + "verification": "", + "history": [], + } + append_history(state, "initialized", args.evidence, args.next_action) + write_state(args.state, state) + return state + + +def command_advance(args: argparse.Namespace) -> dict[str, Any]: + state = read_state(args.state) + require_running(state) + remaining = state["budget"]["limit"] - state["budget"]["used"] + if args.consume > remaining: + raise StateError( + f"transition consumes {args.consume} {state['budget']['unit']}; " + f"only {remaining} remain" + ) + state["progress_evidence"].extend(args.evidence) + state["budget"]["used"] += args.consume + state["next_action"] = args.next_action + if state["budget"]["used"] == state["budget"]["limit"]: + state["terminal_status"] = "budget_exhausted" + append_history(state, "advanced", args.evidence, args.next_action) + write_state(args.state, state) + return state + + +def command_finish(args: argparse.Namespace) -> dict[str, Any]: + state = read_state(args.state) + require_running(state) + if args.status == "complete" and not args.verification: + raise StateError("complete requires --verification from a tool or artifact") + state["progress_evidence"].extend(args.evidence) + state["terminal_status"] = args.status + if args.next_action is not None: + state["next_action"] = args.next_action + state["verification"] = args.verification or "" + append_history( + state, f"finished:{args.status}", args.evidence, state["next_action"] + ) + write_state(args.state, state) + return state + + +def command_annotate(args: argparse.Namespace) -> dict[str, Any]: + state = read_state(args.state) + state["progress_evidence"].extend(args.evidence) + if args.next_action is not None: + state["next_action"] = args.next_action + append_history(state, "annotated", args.evidence, state["next_action"]) + write_state(args.state, state) + return state + + +def summary(state: dict[str, Any]) -> str: + budget = state["budget"] + return "\n".join( + [ + f"goal: {state['goal']}", + f"progress_evidence: {state['progress_evidence'][-1]}", + f"budget: {budget['used']}/{budget['limit']} {budget['unit']}", + f"next_action: {state['next_action']}", + f"terminal_status: {state['terminal_status']}", + ] + ) + + +def add_state_argument(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--state", type=pathlib.Path, required=True) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + initialize = subparsers.add_parser("init", help="initialize a bounded run") + add_state_argument(initialize) + initialize.add_argument("--goal", required=True) + initialize.add_argument("--evidence", action="append", required=True) + initialize.add_argument("--budget-unit", required=True) + initialize.add_argument("--budget-limit", type=int, required=True) + initialize.add_argument("--next-action", required=True) + initialize.add_argument("--allowed-effect", action="append", default=[]) + initialize.add_argument("--approval-boundary", default="") + initialize.add_argument("--force", action="store_true") + initialize.set_defaults(handler=command_init) + + advance = subparsers.add_parser( + "advance", + help="record a failed/nonterminal cycle and consume budget", + ) + add_state_argument(advance) + advance.add_argument("--evidence", action="append", required=True) + advance.add_argument("--next-action", required=True) + advance.add_argument("--consume", type=int, default=1) + advance.set_defaults(handler=command_advance) + + finish = subparsers.add_parser("finish", help="record a terminal outcome") + add_state_argument(finish) + finish.add_argument("--status", choices=sorted(TERMINAL_STATUSES), required=True) + finish.add_argument("--evidence", action="append", required=True) + finish.add_argument("--verification") + finish.add_argument("--next-action") + finish.set_defaults(handler=command_finish) + + annotate = subparsers.add_parser( + "annotate", + help="append corrected evidence without changing status or budget", + ) + add_state_argument(annotate) + annotate.add_argument("--evidence", action="append", required=True) + annotate.add_argument("--next-action") + annotate.set_defaults(handler=command_annotate) + + validate = subparsers.add_parser("validate", help="validate an existing state") + add_state_argument(validate) + validate.set_defaults(handler=lambda args: read_state(args.state)) + + show = subparsers.add_parser("show", help="render the current contract") + add_state_argument(show) + show.add_argument("--json", action="store_true") + show.set_defaults(handler=lambda args: read_state(args.state)) + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + if getattr(args, "consume", 1) < 1: + parser.error("--consume must be a positive integer") + try: + state = args.handler(args) + except StateError as exc: + print(f"loop-state: {exc}", file=sys.stderr) + return 2 + if args.command == "show" and not args.json: + print(summary(state)) + else: + print(json.dumps(state, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/loop-engineering/tests/test_loop_state.py b/skills/loop-engineering/tests/test_loop_state.py new file mode 100755 index 0000000..4e81817 --- /dev/null +++ b/skills/loop-engineering/tests/test_loop_state.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Fixtures for the loop-engineering state CLI.""" + +from __future__ import annotations + +import json +import pathlib +import subprocess +import sys +import tempfile +import unittest + + +SCRIPT = pathlib.Path(__file__).parents[1] / "scripts" / "loop_state.py" + + +class LoopStateTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary_directory.cleanup) + self.state = pathlib.Path(self.temporary_directory.name) / "loop.json" + + def run_cli( + self, + *arguments: str, + expected_returncode: int = 0, + ) -> subprocess.CompletedProcess[str]: + result = subprocess.run( + [sys.executable, str(SCRIPT), *arguments], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual( + result.returncode, + expected_returncode, + msg=f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}", + ) + return result + + def initialize(self, limit: int = 2) -> dict[str, object]: + result = self.run_cli( + "init", + "--state", + str(self.state), + "--goal", + "targeted test passes", + "--evidence", + "failure reproduced", + "--budget-unit", + "hypotheses", + "--budget-limit", + str(limit), + "--next-action", + "test timing hypothesis", + "--allowed-effect", + "edit worktree", + ) + return json.loads(result.stdout) + + def test_init_and_validate(self) -> None: + state = self.initialize() + self.assertEqual(state["terminal_status"], "running") + self.assertEqual(state["budget"]["used"], 0) + self.run_cli("validate", "--state", str(self.state)) + + def test_init_refuses_to_overwrite_existing_state(self) -> None: + self.initialize() + result = self.run_cli( + "init", + "--state", + str(self.state), + "--goal", + "different goal", + "--evidence", + "none", + "--budget-unit", + "turns", + "--budget-limit", + "1", + "--next-action", + "stop", + expected_returncode=2, + ) + self.assertIn("refusing to overwrite", result.stderr) + + def test_advance_exhausts_budget_without_claiming_complete(self) -> None: + self.initialize(limit=1) + result = self.run_cli( + "advance", + "--state", + str(self.state), + "--evidence", + "timing hypothesis falsified", + "--next-action", + "handoff with evidence", + ) + state = json.loads(result.stdout) + self.assertEqual(state["terminal_status"], "budget_exhausted") + self.assertNotEqual(state["terminal_status"], "complete") + + def test_complete_requires_verification_and_failed_transition_is_atomic( + self, + ) -> None: + self.initialize() + before = self.state.read_text() + result = self.run_cli( + "finish", + "--state", + str(self.state), + "--status", + "complete", + "--evidence", + "agent says fixed", + expected_returncode=2, + ) + self.assertIn("requires --verification", result.stderr) + self.assertEqual(self.state.read_text(), before) + + def test_complete_records_external_verification(self) -> None: + self.initialize() + result = self.run_cli( + "finish", + "--state", + str(self.state), + "--status", + "complete", + "--evidence", + "three consecutive passes", + "--verification", + "pytest log artifact: /tmp/targeted-test.log", + "--next-action", + "checkpoint worklog", + ) + state = json.loads(result.stdout) + self.assertEqual(state["terminal_status"], "complete") + self.assertIn("pytest log", state["verification"]) + + def test_annotate_corrects_terminal_evidence_without_reopening(self) -> None: + self.initialize(limit=1) + exhausted = json.loads( + self.run_cli( + "advance", + "--state", + str(self.state), + "--evidence", + "no Git metadata", + "--next-action", + "obtain target repository", + ).stdout + ) + corrected = json.loads( + self.run_cli( + "annotate", + "--state", + str(self.state), + "--evidence", + "correction: cwd has .git but is not the target repository", + ).stdout + ) + self.assertEqual(corrected["terminal_status"], "budget_exhausted") + self.assertEqual(corrected["budget"], exhausted["budget"]) + self.assertEqual(corrected["history"][-1]["event"], "annotated") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/run.sh b/tests/run.sh index dae3556..95fe245 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -112,6 +112,44 @@ PY then ok "which-model lazy-loading contract"; else fail "which-model lazy-loading contract"; fi if python3 - <<'PY' import pathlib + +skill = pathlib.Path("skills/loop-engineering") +root = (skill / "SKILL.md").read_text() +examples = (skill / "references/examples.md").read_text() +hosts = (skill / "references/hosts.md").read_text() +protocol = (skill / "references/protocol.md").read_text() +state_script = (skill / "scripts/loop_state.py").read_text() +references = {path.name for path in (skill / "references").glob("*.md")} +checks = { + "thin portable root": len(root.splitlines()) <= 80, + "standard frontmatter only": root.split("---", 2)[1].count("\n") == 3, + "script-first route": "scripts/loop_state.py init" in root, + "no hand-edited state": "do not hand-edit its JSON" in root, + "worklog context route": " --for=resume" in protocol, + "tracker hydration route": "hydrate the host tracker from the" in protocol, + "worklog delegation route": " --for=compact" in protocol and "spawn " not in protocol, + "worklog creation gate": "slugless `sync`" in protocol, + "worklog checkpoint route": "persist arbitrary evidence or task-body changes" in protocol, + "terminal evidence rule": "model's prose claim is not evidence" in root, + "complete verification guard": "complete requires" in state_script and "--verification" in state_script, + "append-only correction": "def command_annotate" in state_script and "annotated" in state_script, + "atomic state write": "os.replace" in state_script, + "host differences deferred": references == {"examples.md", "hosts.md", "protocol.md"}, + "no host-only injection": "!`" not in root and "allowed-tools:" not in root, + "three contrastive fixtures": examples.count("\n## ") == 4, + "Codex shared install": "~/.agents/skills/" in hosts, + "Claude personal install": "~/.claude/skills/" in hosts, + "Cursor shared install": ".agents/skills/" in hosts, + "Cursor native install": "~/.cursor/skills/" in hosts, +} +missing = [name for name, passed in checks.items() if not passed] +if missing: + print("loop-engineering portability contract failed: " + "; ".join(missing)) + raise SystemExit(1) +PY + then ok "loop-engineering portability contract"; else fail "loop-engineering portability contract"; fi + if python3 - <<'PY' +import pathlib import re text = pathlib.Path("skills/council/SKILL.md").read_text() @@ -202,6 +240,12 @@ PY fail "worklog PR reconciliation fixtures" fi + if python3 -m unittest skills/loop-engineering/tests/test_loop_state.py >/dev/null; then + ok "loop-engineering state transition fixtures" + else + fail "loop-engineering state transition fixtures" + fi + if python3 - <<'PY' import pathlib import subprocess @@ -304,7 +348,7 @@ PY fi rm -rf "$fake_home" - local skill_names skill_name skill_md + local skill_names skill_name skill_md shared_skill_md install_home canonical_skill installed_skill skills_root skill_names=$(python3 - <<'PY' import pathlib import yaml @@ -329,14 +373,59 @@ PY rc=$? set -e skill_md="$fake_home/.claude/skills/$skill_name/SKILL.md" - if [[ $rc -eq 0 && -f "$skill_md" ]]; then - ok "install-skills installs $skill_name" + shared_skill_md="$fake_home/.agents/skills/$skill_name/SKILL.md" + if [[ $rc -eq 0 && -f "$skill_md" && -f "$shared_skill_md" && -f "$fake_home/.cursor/skills/$skill_name/SKILL.md" ]]; then + ok "install-skills installs $skill_name to all user roots" else fail "install-skills failed for $skill_name (rc=$rc)" fi rm -rf "$fake_home" done <<< "$skill_names" + install_home=$(mktemp -d) + if HOME="$install_home" PYTHONPATH="${python_site_path}${PYTHONPATH:+:$PYTHONPATH}" ./bin/install-skills.sh >/dev/null 2>&1; then + rc=0 + for canonical_skill in "$REPO_ROOT"/skills/*/SKILL.md; do + skill_name=$(basename "$(dirname "$canonical_skill")") + for skills_root in .agents/skills .claude/skills .cursor/skills; do + installed_skill="$install_home/$skills_root/$skill_name/SKILL.md" + if [[ ! -f "$installed_skill" || "$(realpath "$installed_skill")" != "$(realpath "$canonical_skill")" ]]; then + rc=1 + fi + done + done + else + rc=1 + fi + if [[ $rc -eq 0 ]]; then + ok "supported installer exposes every canonical skill to all user roots" + else + fail "supported installer cross-host all-skill exposure" + fi + rm -rf "$install_home" + + install_home=$(mktemp -d) + if CODER_SYMLINK_DIR="$install_home" ./install.sh >/dev/null 2>&1; then + rc=0 + for canonical_skill in "$REPO_ROOT"/skills/*/SKILL.md; do + skill_name=$(basename "$(dirname "$canonical_skill")") + for skills_root in .agents/skills .claude/skills .cursor/skills; do + installed_skill="$install_home/$skills_root/$skill_name/SKILL.md" + if [[ ! -f "$installed_skill" || "$(realpath "$installed_skill")" != "$(realpath "$canonical_skill")" ]]; then + rc=1 + fi + done + done + else + rc=1 + fi + if [[ $rc -eq 0 ]]; then + ok "install.sh exposes every canonical skill to all user roots" + else + fail "install.sh cross-host skill exposure" + fi + rm -rf "$install_home" + if python3 - <<'PY' import re From b67a09278fa2da3d56c9cfd5058d721519a019b8 Mon Sep 17 00:00:00 2001 From: cheshirecode Date: Sun, 26 Jul 2026 00:31:18 -0400 Subject: [PATCH 2/2] feat: add evidence gate skill --- manifest/skills.yaml | 8 + skills/evidence-gate/SKILL.md | 54 +++++ skills/evidence-gate/agents/openai.yaml | 4 + skills/evidence-gate/scripts/evidence_gate.py | 214 ++++++++++++++++++ .../evidence-gate/tests/test_evidence_gate.py | 159 +++++++++++++ skills/loop-engineering/SKILL.md | 10 +- tests/run.sh | 29 +++ 7 files changed, 474 insertions(+), 4 deletions(-) create mode 100644 skills/evidence-gate/SKILL.md create mode 100644 skills/evidence-gate/agents/openai.yaml create mode 100755 skills/evidence-gate/scripts/evidence_gate.py create mode 100755 skills/evidence-gate/tests/test_evidence_gate.py diff --git a/manifest/skills.yaml b/manifest/skills.yaml index 182d65e..74f10ca 100644 --- a/manifest/skills.yaml +++ b/manifest/skills.yaml @@ -45,6 +45,14 @@ skills: path: skills/loop-engineering install_to: ~/.claude/skills/loop-engineering + - name: evidence-gate + description: Require typed evidence coverage for every completion criterion. + source: + type: subpath + ref: HEAD + path: skills/evidence-gate + install_to: ~/.claude/skills/evidence-gate + - name: ship-hygiene description: Pre-handoff sweep across worklog + PRs + CI. source: diff --git a/skills/evidence-gate/SKILL.md b/skills/evidence-gate/SKILL.md new file mode 100644 index 0000000..0a284b1 --- /dev/null +++ b/skills/evidence-gate/SKILL.md @@ -0,0 +1,54 @@ +--- +name: evidence-gate +description: Gate completion claims by mapping every observable goal or acceptance criterion to typed command, artifact, Git, GitHub, or URL evidence. Use before marking a multi-clause task, agent loop, deployment, PR, or verification workflow complete, especially when tests passing does not prove delivery, merge, or user-visible success. +--- + +# Evidence Gate + +Require evidence coverage, not a persuasive completion summary. The script +checks that every declared criterion has at least one typed evidence record; the +agent remains responsible for verifying that each record is truthful and +relevant. + +## Declare every goal clause + +Resolve `` as this `SKILL.md` file's directory. Use stable lowercase +criterion IDs and an explicit gate path. + +```bash +python3 /scripts/evidence_gate.py init \ + --gate \ + --goal "" \ + --criterion "tests=full suite passes" \ + --criterion "merge=PR is merged into main" +``` + +Do not collapse distinct outcomes into one criterion. Tests, commit, deployment, +PR state, and user-visible behavior are separate when the goal names them. + +## Record verified evidence + +After checking the source, attach evidence to exactly one criterion: + +```bash +python3 /scripts/evidence_gate.py record \ + --gate \ + --criterion tests \ + --kind command \ + --ref "tests/run.sh all" \ + --result "62 pass, 0 fail" +``` + +Kinds are `command`, `artifact`, `git`, `github`, and `url`. Never record model +prose as evidence. For GitHub completion, inspect the exact PR head, base, +state, non-empty diff, and merged target before recording it. + +## Gate completion + +Run `check --gate `. Exit `1` means criteria remain uncovered; do not +claim completion. Exit `0` returns a `verification` value containing the +gate-file path and SHA-256 digest. Pass that value to the parent workflow's +completion record. + +The digest proves which coverage artifact was checked, not that an evidence +source was interpreted correctly. Preserve the gate file with the task evidence. diff --git a/skills/evidence-gate/agents/openai.yaml b/skills/evidence-gate/agents/openai.yaml new file mode 100644 index 0000000..a947b08 --- /dev/null +++ b/skills/evidence-gate/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Evidence Gate" + short_description: "Require evidence for every completion criterion" + default_prompt: "Use $evidence-gate to verify that every completion criterion has tool or artifact evidence." diff --git a/skills/evidence-gate/scripts/evidence_gate.py b/skills/evidence-gate/scripts/evidence_gate.py new file mode 100755 index 0000000..f62c664 --- /dev/null +++ b/skills/evidence-gate/scripts/evidence_gate.py @@ -0,0 +1,214 @@ +#!/usr/bin/env python3 +"""Require typed evidence coverage for every completion criterion.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import pathlib +import re +import sys +import tempfile +from datetime import datetime, timezone +from typing import Any + + +SCHEMA_VERSION = 1 +EVIDENCE_KINDS = {"command", "artifact", "git", "github", "url"} +CRITERION_ID = re.compile(r"^[a-z][a-z0-9_-]*$") + + +class GateError(ValueError): + """Raised when an evidence-gate operation violates the contract.""" + + +def now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def parse_criterion(value: str) -> tuple[str, str]: + criterion_id, separator, description = value.partition("=") + if not separator or not CRITERION_ID.fullmatch(criterion_id): + raise GateError("criterion must use id=description with a lowercase identifier") + if not description.strip(): + raise GateError(f"criterion description must not be empty: {criterion_id}") + return criterion_id, description + + +def validate_gate(gate: dict[str, Any]) -> None: + if gate.get("schema_version") != SCHEMA_VERSION: + raise GateError(f"schema_version must be {SCHEMA_VERSION}") + if not isinstance(gate.get("goal"), str) or not gate["goal"].strip(): + raise GateError("goal must be a non-empty string") + criteria = gate.get("criteria") + if not isinstance(criteria, dict) or not criteria: + raise GateError("criteria must be a non-empty object") + for criterion_id, criterion in criteria.items(): + if not CRITERION_ID.fullmatch(criterion_id): + raise GateError(f"invalid criterion identifier: {criterion_id}") + if not isinstance(criterion, dict): + raise GateError(f"criterion must be an object: {criterion_id}") + description = criterion.get("description") + if not isinstance(description, str) or not description.strip(): + raise GateError(f"criterion description is empty: {criterion_id}") + evidence = criterion.get("evidence") + if not isinstance(evidence, list): + raise GateError(f"criterion evidence must be a list: {criterion_id}") + for record in evidence: + validate_record(record, criterion_id) + + +def validate_record(record: Any, criterion_id: str) -> None: + if not isinstance(record, dict): + raise GateError(f"evidence record must be an object: {criterion_id}") + if record.get("kind") not in EVIDENCE_KINDS: + raise GateError(f"invalid evidence kind for {criterion_id}") + for field in ("ref", "result", "recorded_at"): + if not isinstance(record.get(field), str) or not record[field].strip(): + raise GateError(f"evidence {field} is empty for {criterion_id}") + + +def read_gate(path: pathlib.Path) -> dict[str, Any]: + try: + gate = json.loads(path.read_text()) + except FileNotFoundError as exc: + raise GateError(f"gate file does not exist: {path}") from exc + except json.JSONDecodeError as exc: + raise GateError(f"gate file is not valid JSON: {path}: {exc}") from exc + validate_gate(gate) + return gate + + +def write_gate(path: pathlib.Path, gate: dict[str, Any]) -> None: + validate_gate(gate) + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".tmp", + ) + temporary_path = pathlib.Path(temporary_name) + try: + with os.fdopen(fd, "w") as handle: + json.dump(gate, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + finally: + temporary_path.unlink(missing_ok=True) + + +def command_init(args: argparse.Namespace) -> dict[str, Any]: + if args.gate.exists() and not args.force: + raise GateError(f"refusing to overwrite existing gate: {args.gate}") + criteria: dict[str, dict[str, Any]] = {} + for raw_criterion in args.criterion: + criterion_id, description = parse_criterion(raw_criterion) + if criterion_id in criteria: + raise GateError(f"duplicate criterion: {criterion_id}") + criteria[criterion_id] = {"description": description, "evidence": []} + gate = { + "schema_version": SCHEMA_VERSION, + "goal": args.goal, + "criteria": criteria, + } + write_gate(args.gate, gate) + return gate + + +def command_record(args: argparse.Namespace) -> dict[str, Any]: + gate = read_gate(args.gate) + if args.criterion not in gate["criteria"]: + raise GateError(f"unknown criterion: {args.criterion}") + gate["criteria"][args.criterion]["evidence"].append( + { + "kind": args.kind, + "ref": args.ref, + "result": args.result, + "recorded_at": now(), + } + ) + write_gate(args.gate, gate) + return gate + + +def gate_report(path: pathlib.Path, gate: dict[str, Any]) -> dict[str, Any]: + missing = [ + criterion_id + for criterion_id, criterion in gate["criteria"].items() + if not criterion["evidence"] + ] + covered = [ + criterion_id + for criterion_id, criterion in gate["criteria"].items() + if criterion["evidence"] + ] + digest = hashlib.sha256(path.read_bytes()).hexdigest() + status = "satisfied" if not missing else "unsatisfied" + verification = ( + f"evidence-gate:{path.resolve()}#sha256={digest}" if not missing else "" + ) + return { + "status": status, + "goal": gate["goal"], + "covered": covered, + "missing": missing, + "verification": verification, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + initialize = subparsers.add_parser("init", help="declare completion criteria") + initialize.add_argument("--gate", type=pathlib.Path, required=True) + initialize.add_argument("--goal", required=True) + initialize.add_argument( + "--criterion", + action="append", + required=True, + help="id=description; repeat for every observable goal clause", + ) + initialize.add_argument("--force", action="store_true") + initialize.set_defaults(handler=command_init) + + record = subparsers.add_parser("record", help="attach evidence to one criterion") + record.add_argument("--gate", type=pathlib.Path, required=True) + record.add_argument("--criterion", required=True) + record.add_argument("--kind", choices=sorted(EVIDENCE_KINDS), required=True) + record.add_argument("--ref", required=True) + record.add_argument("--result", required=True) + record.set_defaults(handler=command_record) + + check = subparsers.add_parser("check", help="require evidence for all criteria") + check.add_argument("--gate", type=pathlib.Path, required=True) + check.set_defaults(handler=lambda args: read_gate(args.gate)) + + show = subparsers.add_parser("show", help="render the full gate") + show.add_argument("--gate", type=pathlib.Path, required=True) + show.set_defaults(handler=lambda args: read_gate(args.gate)) + return parser + + +def main() -> int: + parser = build_parser() + args = parser.parse_args() + try: + gate = args.handler(args) + if args.command == "check": + report = gate_report(args.gate, gate) + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["status"] == "satisfied" else 1 + print(json.dumps(gate, indent=2, sort_keys=True)) + return 0 + except GateError as exc: + print(f"evidence-gate: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/evidence-gate/tests/test_evidence_gate.py b/skills/evidence-gate/tests/test_evidence_gate.py new file mode 100755 index 0000000..67ddc4a --- /dev/null +++ b/skills/evidence-gate/tests/test_evidence_gate.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Fixtures for the evidence-gate CLI.""" + +from __future__ import annotations + +import json +import pathlib +import subprocess +import sys +import tempfile +import unittest + + +SCRIPT = pathlib.Path(__file__).parents[1] / "scripts" / "evidence_gate.py" + + +class EvidenceGateTest(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary_directory.cleanup) + self.gate = pathlib.Path(self.temporary_directory.name) / "gate.json" + + def run_cli( + self, + *arguments: str, + expected_returncode: int = 0, + ) -> subprocess.CompletedProcess[str]: + result = subprocess.run( + [sys.executable, str(SCRIPT), *arguments], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual( + result.returncode, + expected_returncode, + msg=f"stdout:\n{result.stdout}\nstderr:\n{result.stderr}", + ) + return result + + def initialize(self) -> None: + self.run_cli( + "init", + "--gate", + str(self.gate), + "--goal", + "change is verified and merged", + "--criterion", + "tests=full suite passes", + "--criterion", + "merge=PR is merged into main", + ) + + def test_check_rejects_uncovered_criterion(self) -> None: + self.initialize() + self.run_cli( + "record", + "--gate", + str(self.gate), + "--criterion", + "tests", + "--kind", + "command", + "--ref", + "tests/run.sh all", + "--result", + "62 pass, 0 fail", + ) + report = json.loads( + self.run_cli( + "check", + "--gate", + str(self.gate), + expected_returncode=1, + ).stdout + ) + self.assertEqual(report["status"], "unsatisfied") + self.assertEqual(report["missing"], ["merge"]) + self.assertEqual(report["verification"], "") + + def test_check_returns_digest_when_every_criterion_is_covered(self) -> None: + self.initialize() + for criterion, kind, reference, result in ( + ("tests", "command", "tests/run.sh all", "62 pass, 0 fail"), + ( + "merge", + "github", + "https://github.com/cheshirecode/dotfiles/pull/7", + "merged into main", + ), + ): + self.run_cli( + "record", + "--gate", + str(self.gate), + "--criterion", + criterion, + "--kind", + kind, + "--ref", + reference, + "--result", + result, + ) + report = json.loads(self.run_cli("check", "--gate", str(self.gate)).stdout) + self.assertEqual(report["status"], "satisfied") + self.assertEqual(report["missing"], []) + self.assertIn("#sha256=", report["verification"]) + + def test_unknown_criterion_fails_without_mutation(self) -> None: + self.initialize() + before = self.gate.read_text() + self.run_cli( + "record", + "--gate", + str(self.gate), + "--criterion", + "deploy", + "--kind", + "artifact", + "--ref", + "/tmp/deploy.log", + "--result", + "passed", + expected_returncode=2, + ) + self.assertEqual(self.gate.read_text(), before) + + def test_duplicate_criterion_is_rejected(self) -> None: + self.run_cli( + "init", + "--gate", + str(self.gate), + "--goal", + "duplicate test", + "--criterion", + "tests=first", + "--criterion", + "tests=second", + expected_returncode=2, + ) + self.assertFalse(self.gate.exists()) + + def test_init_refuses_overwrite(self) -> None: + self.initialize() + self.run_cli( + "init", + "--gate", + str(self.gate), + "--goal", + "different goal", + "--criterion", + "tests=passes", + expected_returncode=2, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/skills/loop-engineering/SKILL.md b/skills/loop-engineering/SKILL.md index d3ea55b..d8cdc78 100644 --- a/skills/loop-engineering/SKILL.md +++ b/skills/loop-engineering/SKILL.md @@ -42,11 +42,13 @@ manually and label the run as a non-deterministic fallback. 2. Choose the smallest action that advances or falsifies the approach. 3. Check effect scope; serialize writes unless isolation is proven. 4. Execute and verify. A model's prose claim is not evidence. -5. On verified success, run `finish --status complete --verification - "" --evidence ""`. -6. Otherwise run `advance --evidence "" --next-action ""`. +5. On apparent success, invoke `$evidence-gate`. Map every observable goal + clause to typed evidence and require its `check` command to pass. +6. Only then run `finish --status complete --verification + "" --evidence ""`. +7. Otherwise run `advance --evidence "" --next-action ""`. The script emits `budget_exhausted` when the declared ceiling is consumed. -7. Run `show`; continue only while `terminal_status` is `running`. +8. Run `show`; continue only while `terminal_status` is `running`. If later evidence contradicts a recorded fact, use `annotate --evidence ""`. Preserve the audit trail; do not reopen or hand-edit terminal diff --git a/tests/run.sh b/tests/run.sh index 95fe245..addf8bb 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -113,6 +113,28 @@ PY if python3 - <<'PY' import pathlib +skill = pathlib.Path("skills/evidence-gate") +root = (skill / "SKILL.md").read_text() +script = (skill / "scripts/evidence_gate.py").read_text() +checks = { + "thin root": len(root.splitlines()) <= 70, + "standard frontmatter only": root.split("---", 2)[1].count("\n") == 3, + "script route": "scripts/evidence_gate.py init" in root, + "typed evidence": all(kind in root for kind in ("command", "artifact", "git", "github", "url")), + "digest contract": "SHA-256 digest" in root, + "no placeholder": "TODO" not in root, + "atomic gate write": "os.replace" in script, + "coverage check": '"status": status' in script and '"missing": missing' in script, +} +missing = [name for name, passed in checks.items() if not passed] +if missing: + print("evidence-gate contract failed: " + "; ".join(missing)) + raise SystemExit(1) +PY + then ok "evidence-gate contract"; else fail "evidence-gate contract"; fi + if python3 - <<'PY' +import pathlib + skill = pathlib.Path("skills/loop-engineering") root = (skill / "SKILL.md").read_text() examples = (skill / "references/examples.md").read_text() @@ -132,6 +154,7 @@ checks = { "worklog checkpoint route": "persist arbitrary evidence or task-body changes" in protocol, "terminal evidence rule": "model's prose claim is not evidence" in root, "complete verification guard": "complete requires" in state_script and "--verification" in state_script, + "evidence-gate completion route": "$evidence-gate" in root and "evidence-gate verification value" in root, "append-only correction": "def command_annotate" in state_script and "annotated" in state_script, "atomic state write": "os.replace" in state_script, "host differences deferred": references == {"examples.md", "hosts.md", "protocol.md"}, @@ -246,6 +269,12 @@ PY fail "loop-engineering state transition fixtures" fi + if python3 -m unittest skills/evidence-gate/tests/test_evidence_gate.py >/dev/null; then + ok "evidence-gate coverage fixtures" + else + fail "evidence-gate coverage fixtures" + fi + if python3 - <<'PY' import pathlib import subprocess