Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,23 @@ All notable changes to Pi Graph Core are documented here.

The project follows [Semantic Versioning](https://semver.org/).

## [Unreleased]

### Fixed

- Failed durable runs now print the documented `piw resume` remedy instead of
an internal `run_steps.py` invocation with undocumented flags.
- A run missing its required `--input` now fails before creating a run
directory instead of leaving an empty `initialized` bundle behind.

### Documentation

- Documented concurrent node dispatch (`workers:`), the run directory
location, `judge`, `qa`, `cwd`, `preview`, `retry_jitter`,
`retry_max_delay_seconds`, and the `input.description` requirement.
- Added a positioning comparison (LangGraph/Temporal/Prefect/Make) and a real
run-bundle evidence sample to the README.

## [0.1.0] - 2026-08-23

Initial public alpha release.
Expand Down
39 changes: 35 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ execution contract underneath one.

## Try it in two minutes

Install the `piw` command from the v0.1.0 release:
Install `piw` (the **p**i **w**orkflow command) from the v0.1.0 release:

```bash
python3 -m pip install \
Expand All @@ -43,8 +43,23 @@ piw inspect hello --json
```

The run receipt includes a durable run id and every step's terminal status.
The run directory contains the frozen workflow and input, state projection,
append-only trace, ledger, artifacts, logs, and per-step Git history.
Every run lands in `runs/<workflow>-<timestamp>/` next to `steps.yaml` — an
ordinary directory, readable without a database:

```text
$ ls hello/runs/hello-20260823-232621/
input.txt log.md normalize.md result.md state.json workflow.yaml
ledger.json manifest.json run-owner.json run.lock trace.jsonl .git/
```

It holds the frozen workflow and input, the state projection, an append-only
trace, per-step artifacts, diffable Git history, and a ledger that records what
each node actually cost:

```json
{"id": "analyze", "model": "openai-codex/gpt-5.6-luna", "attempts": 1,
"passed": true, "seconds": 8.3, "input": 5288, "output": 168, "cost": 0.0012592}
```

Prefer an isolated CLI install? Use
[`pipx`](https://pipx.pypa.io/stable/installation/):
Expand All @@ -64,6 +79,8 @@ the [setup guide](docs/SETUP.md).
- Typed JSON output contracts and code-owned conditional routing.
- Mechanical gates that check artifacts or side effects—not model confidence.
- Classified, bounded retries with fixed or exponential delay.
- Concurrent dispatch of dependency-ready nodes (`workers:`, default 4);
`needs:` is the serialization guarantee.
- Immutable per-run input and frozen workflow fingerprints.
- Atomic state, contiguous JSONL trace, one-writer locking, and crash recovery.
- Content-addressed cache, per-step ledger, and inspectable Git history.
Expand Down Expand Up @@ -124,10 +141,24 @@ environment, and executes the installed command.
That evidence supports the kernel behavior tested here. It is not a claim that
arbitrary workflows are safe or that model output is deterministic.

## Why not LangGraph, Temporal, or Make?

| If you need | Use |
|---|---|
| Org-wide orchestration with a server, database, and durable timers | Temporal, Prefect, Dagster |
| Graphs wired in Python around one agent framework | LangGraph |
| File-timestamp incremental builds | Make |
| A local, auditable run contract for agent work | **Pi Graph Core** |

Core's bet is auditability over throughput: gates are shell commands, state is
a readable directory, resume is fingerprint-verified, and there is no server,
daemon, or database — a run is a directory you can `ls`, `diff`, and commit.

## Deliberate boundary

Core does **not** include Studio, batch processing, evaluations, optimization,
scheduling, action catalogs, hosted workers, or the software-factory product.
scheduling, action catalogs, hosted worker fleets, or the software-factory
product.
Those live in the full [Pi Graph](https://github.com/ali-abassi/pi-graph)
platform.

Expand Down
9 changes: 6 additions & 3 deletions docs/ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,12 @@ dependency rules mirror the runner: explicit `needs`, artifact references,

### `pi_graph_core.run_steps`

Owns execution. It selects eligible nodes, invokes the weakest configured
runtime, validates schemas, runs gates, classifies failure, applies bounded
retry policy, evaluates conditions, and records artifacts and ledger entries.
Owns execution. It selects eligible nodes and dispatches dependency-ready
nodes concurrently on a bounded pool (top-level `workers:`, default 4),
invokes the weakest configured runtime, validates schemas, runs gates,
classifies failure, applies bounded retry policy, evaluates conditions, and
records artifacts and ledger entries. Declared dependencies are the only
serialization guarantee between nodes.

### `pi_graph_core.run_bundle`

Expand Down
67 changes: 66 additions & 1 deletion docs/USAGE.md
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,10 @@ steps:
gate: tr '[:lower:]' '[:upper:]' < "$INPUT" | cmp - "$OUT"
```

An `input:` block must declare both `required` and `description`—the
description is the contract a calling agent reads, so validation rejects an
input without one.

## 3. Validate and inspect the graph

```bash
Expand Down Expand Up @@ -77,6 +81,15 @@ piw run review/steps.yaml --input "Review this change" --strict --json
piw run review/steps.yaml --input-file request.md --strict --json
```

Each run lands in `runs/<workflow>-<YYYYMMDD-HHMMSS>/` beside `steps.yaml`.
The receipt's `run` field is that directory's name and is the `RUN_ID`
accepted by `inspect` and `resume`.

Dependency-ready nodes run **concurrently**—top-level `workers:` bounds the
pool (default 4, maximum 16). `needs:` is the only serialization guarantee:
nodes that must not overlap (for example commands mutating the same file) must
be ordered with explicit dependencies.

The input is copied into the run and fingerprinted. Each run freezes the
workflow and records:

Expand Down Expand Up @@ -161,11 +174,63 @@ retries: 2
retry_on: [model_error, schema_failed, gate_failed]
retry_delay_seconds: 1
retry_backoff: exponential
retry_jitter: 0.2 # deterministic ±20% spread per step id + attempt
retry_max_delay_seconds: 60 # backoff cap (default 300)
timeout: 900
```

Retry only declared failure classes. Commands and their child process groups
are terminated when their timeout expires.
are terminated when their timeout expires. Jitter is derived from the step id
and attempt number, so a replayed run computes the same delays.

## Judged improvement loops

A step may attach an LLM judge that scores each candidate and iterates until a
target score or the attempt budget is reached:

```yaml
- id: draft
prompt: Write the release announcement for {input}.
gate: test -s "$OUT"
judge:
prompt: >-
Score this draft 0-10 for clarity.
Return JSON: {"score": N}. Candidate: {out}
score: 8 # minimum passing score (default 8)
max_iters: 3 # attempt budget (supersedes retries when larger)
keep_best: false # true keeps the highest-scoring rejected candidate
```

The judge must return JSON containing a numeric `"score"`; each verdict is
stored as `<step>.judge<N>.md`. A judge never replaces the mechanical `gate`—
the gate still runs on every attempt—and a below-target score is an ordinary
retryable failure class (`judge_below_target`). Judge and schema settings are
part of the cache key, so tightening either invalidates cached artifacts.

## Final QA review

A top-level `qa:` block runs one independent model review over all artifacts
after every step has passed (and again during verification):

```yaml
qa:
prompt: >-
Review these artifacts for contradictions.
Return JSON: {"verdict": "pass"} or {"verdict": "fail"}.
{artifacts}
```

The report is written to `qa.md`; a `fail` verdict fails the run even though
every individual gate passed.

## Other schema keys

- `workers:` (top-level, default 4, maximum 16)—concurrent dependency-ready
nodes; see [section 5](#5-run).
- `cwd:` (top-level, default `.`)—execution directory for command nodes,
resolved relative to `steps.yaml`.
- `preview:` (step)—declarative image paths for visual tooling; never affects
execution.

## Runtime choices

Expand Down
7 changes: 5 additions & 2 deletions examples/README.md
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
# Examples

All examples are ordinary `steps.yaml` files. Read them before running them;
command nodes execute with your user permissions.
command nodes execute with your user permissions. Runs are written to
`examples/runs/`, which is gitignored—the demos never dirty the clone.

## Hello: zero-cost linear workflow

Expand Down Expand Up @@ -31,7 +32,9 @@ piw resume examples/recovery.steps.yaml RUN_ID --json
```

Replace `RUN_ID` with the run id from the first receipt. Remove the approval
file before repeating the demonstration.
file before repeating the demonstration. Both `examples/runs/` and
`examples/approvals/*.ok` are gitignored, so the checkpoint demo leaves no
tracked changes behind.

## Agent review: isolated model node

Expand Down
43 changes: 30 additions & 13 deletions src/pi_graph_core/run_steps.py
Original file line number Diff line number Diff line change
Expand Up @@ -1181,6 +1181,24 @@ def main() -> int:
).resolve()
if not cwd.is_dir():
raise SystemExit(f"workflow cwd not found: {cwd}")
incoming_input = None
if args.input is not None:
incoming_input = args.input.encode("utf-8")
elif args.input_file is not None:
source = args.input_file.expanduser().resolve()
if not source.is_file():
raise SystemExit(f"input file not found: {source}")
incoming_input = source.read_bytes()

input_contract = spec.get("input") if isinstance(spec.get("input"), dict) else {}
if (input_contract.get("required") and incoming_input is None
and not (args.resume or args.from_id or args.verify)
and not (args.run_dir
and (args.run_dir.expanduser().resolve() / "input.txt").is_file())):
# Fail before any run directory exists: a fresh run that never had its
# required input must not leave a half-initialized bundle behind.
raise SystemExit("this workflow requires --input or --input-file")

workflow = spec.get("workflow", args.steps_file.stem)
run_dir = (args.run_dir or args.steps_file.parent / "runs" /
f"{workflow}-{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}").resolve()
Expand All @@ -1204,15 +1222,6 @@ def main() -> int:
attempt += 1
run_dir = base.parent / f"{base.name}-{attempt}"

incoming_input = None
if args.input is not None:
incoming_input = args.input.encode("utf-8")
elif args.input_file is not None:
source = args.input_file.expanduser().resolve()
if not source.is_file():
raise SystemExit(f"input file not found: {source}")
incoming_input = source.read_bytes()

workflow_dir = (args.workflow_dir.expanduser().resolve() if args.workflow_dir else cwd)
input_path = run_dir / "input.txt"
durable_existing = any((run_dir / name).exists()
Expand Down Expand Up @@ -1297,7 +1306,7 @@ def main() -> int:
output.flush()
os.fsync(output.fileno())

input_contract = spec.get("input") if isinstance(spec.get("input"), dict) else {}
# Backstop for --run-dir paths where the early guard could not decide.
if input_contract.get("required") and not input_path.is_file():
if bundle:
bundle.close()
Expand Down Expand Up @@ -1448,9 +1457,17 @@ def dispatch_ready() -> None:
payload={"failed": sorted(failed), "skipped": sorted(skipped)})
bundle.close()
ACTIVE_BUNDLE = None
print(f"\nFAILED step(s) {sorted(failed)}. Fix and rerun with:\n"
f" python3 {shlex.quote(sys.argv[0])} {shlex.quote(str(args.steps_file))} "
f"--from {sorted(failed)[0]} --run-dir {shlex.quote(str(run_dir))}", file=sys.stderr)
if bundle is not None:
# Durable runs resume through the documented public surface.
remedy = (f" piw resume {shlex.quote(str(args.steps_file))} "
f"{shlex.quote(str(run_dir))}")
else:
# Legacy run dirs without a durable bundle only support the
# low-level surgical rerun surface.
remedy = (f" python3 {shlex.quote(sys.argv[0])} {shlex.quote(str(args.steps_file))} "
f"--from {sorted(failed)[0]} --run-dir {shlex.quote(str(run_dir))}")
print(f"\nFAILED step(s) {sorted(failed)}. Fix the cause, then resume with:\n{remedy}",
file=sys.stderr)
return 1

# Report what actually ran: with `when:` guards, "all N passed" would be a lie
Expand Down
14 changes: 14 additions & 0 deletions tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,13 +69,27 @@ def test_recovery_example_resumes_after_human_checkpoint(self) -> None:
self.assertNotEqual(first.returncode, 0)
receipt = json.loads(first.stdout)
self.assertEqual(receipt["steps"]["checkpoint"], "failed")
# The failure remedy must be the documented public surface, not
# an internal run_steps.py invocation.
self.assertIn("piw resume", receipt["error"])
self.assertNotIn("run_steps.py", receipt["error"])
approvals = root / "approvals"
approvals.mkdir()
(approvals / "continue.ok").write_text("approved\n")
resumed = run("resume", str(workflow), receipt["run"])
self.assertEqual(resumed.returncode, 0, resumed.stdout + resumed.stderr)
self.assertEqual(json.loads(resumed.stdout)["status"], "completed")

def test_missing_required_input_leaves_no_run_directory(self) -> None:
with tempfile.TemporaryDirectory() as raw:
workflow = Path(raw) / "steps.yaml"
workflow.write_text((ROOT / "examples" / "hello.steps.yaml").read_text())
result = run("run", str(workflow), "--strict")
self.assertNotEqual(result.returncode, 0)
self.assertIn("requires --input", json.loads(result.stdout)["error"])
self.assertFalse((Path(raw) / "runs").exists(),
"a run without its required input must not create a run dir")

def test_version_and_doctor_are_machine_readable(self) -> None:
version = subprocess.run([sys.executable, str(CLI), "--version"],
capture_output=True, text=True, check=False)
Expand Down