# Control Rubric

Use this when setting gates, iteration caps, budgets, and stop conditions.

## Required Guards

- `loop_control.max_iterations`
- `gates.*.max_revisions`
- `loop_control.no_progress.max_stalled_iterations`
- At least one wall-clock, token, or USD budget cap when external models run.
  The generated Python runner enforces wall-clock caps directly; token and USD
  caps are advisory unless the chosen model CLI exposes accounting that the
  loop operator wires in separately.
- A stop condition that describes success.
- A stop condition that describes no-progress or repeated failure.

## Good Gate Design

- Plan gate runs before delivery work.
- Delivery gate runs after each delivery artifact.
- Programmatic checks run before judge calls when possible.
- Human checkpoints sit at high-leverage points, usually after plan approval or
  before external egress.
- Resume happens at gate boundaries unless the user explicitly needs finer
  granularity.

## Execution Boundary

- Name where the loop is allowed to modify files: current workspace, branch,
  worktree, throwaway directory, or an external orchestrator workspace.
- Identify actions with side effects: pushes, PR comments, Slack messages,
  deploys, file deletes, database writes, or vendor sends.
- Decide whether side-effecting actions require approval, idempotency notes, or
  duplicate-action checks.
- If the loop may run on a schedule or in parallel, call out the need for an
  external orchestrator with concurrency controls.

## Failure Behavior

- Stop immediately when a hard cap is reached.
- Write the latest state to `loop-workspace/state.json`.
- Append each meaningful step, decision, check result, and blocker to
  `loop-workspace/run-log.md`.
- Preserve review notes even when the gate fails.
- Stop or ask the human when the same blocker repeats for the configured
  no-progress window.
- Do not let the host keep revising forever.

## Anti-Patterns

- No maximum iteration count.
- A judge gate with no judge.
- A budget cap in prose but not in `loop_control`.
- No no-progress detector.
- A loop that can send duplicate external notifications or repeat destructive
  actions after restart.
- Scheduled or multi-agent work with no durable orchestrator or concurrency
  story.
- Human signoff required but no checkpoint.
- Stop conditions that require subjective self-satisfaction.

# Council Rubric

Use this when selecting reviewers and judges.

## Roles

`reviewer`
: Gives notes only. It may improve quality, but it cannot declare a gate clean.

`judge`
: Gives a structured verdict. It can be used as a gate `verdict_source`.

## Selection Guidance

- Prefer a different model family from the host for blind-spot coverage.
- Prefer local models such as `ollama` when privacy matters more than judgment
  quality.
- Prefer a judge for gates that must block progress.
- Prefer a reviewer for brainstorming, adversarial notes, or tone critique
  where a deterministic pass/fail would be fake precision.
- Keep council scope small: `plan`, `delivery`, or specific paths.

## Gate Rule

`verdict_policy: revise_until_clean` requires `verdict_source` to be either a
judge member or `human`. A reviewer-only gate can use `fixed_passes`, but it
cannot honestly claim clean.

## Judge Rubric Tips

- Name the artifact being judged.
- Name the exact criteria IDs.
- Ask for blocking issues, not general commentary.
- Require the fenced JSON verdict first or last.
- Keep the judge prompt short enough that the artifact, not the instruction
  wrapper, dominates the context.

## Privacy Notes

Cross-vendor review can send project context to another CLI and vendor account.
Always name the destination, scope what it receives, apply redaction globs, and
require consent before the first send.

# Goal Rubric

Use this when shaping the user's loop goal.

## Good Goal Shape

- Names the concrete outcome, not only the activity.
- Defines the artifact or state that proves the loop finished.
- Sets scope boundaries: included work, excluded work, and maximum depth.
- Names context sources the host must gather instead of assumptions it may make.
- Identifies the user, customer, system, or reviewer who will consume the result.

## Critique Prompts

- What would count as done if two competent agents disagreed?
- Which terms are subjective and need a measurable proxy?
- What context must be read before the host drafts a plan?
- What is explicitly out of scope for this loop?
- Can the goal be split into plan, delivery, and verification artifacts?

## Anti-Patterns

- "Improve the project" without a target artifact.
- "Make it good" without criteria.
- "Research X" without the decision the research supports.
- Goals where success depends on information the loop never gathers.
- Goals that require endless polishing with no stop condition.

## Better Examples

Weak: "Make our onboarding better."

Better: "Produce a 5-step onboarding workflow map for new enterprise users,
with each step assigned to a product surface, email, human owner, or missing
capability, and with no unresolved TBDs."

Weak: "Fix the flaky tests."

Better: "Identify and patch the root cause of the checkout test flake, prove it
with 20 local repeats or a CI rerun, and leave a short note explaining the
failure mode and the verification evidence."

# Model Detection and Privacy Notes

Loop-architect detection is intentionally dumb and transparent. It stores
invocation metadata only, never credentials.

## Registry

Default registry path:

```text
~/.loop-architect/models.json
```

Registry entries should look like:

```json
{
  "kiro": {
    "cli": "kiro-cli",
    "invoke": ["kiro-cli", "chat", "--trust-all-tools", "-p"],
    "probe": ["kiro-cli", "--version"],
    "available": true,
    "authed": true,
    "local": false,
    "capabilities": {
      "headless": true,
      "goal": true,
      "subagent": true,
      "review_loop": true
    }
  },
  "claude": {
    "cli": "claude",
    "invoke": ["claude", "-p"],
    "probe": ["claude", "--version"],
    "available": true,
    "authed": true,
    "local": false,
    "capabilities": {
      "headless": true,
      "goal": true,
      "subagent": false,
      "review_loop": false
    }
  }
}
```

## Capabilities

`headless`
: The CLI accepts a prompt via stdin/argument and returns output via stdout
  without interactive prompts. Required for use as host or judge in the
  external Python runner.

`goal`
: The CLI supports a `/goal` command that runs an autonomous loop with
  self-verification. When present, RUN_IN_SESSION.md can emit a `/goal`
  one-liner as an alternative execution path.

`subagent`
: The CLI can spawn isolated sub-agents with their own context. When present,
  the council can use native subagent review loops instead of shelling out.

`review_loop`
: The CLI supports iterative review loops with trigger-based feedback (e.g.
  Kiro's `loop_to` with `NEEDS_CHANGES` trigger). Enables native cross-model
  review without the external runner.

## Kiro CLI Specifics

- Headless mode requires `--trust-all-tools` or the session halts waiting for
  tool approval.
- Full invoke pattern: `["kiro-cli", "chat", "--trust-all-tools", "-p"]`
- The `/goal --max N` command provides native loop execution with configurable
  iteration limits (default 5).
- Subagent review loops use a `trigger` string (e.g. `NEEDS_CHANGES`) and
  `max_iterations` cap.

## `authed` Semantics

`authed` means the basic probe command exited cleanly. It is a convenience
signal, not a guarantee that a future paid model call will succeed.

## Default Redactions

- `.env`
- `.env.*`
- `secrets/**`
- `**/*.key`

Add project-specific globs for customer data, private transcripts, or internal
design docs before sending anything to a non-local council member.

## Local Model UX

Surface `ollama` as the privacy-preserving option when present. It may be lower
quality than frontier hosted models, but it keeps council review in-house.

# Verification Rubric

Use this when converting the user's definition of done into typed criteria.

## Taxonomy

`programmatic`
: A command or deterministic check returns pass/fail. Use this whenever
possible. Examples: tests, build, lint, schema validation, snapshot comparison,
or an extraction script that checks required headings.

`judge`
: A model scores a rubric and returns a structured verdict. Use this for
semantic quality that cannot be cheaply checked by code. The rubric must be
specific enough that a different model can apply it consistently.

`human`
: A person must sign off. Use this for taste, business judgment, private
knowledge, legal risk, or decisions where the user is the true authority.

## Required Fields

- Every criterion needs `id` and `type`.
- `programmatic` needs `check` as an argv array and `expect`.
- `judge` needs `rubric`.
- `human` needs `prompt`.

## Strong Criteria

- Check one thing at a time.
- Say what failure means.
- Prefer deterministic checks before model judgment.
- Make judge rubrics observable against artifacts the judge receives.
- Avoid relying on the host model to grade its own work.

## Anti-Patterns

- All criteria are judge or human criteria when tests or schema checks exist.
- "No errors thrown" as the only success criterion.
- Criteria that require hidden context not sent to the judge.
- Rubrics like "high quality" or "comprehensive" without dimensions.
- Programmatic checks written as shell strings instead of argv arrays.

## Structured Judge Contract

Judges should return a fenced JSON object:

```json
{
  "verdict": "pass",
  "blocking_issues": [],
  "confidence": 0.86,
  "notes": "The artifact satisfies the rubric."
}
```

Valid verdicts are `pass` and `revise`. If output cannot be parsed, the runner
will treat it as `revise` with a warning.

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://github.com/ksimback/looper/schema/loop.resolved.v1.json",
  "title": "Looper resolved spec v1",
  "allOf": [
    { "$ref": "./loop.v1.schema.json" },
    {
      "type": "object",
      "required": ["compiled_at", "source", "criteria_by_id", "council_by_id"],
      "properties": {
        "compiled_at": { "type": "string" },
        "source": { "type": "string" },
        "criteria_by_id": {
          "type": "object",
          "additionalProperties": { "$ref": "./loop.v1.schema.json#/$defs/criterion" }
        },
        "council_by_id": {
          "type": "object",
          "additionalProperties": true
        }
      }
    }
  ]
}

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://github.com/ksimback/looper/schema/loop.v1.json",
  "title": "Looper authoring spec v1",
  "type": "object",
  "required": ["version", "goal", "host", "gates", "loop_control", "workspace"],
  "properties": {
    "version": { "const": 1 },
    "meta": {
      "type": "object",
      "additionalProperties": true
    },
    "goal": {
      "type": "object",
      "required": ["statement", "definition_of_done", "verification"],
      "properties": {
        "statement": { "type": "string", "minLength": 1 },
        "definition_of_done": { "type": "string", "minLength": 1 },
        "context_sources": {
          "type": "array",
          "items": {
            "type": "object",
            "anyOf": [
              { "required": ["file"] },
              { "required": ["cmd"] }
            ]
          }
        },
        "verification": {
          "type": "array",
          "items": { "$ref": "#/$defs/criterion" }
        }
      },
      "additionalProperties": true
    },
    "host": { "$ref": "#/$defs/model_invocation" },
    "council": {
      "type": "array",
      "items": {
        "allOf": [
          { "$ref": "#/$defs/model_invocation" },
          {
            "type": "object",
            "required": ["id", "role"],
            "properties": {
              "id": { "type": "string", "minLength": 1 },
              "role": { "enum": ["reviewer", "judge"] },
              "scope": {
                "type": "array",
                "items": { "type": "string" }
              },
              "local": { "type": "boolean" }
            }
          }
        ]
      }
    },
    "gates": {
      "type": "object",
      "required": ["plan_gate", "delivery_gate"],
      "properties": {
        "plan_gate": { "$ref": "#/$defs/gate" },
        "delivery_gate": { "$ref": "#/$defs/gate" }
      }
    },
    "loop_control": {
      "type": "object",
      "required": ["max_iterations"],
      "properties": {
        "max_iterations": { "type": "integer", "minimum": 1 },
        "budget": { "type": "object" },
        "no_progress": {
          "type": "object",
          "properties": {
            "max_stalled_iterations": { "type": "integer", "minimum": 1 },
            "signals": {
              "type": "array",
              "items": { "type": "string" }
            },
            "action": { "enum": ["stop", "human_checkpoint"] }
          },
          "additionalProperties": true
        },
        "human_checkpoints": {
          "type": "array",
          "items": { "type": "string" }
        },
        "stop_conditions": {
          "type": "array",
          "items": { "type": "string" }
        }
      }
    },
    "execution": {
      "type": "object",
      "properties": {
        "mode": { "enum": ["in_session", "external_runner", "orchestrated"] },
        "isolation": { "enum": ["current_workspace", "branch", "worktree", "sandbox"] },
        "side_effects": { "type": "object" }
      },
      "additionalProperties": true
    },
    "observability": {
      "type": "object",
      "properties": {
        "state_file": { "type": "string" },
        "run_log": { "type": "string" },
        "checkpoint_granularity": { "enum": ["gate", "step"] }
      },
      "additionalProperties": true
    },
    "privacy": { "type": "object" },
    "workspace": {
      "type": "object",
      "required": ["dir"],
      "properties": {
        "dir": { "type": "string", "minLength": 1 },
        "layout": {
          "type": "array",
          "items": { "type": "string" }
        }
      }
    }
  },
  "$defs": {
    "argv": {
      "type": "array",
      "minItems": 1,
      "items": { "type": "string" }
    },
    "model_invocation": {
      "type": "object",
      "required": ["cli", "invoke"],
      "properties": {
        "cli": { "type": "string" },
        "model": { "type": "string" },
        "invoke": { "$ref": "#/$defs/argv" },
        "timeout_sec": { "type": "integer", "minimum": 1 }
      },
      "additionalProperties": true
    },
    "criterion": {
      "type": "object",
      "required": ["id", "type"],
      "oneOf": [
        {
          "properties": {
            "type": { "const": "programmatic" },
            "check": { "$ref": "#/$defs/argv" },
            "expect": { "enum": ["exit_zero", "exit_nonzero", "stdout_contains"] },
            "contains": { "type": "string" }
          },
          "required": ["check", "expect"]
        },
        {
          "properties": {
            "type": { "const": "judge" },
            "rubric": { "type": "string", "minLength": 1 }
          },
          "required": ["rubric"]
        },
        {
          "properties": {
            "type": { "const": "human" },
            "prompt": { "type": "string", "minLength": 1 }
          },
          "required": ["prompt"]
        }
      ]
    },
    "gate": {
      "type": "object",
      "required": ["when", "members", "verdict_policy", "criteria", "max_revisions"],
      "properties": {
        "when": { "type": "string" },
        "members": {
          "type": "array",
          "items": { "type": "string" }
        },
        "verdict_policy": { "enum": ["revise_until_clean", "fixed_passes"] },
        "verdict_source": { "type": "string" },
        "criteria": {
          "type": "array",
          "items": { "type": "string" }
        },
        "max_revisions": { "type": "integer", "minimum": 0 }
      }
    }
  }
}

#!/usr/bin/env python3
"""Loop-architect helper CLI.

This script belongs to the scaffolding side of loop-architect. It may detect
installed CLIs, register invocation metadata, compile loop.yaml to
loop.resolved.json, and render LOOP.md. It must not invoke model CLIs to do
loop work.

Based on Looper by Kevin Simback (https://github.com/ksimback/looper), MIT License.
Adapted for Kiro CLI by ft.ia.br.
"""

from __future__ import annotations

import argparse
import datetime as _dt
import json
import os
from pathlib import Path
import shlex
import shutil
import subprocess
import sys
from typing import Any


DEFAULT_REDACTIONS = [".env", ".env.*", "secrets/**", "**/*.key"]
REGISTRY_PATH = Path.home() / ".loop-architect" / "models.json"

MODEL_PROBES: dict[str, dict[str, Any]] = {
    "kiro": {
        "invoke": ["kiro-cli", "chat", "--trust-all-tools", "-p"],
        "probe": ["kiro-cli", "--version"],
        "local": False,
        "install": "Install Kiro CLI: https://kiro.dev/downloads/",
        "capabilities": ["headless", "goal", "subagent", "review_loop"],
    },
    "claude": {
        "invoke": ["claude", "-p"],
        "probe": ["claude", "--version"],
        "local": False,
        "install": "Install and authenticate the Claude CLI.",
        "capabilities": ["headless", "goal"],
    },
    "codex": {
        "invoke": ["codex", "exec"],
        "probe": ["codex", "--version"],
        "local": False,
        "install": "Install and authenticate the Codex CLI.",
        "capabilities": ["headless", "goal"],
    },
    "gemini": {
        "invoke": ["gemini", "-p"],
        "probe": ["gemini", "--version"],
        "local": False,
        "install": "Install and authenticate the Gemini CLI.",
        "capabilities": ["headless"],
    },
    "llm": {
        "invoke": ["llm"],
        "probe": ["llm", "--version"],
        "local": False,
        "install": "Install llm and configure a model/provider.",
        "capabilities": ["headless"],
    },
    "ollama": {
        "invoke": ["ollama", "run"],
        "probe": ["ollama", "--version"],
        "local": True,
        "install": "Install Ollama and pull a local model.",
        "capabilities": ["headless"],
    },
}


class LooperError(RuntimeError):
    pass


def load_yaml(path: Path) -> dict[str, Any]:
    try:
        import yaml  # type: ignore
    except ImportError as exc:
        raise LooperError(
            "PyYAML is required to compile loop.yaml. Install with: python -m pip install PyYAML"
        ) from exc

    try:
        with path.open("r", encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
    except OSError as exc:
        raise LooperError(f"Could not read {path}: {exc}") from exc
    except yaml.YAMLError as exc:
        raise LooperError(f"Could not parse YAML in {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise LooperError(f"{path} must contain a YAML mapping at the top level")
    return data


def load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict):
        raise LooperError(f"{path} must contain a JSON object")
    return data


def write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(to_jsonable(data), indent=2, sort_keys=True) + "\n", encoding="utf-8")


def to_jsonable(value: Any) -> Any:
    if isinstance(value, dict):
        return {str(key): to_jsonable(item) for key, item in value.items()}
    if isinstance(value, list):
        return [to_jsonable(item) for item in value]
    if isinstance(value, (_dt.date, _dt.datetime)):
        return value.isoformat()
    return value


def read_registry(path: Path = REGISTRY_PATH) -> dict[str, Any]:
    if not path.exists():
        return {}
    with path.open("r", encoding="utf-8") as fh:
        data = json.load(fh)
    if not isinstance(data, dict):
        raise LooperError(f"Registry {path} must contain a JSON object")
    return data


def write_registry(data: dict[str, Any], path: Path = REGISTRY_PATH) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    write_json(path, data)


def run_probe(argv: list[str], timeout_sec: int = 5) -> tuple[bool, str]:
    probe_argv = list(argv)
    if os.name == "nt":
        resolved = shutil.which(argv[0])
        if resolved and Path(resolved).suffix.lower() in {".cmd", ".bat"}:
            probe_argv = ["cmd", "/d", "/c", *argv]
    try:
        completed = subprocess.run(
            probe_argv,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            timeout=timeout_sec,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        return False, str(exc)

    output = (completed.stdout or completed.stderr or "").strip()
    return completed.returncode == 0, output.splitlines()[0] if output else ""


def detect_models() -> dict[str, Any]:
    registry: dict[str, Any] = {}
    for model_id, meta in MODEL_PROBES.items():
        cli = meta["invoke"][0]
        path = shutil.which(cli)
        available = path is not None
        authed = False
        version = ""
        if available:
            authed, version = run_probe(meta["probe"])
        registry[model_id] = {
            "cli": cli,
            "path": path,
            "invoke": meta["invoke"],
            "available": available,
            "authed": authed,
            "local": meta["local"],
            "probe": meta["probe"],
            "version": version,
            "install": meta["install"],
            "capabilities": meta.get("capabilities", []),
        }
    return registry


def normalize_argv(value: Any, field: str) -> list[str]:
    if isinstance(value, list) and all(isinstance(item, str) for item in value):
        return value
    if isinstance(value, str):
        return shlex.split(value, posix=os.name != "nt")
    raise LooperError(f"{field} must be an argv array or string")


def criteria_by_id(spec: dict[str, Any]) -> dict[str, dict[str, Any]]:
    criteria = spec.get("goal", {}).get("verification", [])
    if not isinstance(criteria, list):
        raise LooperError("goal.verification must be a list")
    result: dict[str, dict[str, Any]] = {}
    for item in criteria:
        if not isinstance(item, dict):
            raise LooperError("Each verification criterion must be an object")
        cid = item.get("id")
        ctype = item.get("type")
        if not isinstance(cid, str) or not cid:
            raise LooperError("Each verification criterion needs a non-empty id")
        if cid in result:
            raise LooperError(f"Duplicate verification criterion id: {cid}")
        if ctype not in {"programmatic", "judge", "human"}:
            raise LooperError(f"Criterion {cid} has invalid type: {ctype}")
        if ctype == "programmatic":
            item["check"] = normalize_argv(item.get("check"), f"criterion {cid}.check")
            if item.get("expect") not in {"exit_zero", "exit_nonzero", "stdout_contains"}:
                raise LooperError(
                    f"Criterion {cid}.expect must be exit_zero, exit_nonzero, or stdout_contains"
                )
            if item.get("expect") == "stdout_contains" and not isinstance(item.get("contains"), str):
                raise LooperError(f"Criterion {cid} with stdout_contains needs contains")
        elif ctype == "judge" and not isinstance(item.get("rubric"), str):
            raise LooperError(f"Criterion {cid} needs a judge rubric")
        elif ctype == "human" and not isinstance(item.get("prompt"), str):
            raise LooperError(f"Criterion {cid} needs a human prompt")
        result[cid] = item
    return result


def validate_member(member: dict[str, Any]) -> None:
    mid = member.get("id")
    role = member.get("role")
    if not isinstance(mid, str) or not mid:
        raise LooperError("Each council member needs a non-empty id")
    if role not in {"reviewer", "judge"}:
        raise LooperError(f"Council member {mid} role must be reviewer or judge")
    member["invoke"] = normalize_argv(member.get("invoke"), f"council.{mid}.invoke")
    timeout = member.get("timeout_sec", 600)
    if not isinstance(timeout, int) or timeout <= 0:
        raise LooperError(f"Council member {mid}.timeout_sec must be a positive integer")
    member.setdefault("scope", ["plan", "delivery"])
    member.setdefault("local", member.get("cli") == "ollama")


def validate_gate(
    name: str,
    gate: dict[str, Any],
    criteria: dict[str, dict[str, Any]],
    members: dict[str, dict[str, Any]],
) -> None:
    if not isinstance(gate, dict):
        raise LooperError(f"{name} must be an object")
    policy = gate.get("verdict_policy")
    if policy not in {"revise_until_clean", "fixed_passes"}:
        raise LooperError(f"{name}.verdict_policy must be revise_until_clean or fixed_passes")
    max_revisions = gate.get("max_revisions", 1)
    if not isinstance(max_revisions, int) or max_revisions < 0:
        raise LooperError(f"{name}.max_revisions must be a non-negative integer")
    for cid in gate.get("criteria", []):
        if cid not in criteria:
            raise LooperError(f"{name} references unknown criterion: {cid}")
    for mid in gate.get("members", []):
        if mid not in members:
            raise LooperError(f"{name} references unknown council member: {mid}")
    if policy == "revise_until_clean":
        source = gate.get("verdict_source")
        if source == "human":
            return
        if source not in members:
            raise LooperError(f"{name}.verdict_source must be a judge member or human")
        if members[source].get("role") != "judge":
            raise LooperError(f"{name}.verdict_source must name a judge, not a reviewer")


def normalize_spec(spec: dict[str, Any], source_path: Path) -> dict[str, Any]:
    if spec.get("version") != 1:
        raise LooperError("Only loop.yaml version: 1 is supported")

    goal = spec.get("goal")
    if not isinstance(goal, dict):
        raise LooperError("goal must be an object")
    if not isinstance(goal.get("statement"), str) or not goal["statement"].strip():
        raise LooperError("goal.statement is required")
    if not isinstance(goal.get("definition_of_done"), str) or not goal["definition_of_done"].strip():
        raise LooperError("goal.definition_of_done is required")

    for index, source in enumerate(goal.get("context_sources", [])):
        if not isinstance(source, dict):
            raise LooperError("goal.context_sources entries must be objects")
        if "cmd" in source:
            source["cmd"] = normalize_argv(source["cmd"], f"context_sources[{index}].cmd")

    criteria = criteria_by_id(spec)

    host = spec.get("host")
    if not isinstance(host, dict):
        raise LooperError("host must be an object")
    host["invoke"] = normalize_argv(host.get("invoke"), "host.invoke")
    host.setdefault("timeout_sec", 600)
    if not isinstance(host["timeout_sec"], int) or host["timeout_sec"] <= 0:
        raise LooperError("host.timeout_sec must be a positive integer")

    council_list = spec.get("council", [])
    if not isinstance(council_list, list):
        raise LooperError("council must be a list")
    for member in council_list:
        if not isinstance(member, dict):
            raise LooperError("council entries must be objects")
        validate_member(member)
    members = {member["id"]: member for member in council_list}

    gates = spec.get("gates")
    if not isinstance(gates, dict):
        raise LooperError("gates must be an object")
    for gate_name in ("plan_gate", "delivery_gate"):
        validate_gate(gate_name, gates.get(gate_name), criteria, members)

    control = spec.get("loop_control")
    if not isinstance(control, dict):
        raise LooperError("loop_control must be an object")
    max_iterations = control.get("max_iterations")
    if not isinstance(max_iterations, int) or max_iterations <= 0:
        raise LooperError("loop_control.max_iterations must be a positive integer")
    budget = control.setdefault("budget", {})
    if not isinstance(budget, dict):
        raise LooperError("loop_control.budget must be an object")
    if "wall_clock_min" not in budget:
        budget["wall_clock_min"] = 30
    no_progress = control.setdefault(
        "no_progress",
        {
            "max_stalled_iterations": 2,
            "signals": [
                "same blocking issue repeats",
                "delivery artifact has no material change",
                "verifier output is unchanged",
            ],
            "action": "stop",
        },
    )
    if not isinstance(no_progress, dict):
        raise LooperError("loop_control.no_progress must be an object")
    stalled = no_progress.setdefault("max_stalled_iterations", 2)
    if not isinstance(stalled, int) or stalled <= 0:
        raise LooperError("loop_control.no_progress.max_stalled_iterations must be a positive integer")
    signals = no_progress.setdefault("signals", ["same blocking issue repeats"])
    if not isinstance(signals, list) or not all(isinstance(item, str) for item in signals):
        raise LooperError("loop_control.no_progress.signals must be a list of strings")
    action = no_progress.setdefault("action", "stop")
    if action not in {"stop", "human_checkpoint"}:
        raise LooperError("loop_control.no_progress.action must be stop or human_checkpoint")

    execution = spec.setdefault(
        "execution",
        {
            "mode": "in_session",
            "isolation": "current_workspace",
            "side_effects": {"requires_approval": True, "duplicate_action_check": True},
        },
    )
    if not isinstance(execution, dict):
        raise LooperError("execution must be an object")
    execution.setdefault("mode", "in_session")
    execution.setdefault("isolation", "current_workspace")
    if execution["mode"] not in {"in_session", "external_runner", "orchestrated"}:
        raise LooperError("execution.mode must be in_session, external_runner, or orchestrated")
    if execution["isolation"] not in {"current_workspace", "branch", "worktree", "sandbox"}:
        raise LooperError("execution.isolation must be current_workspace, branch, worktree, or sandbox")
    side_effects = execution.setdefault("side_effects", {})
    if not isinstance(side_effects, dict):
        raise LooperError("execution.side_effects must be an object")
    side_effects.setdefault("requires_approval", True)
    side_effects.setdefault("duplicate_action_check", True)

    observability = spec.setdefault(
        "observability",
        {"state_file": "state.json", "run_log": "run-log.md", "checkpoint_granularity": "gate"},
    )
    if not isinstance(observability, dict):
        raise LooperError("observability must be an object")
    observability.setdefault("state_file", "state.json")
    observability.setdefault("run_log", "run-log.md")
    observability.setdefault("checkpoint_granularity", "gate")
    if not isinstance(observability["state_file"], str) or not observability["state_file"]:
        raise LooperError("observability.state_file must be a non-empty string")
    if not isinstance(observability["run_log"], str) or not observability["run_log"]:
        raise LooperError("observability.run_log must be a non-empty string")
    if observability["checkpoint_granularity"] not in {"gate", "step"}:
        raise LooperError("observability.checkpoint_granularity must be gate or step")

    workspace = spec.setdefault("workspace", {})
    if not isinstance(workspace, dict):
        raise LooperError("workspace must be an object")
    workspace.setdefault("dir", "./loop-workspace")
    layout = workspace.setdefault("layout", ["plan.md", "delivery-{n}.md", "review-{n}.md", "state.json", "run-log.md"])
    if not isinstance(layout, list) or not all(isinstance(item, str) for item in layout):
        raise LooperError("workspace.layout must be a list of strings")
    for required_file in (observability["state_file"], observability["run_log"]):
        if required_file not in layout:
            layout.append(required_file)

    privacy = spec.setdefault("privacy", {})
    if not isinstance(privacy, dict):
        raise LooperError("privacy must be an object")
    egress = privacy.setdefault("egress", [])
    if not isinstance(egress, list):
        raise LooperError("privacy.egress must be a list")
    for entry in egress:
        if not isinstance(entry, dict):
            raise LooperError("privacy.egress entries must be objects")
        entry.setdefault("redact", DEFAULT_REDACTIONS)
        entry.setdefault("consent", "required")

    resolved = {
        "$schema": "https://github.com/ksimback/looper/schema/loop.resolved.v1.json",
        "compiled_at": _dt.datetime.now(_dt.UTC).replace(microsecond=0).isoformat(),
        "source": str(source_path),
        **spec,
        "criteria_by_id": criteria,
        "council_by_id": members,
    }
    return to_jsonable(resolved)


def clip(text: Any, width: int) -> str:
    value = str(text or "")
    return value if len(value) <= width else value[: width - 1] + "~"


def ascii_box(*rows: str, width: int = 30) -> list[str]:
    border = "+" + "-" * (width + 2) + "+"
    body = [f"| {clip(row, width):<{width}} |" for row in rows if row is not None]
    return [border, *body, border]


def render_ascii_diagram(resolved: dict[str, Any]) -> str:
    gates = resolved.get("gates", {})
    control = resolved.get("loop_control", {})
    observability = resolved.get("observability", {})
    plan_gate = gates.get("plan_gate", {})
    delivery_gate = gates.get("delivery_gate", {})
    plan_revisions = plan_gate.get("max_revisions", 0)
    delivery_revisions = delivery_gate.get("max_revisions", 0)
    plan_source = plan_gate.get("verdict_source", "human")
    delivery_source = delivery_gate.get("verdict_source", "human")
    no_progress = control.get("no_progress", {})
    stalled = no_progress.get("max_stalled_iterations", 2)
    budget = control.get("budget", {})
    budget_bits = []
    if budget.get("wall_clock_min") is not None:
        budget_bits.append(f"{budget.get('wall_clock_min')}m")
    if budget.get("usd") is not None:
        budget_bits.append(f"${budget.get('usd')}")
    if budget.get("tokens") is not None:
        budget_bits.append(f"{budget.get('tokens')} tokens")
    budget_text = ", ".join(budget_bits) or "configured caps"

    lines: list[str] = []
    lines.extend(ascii_box("1. Goal + context", "read sources"))
    lines.extend(["               |", "               v"])
    lines.extend(ascii_box("2. Draft plan.md", f"state -> {observability.get('state_file', 'state.json')}"))
    lines.extend(["               |", "               v"])
    lines.extend(ascii_box("3. Plan gate", f"verdict: {plan_source}"))
    lines.extend([f"               | needs work -> revise <= {plan_revisions} -> step 2", "               | pass", "               v"])
    lines.extend(ascii_box("4. Write delivery-N.md", f"log -> {observability.get('run_log', 'run-log.md')}"))
    lines.extend(["               |", "               v"])
    lines.extend(ascii_box("5. Delivery gate", f"verdict: {delivery_source}"))
    lines.extend([f"               | needs work -> revise <= {delivery_revisions} -> step 4", "               | pass", "               v"])
    lines.extend(ascii_box("6. Final output", "all gates clean"))
    lines.extend(
        [
            "",
            f"Stops: pass gates | max {control.get('max_iterations')} iterations | "
            f"no progress x{stalled} | budget {budget_text}",
        ]
    )
    return "\n".join(lines)


def render_loop(resolved: dict[str, Any]) -> str:
    meta = resolved.get("meta", {})
    goal = resolved.get("goal", {})
    gates = resolved.get("gates", {})
    control = resolved.get("loop_control", {})
    execution = resolved.get("execution", {})
    observability = resolved.get("observability", {})
    title = meta.get("name") or "Looper Generated Loop"
    criteria = goal.get("verification", [])
    council = resolved.get("council", [])

    lines = [
        f"# {title}",
        "",
        meta.get("description", "").strip(),
        "",
        "## Goal",
        "",
        goal.get("statement", "").strip(),
        "",
        "## Definition of Done",
        "",
        goal.get("definition_of_done", "").strip(),
        "",
        "## Verification",
        "",
    ]
    for item in criteria:
        lines.append(f"- `{item['id']}` ({item['type']})")
    lines.extend(["", "## Council", ""])
    if council:
        for member in council:
            lines.append(
                f"- `{member['id']}`: {member.get('role')} via {member.get('cli')} "
                f"({member.get('model', 'default')})"
            )
    else:
        lines.append("- No council members configured.")
    lines.extend(
        [
            "",
            "## Gates",
            "",
            f"- Plan gate: {gates.get('plan_gate', {}).get('verdict_policy')}",
            f"- Delivery gate: {gates.get('delivery_gate', {}).get('verdict_policy')}",
            "",
            "## Loop Control",
            "",
            f"- Max iterations: {control.get('max_iterations')}",
            f"- Budget: `{json.dumps(control.get('budget', {}), sort_keys=True)}`",
            f"- No-progress: `{json.dumps(control.get('no_progress', {}), sort_keys=True)}`",
            "",
            "## Execution Boundary",
            "",
            f"- Mode: `{execution.get('mode', 'in_session')}`",
            f"- Isolation: `{execution.get('isolation', 'current_workspace')}`",
            f"- Side effects: `{json.dumps(execution.get('side_effects', {}), sort_keys=True)}`",
            "",
            "## Observability",
            "",
            f"- State file: `{observability.get('state_file', 'state.json')}`",
            f"- Run log: `{observability.get('run_log', 'run-log.md')}`",
            f"- Checkpoint granularity: `{observability.get('checkpoint_granularity', 'gate')}`",
            "",
            "## Flow Preview",
            "",
            "```text",
            render_ascii_diagram(resolved),
            "```",
            "",
        ]
    )
    return "\n".join(line for line in lines if line is not None)


def render_session_prompt(resolved: dict[str, Any]) -> str:
    meta = resolved.get("meta", {})
    goal = resolved.get("goal", {})
    gates = resolved.get("gates", {})
    control = resolved.get("loop_control", {})
    workspace = resolved.get("workspace", {})
    execution = resolved.get("execution", {})
    observability = resolved.get("observability", {})
    criteria = goal.get("verification", [])
    council = resolved.get("council", [])
    title = meta.get("name") or "Looper Generated Loop"

    lines = [
        f"# Run `{title}` In This Session",
        "",
        "Use this prompt when the user wants to run the Looper-designed loop in the current LLM session.",
        "This is the default/easy execution path. The Python runner is the advanced path for running later or outside the session.",
        "",
        "## Operator Instructions",
        "",
        "You are executing a Looper-designed loop in this current session.",
        "Follow the resolved spec below, write handoff files into the workspace, and enforce the caps manually.",
        "Do not use `run-loop.py` unless the user explicitly asks for the advanced external runner.",
        "",
        "1. Create the workspace directory if it does not exist.",
        "2. Read the context sources before drafting the plan.",
        "3. Draft `plan.md` in the workspace.",
        "4. Run the plan gate. Apply programmatic checks when available. For judge criteria, use the configured judge only after consent for any non-local egress; otherwise ask the user to approve a human/current-session substitute.",
        "5. Revise until the gate passes or `max_revisions` is reached.",
        "6. Produce `delivery-N.md` in the workspace.",
        "7. Run the delivery gate after each delivery.",
        "8. Stop when all delivery criteria pass, a cap is reached, or the user stops the loop.",
        "9. Keep `state.json` current with status, iteration, last gate, consent, and blockers.",
        "10. Append a compact entry to `run-log.md` after every context read, model call, check, gate verdict, revision, blocker, and stop decision.",
        "11. Compare each blocker against the previous blocker. If the same blocker repeats for the configured no-progress window, stop or ask for the configured human checkpoint instead of revising again.",
        "12. Treat token and USD budgets as operator limits in this session: if exact accounting is unavailable, stop and ask before continuing when the loop appears likely to exceed them.",
        "",
        "## Files",
        "",
        f"- Source spec: `{Path(resolved.get('source', 'loop.yaml')).name}`",
        "- Human summary: `LOOP.md`",
        "- Resolved spec: `loop.resolved.json`",
        f"- Workspace: `{workspace.get('dir', './loop-workspace')}`",
        f"- State file: `{observability.get('state_file', 'state.json')}`",
        f"- Run log: `{observability.get('run_log', 'run-log.md')}`",
        "",
        "## Goal",
        "",
        goal.get("statement", "").strip(),
        "",
        "## Definition Of Done",
        "",
        goal.get("definition_of_done", "").strip(),
        "",
        "## Context Sources",
        "",
    ]

    context_sources = goal.get("context_sources", [])
    if context_sources:
        for source in context_sources:
            if "file" in source:
                lines.append(f"- Read file `{source['file']}`")
            elif "cmd" in source:
                lines.append(f"- Run command `{json.dumps(source['cmd'])}`")
    else:
        lines.append("- No context sources configured.")

    lines.extend(["", "## Verification Criteria", ""])
    for item in criteria:
        if item["type"] == "programmatic":
            lines.append(
                f"- `{item['id']}` programmatic: run `{json.dumps(item['check'])}` and expect `{item['expect']}`"
            )
        elif item["type"] == "judge":
            lines.append(f"- `{item['id']}` judge rubric: {item['rubric']}")
        elif item["type"] == "human":
            lines.append(f"- `{item['id']}` human signoff: {item['prompt']}")

    lines.extend(["", "## Council", ""])
    if council:
        for member in council:
            locality = "local" if member.get("local") else "non-local"
            lines.append(
                f"- `{member['id']}` {member.get('role')} via `{json.dumps(member.get('invoke', []))}` "
                f"({locality}; timeout {member.get('timeout_sec', 600)}s)"
            )
    else:
        lines.append("- No council members configured.")

    lines.extend(["", "## Gates", ""])
    for gate_name in ("plan_gate", "delivery_gate"):
        gate = gates.get(gate_name, {})
        lines.extend(
            [
                f"### {gate_name}",
                "",
                f"- When: `{gate.get('when')}`",
                f"- Policy: `{gate.get('verdict_policy')}`",
                f"- Verdict source: `{gate.get('verdict_source', 'none')}`",
                f"- Criteria: `{', '.join(gate.get('criteria', []))}`",
                f"- Max revisions: `{gate.get('max_revisions')}`",
                "",
            ]
        )

    lines.extend(
        [
            "## Loop Control",
            "",
            f"- Max iterations: `{control.get('max_iterations')}`",
            f"- Budget: `{json.dumps(control.get('budget', {}), sort_keys=True)}`",
            f"- No-progress: `{json.dumps(control.get('no_progress', {}), sort_keys=True)}`",
            f"- Human checkpoints: `{', '.join(control.get('human_checkpoints', [])) or 'none'}`",
            "- Stop conditions:",
        ]
    )
    for condition in control.get("stop_conditions", []):
        lines.append(f"  - {condition}")

    lines.extend(
        [
            "",
            "## Execution Boundary",
            "",
            f"- Mode: `{execution.get('mode', 'in_session')}`",
            f"- Isolation: `{execution.get('isolation', 'current_workspace')}`",
            f"- Side effects: `{json.dumps(execution.get('side_effects', {}), sort_keys=True)}`",
            "",
            "If the loop needs scheduled runs, child-agent lifecycle management, concurrency control, or restart-safe step retries, stop and tell the user this Looper spec should be handed to a durable orchestrator.",
            "",
            "## Observability",
            "",
            f"- State file: `{observability.get('state_file', 'state.json')}`",
            f"- Run log: `{observability.get('run_log', 'run-log.md')}`",
            f"- Checkpoint granularity: `{observability.get('checkpoint_granularity', 'gate')}`",
            "",
            "Use `state.json` for the latest resumable status and `run-log.md` for the append-only history of what happened.",
        ]
    )

    lines.extend(["", "## Privacy", ""])
    egress = resolved.get("privacy", {}).get("egress", [])
    if egress:
        for entry in egress:
            lines.append(
                f"- Before sending `{', '.join(entry.get('sends', []))}` to `{entry.get('to')}`, "
                f"confirm consent and apply redactions `{', '.join(entry.get('redact', []))}`."
            )
    else:
        lines.append("- No cross-vendor egress configured.")

    lines.extend(
        [
            "",
            "## Start Now",
            "",
            "If the user asked to run now, begin at step 1 under Operator Instructions and keep going until a stop condition is reached.",
            "",
        ]
    )
    return "\n".join(lines)


def cmd_detect(args: argparse.Namespace) -> int:
    registry = detect_models()
    if args.write:
        existing = read_registry(args.registry)
        existing.update(registry)
        write_registry(existing, args.registry)
    print(json.dumps(registry, indent=2, sort_keys=True))
    return 0


def cmd_register(args: argparse.Namespace) -> int:
    if not args.invoke:
        raise LooperError("--invoke needs at least one command token")
    registry = read_registry(args.registry)
    registry[args.model_id] = {
        "cli": args.invoke[0],
        "invoke": args.invoke,
        "available": shutil.which(args.invoke[0]) is not None,
        "authed": args.authed,
        "local": args.local,
        "model": args.model,
        "notes": args.notes or "",
    }
    write_registry(registry, args.registry)
    print(f"Registered {args.model_id} in {args.registry}")
    return 0


def cmd_compile(args: argparse.Namespace) -> int:
    source = args.loop_yaml.resolve()
    spec = load_yaml(source)
    resolved = normalize_spec(spec, source)
    out = args.out or source.with_name("loop.resolved.json")
    write_json(out, resolved)
    if args.render:
        args.render.parent.mkdir(parents=True, exist_ok=True)
        args.render.write_text(render_loop(resolved), encoding="utf-8")
    if args.session_prompt:
        args.session_prompt.parent.mkdir(parents=True, exist_ok=True)
        args.session_prompt.write_text(render_session_prompt(resolved), encoding="utf-8")
    print(f"Wrote {out}")
    if args.render:
        print(f"Wrote {args.render}")
    if args.session_prompt:
        print(f"Wrote {args.session_prompt}")
    return 0


def cmd_session_prompt(args: argparse.Namespace) -> int:
    resolved = load_json(args.resolved_json)
    prompt = render_session_prompt(resolved)
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(prompt, encoding="utf-8")
        print(f"Wrote {args.out}")
    else:
        print(prompt)
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="looper", description="Looper scaffolding helpers")
    sub = parser.add_subparsers(dest="command", required=True)

    detect = sub.add_parser("detect-models", help="Detect model CLIs and print registry JSON")
    detect.add_argument("--write", action="store_true", help="Merge results into the model registry")
    detect.add_argument("--registry", type=Path, default=REGISTRY_PATH)
    detect.set_defaults(func=cmd_detect)

    register = sub.add_parser("register-model", help="Register custom model CLI invocation metadata")
    register.add_argument("model_id")
    register.add_argument("--invoke", nargs="+", required=True)
    register.add_argument("--model", default="")
    register.add_argument("--local", action="store_true")
    register.add_argument("--authed", action="store_true")
    register.add_argument("--notes", default="")
    register.add_argument("--registry", type=Path, default=REGISTRY_PATH)
    register.set_defaults(func=cmd_register)

    compile_cmd = sub.add_parser("compile", help="Compile loop.yaml to loop.resolved.json")
    compile_cmd.add_argument("loop_yaml", type=Path)
    compile_cmd.add_argument("--out", type=Path)
    compile_cmd.add_argument("--render", type=Path)
    compile_cmd.add_argument("--session-prompt", type=Path)
    compile_cmd.set_defaults(func=cmd_compile)

    session_prompt = sub.add_parser(
        "session-prompt", help="Render the in-session execution prompt from loop.resolved.json"
    )
    session_prompt.add_argument("resolved_json", type=Path)
    session_prompt.add_argument("--out", type=Path)
    session_prompt.set_defaults(func=cmd_session_prompt)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        return int(args.func(args))
    except LooperError as exc:
        print(f"looper: error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Generated Looper runner.

This file executes a resolved loop spec. It intentionally reads only
loop.resolved.json and uses only Python stdlib.
"""

from __future__ import annotations

import argparse
import datetime as _dt
import fnmatch
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import time
from typing import Any


PASS = "pass"
REVISE = "revise"

DEFAULT_REDACTIONS = [".env", ".env.*", "secrets/**", "**/*.key"]

SKIP_DIRS = {".git", "__pycache__", "node_modules", ".venv"}


class RunnerError(RuntimeError):
    pass


def utc_now() -> str:
    return _dt.datetime.now(_dt.UTC).replace(microsecond=0).isoformat()


def load_json(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as fh:
            data = json.load(fh)
    except OSError as exc:
        raise RunnerError(f"Could not read {path}: {exc}") from exc
    except json.JSONDecodeError as exc:
        raise RunnerError(f"Could not parse JSON in {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise RunnerError(f"{path} must contain a JSON object")
    return data


def write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text.rstrip() + "\n", encoding="utf-8")


def write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")


def ensure_argv(value: Any, field: str) -> list[str]:
    if isinstance(value, list) and value and all(isinstance(item, str) for item in value):
        return value
    raise RunnerError(f"{field} must be a non-empty argv array")


def relative_to_base(path_text: str, base_dir: Path) -> Path:
    path = Path(path_text)
    resolved = path if path.is_absolute() else base_dir / path
    try:
        resolved.resolve().relative_to(base_dir.resolve())
    except ValueError:
        raise RunnerError(
            f"Path {path_text!r} escapes the loop directory {base_dir}; "
            "workspace and context paths must stay inside the loop directory"
        ) from None
    return resolved


def is_redacted(path: Path, base_dir: Path, globs: list[str]) -> bool:
    try:
        rel = path.resolve().relative_to(base_dir.resolve()).as_posix()
    except ValueError:
        rel = path.name
    # Match each pattern against the relative path and every path suffix so
    # bare patterns like ".env" also cover nested files like "config/.env".
    parts = rel.split("/")
    suffixes = {"/".join(parts[i:]) for i in range(len(parts))}
    for pattern in globs:
        normalized = pattern[3:] if pattern.startswith("**/") else pattern
        if any(fnmatch.fnmatch(candidate, normalized) for candidate in suffixes):
            return True
    return False


def run_argv(
    argv: list[str],
    *,
    cwd: Path,
    timeout_sec: int,
    stdin: str = "",
) -> subprocess.CompletedProcess[str]:
    try:
        return subprocess.run(
            argv,
            input=stdin,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            encoding="utf-8",
            errors="replace",
            cwd=str(cwd),
            timeout=timeout_sec,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:
        completed = subprocess.CompletedProcess(argv, 124, exc.stdout or "", exc.stderr or "")
        return completed
    except OSError as exc:
        return subprocess.CompletedProcess(argv, 127, "", str(exc))


def call_model(member: dict[str, Any], prompt: str, base_dir: Path) -> str:
    argv = ensure_argv(member.get("invoke"), f"{member.get('id', member.get('cli', 'model'))}.invoke")
    timeout_sec = int(member.get("timeout_sec", 600))
    result = run_argv(argv, cwd=base_dir, timeout_sec=timeout_sec, stdin=prompt)
    if result.returncode != 0:
        raise RunnerError(
            f"Model invocation failed ({' '.join(argv)}): exit {result.returncode}\n{result.stderr}"
        )
    return result.stdout.strip()


def parse_judge_output(text: str) -> dict[str, Any]:
    fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
    candidate = fenced.group(1) if fenced else text.strip()
    try:
        parsed = json.loads(candidate)
    except json.JSONDecodeError:
        return {
            "verdict": REVISE,
            "blocking_issues": ["Judge output was not parseable JSON."],
            "confidence": 0.0,
            "notes": text.strip(),
            "warning": "unparseable_judge_output",
        }
    if not isinstance(parsed, dict):
        return {
            "verdict": REVISE,
            "blocking_issues": ["Judge output was not a JSON object."],
            "confidence": 0.0,
            "notes": text.strip(),
            "warning": "invalid_judge_output",
        }
    verdict = parsed.get("verdict")
    if verdict not in {PASS, REVISE}:
        parsed["verdict"] = REVISE
        parsed.setdefault("blocking_issues", []).append("Judge verdict was not pass or revise.")
    parsed.setdefault("blocking_issues", [])
    parsed.setdefault("confidence", 0.0)
    parsed.setdefault("notes", "")
    return parsed


class Runner:
    def __init__(self, spec_path: Path) -> None:
        self.spec_path = spec_path.resolve()
        self.base_dir = self.spec_path.parent
        self.spec = load_json(self.spec_path)
        self.workspace = relative_to_base(self.spec["workspace"]["dir"], self.base_dir)
        self.workspace.mkdir(parents=True, exist_ok=True)
        self.observability = self.spec.get("observability", {})
        self.run_log_path = self.workspace / self.observability.get("run_log", "run-log.md")
        self.state_path = self.workspace / self.observability.get("state_file", "state.json")
        self.state = self.load_state()
        self.started = time.monotonic()
        # Per-run caches for the scrub layer: flagged-file contents are read
        # once per glob set, surfaced events are logged once per destination.
        self._flagged_cache: dict[tuple[str, ...], list[tuple[str, str]]] = {}
        self._unscrubbable_reported: set[str] = set()
        self._surfaced_events: set[tuple[str, tuple[str, ...]]] = set()

    def load_state(self) -> dict[str, Any]:
        if self.state_path.exists():
            return load_json(self.state_path)
        return {
            "status": "initialized",
            "started_at": utc_now(),
            "iteration": 0,
            "warnings": [],
            "consent": {},
        }

    def save_state(self, **updates: Any) -> None:
        self.state.update(updates)
        self.state["updated_at"] = utc_now()
        write_json(self.state_path, self.state)

    def append_log(self, event: str, **fields: Any) -> None:
        self.run_log_path.parent.mkdir(parents=True, exist_ok=True)
        payload = f" {json.dumps(fields, sort_keys=True)}" if fields else ""
        with self.run_log_path.open("a", encoding="utf-8") as fh:
            fh.write(f"- {utc_now()} `{event}`{payload}\n")

    def enforce_wall_clock(self) -> None:
        budget = self.spec.get("loop_control", {}).get("budget", {})
        wall_clock_min = budget.get("wall_clock_min")
        if wall_clock_min is None:
            return
        if time.monotonic() - self.started > float(wall_clock_min) * 60:
            self.save_state(status="failed", failure="wall_clock_budget_exceeded")
            self.append_log("stop", reason="wall_clock_budget_exceeded")
            raise RunnerError("Wall-clock budget exceeded")

    def no_progress_reached(self, gate_name: str, failures: list[str]) -> bool:
        if not failures:
            self.save_state(no_progress={"count": 0, "signature": "", "gate": gate_name})
            return False
        config = self.spec.get("loop_control", {}).get("no_progress", {})
        threshold = int(config.get("max_stalled_iterations", 2))
        signature = "\n".join(sorted(failures))
        previous = self.state.get("no_progress", {})
        same_gate = previous.get("gate") == gate_name
        same_signature = previous.get("signature") == signature
        count = int(previous.get("count", 0)) + 1 if same_gate and same_signature else 1
        progress = {
            "gate": gate_name,
            "signature": signature,
            "count": count,
            "threshold": threshold,
            "updated_at": utc_now(),
        }
        self.save_state(no_progress=progress)
        if count < threshold:
            return False
        self.append_log("no_progress_detected", gate=gate_name, count=count, failures=failures)
        if config.get("action", "stop") == "human_checkpoint":
            answer = input("No-progress detected. Type 'continue' to allow one more revision: ").strip().lower()
            if answer == "continue":
                progress["count"] = 0
                self.save_state(no_progress=progress)
                self.append_log("no_progress_override", gate=gate_name)
                return False
        self.save_state(status="failed", failure="no_progress_detected", blocking_issues=failures)
        return True

    def criteria(self, ids: list[str]) -> list[dict[str, Any]]:
        by_id = self.spec.get("criteria_by_id", {})
        return [by_id[item] for item in ids]

    def member(self, member_id: str) -> dict[str, Any]:
        return self.spec["council_by_id"][member_id]

    def redactions_for(self, member_id: str) -> list[str]:
        # Defaults always apply; configured egress redactions extend them.
        redactions = list(DEFAULT_REDACTIONS)
        for entry in self.spec.get("privacy", {}).get("egress", []):
            if entry.get("to") == member_id:
                for pattern in entry.get("redact", []):
                    if pattern not in redactions:
                        redactions.append(pattern)
        return redactions

    def all_redaction_globs(self) -> list[str]:
        globs = list(DEFAULT_REDACTIONS)
        for entry in self.spec.get("privacy", {}).get("egress", []):
            for pattern in entry.get("redact", []):
                if pattern not in globs:
                    globs.append(pattern)
        return globs

    def iter_redaction_files(self, globs: list[str]):
        for root, dirs, files in os.walk(self.base_dir):
            dirs[:] = [d for d in dirs if d not in SKIP_DIRS]
            for name in files:
                path = Path(root) / name
                if is_redacted(path, self.base_dir, globs):
                    yield path

    def flagged_file_contents(self, globs: list[str]) -> list[tuple[str, str]]:
        """Read flagged files once per glob set; surface the unreadable ones.

        A flagged file the scrub cannot read (too large, not UTF-8) cannot be
        detected if its content leaks, so that blind spot is reported instead
        of silently skipped.
        """
        key = tuple(sorted(globs))
        if key in self._flagged_cache:
            return self._flagged_cache[key]
        contents: list[tuple[str, str]] = []
        for path in self.iter_redaction_files(globs):
            rel = path.relative_to(self.base_dir).as_posix()
            reason = ""
            try:
                if path.stat().st_size > 1_000_000:
                    reason = "larger than 1MB"
                else:
                    secret_text = path.read_text(encoding="utf-8")
                    if secret_text.strip():
                        contents.append((rel, secret_text))
                    continue
            except UnicodeDecodeError:
                reason = "not valid UTF-8"
            except OSError as exc:
                reason = f"unreadable ({exc})"
            if reason and rel not in self._unscrubbable_reported:
                self._unscrubbable_reported.add(rel)
                self.append_log("redaction_unscrubbable", source=rel, reason=reason)
                self.add_state_warning(
                    f"flagged file {rel} is {reason}; its content cannot be "
                    "detected by the scrub layer if it leaks into artifacts"
                )
        self._flagged_cache[key] = contents
        return contents

    def scrub_flagged_content(self, text: str, globs: list[str]) -> tuple[str, list[str]]:
        """Remove content originating from redaction-glob files.

        The flagged files themselves are never read into prompts (path-based
        non-send in gather_context); this second layer catches their content
        when it re-surfaces elsewhere - a cmd context source that printed it,
        or an artifact a model copied it into. Returns the scrubbed text and
        the relative paths whose content was found. Best effort by design,
        erring toward over-redaction: reformatted content or lines shorter
        than 8 characters can survive, and a flagged-file line that
        legitimately appears elsewhere is masked too.
        """
        original = text
        hits: list[str] = []
        for rel, secret_text in self.flagged_file_contents(globs):
            marker = f"[redacted:{rel}]"
            # Detect against the original text so a secret shared by two
            # flagged files attributes both, not just the first replaced.
            lines = [line.strip() for line in secret_text.splitlines() if len(line.strip()) >= 8]
            if secret_text in original or any(line in original for line in lines):
                hits.append(rel)
            text = text.replace(secret_text, marker)
            for line in lines:
                text = text.replace(line, marker)
        return text, hits

    def add_state_warning(self, note: str) -> None:
        warnings = list(self.state.get("warnings", []) or [])
        if note not in warnings:
            warnings.append(note)
            self.save_state(warnings=warnings)

    def surface_redaction(self, where: str, hits: list[str]) -> None:
        if not hits:
            return
        event_key = (where, tuple(hits))
        if event_key not in self._surfaced_events:
            self._surfaced_events.add(event_key)
            self.append_log("redaction_applied", where=where, sources=hits)
        self.add_state_warning(
            f"flagged content from {', '.join(hits)} appeared in {where}; "
            "scrubbed before use"
        )

    def redact_prompt_for_member(self, member_id: str, prompt: str) -> str:
        scrubbed, hits = self.scrub_flagged_content(prompt, self.redactions_for(member_id))
        self.surface_redaction(f"prompt for {member_id}", hits)
        return scrubbed

    def ensure_consent(self, member_id: str) -> None:
        member = self.member(member_id)
        if member.get("local"):
            return
        if self.state.get("consent", {}).get(member_id):
            return
        matching = [
            entry
            for entry in self.spec.get("privacy", {}).get("egress", [])
            if entry.get("to") == member_id
        ]
        # Consent fails closed: a non-local member always needs consent unless
        # every egress entry for it explicitly pre-grants with consent: granted.
        if matching and all(entry.get("consent") == "granted" for entry in matching):
            return
        sends = sorted({item for entry in matching for item in entry.get("sends", [])})
        redactions = self.redactions_for(member_id)
        print()
        print(f"Looper is about to send {', '.join(sends) or 'loop artifacts'} to {member_id}.")
        print(f"CLI: {member.get('cli')} / model: {member.get('model', 'default')}")
        print(f"Redactions: {', '.join(redactions)}")
        for note in self.state.get("warnings", []) or []:
            if f"prompt for {member_id}" in note:
                print(f"Warning: {note}")
        if not matching:
            print("No privacy.egress entry covers this member; consent is required by default.")
        answer = input("Type 'yes' to consent to this first send: ").strip().lower()
        if answer != "yes":
            self.save_state(status="blocked", failure=f"consent_refused:{member_id}")
            raise RunnerError(f"Consent refused for {member_id}")
        consent = dict(self.state.get("consent", {}))
        consent[member_id] = {"granted_at": utc_now(), "sends": sends, "redact": redactions}
        self.save_state(consent=consent)

    def gather_context(self) -> str:
        goal = self.spec["goal"]
        redaction_globs = self.all_redaction_globs()
        chunks: list[str] = []
        for index, source in enumerate(goal.get("context_sources", []), start=1):
            self.enforce_wall_clock()
            if "file" in source:
                try:
                    path = relative_to_base(source["file"], self.base_dir)
                except RunnerError:
                    chunks.append(f"## Context source {index}: {source['file']}\n[blocked: outside loop directory]\n")
                    self.append_log("context", source=source["file"], status="blocked_outside_base")
                    continue
                if is_redacted(path, self.base_dir, redaction_globs):
                    chunks.append(f"## Context source {index}: {source['file']}\n[redacted]\n")
                    self.append_log("context", source=source["file"], status="redacted")
                elif path.exists():
                    chunks.append(f"## Context source {index}: {source['file']}\n{path.read_text(encoding='utf-8')}\n")
                    self.append_log("context", source=source["file"], status="read")
                else:
                    chunks.append(f"## Context source {index}: {source['file']}\n[missing]\n")
                    self.append_log("context", source=source["file"], status="missing")
            elif "cmd" in source:
                argv = ensure_argv(source["cmd"], f"context_sources[{index}].cmd")
                result = run_argv(argv, cwd=self.base_dir, timeout_sec=int(source.get("timeout_sec", 60)))
                # Command output can reproduce flagged-file content (cat, git
                # log, env dumps); scrub it before it enters any prompt.
                block = (
                    f"exit={result.returncode}\nstdout:\n{result.stdout}\n"
                    f"stderr:\n{result.stderr}\n"
                )
                block, hits = self.scrub_flagged_content(block, redaction_globs)
                self.surface_redaction(f"context command output ({' '.join(argv)})", hits)
                chunks.append(f"## Context source {index}: {' '.join(argv)}\n{block}")
                self.append_log("context_cmd", argv=argv, returncode=result.returncode)
        context = "\n".join(chunks).strip()
        write_text(self.workspace / "context.md", context or "No context sources configured.")
        return context

    def host_prompt(self, phase: str, artifact: str = "", review: str = "") -> str:
        goal = self.spec["goal"]
        if phase == "plan":
            return (
                "Draft plan.md for this loop.\n\n"
                f"Goal:\n{goal['statement']}\n\n"
                f"Definition of done:\n{goal['definition_of_done']}\n\n"
                f"Context:\n{(self.workspace / 'context.md').read_text(encoding='utf-8')}\n"
            )
        if phase == "delivery":
            return (
                "Write the next delivery artifact for this loop.\n\n"
                f"Goal:\n{goal['statement']}\n\n"
                f"Definition of done:\n{goal['definition_of_done']}\n\n"
                f"Plan:\n{(self.workspace / 'plan.md').read_text(encoding='utf-8')}\n"
            )
        if phase == "revise":
            return (
                "Revise the artifact to address the review. Return only the revised artifact.\n\n"
                f"Artifact:\n{artifact}\n\nReview:\n{review}\n"
            )
        raise RunnerError(f"Unknown host phase: {phase}")

    def run_host(self, phase: str, target: Path, artifact: str = "", review: str = "") -> None:
        self.enforce_wall_clock()
        self.append_log("host_start", phase=phase, target=target.name)
        output = call_model(self.spec["host"], self.host_prompt(phase, artifact, review), self.base_dir)
        write_text(target, output)
        self.append_log("host_done", phase=phase, target=target.name)

    def run_programmatic(self, criterion: dict[str, Any]) -> dict[str, Any]:
        argv = ensure_argv(criterion["check"], f"{criterion['id']}.check")
        result = run_argv(argv, cwd=self.base_dir, timeout_sec=int(criterion.get("timeout_sec", 300)))
        expect = criterion.get("expect")
        passed = False
        if expect == "exit_zero":
            passed = result.returncode == 0
        elif expect == "exit_nonzero":
            passed = result.returncode != 0
        elif expect == "stdout_contains":
            passed = criterion.get("contains", "") in result.stdout
        self.append_log(
            "programmatic_check",
            criterion=criterion["id"],
            passed=passed,
            returncode=result.returncode,
        )
        return {
            "id": criterion["id"],
            "type": "programmatic",
            "passed": passed,
            "returncode": result.returncode,
            "stdout": result.stdout,
            "stderr": result.stderr,
        }

    def judge_prompt(
        self,
        gate_name: str,
        artifact_label: str,
        artifact_text: str,
        criteria: list[dict[str, Any]],
    ) -> str:
        rubric_lines = []
        for criterion in criteria:
            if criterion["type"] == "judge":
                rubric_lines.append(f"- {criterion['id']}: {criterion['rubric']}")
            elif criterion["type"] == "programmatic":
                rubric_lines.append(f"- {criterion['id']}: programmatic check result is included below.")
            elif criterion["type"] == "human":
                rubric_lines.append(f"- {criterion['id']}: human signoff is required separately.")
        return (
            "You are the Looper judge. Return only a fenced JSON object with keys "
            "verdict, blocking_issues, confidence, and notes. verdict must be pass or revise.\n\n"
            f"Gate: {gate_name}\n"
            f"Artifact: {artifact_label}\n\n"
            "Criteria:\n" + "\n".join(rubric_lines) + "\n\n"
            f"Artifact content:\n{artifact_text}\n"
        )

    def run_judge(
        self,
        member_id: str,
        gate_name: str,
        artifact_label: str,
        artifact_text: str,
        criteria: list[dict[str, Any]],
    ) -> dict[str, Any]:
        # Scrub (and surface any leak) before asking for consent, so the
        # consent decision is made with the leak warning already visible.
        prompt = self.redact_prompt_for_member(
            member_id,
            self.judge_prompt(gate_name, artifact_label, artifact_text, criteria),
        )
        self.ensure_consent(member_id)
        output = call_model(self.member(member_id), prompt, self.base_dir)
        verdict = parse_judge_output(output)
        verdict["member"] = member_id
        self.append_log("judge_verdict", gate=gate_name, member=member_id, verdict=verdict.get("verdict"))
        return verdict

    def run_reviewers(
        self,
        gate_name: str,
        artifact_label: str,
        artifact_text: str,
        member_ids: list[str],
    ) -> list[str]:
        notes = []
        for member_id in member_ids:
            member = self.member(member_id)
            if member.get("role") != "reviewer":
                continue
            prompt = (
                "You are a Looper reviewer. Return concise blocking and non-blocking notes. "
                "Do not return a verdict.\n\n"
                f"Gate: {gate_name}\nArtifact: {artifact_label}\n\n{artifact_text}\n"
            )
            prompt = self.redact_prompt_for_member(member_id, prompt)
            self.ensure_consent(member_id)
            notes.append(f"## {member_id}\n\n{call_model(member, prompt, self.base_dir)}")
            self.append_log("reviewer_notes", gate=gate_name, member=member_id)
        return notes

    def human_check(self, criterion: dict[str, Any]) -> dict[str, Any]:
        print()
        print(criterion["prompt"])
        answer = input("Type 'pass' to approve, anything else to request revision: ").strip().lower()
        return {
            "id": criterion["id"],
            "type": "human",
            "passed": answer == PASS,
            "notes": "approved" if answer == PASS else "human requested revision",
        }

    def run_gate(self, gate_name: str, artifact_path: Path, artifact_label: str) -> bool:
        gate = self.spec["gates"][gate_name]
        criteria = self.criteria(gate.get("criteria", []))
        max_revisions = int(gate.get("max_revisions", 0))
        revision = 0
        self.append_log("gate_start", gate=gate_name, artifact=artifact_label)

        while True:
            self.enforce_wall_clock()
            artifact_text = artifact_path.read_text(encoding="utf-8")
            review_parts: list[str] = []
            failures: list[str] = []

            for criterion in criteria:
                if criterion["type"] == "programmatic":
                    result = self.run_programmatic(criterion)
                    review_parts.append(f"## Programmatic {criterion['id']}\n\n```json\n{json.dumps(result, indent=2)}\n```")
                    if not result["passed"]:
                        failures.append(f"Programmatic check failed: {criterion['id']}")
                elif criterion["type"] == "human":
                    result = self.human_check(criterion)
                    review_parts.append(f"## Human {criterion['id']}\n\n{result['notes']}")
                    if not result["passed"]:
                        failures.append(f"Human check failed: {criterion['id']}")

            reviewer_notes = self.run_reviewers(
                gate_name,
                artifact_label,
                artifact_text,
                list(gate.get("members", [])),
            )
            review_parts.extend(reviewer_notes)

            policy = gate.get("verdict_policy")
            verdict: dict[str, Any] | None = None
            if policy == "revise_until_clean" and not failures:
                source = gate.get("verdict_source")
                if source == "human":
                    answer = input(f"Type 'pass' if {artifact_label} is clean: ").strip().lower()
                    verdict = {
                        "verdict": PASS if answer == PASS else REVISE,
                        "blocking_issues": [] if answer == PASS else ["human requested revision"],
                        "confidence": 1.0,
                        "notes": "human verdict",
                    }
                else:
                    verdict = self.run_judge(source, gate_name, artifact_label, artifact_text, criteria)
                review_parts.append(f"## Verdict\n\n```json\n{json.dumps(verdict, indent=2)}\n```")
                if verdict.get("verdict") == REVISE:
                    failures.extend(verdict.get("blocking_issues") or ["Judge requested revision"])

            if policy == "fixed_passes":
                if failures:
                    pass
                elif revision >= max_revisions:
                    return True
                else:
                    failures.append("fixed_passes reviewer pass")

            if not failures:
                self.save_state(status=f"{gate_name}_passed", **{gate_name: {"passed_at": utc_now()}})
                self.append_log("gate_passed", gate=gate_name, artifact=artifact_label)
                return True

            review_text = "\n\n".join(review_parts + ["## Blocking Issues", "\n".join(f"- {item}" for item in failures)])
            review_path = self.workspace / f"review-{gate_name}-{revision + 1}.md"
            write_text(review_path, review_text)
            self.append_log("gate_blocked", gate=gate_name, review=review_path.name, failures=failures)

            if self.no_progress_reached(gate_name, failures):
                return False

            if revision >= max_revisions:
                self.save_state(
                    status="failed",
                    failure=f"{gate_name}_max_revisions_reached",
                    last_review=str(review_path),
                )
                self.append_log("stop", reason=f"{gate_name}_max_revisions_reached")
                return False

            revised = call_model(
                self.spec["host"],
                self.host_prompt("revise", artifact_text, review_text),
                self.base_dir,
            )
            write_text(artifact_path, revised)
            revision += 1
            self.save_state(status=f"{gate_name}_revision_{revision}", last_review=str(review_path))
            self.append_log("revision", gate=gate_name, revision=revision, artifact=artifact_label)

    def run(self) -> int:
        self.save_state(status="running")
        self.append_log("run_start", spec=str(self.spec_path))
        self.gather_context()

        plan_path = self.workspace / "plan.md"
        if not plan_path.exists():
            self.run_host("plan", plan_path)
        if not self.run_gate("plan_gate", plan_path, "plan.md"):
            return 1

        max_iterations = int(self.spec["loop_control"]["max_iterations"])
        for iteration in range(1, max_iterations + 1):
            self.enforce_wall_clock()
            self.save_state(status="delivery", iteration=iteration)
            delivery_path = self.workspace / f"delivery-{iteration}.md"
            self.run_host("delivery", delivery_path)
            if self.run_gate("delivery_gate", delivery_path, delivery_path.name):
                self.save_state(status="passed", final_delivery=str(delivery_path), completed_at=utc_now())
                self.append_log("run_passed", final_delivery=str(delivery_path))
                print(f"Looper run passed. Final delivery: {delivery_path}")
                return 0

        self.save_state(status="failed", failure="max_iterations_reached")
        self.append_log("stop", reason="max_iterations_reached")
        return 1


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Run a compiled Looper loop.")
    parser.add_argument(
        "spec_path",
        nargs="?",
        type=Path,
        default=Path(__file__).with_name("loop.resolved.json"),
        help="Path to loop.resolved.json (defaults to the file next to run-loop.py).",
    )
    args = parser.parse_args(sys.argv[1:] if argv is None else argv)
    try:
        return Runner(args.spec_path).run()
    except RunnerError as exc:
        print(f"run-loop: error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())