From eedc417586772b4f3991db317c9d3bb7209c7901 Mon Sep 17 00:00:00 2001 From: quantamixsol Date: Wed, 29 Jul 2026 16:14:47 +0200 Subject: [PATCH] CR-010.R6: make the CLI drivable by a scheduler, not just a human MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `graq rebuild` returns an int that Typer discards, so every invocation exits 0 — including "Graph file not found". Measured: a missing graph prints an error in red and exits 0, so an unattended scheduler cannot detect the failure at all. R6's acceptance criterion is that a failed step be distinguishable from an empty delta by exit code alone; today they are both 0. Adds graqle/cli/headless.py as the one place the machine contract is defined: 0 success, 1 failure, 2 usage, 3 empty delta. EMPTY_DELTA is deliberately distinct from SUCCESS so a scheduler can also gate downstream work on whether anything actually changed. RunReport is frozen and derives exit_code from status, so a serialized report can never carry a pair that disagree; errors carry exception TYPE names only, inheriting the PII rule already enforced on .graqle/govern.health.json. --headless and --json stay orthogonal: --headless is about interactivity, --json about format. Coupling them would contradict the standalone --json convention already used at 15+ sites and hand JSON to scripts that never asked for it. Incremental rebuild becomes change-based. rebuild_chunks previously skipped any node that already had chunks, so a node whose source was edited kept stale evidence — a correctness gap, not just a perf one. --incremental compares a stored SHA-256 of the source content. Content, not mtime: git checkouts, CI clones and Docker layer caching all rewrite mtime, which is exactly where a scheduler runs. Graphs with no stored hash fall back to the old behaviour. Backwards compatibility: the corrected exit codes apply only when a machine flag is passed. Bare `graq rebuild` behaves byte-for-byte as before and gets a stderr DeprecationWarning naming the future change. `govern serve --once` without --json is untouched. Sentinel (graq_reason, 2 passes; graq_review is down). Pass 1 raised 4 BLOCKERs; each was reproduced before being fixed: - BLOCKER-1 REAL: a run where every node failed reported empty_delta/exit 3 with errors:[] — "healthy, nothing to do" while nothing worked. Fixed: rebuild_chunks counts failures into Graqle.last_rebuild_failed_nodes and any failure now yields FAILURE. - BLOCKER-2 REFUTED: the two property writes are adjacent in-memory dict assignments with no I/O between them, and the worst case (hash missing) re-chunks next run rather than skipping stale evidence. - BLOCKER-3 REFUTED by measurement: bare rebuild updating 3 nodes exits 0, not 3 — Typer 0.24.1 discards the return value. That IS the defect here. - BLOCKER-4 REFUTED by measurement on win32: NamedTemporaryFile is used as a context manager, so the handle closes before os.replace. Report written, zero orphan tempfiles. Pass 2 found a further real BLOCKER: partial failure (some nodes rebuilt, some failed) fell into the "updated" branch and exited 0. A scheduler routes on the exit code, so that is the same silent success in a smaller costume — any failure is now a FAILURE. graq_predict then surfaced a defect neither pass caught: rebuild_command is also called programmatically by `graq init`, where unfilled typer.Option defaults arrive as truthy OptionInfo sentinels. Those wrongly selected machine mode and then raised TypeError on Path(OptionInfo), which would have broken init's auto-rebuild. Flags are now normalised before use. Verified the direct call returns a plain int again. Also ships cron / Airflow / GitHub Actions reference recipes, as the requirement asks. Tests: 913 passed across tests/test_core + tests/test_cli. The 10 failures in test_gate_install.py / test_g5_vscode_gate_install.py are pre-existing — verified by stashing this branch and reproducing them on untouched private/master; they reference none of these modules. TS-screen: 0 hits. Plan: plan_b48cef6f CR: .gsm/external/Change Requests/CR-010.R6-scheduler-pipeline-contract.md (cherry picked from commit 95813097d9a15fbcf538a8881df6e40af4d13268) --- docs/scheduler-recipes.md | 165 +++++++++ graqle/cli/commands/govern_serve.py | 40 +++ graqle/cli/commands/rebuild.py | 184 ++++++++++- graqle/cli/headless.py | 234 +++++++++++++ graqle/core/graph.py | 87 ++++- tests/test_cli/test_govern_serve.py | 37 +++ tests/test_cli/test_headless_contract.py | 404 +++++++++++++++++++++++ 7 files changed, 1140 insertions(+), 11 deletions(-) create mode 100644 docs/scheduler-recipes.md create mode 100644 graqle/cli/headless.py create mode 100644 tests/test_cli/test_headless_contract.py diff --git a/docs/scheduler-recipes.md b/docs/scheduler-recipes.md new file mode 100644 index 00000000..7c02565c --- /dev/null +++ b/docs/scheduler-recipes.md @@ -0,0 +1,165 @@ +# Running GraQle on a scheduler + +GraQle's rebuild and anchoring commands are designed to run unattended. This page +is the reference contract plus copy-paste recipes for cron, DAG platforms, and CI +runners. + +## The contract + +Pass any of `--headless`, `--json`, or `--report-json` to opt a command into the +scheduler contract. You then get **meaningful exit codes** and a **machine-readable +run report**. + +| Exit code | Meaning | What a scheduler should do | +|---|---|---| +| `0` | Success — work was performed, nothing failed | Continue; downstream steps may run | +| `1` | Failure — the step did not complete, **or any node failed** | Alert / retry | +| `2` | Usage error — bad invocation | Fix the job definition; do not retry | +| `3` | Empty delta — ran fine, nothing to do | Continue, but skip downstream work | + +A **partial** failure (some nodes rebuilt, some failed) exits `1`, not `0`. Check +`counters.nodes_updated` and `counters.nodes_failed` in the report to see how much +got through. + +`3` is deliberately separate from `0`. A failed step is distinguishable from an +empty delta by exit code alone, and so is "did work actually happen?" — which is +what lets you gate an expensive downstream step on whether the graph changed. + +### Flags + +| Flag | Effect | +|---|---| +| `--headless` | Never prompt; no colour or progress output. Keeps stdout clean for parsing. | +| `--json` | Write the run report to stdout. | +| `--report-json PATH` | Write the run report to a file (atomically). | +| `--incremental` | `rebuild` only: rebuild nodes whose **source content changed**, not just those missing chunks. | + +`--headless` and `--json` are independent on purpose: `--headless` is about +interactivity, `--json` is about output format. Combine them for a scheduler. + +### Run report + +```json +{ + "schema_version": "1", + "command": "rebuild", + "status": "empty_delta", + "exit_code": 3, + "started_at": "2026-07-29T13:41:32+00:00", + "duration_s": 0.42, + "counters": { + "nodes_total": 1200, + "nodes_updated": 0, + "nodes_with_chunks_after": 1200, + "nodes_with_chunks_before": 1200 + }, + "errors": [] +} +``` + +`errors` carries exception **type names only** — never messages, paths, or +credentials — so a report is safe to archive as a build artefact. + +### Incremental rebuilds + +Without `--incremental`, `graq rebuild` only fills in *missing* chunks: a node +whose source file changed but which already has chunks is skipped, so its +evidence goes stale. `--incremental` compares a stored SHA-256 of the source +content and rebuilds what actually changed. + +Detection is content-based, not mtime-based, because git checkouts, CI clones, +and Docker layer caching all rewrite mtime — exactly the environments a scheduler +runs in. A graph built before this feature has no stored hash and simply falls +back to the previous behaviour on its first run. + +--- + +## cron + +```bash +#!/usr/bin/env bash +# /etc/cron.daily/graqle-rebuild +set -euo pipefail + +cd /srv/myproject +REPORT=/var/log/graqle/rebuild-$(date +%F).json + +set +e +graq rebuild --headless --json --incremental --report-json "$REPORT" +code=$? +set -e + +case $code in + 0) logger -t graqle "rebuild: graph updated" ;; + 3) logger -t graqle "rebuild: no changes" ;; + 2) logger -t graqle -p user.err "rebuild: bad invocation"; exit 2 ;; + *) logger -t graqle -p user.err "rebuild: FAILED (exit $code)"; exit 1 ;; +esac + +# Anchor whatever the rebuild produced. +graq govern serve --once --json --report-json "${REPORT%.json}-anchor.json" +``` + +`set +e` around the call matters: under `set -e` a non-zero exit (including the +perfectly healthy `3`) would abort the script before you can branch on it. + +## Airflow + +```python +from airflow.decorators import task +from airflow.exceptions import AirflowSkipException +import subprocess, json + +@task +def rebuild_graph() -> dict: + proc = subprocess.run( + ["graq", "rebuild", "--headless", "--json", "--incremental"], + capture_output=True, text=True, cwd="/srv/myproject", + ) + if proc.returncode == 2: + raise ValueError(f"bad invocation: {proc.stderr}") + if proc.returncode not in (0, 3): + raise RuntimeError(f"rebuild failed (exit {proc.returncode})") + + report = json.loads(proc.stdout) + if proc.returncode == 3: + # Nothing changed — skip the downstream anchor/publish tasks. + raise AirflowSkipException("graph unchanged") + return report +``` + +Mapping exit `3` onto `AirflowSkipException` is the idiomatic way to express +"healthy, but there was nothing to do" — it keeps the DAG green while correctly +short-circuiting downstream work. + +## GitHub Actions + +```yaml +- name: Rebuild knowledge graph + id: rebuild + run: | + set +e + graq rebuild --headless --json --incremental --report-json rebuild.json + echo "exit_code=$?" >> "$GITHUB_OUTPUT" + set -e + +- name: Fail on rebuild error + if: steps.rebuild.outputs.exit_code == '1' || steps.rebuild.outputs.exit_code == '2' + run: exit 1 + +- name: Publish (only when the graph actually changed) + if: steps.rebuild.outputs.exit_code == '0' + run: graq govern serve --once --json + +- uses: actions/upload-artifact@v4 + if: always() + with: + name: graqle-run-report + path: rebuild.json +``` + +## Idempotency + +Re-running any of these is safe. A second `graq rebuild --incremental` over an +unchanged tree does no work and exits `3`, so an overlapping or retried job +cannot corrupt the graph or double-anchor. diff --git a/graqle/cli/commands/govern_serve.py b/graqle/cli/commands/govern_serve.py index 97c7aae2..08642fce 100644 --- a/graqle/cli/commands/govern_serve.py +++ b/graqle/cli/commands/govern_serve.py @@ -26,6 +26,7 @@ import os import signal import sys +import time from pathlib import Path from typing import Any @@ -273,6 +274,15 @@ def govern_serve( None, "--tick-seconds", help="Override the loop interval (default: attestation.batch_max_seconds).", ), + json_out: bool = typer.Option( + False, "--json", + help="With --once: emit a machine-readable JSON run report on stdout " + "(CR-010.R6 scheduler contract).", + ), + report_json: str | None = typer.Option( + None, "--report-json", + help="With --once: write the JSON run report to this path (atomic).", + ), ) -> None: """Run the AnchoringWorker as a long-lived service. @@ -310,8 +320,38 @@ def govern_serve( if once: # One-shot mode: run a single tick. No signal handlers (the process exits # immediately after) and no run-loop lifecycle. + from graqle.cli.headless import ( + RunReport, + RunStatus, + emit_and_exit, + utc_now_iso, + ) + + started_at = utc_now_iso() + t_start = time.monotonic() committed = worker.tick() h = worker.health() + + if json_out or report_json: + # CR-010.R6: nothing committed is an EMPTY_DELTA (exit 3), not a + # failure — a cron catch-up that finds an empty queue succeeded. + emit_and_exit( + RunReport( + command="govern serve --once", + status=RunStatus.SUCCESS if committed else RunStatus.EMPTY_DELTA, + started_at=started_at, + duration_s=time.monotonic() - t_start, + counters={ + "committed": int(committed), + "backfill_count": int(h.backfill_count), + "replay_queue_depth": int(h.replay_queue_depth), + }, + ), + json_out=json_out, + report_path=report_json, + ) + + # Legacy path: unchanged human output, still exits 0. console.print( f"[green]✓ once: committed={committed} backfill={h.backfill_count} " f"queue_depth={h.replay_queue_depth}[/green]" diff --git a/graqle/cli/commands/rebuild.py b/graqle/cli/commands/rebuild.py index 7edebc56..ef9ad1ad 100644 --- a/graqle/cli/commands/rebuild.py +++ b/graqle/cli/commands/rebuild.py @@ -12,6 +12,14 @@ graq rebuild --graph my.json # specify a different graph graq rebuild --re-embed # dry-run: show what re-embed would do (safe) graq rebuild --re-embed --force # actually re-embed all nodes (writes to disk) + +Scheduler use (CR-010.R6): + graq rebuild --headless --json # machine contract: JSON report + exit code + graq rebuild --headless --json --incremental # rebuild only nodes whose source CHANGED + graq rebuild --json --report-json run.json # also archive the report + +Exit codes apply when any of --headless/--json/--report-json is passed: + 0 success (work done) · 1 failure · 2 usage error · 3 empty delta (nothing to do) """ # ── graqle:intelligence ── @@ -29,14 +37,51 @@ import time from pathlib import Path +import typer + logger = logging.getLogger("graqle.cli.rebuild") +def _flag(value: object) -> bool: + """Coerce a possibly-unfilled Typer option to a plain bool. + + Typer fills these in when the command is invoked from the CLI, but a direct + Python call (``graq init`` auto-rebuilds this way) leaves the ``OptionInfo`` + sentinel in place. ``OptionInfo`` is truthy, so a naive ``bool()`` would read + an unsupplied flag as enabled. + """ + return value is True + + def rebuild_command( graph_path: str = "graqle.json", config_path: str = "graqle.yaml", force: bool = False, re_embed: bool = False, + incremental: bool = typer.Option( + False, + "--incremental", + help="Rebuild only nodes whose source content CHANGED (hash-based), " + "not merely those missing chunks.", + ), + headless: bool = typer.Option( + False, + "--headless", + help="Non-interactive: never prompt, no colour/progress output. " + "Enables the scheduler exit-code contract.", + ), + json_out: bool = typer.Option( + False, + "--json", + help="Emit a machine-readable JSON run report on stdout. " + "Enables the scheduler exit-code contract.", + ), + report_json: str | None = typer.Option( + None, + "--report-json", + help="Write the JSON run report to this path (atomic). " + "Enables the scheduler exit-code contract.", + ), ) -> int: """Rebuild chunks for all nodes in the KG. @@ -44,34 +89,123 @@ def rebuild_command( would happen without writing anything to disk. Pass force=True to commit. Returns the number of nodes updated. + + Machine contract (CR-010.R6) + ---------------------------- + Passing any of *headless*, *json_out* or *report_json* opts this invocation + into the scheduler contract: a :class:`RunReport` is produced and the process + exits 0/1/2/3 (see :mod:`graqle.cli.headless`). + + Why opt-in: Typer discards a command's return value, so today **every** + ``graq rebuild`` exits 0 — including the "graph file not found" path. Fixing + that unconditionally would change the exit code of a published CLI under + anyone's existing cron entry. Callers who pass a machine flag are new by + definition and have no legacy expectation, so they get the corrected codes + immediately; bare invocations keep exiting 0 and get a DeprecationWarning + naming the release that changes it. """ + from graqle.cli.headless import ( + RunReport, + RunStatus, + emit_and_exit, + utc_now_iso, + ) + + # `rebuild_command` is also called PROGRAMMATICALLY (graq init's auto-rebuild, + # init.py). Such a caller does not go through Typer, so the unfilled + # ``typer.Option(...)`` defaults arrive as OptionInfo objects — which are + # truthy, and would wrongly select machine mode and then blow up on + # Path(OptionInfo). Normalise first: an OptionInfo means "not supplied". + headless = _flag(headless) + json_out = _flag(json_out) + incremental = _flag(incremental) + report_json = report_json if isinstance(report_json, (str, Path)) else None + + # Any machine flag selects the scheduler contract. + machine_mode = bool(headless or json_out or report_json) + started_at = utc_now_iso() + t_start = time.monotonic() + + def _finish( + status: RunStatus, + counters: dict[str, int], + errors: tuple[str, ...] = (), + ) -> None: + """Emit the run report and exit. Only called in machine mode.""" + emit_and_exit( + RunReport( + command="rebuild", + status=status, + started_at=started_at, + duration_s=time.monotonic() - t_start, + counters=counters, + errors=errors, + ), + json_out=json_out, + report_path=report_json, + ) + try: from rich.console import Console console = Console() except ImportError: console = None + # Under --headless: no colour, no progress rendering, and stdout stays clean + # so a scheduler parsing --json output is never fed decorative text. def _print(msg: str) -> None: + if headless: + return if console: console.print(msg) else: print(msg) + if not machine_mode: + # Advance notice: this path is scheduled to start exiting non-zero. + # stderr, so it can never contaminate stdout that a script is parsing. + import warnings + + warnings.warn( + "graq rebuild currently exits 0 even when it fails (e.g. a missing " + "graph file). A future release will return a meaningful exit code " + "for bare invocations. Pass --headless/--json today to opt into the " + "scheduler contract (0 success, 1 failure, 2 usage, 3 empty delta).", + DeprecationWarning, + stacklevel=2, + ) + gp = Path(graph_path) if not gp.exists(): _print(f"[red]Graph file not found: {graph_path}[/red]") _print("Run [cyan]graq init[/cyan] first to create a graph.") + if machine_mode: + _finish( + RunStatus.FAILURE, + {"nodes_total": 0, "nodes_updated": 0}, + errors=("GraphFileNotFound",), + ) return 0 from graqle.config.settings import GraqleConfig from graqle.core.graph import Graqle - # Load config - cp = Path(config_path) - config = GraqleConfig.from_yaml(str(cp)) if cp.exists() else GraqleConfig.default() - - # Load graph - graph = Graqle.from_json(str(gp), config=config) + # Load config + graph. A corrupt or unreadable graph is a hard failure: in + # machine mode it must surface as exit 1, never as an unhandled traceback a + # scheduler would record as a crash with no report. + try: + cp = Path(config_path) + config = GraqleConfig.from_yaml(str(cp)) if cp.exists() else GraqleConfig.default() + graph = Graqle.from_json(str(gp), config=config) + except Exception as exc: + _print(f"[red]Could not load graph: {type(exc).__name__}[/red]") + if machine_mode: + _finish( + RunStatus.FAILURE, + {"nodes_total": 0, "nodes_updated": 0}, + errors=(type(exc).__name__,), + ) + raise _print(f"[bold cyan]Rebuilding chunks[/bold cyan] for {len(graph.nodes)} nodes...") if force: @@ -86,7 +220,7 @@ def _print(msg: str) -> None: ) # Rebuild - updated = graph.rebuild_chunks(force=force) + updated = graph.rebuild_chunks(force=force, incremental=incremental) # Count after after_count = sum( @@ -148,6 +282,42 @@ def _print(msg: str) -> None: total_time = time.monotonic() - t0 _print(f"\n [bold]Total rebuild time: {total_time:.1f}s[/bold]") + if machine_mode: + # EMPTY_DELTA is the whole point of the contract: "ran fine, nothing to + # do" must be distinguishable from "failed" by exit code alone, and also + # from "did work" — so a scheduler can gate a downstream step on whether + # the graph actually changed. + # + # Critically, a run in which every node raised also produces updated == 0. + # Reporting that as EMPTY_DELTA would tell the scheduler "all healthy, + # nothing to do" while nothing worked at all, so nodes_failed decides: + # any failure with no successful update is a FAILURE, not an empty delta. + # Any failure at all is a FAILURE, whether or not other nodes succeeded. + # A partial failure reported as exit 0 would be the same silent-success + # bug in a smaller costume: a scheduler routes on the exit code, so + # "most of the graph rebuilt, some of it is broken" must not read as + # healthy. nodes_failed/nodes_updated in the report tell the operator + # how much got through. + failed = int(getattr(graph, "last_rebuild_failed_nodes", 0)) + if failed: + status = RunStatus.FAILURE + elif updated: + status = RunStatus.SUCCESS + else: + status = RunStatus.EMPTY_DELTA + + _finish( + status, + { + "nodes_total": len(graph.nodes), + "nodes_updated": updated, + "nodes_failed": failed, + "nodes_with_chunks_before": before_count, + "nodes_with_chunks_after": after_count, + }, + errors=("ChunkRebuildFailed",) if failed else (), + ) + return updated diff --git a/graqle/cli/headless.py b/graqle/cli/headless.py new file mode 100644 index 00000000..c981ea9e --- /dev/null +++ b/graqle/cli/headless.py @@ -0,0 +1,234 @@ +"""Scheduler-grade CLI contract — exit codes and machine-readable run reports. + +CR-010.R6. Enterprise schedulers (cron, Airflow/DAG platforms, CI runners) drive +GraQle unattended. They cannot read rich text: they branch on the **exit code** and +parse a **JSON run report**. This module is the single place both are defined, so +every command that opts in speaks one contract. + +The contract +------------ +===== =================================================================== +Code Meaning +===== =================================================================== +0 SUCCESS — the command ran and *work was performed* +1 FAILURE — a hard error; the step did not complete +2 USAGE — the invocation itself was wrong (bad flags, no TTY) +3 EMPTY_DELTA — the command ran, there was *nothing to do*, no error +===== =================================================================== + +``EMPTY_DELTA`` is deliberately distinct from ``SUCCESS``. R6's acceptance +criterion is that "a failed step is distinguishable from an empty delta by exit +code alone"; collapsing both onto 0 would also make "did work actually happen?" +unanswerable without parsing stdout, which defeats the point of an exit-code +contract (a scheduler cannot gate a downstream deploy on "the graph changed"). + +Design notes +------------ +* ``exit_code`` is **derived** from ``status`` (never set independently), so the + two can never disagree inside a serialized report. +* Reports are PII-safe by construction: ``errors`` carries exception **type** + names only — never messages, paths, or credential material. This mirrors the + rule already enforced on ``.graqle/govern.health.json`` + (``govern_serve.py`` — "counts, ticks, queue depth, exception TYPE names"). +* ``--report-json`` writes atomically (tempfile in the destination directory + + ``os.replace``) and cleans up an orphaned tempfile on failure — the same + pattern the governance health snapshot uses, for the same reason: a scheduler + must never read a half-written file. +""" + +# ── graqle:intelligence ── +# module: graqle.cli.headless +# risk: LOW (new module; no existing consumers) +# consumers: cli.commands.rebuild, cli.commands.govern_serve +# dependencies: __future__, dataclasses, datetime, enum, json, os, sys, tempfile +# constraints: report payloads must stay PII-safe (exception type names only) +# ── /graqle:intelligence ── + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime, timezone +from enum import Enum, IntEnum +from pathlib import Path +from typing import Any, NoReturn + +__all__ = [ + "REPORT_SCHEMA_VERSION", + "ExitCode", + "RunReport", + "RunStatus", + "HeadlessPromptError", + "emit_and_exit", + "guard_no_prompt", + "utc_now_iso", +] + +#: Report schema version. Matches the existing SDK idiom (``compliance.py`` emits +#: ``"schema_version": "1"``). Bump only on a breaking payload change. +REPORT_SCHEMA_VERSION = "1" + + +class ExitCode(IntEnum): + """Process exit codes for scheduler-driven invocations.""" + + SUCCESS = 0 + FAILURE = 1 + USAGE = 2 + EMPTY_DELTA = 3 + + +class RunStatus(str, Enum): + """Outcome of a run. ``ExitCode`` is derived from this — never set directly.""" + + SUCCESS = "success" + EMPTY_DELTA = "empty_delta" + FAILURE = "failure" + USAGE_ERROR = "usage_error" + + @property + def exit_code(self) -> ExitCode: + """The exit code this status maps to. Total function — no default branch.""" + return _STATUS_TO_EXIT[self] + + +_STATUS_TO_EXIT: dict[RunStatus, ExitCode] = { + RunStatus.SUCCESS: ExitCode.SUCCESS, + RunStatus.EMPTY_DELTA: ExitCode.EMPTY_DELTA, + RunStatus.FAILURE: ExitCode.FAILURE, + RunStatus.USAGE_ERROR: ExitCode.USAGE, +} + + +class HeadlessPromptError(RuntimeError): + """Raised when a code path would prompt while running under ``--headless``. + + Scheduler runs have no TTY. Blocking on input would hang the job until the + platform's timeout fires, which reads as a stall rather than a failure. We + fail fast with ``USAGE`` instead. + """ + + +def utc_now_iso() -> str: + """Timezone-aware UTC timestamp, ISO-8601, second precision.""" + return datetime.now(timezone.utc).replace(microsecond=0).isoformat() + + +@dataclass(frozen=True) +class RunReport: + """A machine-readable record of one command invocation. + + Frozen: a report describes a run that already happened, so it is a value + object. Build it once at the end of the run and hand it to + :func:`emit_and_exit`. + """ + + command: str + status: RunStatus + started_at: str + duration_s: float + counters: Mapping[str, int] = field(default_factory=dict) + errors: tuple[str, ...] = () + schema_version: str = REPORT_SCHEMA_VERSION + + @property + def exit_code(self) -> ExitCode: + """Derived from :attr:`status`. Cannot disagree with it.""" + return self.status.exit_code + + def to_dict(self) -> dict[str, Any]: + """Serializable payload. Key order is stable for golden-file tests.""" + return { + "schema_version": self.schema_version, + "command": self.command, + "status": self.status.value, + "exit_code": int(self.exit_code), + "started_at": self.started_at, + "duration_s": round(float(self.duration_s), 3), + "counters": {str(k): int(v) for k, v in sorted(self.counters.items())}, + "errors": list(self.errors), + } + + def to_json(self) -> str: + """Compact-but-readable JSON. ``sort_keys=False`` preserves :meth:`to_dict` order.""" + return json.dumps(self.to_dict(), indent=2, sort_keys=False) + + +def guard_no_prompt(headless: bool, what: str) -> None: + """Refuse to prompt under ``--headless``. + + Call this immediately before any interactive read. ``rebuild`` and + ``govern serve`` have no prompts today, so this is a forward-looking guard: + it makes a future prompt fail loudly in CI instead of hanging a scheduled job. + """ + if headless: + raise HeadlessPromptError( + f"{what} requires interactive input, which is unavailable under --headless" + ) + + +def _write_report_atomically(report: RunReport, destination: Path) -> None: + """Write *report* to *destination* atomically. + + Tempfile lives in the destination directory so ``os.replace`` is atomic on + every platform (a cross-device rename is not). An orphaned tempfile is + removed on failure so repeated errors cannot fill the directory. + """ + destination.parent.mkdir(parents=True, exist_ok=True) + tmp_path: str | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(destination.parent), + prefix=destination.name + ".", + suffix=".tmp", + delete=False, + ) as fh: + fh.write(report.to_json()) + tmp_path = fh.name + os.replace(tmp_path, destination) + tmp_path = None + finally: + if tmp_path is not None: + try: + os.unlink(tmp_path) + except OSError: + pass + + +def emit_and_exit( + report: RunReport, + *, + json_out: bool = False, + report_path: Path | str | None = None, +) -> NoReturn: + """Emit *report* per the caller's flags, then exit with its derived code. + + ``json_out`` and ``report_path`` are independent: stdout JSON is for a + scheduler reading the pipe, the file is for one archiving the artefact. + + A failure to persist the report never masks the run's own outcome — the + write error is surfaced on stderr and the original exit code still stands. + A scheduler must see why the *step* failed, not why the bookkeeping did. + """ + import typer + + if json_out: + sys.stdout.write(report.to_json() + "\n") + sys.stdout.flush() + + if report_path is not None: + try: + _write_report_atomically(report, Path(report_path)) + except OSError as exc: + sys.stderr.write( + f"warning: could not write run report to {report_path}: " + f"{type(exc).__name__}\n" + ) + + raise typer.Exit(int(report.exit_code)) diff --git a/graqle/core/graph.py b/graqle/core/graph.py index 2a748a3a..2b290cb7 100644 --- a/graqle/core/graph.py +++ b/graqle/core/graph.py @@ -33,6 +33,28 @@ logger = logging.getLogger("graqle") +# ── CR-010.R6: change-based chunk rebuild ──────────────────── +#: Node property holding the SHA-256 of the source content at the last chunk +#: rebuild. Leading underscore marks it SDK-internal. Additive — graphs written +#: by earlier releases simply lack it and fall back to presence-based selection. +_CHUNK_HASH_PROPERTY = "_chunk_content_hash" + + +def _content_hash(content: str) -> str: + """SHA-256 of *content*, hex-encoded. + + Content-based rather than mtime-based on purpose: git checkouts, CI clones + and Docker layer caching all rewrite mtime, so mtime is unreliable in + exactly the environments a scheduler runs in. + """ + import hashlib + + # errors="replace", not "ignore": dropping undecodable bytes would let two + # files that differ only in those bytes hash identically, so a genuinely + # modified file would be skipped as unchanged. + return hashlib.sha256(content.encode("utf-8", errors="replace")).hexdigest() + + # ── (v0.47.3): batch error fallback helper ─────────────────── # # graqle.core.graph.areason_batch() previously constructed a ReasoningResult @@ -588,6 +610,12 @@ class Graqle: provides reasoning capabilities through distributed model agents. """ + #: Nodes that raised during the most recent :meth:`rebuild_chunks` call. + #: Class-level default so the attribute is always readable, even before a + #: rebuild has run (CR-010.R6 — the scheduler contract reads this to tell an + #: empty delta apart from a run in which every node failed). + last_rebuild_failed_nodes: int = 0 + def __init__( self, nodes: dict[str, CogniNode] | None = None, @@ -1082,20 +1110,50 @@ def to_neo4j( # --- Public chunk management --- - def rebuild_chunks(self, force: bool = False) -> int: + def rebuild_chunks(self, force: bool = False, incremental: bool = False) -> int: """Rebuild chunks for all nodes from their source files. Use this after ``graq init`` or when source files have changed. By default only fills in missing chunks; set *force=True* to re-read even nodes that already have chunks. - Returns the number of nodes updated. + Selection modes (CR-010.R6): + + ``force=False, incremental=False`` (default, unchanged) + *Presence*-based: skip any node that already has chunks. Cannot + notice that a source file changed — a node whose file was edited + keeps its stale chunks. + ``incremental=True`` + *Change*-based: rebuild a node when its source content hash differs + from the hash recorded at the last rebuild. A node with no recorded + hash (any graph built before this release) falls back to the + presence-based rule, so an existing graph degrades gracefully rather + than triggering a full rebuild. + ``force=True`` + Rebuild everything. Wins over *incremental*. + + The hash is content-based (SHA-256), not mtime-based: git checkouts, CI + clones and Docker layer caching all rewrite mtime, and those are exactly + the environments a scheduler runs in. + + Returns the number of nodes updated. The count of nodes that raised while + being processed is recorded on :attr:`last_rebuild_failed_nodes` — a node + failing is not fatal to the run (one unreadable file should not abort a + 10,000-node rebuild), but a caller reporting to a scheduler MUST be able + to tell "nothing needed doing" from "every node failed". Both look like + ``updated == 0`` otherwise. """ from pathlib import Path as _P updated = 0 + failed = 0 for node in self.nodes.values(): - if not force and node.properties.get("chunks"): + has_chunks = bool(node.properties.get("chunks")) + stored_hash = node.properties.get(_CHUNK_HASH_PROPERTY) + + # Cheap skip first: only `incremental` needs to read the file to + # decide, and only when a prior hash exists to compare against. + if not force and has_chunks and not (incremental and stored_hash): continue file_path = ( @@ -1113,6 +1171,12 @@ def rebuild_chunks(self, force: bool = False) -> int: if not content.strip(): continue + content_hash = _content_hash(content) + + # Change-based skip: content is byte-identical to last rebuild. + if not force and incremental and stored_hash == content_hash and has_chunks: + continue + suffix = fp.suffix.lower() if suffix in (".py", ".js", ".ts", ".tsx", ".jsx"): chunks = self._chunk_source_code(content) @@ -1121,11 +1185,26 @@ def rebuild_chunks(self, force: bool = False) -> int: if chunks: node.properties["chunks"] = chunks + # Record the hash on every successful rebuild — including + # non-incremental ones — so the first `--incremental` run + # after an ordinary rebuild already has a baseline. + node.properties[_CHUNK_HASH_PROPERTY] = content_hash updated += 1 except Exception: + # One bad file must not abort the whole rebuild, but it must not + # vanish either — see last_rebuild_failed_nodes. + failed += 1 continue - logger.info("rebuild_chunks: updated %d nodes (force=%s)", updated, force) + self.last_rebuild_failed_nodes = failed + + logger.info( + "rebuild_chunks: updated %d nodes, %d failed (force=%s, incremental=%s)", + updated, + failed, + force, + incremental, + ) return updated # --- Node Enrichment & Validation --- diff --git a/tests/test_cli/test_govern_serve.py b/tests/test_cli/test_govern_serve.py index 138ccd34..bd71fd48 100644 --- a/tests/test_cli/test_govern_serve.py +++ b/tests/test_cli/test_govern_serve.py @@ -95,6 +95,43 @@ def test_once_runs_one_tick_and_exits(self, tmp_path, monkeypatch): worker.run.assert_not_called() assert "once: committed=3" in res.output + def test_once_json_emits_a_run_report(self, tmp_path, monkeypatch): + """CR-010.R6: --once --json speaks the scheduler contract.""" + import json as _json + + monkeypatch.chdir(tmp_path) + cfg = _disabled_config_yaml(tmp_path) + worker = _fake_worker(committed=3) + + with patch( + "graqle.cli.commands.govern_serve._build_worker", return_value=worker + ): + res = _invoke(["--config", str(cfg), "--once", "--json"]) + + assert res.exit_code == 0, res.output + payload = _json.loads(res.stdout) + assert payload["command"] == "govern serve --once" + assert payload["status"] == "success" + assert payload["counters"]["committed"] == 3 + + def test_once_json_reports_empty_delta_when_nothing_committed( + self, tmp_path, monkeypatch + ): + """An empty queue is a successful catch-up, not a failure — exit 3, not 1.""" + import json as _json + + monkeypatch.chdir(tmp_path) + cfg = _disabled_config_yaml(tmp_path) + worker = _fake_worker(committed=0) + + with patch( + "graqle.cli.commands.govern_serve._build_worker", return_value=worker + ): + res = _invoke(["--config", str(cfg), "--once", "--json"]) + + assert res.exit_code == 3, res.output + assert _json.loads(res.stdout)["status"] == "empty_delta" + # -- fail-closed / misconfig surfaces --------------------------------------- diff --git a/tests/test_cli/test_headless_contract.py b/tests/test_cli/test_headless_contract.py new file mode 100644 index 00000000..c7b223c4 --- /dev/null +++ b/tests/test_cli/test_headless_contract.py @@ -0,0 +1,404 @@ +"""CR-010.R6 — scheduler-grade CLI contract. + +Covers the machine contract itself (``graqle.cli.headless``) and its two current +consumers (``graq rebuild``, ``graq govern serve --once``). + +The load-bearing assertion in this file is that **a failed step is +distinguishable from an empty delta by exit code alone** — R6's literal +acceptance criterion — and that opting into the contract does not change the +behaviour of any existing bare invocation. +""" + +from __future__ import annotations + +import json +import os +import stat +import sys +from pathlib import Path + +import pytest +from typer.testing import CliRunner + +from graqle.cli.headless import ( + REPORT_SCHEMA_VERSION, + ExitCode, + HeadlessPromptError, + RunReport, + RunStatus, + guard_no_prompt, + utc_now_iso, +) + +runner = CliRunner() + + +def _report(status: RunStatus, **kw) -> RunReport: + return RunReport( + command=kw.pop("command", "rebuild"), + status=status, + started_at=kw.pop("started_at", utc_now_iso()), + duration_s=kw.pop("duration_s", 0.5), + **kw, + ) + + +# ── the contract itself ────────────────────────────────────────────────────── + + +class TestExitCodeContract: + def test_the_four_codes_are_the_documented_values(self): + assert int(ExitCode.SUCCESS) == 0 + assert int(ExitCode.FAILURE) == 1 + assert int(ExitCode.USAGE) == 2 + assert int(ExitCode.EMPTY_DELTA) == 3 + + def test_failure_is_distinguishable_from_empty_delta(self): + """R6's literal acceptance criterion.""" + assert ExitCode.FAILURE != ExitCode.EMPTY_DELTA + + def test_empty_delta_is_also_distinguishable_from_success(self): + """Otherwise a scheduler cannot gate a downstream step on 'did work happen?'.""" + assert ExitCode.EMPTY_DELTA != ExitCode.SUCCESS + + @pytest.mark.parametrize("status", list(RunStatus)) + def test_every_status_maps_to_an_exit_code(self, status): + """Total function — a new status cannot silently fall through to 0.""" + assert isinstance(status.exit_code, ExitCode) + + def test_exit_code_is_derived_not_settable(self): + """The report cannot carry a status/exit_code pair that disagree.""" + assert "exit_code" not in RunReport.__dataclass_fields__ + assert _report(RunStatus.FAILURE).exit_code == ExitCode.FAILURE + + def test_report_is_frozen(self): + with pytest.raises(Exception): + _report(RunStatus.SUCCESS).status = RunStatus.FAILURE # type: ignore[misc] + + +class TestRunReportSerialization: + def test_payload_shape_and_schema_version(self): + payload = json.loads(_report(RunStatus.SUCCESS, counters={"n": 2}).to_json()) + assert payload["schema_version"] == REPORT_SCHEMA_VERSION + assert payload["status"] == "success" + assert payload["exit_code"] == 0 + assert payload["counters"] == {"n": 2} + assert payload["errors"] == [] + + def test_serialized_exit_code_matches_status(self): + for status in RunStatus: + payload = json.loads(_report(status).to_json()) + assert payload["exit_code"] == int(status.exit_code) + + def test_counters_are_sorted_for_stable_diffs(self): + payload = json.loads(_report(RunStatus.SUCCESS, counters={"z": 1, "a": 2}).to_json()) + assert list(payload["counters"]) == ["a", "z"] + + def test_output_is_valid_json(self): + json.loads(_report(RunStatus.EMPTY_DELTA).to_json()) + + +# ── adversarial / negative ─────────────────────────────────────────────────── + + +class TestHeadlessGuard: + def test_guard_raises_under_headless(self): + with pytest.raises(HeadlessPromptError): + guard_no_prompt(True, "API key") + + def test_guard_is_a_noop_when_interactive(self): + guard_no_prompt(False, "API key") # must not raise + + +class TestReportIsPiiSafe: + def test_errors_carry_type_names_not_messages(self): + """A report may be archived by a scheduler — it must never carry secrets. + + Mirrors the rule already enforced on .graqle/govern.health.json. + """ + secret = "sk-live-DEADBEEF-do-not-leak" + try: + raise ValueError(f"auth failed for token {secret}") + except ValueError as exc: + report = _report(RunStatus.FAILURE, errors=(type(exc).__name__,)) + + rendered = report.to_json() + assert secret not in rendered + assert "ValueError" in rendered + + +class TestReportPersistence: + def test_report_json_is_written_and_parseable(self, tmp_path): + from graqle.cli.headless import _write_report_atomically + + dest = tmp_path / "nested" / "run.json" + _write_report_atomically(_report(RunStatus.SUCCESS, counters={"n": 1}), dest) + + assert json.loads(dest.read_text(encoding="utf-8"))["counters"] == {"n": 1} + + def test_no_orphan_tempfile_is_left_behind(self, tmp_path): + from graqle.cli.headless import _write_report_atomically + + dest = tmp_path / "run.json" + _write_report_atomically(_report(RunStatus.SUCCESS), dest) + + leftovers = [p.name for p in tmp_path.iterdir() if p.name.endswith(".tmp")] + assert leftovers == [] + + def test_unwritable_report_path_does_not_mask_the_run_outcome(self, tmp_path, capsys): + """A bookkeeping failure must not hide why the step itself failed.""" + import typer + + from graqle.cli.headless import emit_and_exit + + # A directory where the report file should be → the write must fail. + blocked = tmp_path / "run.json" + blocked.mkdir() + + with pytest.raises(typer.Exit) as excinfo: + emit_and_exit(_report(RunStatus.FAILURE), json_out=False, report_path=blocked) + + assert excinfo.value.exit_code == int(ExitCode.FAILURE) + assert "could not write run report" in capsys.readouterr().err + + +# ── graq rebuild: the end-to-end contract ──────────────────────────────────── + + +def _seed_graph(tmp_path: Path, body: str) -> tuple[Path, Path]: + src = tmp_path / "src_a.py" + src.write_text(body, encoding="utf-8") + graph = tmp_path / "g.json" + graph.write_text( + json.dumps( + { + "directed": True, + "multigraph": False, + "graph": {}, + "nodes": [ + { + "id": "mod::a", + "label": "a", + "type": "PythonModule", + "description": "sample module a", + "properties": {"file_path": str(src)}, + } + ], + "links": [], + } + ), + encoding="utf-8", + ) + return graph, src + + +def _run_rebuild(*args: str): + """Invoke the real registered CLI command.""" + from graqle.cli.main import app + + return runner.invoke(app, ["rebuild", *args]) + + +class TestProgrammaticCaller: + """`graq init` calls rebuild_command() directly (init.py) — not via Typer. + + Unfilled ``typer.Option(...)`` defaults arrive as truthy ``OptionInfo`` + sentinels on that path. Left unnormalised they select machine mode and then + fail on ``Path(OptionInfo)``, breaking `graq init`'s auto-rebuild. + """ + + def test_direct_call_returns_int_and_does_not_exit(self, tmp_path): + import warnings + + from graqle.cli.commands.rebuild import rebuild_command + + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + result = rebuild_command( + graph_path=str(graph), + config_path=str(tmp_path / "absent.yaml"), + force=True, + ) + + assert isinstance(result, int) + + def test_option_sentinels_do_not_enable_machine_mode(self): + """The OptionInfo sentinel must read as 'not supplied', not as True.""" + import typer + + from graqle.cli.commands.rebuild import _flag + + assert _flag(typer.Option(False, "--headless")) is False + assert _flag(False) is False + assert _flag(True) is True + + +class TestFailedNodeAccounting: + """Sentinel BLOCKER-1: a swallowed per-node error must remain countable.""" + + def test_failed_nodes_are_counted_not_swallowed(self, tmp_path): + from unittest.mock import patch + + from graqle.core.graph import Graqle + + graph_path, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + graph = Graqle.from_json(str(graph_path)) + + with patch("pathlib.Path.read_text", side_effect=PermissionError("denied")): + updated = graph.rebuild_chunks(force=True) + + assert updated == 0 + assert graph.last_rebuild_failed_nodes >= 1 + + def test_attribute_exists_before_any_rebuild(self, tmp_path): + """Class-level default — readable even if rebuild_chunks never ran.""" + from graqle.core.graph import Graqle + + assert Graqle().last_rebuild_failed_nodes == 0 + + def test_clean_run_reports_zero_failures(self, tmp_path): + from graqle.core.graph import Graqle + + graph_path, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + graph = Graqle.from_json(str(graph_path)) + graph.rebuild_chunks(force=True) + + assert graph.last_rebuild_failed_nodes == 0 + + +class TestContentHashing: + def test_undecodable_bytes_do_not_collide(self): + """Sentinel MINOR-1: errors='ignore' silently DROPS undecodable bytes. + + Under ``ignore`` these two hash identically (both encode to ``b"ab"``), + so editing a file to add or remove an undecodable byte would be seen as + "unchanged" and skipped. ``replace`` substitutes rather than drops, so + the difference survives into the hash. + """ + from graqle.core.graph import _content_hash + + assert _content_hash("a\udce9b") != _content_hash("ab") + + +class TestRebuildMachineContract: + def test_missing_graph_exits_failure_not_zero(self, tmp_path): + """The measured defect R6 exists to fix: this used to exit 0.""" + res = _run_rebuild( + "--graph-path", str(tmp_path / "missing.json"), "--headless", "--json" + ) + assert res.exit_code == int(ExitCode.FAILURE) + payload = json.loads(res.stdout) + assert payload["status"] == "failure" + assert payload["errors"] == ["GraphFileNotFound"] + + def test_bare_invocation_on_missing_graph_still_exits_zero(self, tmp_path): + """Backwards compatibility: a published CLI's exit code must not move.""" + res = _run_rebuild("--graph-path", str(tmp_path / "missing.json")) + assert res.exit_code == 0 + + def test_unchanged_source_is_empty_delta(self, tmp_path): + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--force") + + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--incremental") + assert res.exit_code == int(ExitCode.EMPTY_DELTA) + assert json.loads(res.stdout)["counters"]["nodes_updated"] == 0 + + def test_changed_source_is_rebuilt(self, tmp_path): + """Change-based, not presence-based: the node already HAS chunks.""" + graph, src = _seed_graph(tmp_path, "def alpha():\n return 1\n") + _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--force") + + src.write_text("def alpha():\n return 99\n\ndef beta():\n return 2\n", encoding="utf-8") + + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--incremental") + assert res.exit_code == int(ExitCode.SUCCESS) + assert json.loads(res.stdout)["counters"]["nodes_updated"] >= 1 + + def test_rerunning_is_idempotent(self, tmp_path): + """Second identical run must report no work — the cron-safety property.""" + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--force") + + first = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--incremental") + second = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--incremental") + assert first.exit_code == second.exit_code == int(ExitCode.EMPTY_DELTA) + + def test_report_json_flag_writes_the_file(self, tmp_path): + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + dest = tmp_path / "reports" / "run.json" + + res = _run_rebuild("--graph-path", str(graph), "--headless", "--report-json", str(dest)) + assert dest.exists() + payload = json.loads(dest.read_text(encoding="utf-8")) + assert payload["command"] == "rebuild" + assert payload["exit_code"] == res.exit_code + + def test_headless_keeps_stdout_machine_clean(self, tmp_path): + """Decorative output must never contaminate a parsed stdout stream.""" + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json") + json.loads(res.stdout) # parses as pure JSON, nothing else on the stream + + def test_corrupt_graph_is_a_failure_not_a_traceback(self, tmp_path): + bad = tmp_path / "bad.json" + bad.write_text("{ this is not valid json", encoding="utf-8") + + res = _run_rebuild("--graph-path", str(bad), "--headless", "--json") + assert res.exit_code == int(ExitCode.FAILURE) + assert json.loads(res.stdout)["status"] == "failure" + + def test_total_node_failure_is_not_reported_as_empty_delta(self, tmp_path): + """Sentinel BLOCKER-1 (reproduced, real). + + Every node raising also yields ``updated == 0``. Reporting that as + EMPTY_DELTA would tell a scheduler "all healthy, nothing to do" when in + fact nothing worked. It must be a FAILURE. + """ + from unittest.mock import patch + + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + + # Drive the real code path: make the per-node file read raise. + with patch("pathlib.Path.read_text", side_effect=PermissionError("denied")): + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--force") + + assert res.exit_code == int(ExitCode.FAILURE) + assert json.loads(res.stdout)["status"] == "failure" + + def test_partial_failure_is_not_reported_as_success(self, tmp_path): + """Sentinel pass-2 BLOCKER: some nodes fail, some succeed. + + Falling into the "updated > 0" branch would exit 0 and tell the + scheduler the run was healthy while part of the graph is broken. + """ + from unittest.mock import patch + + from graqle.core.graph import Graqle + + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + + real_rebuild = Graqle.rebuild_chunks + + def _one_ok_one_failed(self, *args, **kwargs): + updated = real_rebuild(self, *args, **kwargs) + self.last_rebuild_failed_nodes = 1 # simulate a mixed outcome + return max(updated, 1) + + with patch.object(Graqle, "rebuild_chunks", _one_ok_one_failed): + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--force") + + assert res.exit_code == int(ExitCode.FAILURE) + payload = json.loads(res.stdout) + assert payload["status"] == "failure" + assert payload["counters"]["nodes_failed"] == 1 + assert payload["counters"]["nodes_updated"] >= 1 # work DID happen, still a failure + + def test_missing_hash_falls_back_safely(self, tmp_path): + """A graph from an older release has no hash — it must still rebuild.""" + graph, _ = _seed_graph(tmp_path, "def alpha():\n return 1\n") + res = _run_rebuild("--graph-path", str(graph), "--headless", "--json", "--incremental") + assert res.exit_code in (int(ExitCode.SUCCESS), int(ExitCode.EMPTY_DELTA)) + assert json.loads(res.stdout)["status"] in ("success", "empty_delta")