diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c27ff82..9c8920f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,3 +47,16 @@ jobs: run: | python -B -m loop doctor examples/coverage-repair python -B -m loop inspect examples/coverage-repair + + recipe-langgraph: + name: recipe (langgraph) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: Install recipe dependencies + run: python -m pip install --upgrade pip pyyaml pytest jsonschema langgraph + - name: LangGraph recipe end-to-end + run: python -B -m pytest -q -p no:cacheprovider scripts/test_langgraph_recipe.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ef3a11d..4d86bf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,31 @@ All notable changes to `loop-engineer` are documented here. `WORKFLOW.md` and `README.md` are reworded to describe the mechanism; the 0.3.4 history is left intact. +## Unreleased + +**B1 — the writer API.** `loop.emit` lets a foreign runtime (LangGraph, a plain +script, any orchestrator) record an evidence-backed loop contract without +adopting the loop-engineer runtime. It is a writer, never a runtime: it renders +the contract artifacts and refuses a dishonest `Succeeded` at write time — the +same evidence cross-check `loop doctor` enforces, applied before the file exists. + +### Added +- **`loop/emit.py` writer API** — `open_contract`, `append_iteration`, + `append_receipt`, and `terminate`, plus the `EmitError` raised when a write + would produce a dishonest or schema-invalid artifact. `terminate` refuses an + evidence-free `Succeeded` (also no-met-criterion or false-completion-flagged), + so the honesty gate runs at write time rather than only at validate time; + every artifact it writes passes `doctor` by construction. +- **LangGraph recipe** (`examples/langgraph-emit/`) — a runnable three-node + graph whose terminal node ships proof-of-done through `loop.emit`; the emitted + contract passes `loop doctor` independently of the graph that wrote it. Paired + with the 10-line integration guide `docs/integrations/langgraph.md`. +- **Recipe acceptance test** (`scripts/test_langgraph_recipe.py`) — runs the + example end-to-end and asserts the emitted contract passes `doctor` and ends + `Succeeded` with evidence. Env-guarded on `langgraph` (skips when absent), so + the package stays zero-dependency; a dedicated `recipe (langgraph)` CI job + installs LangGraph and runs it. + ## 0.6.1 — 2026-07-04 **PyPI substrate.** `loop-engineer` becomes a self-contained wheel that runs from diff --git a/docs/integrations/langgraph.md b/docs/integrations/langgraph.md new file mode 100644 index 0000000..a42edbb --- /dev/null +++ b/docs/integrations/langgraph.md @@ -0,0 +1,30 @@ +# LangGraph — proof-of-done in 10 lines + +`loop.emit` is a pure-stdlib writer: your graph keeps its own runtime, and the +terminal node records evidence-backed state the `loop` CLI can independently +validate. `pip install loop-engineer` (LangGraph itself stays your dependency). + +```python +from loop import emit + +emit.open_contract("run/") # once, before the graph runs + +def conclude(state): # your graph's terminal node + emit.append_iteration("run/", iteration_id=1, outcome="task_passed", + task_id="T1", verify_cmd="pytest -q", verify_outcome="pass") + emit.terminate("run/", state="Succeeded", + criteria_met={"tests": True}, evidence=["reports/pytest.txt"]) + return {} +``` + +`emit.terminate` **refuses an evidence-free `Succeeded`** (raises `EmitError`) — +the same cross-check `loop doctor` enforces, applied before the file exists. + +Gate it in CI: + +```yaml +- run: pip install loop-engineer +- run: loop doctor run/ +``` + +Full runnable example: [`examples/langgraph-emit/`](../../examples/langgraph-emit/). diff --git a/examples/langgraph-emit/README.md b/examples/langgraph-emit/README.md new file mode 100644 index 0000000..064677c --- /dev/null +++ b/examples/langgraph-emit/README.md @@ -0,0 +1,34 @@ +# LangGraph recipe — proof-of-done through `loop.emit` + +A runnable [LangGraph](https://github.com/langchain-ai/langgraph) graph whose +**terminal node writes the loop contract** — evidence-backed state the `loop` +CLI can independently validate. LangGraph keeps its own runtime; `loop.emit` is +a pure-stdlib writer that refuses to record a dishonest result. + +## What it shows + +`graph_example.py` runs three plain-function nodes — `do_work` writes +`artifact.txt`, `verify` re-reads it from disk, and `conclude` records the +outcome: + +- On a real pass, `conclude` calls `emit.terminate(..., state="Succeeded", + evidence=["artifact.txt"])`. +- A lying `Succeeded` — no evidence, or no met criterion — raises `EmitError` + **before anything hits disk**. That is the same cross-check `loop doctor` + enforces, applied at write time. + +## Run it + +```bash +pip install loop-engineer langgraph +python graph_example.py demo-run/ +loop doctor demo-run/ # -> {"ok": true, ...} +``` + +`demo-run/.loop/terminal_state.json` ends `Succeeded` with `evidence`; `loop +doctor` validates it independently of the graph that wrote it. + +## The 10-line integration + +The general pattern (any graph, any terminal node) lives in +[`docs/integrations/langgraph.md`](../../docs/integrations/langgraph.md). diff --git a/examples/langgraph-emit/graph_example.py b/examples/langgraph-emit/graph_example.py new file mode 100644 index 0000000..b42f602 --- /dev/null +++ b/examples/langgraph-emit/graph_example.py @@ -0,0 +1,82 @@ +"""A minimal LangGraph graph that ships proof-of-done through loop.emit. + +The graph does real (tiny) work, verifies it from the filesystem, and the +terminal node records the outcome via emit.terminate(...) — which refuses an +evidence-free Succeeded. Run: + + python graph_example.py +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import TypedDict + +from langgraph.graph import END, START, StateGraph + +from loop import emit + + +class State(TypedDict): + workspace: str + verified: bool + + +def do_work(state: State) -> dict: + out = Path(state["workspace"]) / "artifact.txt" + out.write_text("hello from langgraph\n", encoding="utf-8") + return {} + + +def verify(state: State) -> dict: + artifact = Path(state["workspace"]) / "artifact.txt" + ok = artifact.is_file() and "hello" in artifact.read_text(encoding="utf-8") + return {"verified": ok} + + +def conclude(state: State) -> dict: + ws = state["workspace"] + passed = state["verified"] + emit.append_iteration( + ws, iteration_id=1, outcome="task_passed" if passed else "task_failed", + task_id="T1", actions=["wrote artifact.txt", "re-read and checked content"], + verify_cmd="verify node (filesystem re-read)", verify_outcome="pass" if passed else "fail", + ) + if passed: + emit.terminate( + ws, state="Succeeded", criteria_met={"1": True}, + evidence=["artifact.txt"], reason="artifact written and independently re-read", + iteration_id=1, + ) + else: + emit.terminate( + ws, state="FailedUnverifiable", criteria_met={"1": False}, + evidence=[], reason="verification failed", iteration_id=1, + ) + return {} + + +def main(workspace: str) -> int: + emit.open_contract(workspace) + graph = ( + StateGraph(State) + .add_node(do_work) + .add_node(verify) + .add_node(conclude) + .add_edge(START, "do_work") + .add_edge("do_work", "verify") + .add_edge("verify", "conclude") + .add_edge("conclude", END) + .compile() + ) + graph.invoke({"workspace": workspace, "verified": False}) + print(f"contract emitted at {workspace}/.loop — run: python3 -m loop doctor {workspace}") + return 0 + + +if __name__ == "__main__": + if len(sys.argv) != 2: + print("usage: python graph_example.py ", file=sys.stderr) + raise SystemExit(2) + raise SystemExit(main(sys.argv[1])) diff --git a/loop/emit.py b/loop/emit.py new file mode 100644 index 0000000..520336a --- /dev/null +++ b/loop/emit.py @@ -0,0 +1,229 @@ +"""Writer API for foreign runtimes (B1). A writer, never a runtime: it renders +contract artifacts and refuses dishonest ones — no orchestration, no execution. + +The G1 cross-check (a Succeeded terminal needs evidence and a met criterion) +is enforced HERE, at write time, before doctor ever sees the file. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Sequence + +from .contract import ( + TERMINAL_STATES, + _validate_record, + _validate_terminal, + _validation_mode, +) +from .paths import resolve_loop_paths +from .scaffold import scaffold + +_ITERATION_OUTCOMES = ( + "task_passed", + "task_failed", + "repair_triggered", + "approval_requested", + "replanned", + "terminal", +) +_RECEIPT_ROLES = ("read", "reason", "write", "orchestrate") +_RECEIPT_OUTCOMES = ("ok", "fail", "escalated") + + +class EmitError(ValueError): + """A write was refused: it would produce a dishonest or schema-invalid artifact.""" + + +def open_contract(target: str | Path) -> dict[str, Any]: + """Render a fresh, doctor-clean contract. Delegates to the scaffold renderer.""" + return scaffold(target) + + +def _require_contract(target: str | Path): + paths = resolve_loop_paths(target) + if not paths.state.is_file(): + raise EmitError( + f"no loop contract at {paths.workspace} (missing .loop/state.json) — " + f"call emit.open_contract() first" + ) + return paths + + +def _read_state(paths) -> dict[str, Any]: + try: + data = json.loads(paths.state.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise EmitError(f"unreadable state.json: {exc}") from exc + if not isinstance(data, dict): + raise EmitError("state.json must hold a JSON object") + return data + + +def _write_state(paths, state: dict[str, Any]) -> None: + paths.state.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8") + + +def append_iteration( + target: str | Path, + *, + iteration_id: int, + outcome: str, + task_id: str = "", + actions: Sequence[str] = (), + verify_cmd: str = "", + verify_outcome: str = "", + notes: str = "", +) -> Path: + """Append one iteration block to RUNLOG.md (the shape scripts/metrics.py + parses: `## Iteration ` header + a backticked outcome token) and advance + .loop/state.json's iteration_id/active_task.""" + if outcome not in _ITERATION_OUTCOMES: + raise EmitError(f"unknown iteration outcome {outcome!r}; expected one of {_ITERATION_OUTCOMES}") + paths = _require_contract(target) + + lines = [ + "", + f"## Iteration {iteration_id} — {datetime.now(timezone.utc).date().isoformat()}", + "", + ] + if task_id: + lines.append(f"**Active task:** `{task_id}`") + lines.append("") + if actions: + lines.append("### Actions taken") + lines.append("") + lines.extend(f"- {a}" for a in actions) + lines.append("") + if verify_cmd or verify_outcome: + lines.append("### Verification result") + lines.append("") + lines.append(f"- **Gate:** `{verify_cmd}` — {verify_outcome}") + lines.append("") + lines.append("### Outcome") + lines.append("") + lines.append(f"`{outcome}`") + lines.append("") + if notes: + lines.append("### Notes") + lines.append("") + lines.append(notes) + lines.append("") + + runlog = paths.runlog + if not runlog.exists(): + runlog = paths.workspace / "RUNLOG.md" + runlog.write_text(f"# RUNLOG.md — {paths.workspace.name}\n", encoding="utf-8") + with runlog.open("a", encoding="utf-8") as fh: + fh.write("\n".join(lines)) + + state = _read_state(paths) + state["iteration_id"] = str(iteration_id) + if task_id: + state["active_task"] = task_id + _write_state(paths, state) + return runlog + + +def append_receipt( + target: str | Path, + *, + iteration_id: int, + role: str, + model: str, + outcome: str, + dispatch_id: str | None = None, + tokens: int | None = None, + cost_usd: float | None = None, + ts: str | None = None, +) -> Path: + """Append one loop-engineer/receipt@1 line to .loop/receipts/receipts.jsonl.""" + if role not in _RECEIPT_ROLES: + raise EmitError(f"unknown receipt role {role!r}; expected one of {_RECEIPT_ROLES}") + if outcome not in _RECEIPT_OUTCOMES: + raise EmitError(f"unknown receipt outcome {outcome!r}; expected one of {_RECEIPT_OUTCOMES}") + if not isinstance(iteration_id, int) or isinstance(iteration_id, bool) or iteration_id < 0: + raise EmitError("iteration_id must be a non-negative integer") + paths = _require_contract(target) + + record: dict[str, Any] = { + "schema": "loop-engineer/receipt@1", + "iteration_id": iteration_id, + "dispatch_id": dispatch_id, + "role": role, + "model": model, + "outcome": outcome, + "tokens": tokens, + "cost_usd": cost_usd, + "ts": ts, + } + receipts = paths.loop_dir / "receipts" / "receipts.jsonl" + issues: list[dict] = [] + _validate_record(record, "receipt", receipts, _validation_mode(), issues) + if issues: + raise EmitError(f"receipt failed schema validation: {issues}") + receipts.parent.mkdir(parents=True, exist_ok=True) + with receipts.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(record, sort_keys=True) + "\n") + return receipts + + +def terminate( + target: str | Path, + *, + state: str, + criteria_met: dict[str, bool], + evidence: list[str], + reason: str = "", + iteration_id: int | None = None, + false_completion: bool = False, + lessons_ref: str | None = None, +) -> Path: + """Write .loop/terminal_state.json (and stamp state.json.terminal_state). + + Refuses an evidence-free Succeeded — the G1 cross-check at write time: + Succeeded requires non-empty evidence, at least one met criterion, and + false_completion=False. + """ + if state not in TERMINAL_STATES: + raise EmitError(f"unknown terminal state {state!r}; expected one of {TERMINAL_STATES}") + if state == "Succeeded": + if false_completion: + raise EmitError("refusing Succeeded with false_completion=True (G1 contradiction)") + if not evidence: + raise EmitError("refusing evidence-free Succeeded: evidence[] is empty (G1)") + if not any(v is True for v in criteria_met.values()): + raise EmitError("refusing Succeeded with no met (true) entry in criteria_met (G1)") + if not all(isinstance(v, bool) for v in criteria_met.values()): + raise EmitError("criteria_met values must be booleans") + paths = _require_contract(target) + current = _read_state(paths) + + terminal: dict[str, Any] = { + "schema": "loop-engineer/terminal@1", + "project": paths.workspace.name, + "state": state, + "criteria_met": dict(criteria_met), + "evidence": list(evidence), + "false_completion": false_completion, + "terminated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + } + if reason: + terminal["reason"] = reason + if iteration_id is not None: + terminal["iteration_id"] = iteration_id + if lessons_ref is not None: + terminal["lessons_ref"] = lessons_ref + + terminal_path = paths.loop_dir / "terminal_state.json" + issues: list[dict] = [] + _validate_terminal(terminal, terminal_path, issues) + if issues: + raise EmitError(f"terminal failed validation before write: {issues}") + + terminal_path.write_text(json.dumps(terminal, indent=2) + "\n", encoding="utf-8") + current["terminal_state"] = state + _write_state(paths, current) + return terminal_path diff --git a/scripts/test_emit.py b/scripts/test_emit.py new file mode 100644 index 0000000..69f7327 --- /dev/null +++ b/scripts/test_emit.py @@ -0,0 +1,115 @@ +"""B1 acceptance: emit writes schema-valid artifacts by construction and refuses +an evidence-free Succeeded at write time (G1 enforced before validate time).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loop import emit +from loop.contract import validate_contract + + +@pytest.fixture() +def workspace(tmp_path): + ws = tmp_path / "demo" + report = emit.open_contract(ws) + assert report["ok"] is True + return ws + + +def test_open_contract_is_doctor_clean(workspace): + assert validate_contract(workspace)["ok"] is True + + +def test_append_iteration_writes_parseable_runlog_and_updates_state(workspace): + runlog = emit.append_iteration( + workspace, iteration_id=1, outcome="task_passed", task_id="T1", + actions=["did the thing"], verify_cmd="scripts/verify-fast", verify_outcome="pass", + ) + text = runlog.read_text(encoding="utf-8") + assert "## Iteration 1" in text + assert "`task_passed`" in text + + state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8")) + assert state["iteration_id"] == "1" + assert state["active_task"] == "T1" + assert validate_contract(workspace)["ok"] is True + + +def test_append_iteration_rejects_unknown_outcome(workspace): + with pytest.raises(emit.EmitError): + emit.append_iteration(workspace, iteration_id=1, outcome="totally_done") + + +def test_append_receipt_is_schema_valid(workspace): + path = emit.append_receipt( + workspace, iteration_id=1, role="write", model="claude-opus", outcome="ok" + ) + assert path == workspace / ".loop" / "receipts" / "receipts.jsonl" + # doctor validates .loop/receipts/*.jsonl against loop-engineer/receipt@1 + report = validate_contract(workspace) + assert report["ok"] is True + assert "loop-engineer/receipt@1" in report["schemas_checked"] + + +def test_append_receipt_rejects_bad_role(workspace): + with pytest.raises(emit.EmitError): + emit.append_receipt(workspace, iteration_id=1, role="wizard", model="m", outcome="ok") + + +def test_terminate_succeeded_with_evidence_passes_doctor(workspace): + terminal = emit.terminate( + workspace, state="Succeeded", criteria_met={"1": True}, + evidence=["artifact.txt"], reason="verified", iteration_id=1, + ) + data = json.loads(terminal.read_text(encoding="utf-8")) + assert data["schema"] == "loop-engineer/terminal@1" + assert data["false_completion"] is False + state = json.loads((workspace / ".loop" / "state.json").read_text(encoding="utf-8")) + assert state["terminal_state"] == "Succeeded" + assert validate_contract(workspace)["ok"] is True + + +@pytest.mark.parametrize( + "kwargs", + [ + dict(criteria_met={"1": True}, evidence=[]), # evidence-free + dict(criteria_met={"1": False}, evidence=["a.txt"]), # no met criterion + dict(criteria_met={}, evidence=["a.txt"]), # empty criteria + dict(criteria_met={"1": True}, evidence=["a.txt"], false_completion=True), # G1 contradiction + ], +) +def test_terminate_refuses_dishonest_succeeded(workspace, kwargs): + with pytest.raises(emit.EmitError): + emit.terminate(workspace, state="Succeeded", reason="claimed", **kwargs) + assert not (workspace / ".loop" / "terminal_state.json").exists() + + +def test_terminate_honest_failure_needs_no_evidence(workspace): + emit.terminate( + workspace, state="FailedUnverifiable", criteria_met={"1": False}, + evidence=[], reason="could not verify", + ) + assert validate_contract(workspace)["ok"] is True + + +def test_terminate_rejects_unknown_state(workspace): + with pytest.raises(emit.EmitError): + emit.terminate(workspace, state="Done", criteria_met={"1": True}, evidence=["a"]) + + +def test_terminate_rejects_non_boolean_criteria_met_value(workspace): + with pytest.raises(emit.EmitError): + emit.terminate( + workspace, state="FailedUnverifiable", criteria_met={"done": "yes"}, + evidence=[], reason="could not verify", + ) + assert not (workspace / ".loop" / "terminal_state.json").exists() + + +def test_writes_refused_without_a_contract(tmp_path): + with pytest.raises(emit.EmitError): + emit.append_iteration(tmp_path / "nowhere", iteration_id=1, outcome="task_passed") diff --git a/scripts/test_langgraph_recipe.py b/scripts/test_langgraph_recipe.py new file mode 100644 index 0000000..fd700a3 --- /dev/null +++ b/scripts/test_langgraph_recipe.py @@ -0,0 +1,39 @@ +"""B1 acceptance: the LangGraph recipe example runs end-to-end and its emitted +contract passes doctor. Env-guarded: langgraph is a dev dependency of the +example only — the package stays zero-dependency.""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +pytest.importorskip("langgraph") + +REPO_ROOT = Path(__file__).resolve().parent.parent +EXAMPLE = REPO_ROOT / "examples" / "langgraph-emit" / "graph_example.py" + + +def test_recipe_runs_end_to_end_and_passes_doctor(tmp_path): + workspace = tmp_path / "graph-run" + env = dict(os.environ, PYTHONPATH=str(REPO_ROOT)) + proc = subprocess.run( + [sys.executable, "-B", str(EXAMPLE), str(workspace)], + cwd=tmp_path, env=env, capture_output=True, text=True, + ) + assert proc.returncode == 0, proc.stdout + proc.stderr + + doctored = subprocess.run( + [sys.executable, "-B", "-m", "loop", "doctor", str(workspace)], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert doctored.returncode == 0, doctored.stdout + assert json.loads(doctored.stdout)["ok"] is True + + terminal = json.loads((workspace / ".loop" / "terminal_state.json").read_text()) + assert terminal["state"] == "Succeeded" + assert terminal["evidence"], "Succeeded must carry evidence"