Runner.run / 终止守卫 / 隐私边界 深度解析

   1#!/usr/bin/env python3
   2"""Generated Looper runner.
   3
   4This file executes a resolved loop spec. It intentionally reads only
   5loop.resolved.json and uses only Python stdlib.
   6"""
   7
   8from __future__ import annotations
   9
  10import argparse
  11import datetime as _dt
  12import fnmatch
  13import json
  14from pathlib import Path
  15import re
  16import subprocess
  17import sys
  18import time
  19from typing import Any
  20
  21
  22PASS = "pass"
  23REVISE = "revise"
  24
  25
  26class RunnerError(RuntimeError):
  27    pass
  28
  29
  30def utc_now() -> str:
  31    return _dt.datetime.now(_dt.UTC).replace(microsecond=0).isoformat()
  32
  33
  34def load_json(path: Path) -> dict[str, Any]:
  35    try:
  36        with path.open("r", encoding="utf-8") as fh:
  37            data = json.load(fh)
  38    except OSError as exc:
  39        raise RunnerError(f"Could not read {path}: {exc}") from exc
  40    except json.JSONDecodeError as exc:
  41        raise RunnerError(f"Could not parse JSON in {path}: {exc}") from exc
  42    if not isinstance(data, dict):
  43        raise RunnerError(f"{path} must contain a JSON object")
  44    return data
  45
  46
  47def write_text(path: Path, text: str) -> None:
  48    path.parent.mkdir(parents=True, exist_ok=True)
  49    path.write_text(text.rstrip() + "\n", encoding="utf-8")
  50
  51
  52def write_json(path: Path, data: Any) -> None:
  53    path.parent.mkdir(parents=True, exist_ok=True)
  54    path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8")
  55
  56
  57def ensure_argv(value: Any, field: str) -> list[str]:
  58    if isinstance(value, list) and value and all(isinstance(item, str) for item in value):
  59        return value
  60    raise RunnerError(f"{field} must be a non-empty argv array")

 143class Runner:
 144    def __init__(self, spec_path: Path) -> None:
 145        self.spec_path = spec_path.resolve()
 146        self.base_dir = self.spec_path.parent
 147        self.spec = load_json(self.spec_path)
 148        self.workspace = relative_to_base(self.spec["workspace"]["dir"], self.base_dir)
 149        self.workspace.mkdir(parents=True, exist_ok=True)
 150        self.observability = self.spec.get("observability", {})
 151        self.run_log_path = self.workspace / self.observability.get("run_log", "run-log.md")
 152        self.state_path = self.workspace / self.observability.get("state_file", "state.json")
 153        self.state = self.load_state()
 154        self.started = time.monotonic()
 155
 156    def load_state(self) -> dict[str, Any]:
 157        if self.state_path.exists():
 158            return load_json(self.state_path)
 159        return {
 160            "status": "initialized",
 161            "started_at": utc_now(),
 162            "iteration": 0,
 163            "warnings": [],
 164            "consent": {},
 165        }
 166
 167    def save_state(self, **updates: Any) -> None:
 168        self.state.update(updates)
 169        self.state["updated_at"] = utc_now()
 170        write_json(self.state_path, self.state)
 171
 172    def append_log(self, event: str, **fields: Any) -> None:
 173        self.run_log_path.parent.mkdir(parents=True, exist_ok=True)
 174        payload = f" {json.dumps(fields, sort_keys=True)}" if fields else ""
 175        with self.run_log_path.open("a", encoding="utf-8") as fh:

 178    def enforce_wall_clock(self) -> None:
 179        budget = self.spec.get("loop_control", {}).get("budget", {})
 180        wall_clock_min = budget.get("wall_clock_min")
 181        if wall_clock_min is None:
 182            return
 183        if time.monotonic() - self.started > float(wall_clock_min) * 60:
 184            self.save_state(status="failed", failure="wall_clock_budget_exceeded")
 185            self.append_log("stop", reason="wall_clock_budget_exceeded")
 186            raise RunnerError("Wall-clock budget exceeded")
 187
 188    def no_progress_reached(self, gate_name: str, failures: list[str]) -> bool:
 189        if not failures:
 190            self.save_state(no_progress={"count": 0, "signature": "", "gate": gate_name})
 191            return False
 192        config = self.spec.get("loop_control", {}).get("no_progress", {})
 193        threshold = int(config.get("max_stalled_iterations", 2))
 194        signature = "\n".join(sorted(failures))
 195        previous = self.state.get("no_progress", {})
 196        same_gate = previous.get("gate") == gate_name
 197        same_signature = previous.get("signature") == signature
 198        count = int(previous.get("count", 0)) + 1 if same_gate and same_signature else 1
 199        progress = {
 200            "gate": gate_name,
 201            "signature": signature,
 202            "count": count,
 203            "threshold": threshold,
 204            "updated_at": utc_now(),
 205        }
 206        self.save_state(no_progress=progress)
 207        if count < threshold:
 208            return False
 209        self.append_log("no_progress_detected", gate=gate_name, count=count, failures=failures)
 210        if config.get("action", "stop") == "human_checkpoint":
 211            answer = input("No-progress detected. Type 'continue' to allow one more revision: ").strip().lower()
 212            if answer == "continue":
 213                progress["count"] = 0
 214                self.save_state(no_progress=progress)
 215                self.append_log("no_progress_override", gate=gate_name)
 216                return False
 217        self.save_state(status="failed", failure="no_progress_detected", blocking_issues=failures)
 218        return True

 227    def redactions_for(self, member_id: str) -> list[str]:
 228        redactions: list[str] = []
 229        for entry in self.spec.get("privacy", {}).get("egress", []):
 230            if entry.get("to") == member_id:
 231                redactions.extend(entry.get("redact", []))
 232        return redactions or [".env", ".env.*", "secrets/**", "**/*.key"]
 233
 234    def redact_prompt_for_member(self, member_id: str, prompt: str) -> str:
 235        redactions = self.redactions_for(member_id)
 236        redacted = prompt
 237        for pattern in redactions:
 238            paths = list(self.base_dir.glob(pattern))
 239            if pattern.endswith("/**"):
 240                root = self.base_dir / pattern[:-3]
 241                if root.exists():
 242                    paths.extend(root.rglob("*"))
 243            for path in paths:
 244                if not path.is_file():
 245                    continue
 246                try:
 247                    secret_text = path.read_text(encoding="utf-8")
 248                except UnicodeDecodeError:
 249                    continue
 250                if not secret_text.strip() or len(secret_text) > 1_000_000:
 251                    continue
 252                marker = f"[redacted:{path.relative_to(self.base_dir).as_posix()}]"
 253                redacted = redacted.replace(secret_text, marker)
 254                for line in secret_text.splitlines():
 255                    stripped = line.strip()
 256                    if len(stripped) >= 8:
 257                        redacted = redacted.replace(stripped, marker)
 258        return redacted

 260    def ensure_consent(self, member_id: str) -> None:
 261        member = self.member(member_id)
 262        if member.get("local"):
 263            return
 264        matching = [
 265            entry
 266            for entry in self.spec.get("privacy", {}).get("egress", [])
 267            if entry.get("to") == member_id and entry.get("consent") == "required"
 268        ]
 269        if not matching:
 270            return
 271        if self.state.get("consent", {}).get(member_id):
 272            return
 273        sends = sorted({item for entry in matching for item in entry.get("sends", [])})
 274        redactions = sorted({item for entry in matching for item in entry.get("redact", [])})
 275        print()
 276        print(f"Looper is about to send {', '.join(sends) or 'context'} to {member_id}.")
 277        print(f"CLI: {member.get('cli')} / model: {member.get('model', 'default')}")
 278        print(f"Redactions: {', '.join(redactions) or '(none)'}")
 279        answer = input("Type 'yes' to consent to this first send: ").strip().lower()
 280        if answer != "yes":
 281            self.save_state(status="blocked", failure=f"consent_refused:{member_id}")
 282            raise RunnerError(f"Consent refused for {member_id}")
 283        consent = dict(self.state.get("consent", {}))
 284        consent[member_id] = {"granted_at": utc_now(), "sends": sends, "redact": redactions}
 285        self.save_state(consent=consent)
 286

 287    def gather_context(self) -> str:
 288        goal = self.spec["goal"]
 289        chunks: list[str] = []
 290        for index, source in enumerate(goal.get("context_sources", []), start=1):
 291            self.enforce_wall_clock()
 292            if "file" in source:
 293                path = relative_to_base(source["file"], self.base_dir)
 294                if is_redacted(path, self.base_dir, [".env", ".env.*", "secrets/**", "**/*.key"]):
 295                    chunks.append(f"## Context source {index}: {source['file']}\n[redacted]\n")
 296                    self.append_log("context", source=source["file"], status="redacted")
 297                elif path.exists():
 298                    chunks.append(f"## Context source {index}: {source['file']}\n{path.read_text(encoding='utf-8')}\n")
 299                    self.append_log("context", source=source["file"], status="read")
 300                else:
 301                    chunks.append(f"## Context source {index}: {source['file']}\n[missing]\n")
 302                    self.append_log("context", source=source["file"], status="missing")
 303            elif "cmd" in source:
 304                argv = ensure_argv(source["cmd"], f"context_sources[{index}].cmd")
 305                result = run_argv(argv, cwd=self.base_dir, timeout_sec=int(source.get("timeout_sec", 60)))
 306                chunks.append(
 307                    f"## Context source {index}: {' '.join(argv)}\n"
 308                    f"exit={result.returncode}\nstdout:\n{result.stdout}\nstderr:\n{result.stderr}\n"
 309                )
 310                self.append_log("context_cmd", argv=argv, returncode=result.returncode)
 311        context = "\n".join(chunks).strip()
 312        write_text(self.workspace / "context.md", context or "No context sources configured.")
 313        return context
 314
 315    def host_prompt(self, phase: str, artifact: str = "", review: str = "") -> str:
 316        goal = self.spec["goal"]
 317        if phase == "plan":
 318            return (
 319                "Draft plan.md for this loop.\n\n"
 320                f"Goal:\n{goal['statement']}\n\n"
 321                f"Definition of done:\n{goal['definition_of_done']}\n\n"
 322                f"Context:\n{(self.workspace / 'context.md').read_text(encoding='utf-8')}\n"
 323            )
 324        if phase == "delivery":
 325            return (
 326                "Write the next delivery artifact for this loop.\n\n"
 327                f"Goal:\n{goal['statement']}\n\n"
 328                f"Definition of done:\n{goal['definition_of_done']}\n\n"
 329                f"Plan:\n{(self.workspace / 'plan.md').read_text(encoding='utf-8')}\n"
 330            )
 331        if phase == "revise":
 332            return (
 333                "Revise the artifact to address the review. Return only the revised artifact.\n\n"
 334                f"Artifact:\n{artifact}\n\nReview:\n{review}\n"
 335            )
 336        raise RunnerError(f"Unknown host phase: {phase}")

 357            "programmatic_check",
 358            criterion=criterion["id"],
 359            passed=passed,
 360            returncode=result.returncode,
 361        )
 362        return {
 363            "id": criterion["id"],
 364            "type": "programmatic",
 365            "passed": passed,
 366            "returncode": result.returncode,
 367            "stdout": result.stdout,
 368            "stderr": result.stderr,
 369        }
 370
 371    def judge_prompt(
 372        self,
 373        gate_name: str,
 374        artifact_label: str,
 375        artifact_text: str,
 376        criteria: list[dict[str, Any]],
 377    ) -> str:
 378        rubric_lines = []
 379        for criterion in criteria:
 380            if criterion["type"] == "judge":
 381                rubric_lines.append(f"- {criterion['id']}: {criterion['rubric']}")
 382            elif criterion["type"] == "programmatic":
 383                rubric_lines.append(f"- {criterion['id']}: programmatic check result is included below.")
 384            elif criterion["type"] == "human":
 385                rubric_lines.append(f"- {criterion['id']}: human signoff is required separately.")
 386        return (
 387            "You are the Looper judge. Return only a fenced JSON object with keys "
 388            "verdict, blocking_issues, confidence, and notes. verdict must be pass or revise.\n\n"
 389            f"Gate: {gate_name}\n"
 390            f"Artifact: {artifact_label}\n\n"
 391            "Criteria:\n" + "\n".join(rubric_lines) + "\n\n"
 392            f"Artifact content:\n{artifact_text}\n"
 393        )

 395    def run_judge(
 396        self,
 397        member_id: str,
 398        gate_name: str,
 399        artifact_label: str,
 400        artifact_text: str,
 401        criteria: list[dict[str, Any]],
 402    ) -> dict[str, Any]:
 403        self.ensure_consent(member_id)
 404        output = call_model(
 405            self.member(member_id),
 406            self.redact_prompt_for_member(
 407                member_id,
 408                self.judge_prompt(gate_name, artifact_label, artifact_text, criteria),
 409            ),
 410            self.base_dir,
 411        )
 412        verdict = parse_judge_output(output)
 413        verdict["member"] = member_id
 414        self.append_log("judge_verdict", gate=gate_name, member=member_id, verdict=verdict.get("verdict"))
 415        return verdict
 416
 417    def run_reviewers(
 418        self,
 419        gate_name: str,
 420        artifact_label: str,
 421        artifact_text: str,
 422        member_ids: list[str],
 423    ) -> list[str]:
 424        notes = []
 425        for member_id in member_ids:
 426            member = self.member(member_id)
 427            if member.get("role") != "reviewer":
 428                continue
 429            self.ensure_consent(member_id)
 430            prompt = (
 431                "You are a Looper reviewer. Return concise blocking and non-blocking notes. "
 432                "Do not return a verdict.\n\n"
 433                f"Gate: {gate_name}\nArtifact: {artifact_label}\n\n{artifact_text}\n"
 434            )
 435            prompt = self.redact_prompt_for_member(member_id, prompt)
 436            notes.append(f"## {member_id}\n\n{call_model(member, prompt, self.base_dir)}")
 437            self.append_log("reviewer_notes", gate=gate_name, member=member_id)
 438        return notes
 439
 440    def human_check(self, criterion: dict[str, Any]) -> dict[str, Any]:
 441        print()
 442        print(criterion["prompt"])
 443        answer = input("Type 'pass' to approve, anything else to request revision: ").strip().lower()
 444        return {
 445            "id": criterion["id"],
 446            "type": "human",
 447            "passed": answer == PASS,
 448            "notes": "approved" if answer == PASS else "human requested revision",
 449        }

 451    def run_gate(self, gate_name: str, artifact_path: Path, artifact_label: str) -> bool:
 452        gate = self.spec["gates"][gate_name]
 453        criteria = self.criteria(gate.get("criteria", []))
 454        max_revisions = int(gate.get("max_revisions", 0))
 455        revision = 0
 456        self.append_log("gate_start", gate=gate_name, artifact=artifact_label)
 457
 458        while True:
 459            self.enforce_wall_clock()
 460            artifact_text = artifact_path.read_text(encoding="utf-8")
 461            review_parts: list[str] = []
 462            failures: list[str] = []
 463
 464            for criterion in criteria:
 465                if criterion["type"] == "programmatic":
 466                    result = self.run_programmatic(criterion)
 467                    review_parts.append(f"## Programmatic {criterion['id']}\n\n```json\n{json.dumps(result, indent=2)}\n```")
 468                    if not result["passed"]:
 469                        failures.append(f"Programmatic check failed: {criterion['id']}")
 470                elif criterion["type"] == "human":
 471                    result = self.human_check(criterion)
 472                    review_parts.append(f"## Human {criterion['id']}\n\n{result['notes']}")
 473                    if not result["passed"]:
 474                        failures.append(f"Human check failed: {criterion['id']}")
 475
 476            reviewer_notes = self.run_reviewers(
 477                gate_name,
 478                artifact_label,
 479                artifact_text,
 480                list(gate.get("members", [])),
 481            )
 482            review_parts.extend(reviewer_notes)
 483
 484            policy = gate.get("verdict_policy")
 485            verdict: dict[str, Any] | None = None
 486            if policy == "revise_until_clean" and not failures:
 487                source = gate.get("verdict_source")
 488                if source == "human":
 489                    answer = input(f"Type 'pass' if {artifact_label} is clean: ").strip().lower()
 490                    verdict = {
 491                        "verdict": PASS if answer == PASS else REVISE,
 492                        "blocking_issues": [] if answer == PASS else ["human requested revision"],
 493                        "confidence": 1.0,
 494                        "notes": "human verdict",
 495                    }
 496                else:
 497                    verdict = self.run_judge(source, gate_name, artifact_label, artifact_text, criteria)
 498                review_parts.append(f"## Verdict\n\n```json\n{json.dumps(verdict, indent=2)}\n```")
 499                if verdict.get("verdict") == REVISE:
 500                    failures.extend(verdict.get("blocking_issues") or ["Judge requested revision"])
 501
 502            if policy == "fixed_passes":
 503                if failures:
 504                    pass
 505                elif revision >= max_revisions:
 506                    return True
 507                else:
 508                    failures.append("fixed_passes reviewer pass")
 509
 510            if not failures:
 511                self.save_state(status=f"{gate_name}_passed", **{gate_name: {"passed_at": utc_now()}})
 512                self.append_log("gate_passed", gate=gate_name, artifact=artifact_label)
 513                return True
 514
 515            review_text = "\n\n".join(review_parts + ["## Blocking Issues", "\n".join(f"- {item}" for item in failures)])
 516            review_path = self.workspace / f"review-{gate_name}-{revision + 1}.md"
 517            write_text(review_path, review_text)
 518            self.append_log("gate_blocked", gate=gate_name, review=review_path.name, failures=failures)
 519
 520            if self.no_progress_reached(gate_name, failures):
 521                return False
 522
 523            if revision >= max_revisions:
 524                self.save_state(
 525                    status="failed",
 526                    failure=f"{gate_name}_max_revisions_reached",
 527                    last_review=str(review_path),
 528                )
 529                self.append_log("stop", reason=f"{gate_name}_max_revisions_reached")
 530                return False
 531
 532            revised = call_model(
 533                self.spec["host"],
 534                self.host_prompt("revise", artifact_text, review_text),
 535                self.base_dir,
 536            )
 537            write_text(artifact_path, revised)
 538            revision += 1
 539            self.save_state(status=f"{gate_name}_revision_{revision}", last_review=str(review_path))
 540            self.append_log("revision", gate=gate_name, revision=revision, artifact=artifact_label)

 542    def run(self) -> int:
 543        self.save_state(status="running")
 544        self.append_log("run_start", spec=str(self.spec_path))
 545        self.gather_context()
 546
 547        plan_path = self.workspace / "plan.md"
 548        if not plan_path.exists():
 549            self.run_host("plan", plan_path)
 550        if not self.run_gate("plan_gate", plan_path, "plan.md"):
 551            return 1
 552
 553        max_iterations = int(self.spec["loop_control"]["max_iterations"])
 554        for iteration in range(1, max_iterations + 1):
 555            self.enforce_wall_clock()
 556            self.save_state(status="delivery", iteration=iteration)
 557            delivery_path = self.workspace / f"delivery-{iteration}.md"
 558            self.run_host("delivery", delivery_path)
 559            if self.run_gate("delivery_gate", delivery_path, delivery_path.name):
 560                self.save_state(status="passed", final_delivery=str(delivery_path), completed_at=utc_now())
 561                self.append_log("run_passed", final_delivery=str(delivery_path))
 562                print(f"Looper run passed. Final delivery: {delivery_path}")
 563                return 0
 564
 565        self.save_state(status="failed", failure="max_iterations_reached")
 566        self.append_log("stop", reason="max_iterations_reached")
 567        return 1
 568
 569
 570def main(argv: list[str] | None = None) -> int:
 571    parser = argparse.ArgumentParser(description="Run a compiled Looper loop.")
 572    parser.add_argument(
 573        "spec_path",
 574        nargs="?",
 575        type=Path,
 576        default=Path(__file__).with_name("loop.resolved.json"),
 577        help="Path to loop.resolved.json (defaults to the file next to run-loop.py).",
 578    )
 579    args = parser.parse_args(sys.argv[1:] if argv is None else argv)
 580    try:
 581        return Runner(args.spec_path).run()
 582    except RunnerError as exc:
 583        print(f"run-loop: error: {exc}", file=sys.stderr)
 584        return 2
 585
 586
 587if __name__ == "__main__":
 588    raise SystemExit(main())

   1#!/usr/bin/env python3
   2"""Fake judge model for Looper runner tests."""
   3
   4from __future__ import annotations
   5
   6import json
   7from pathlib import Path
   8import sys
   9
  10
  11def main() -> int:
  12    if len(sys.argv) < 2:
  13        print("usage: fake_judge.py <state-file> [revise-count]", file=sys.stderr)
  14        return 2
  15    state_path = Path(sys.argv[1])
  16    revise_count = int(sys.argv[2]) if len(sys.argv) > 2 else 0
  17    state_path.parent.mkdir(parents=True, exist_ok=True)
  18    state = {"count": 0}
  19    if state_path.exists():
  20        state = json.loads(state_path.read_text(encoding="utf-8"))
  21    count = int(state.get("count", 0))
  22    state["count"] = count + 1
  23    state_path.write_text(json.dumps(state), encoding="utf-8")
  24
  25    if count < revise_count:
  26        verdict = {
  27            "verdict": "revise",
  28            "blocking_issues": ["Add an explicit owner."],
  29            "confidence": 0.91,
  30            "notes": "The artifact is close but needs one concrete owner.",
  31        }
  32    else:
  33        verdict = {
  34            "verdict": "pass",
  35            "blocking_issues": [],
  36            "confidence": 0.94,
  37            "notes": "The artifact satisfies the rubric.",
  38        }
  39    print("```json")
  40    print(json.dumps(verdict))
  41    print("```")
  42    return 0
  43
  44
  45if __name__ == "__main__":
  46    raise SystemExit(main())
  47

   1#!/usr/bin/env python3
   2"""Fake host model for Looper runner tests."""
   3
   4from __future__ import annotations
   5
   6import sys
   7
   8
   9def main() -> int:
  10    prompt = sys.stdin.read()
  11    if "Revise the artifact" in prompt:
  12        print("Revised artifact")
  13        print()
  14        print("Owner: host")
  15        print("Resolved reviewer issue.")
  16        print("No TBD")
  17    elif "Draft plan.md" in prompt:
  18        print("# Plan")
  19        print()
  20        print("Owner: host")
  21        print("1. Read the context.")
  22        print("2. Produce the delivery.")
  23        print("No TBD")
  24    else:
  25        print("# LOOP.md")
  26        print()
  27        print("Owner: host")
  28        print("Diagram: included")
  29        print("No TBD")
  30    return 0
  31
  32
  33if __name__ == "__main__":
  34    raise SystemExit(main())
  35