From 1bc7ef7ea8efbe26520421d9b11b7339f4fe0530 Mon Sep 17 00:00:00 2001 From: Jammy2211 Date: Fri, 30 Oct 2026 16:06:02 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20cortex=20sync-runs=20=E2=80=94=20ru?= =?UTF-8?q?n=20status=20from=20a=20campaign=20digest=20(schema=20decision?= =?UTF-8?q?=2064)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cortex.py sync-runs --digest [--dry-run]` reads a PyAutoFit autofit.campaign digest and the tasks.jsonl beside it, matches run lines by SLURM job id (narrowed to the array indices the ident names), flips open -> running with a run entry, and calls done with the task counts (wall from first start to last end; failed when none completed) once every matched task has ended. Writes run entries only, never result/lesson/Now; never adds or drops a run it cannot account for; idempotent; first stdout line `sync-runs: : changed`; exit 1 on a malformed digest. projects.yaml gains optional campaign_file and digest (paths relative to local_path), validated by check. REFERENCE.md documents the verb, the key and the boundary (status is a cluster fact; results and lessons are the human's); decision 64 records hpc-campaign ruling F3. Refs #82 Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 5 + REFERENCE.md | 35 ++++++- docs/schema_decisions.md | 16 +++ scripts/cortex.py | 190 +++++++++++++++++++++++++++++++++- tests/test_cortex.py | 2 +- tests/test_sync_runs.py | 216 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 459 insertions(+), 5 deletions(-) create mode 100644 tests/test_sync_runs.py diff --git a/AGENTS.md b/AGENTS.md index 20f9a26..e6f5523 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -102,6 +102,11 @@ not re-derive them. - `running ` / `done [--failed] [--wall H:MM] [--note …]` — what the cluster said; `done` moves the run into the log. + - `sync-runs --digest [--dry-run]` — the same two + facts, read from a PyAutoFit campaign digest (`autofit.campaign`): `open` + runs whose tasks started become `running`, runs whose tasks all ended go + through `done` with the task counts. Matched by SLURM job id; never writes + a `result`, `lesson` or `## Now` (decision 64). - `log "" [--kind note|result|lesson]` — the human's words, dated today, at the head of the log. - `now ""` — rewrites `## Now`. diff --git a/REFERENCE.md b/REFERENCE.md index d9490b2..000c2b2 100644 --- a/REFERENCE.md +++ b/REFERENCE.md @@ -56,7 +56,7 @@ Four kinds: | Kind | Written by | What it is | |---|---|---| -| `run` | `cortex.py run` / `done` | ` submitted: ` · ` finished — wall H:MM: ` · ` failed: ` | +| `run` | `cortex.py run` / `done` / `sync-runs` | ` submitted: ` · ` running: — (digest …)` · ` finished — wall H:MM: ` · ` failed: ` | | `result` | `cortex.py log --kind result` | what the human read off the results, in their words | | `lesson` | `cortex.py log --kind lesson` | something the human wants kept — a trap, a rule, a fact about the setup | | `note` | `cortex.py log` (default), `new`, `retire` | anything else: a thought, a plan, `ledger opened`, `retired: ` | @@ -96,6 +96,7 @@ creates an issue; `Issue:` is set with `new --issue` or `link`. | `issue [-n N]` | nothing (prints) | — | | `link ` | the `Issue:` header line | the ref is not `Repo#N`, an issue URL or `none` | | `retire --why "…"` | `projects.yaml` (`status: retired`, `note:`), one `note` entry | the row is already retired; the ledger still lists a run; `--why` holds `"` | +| `sync-runs [--digest PATH] [--dry-run]` | each run line the digest settles: `open` → `running` (+ a `run` entry), or `done` with the task counts | the digest is not a schema-1 `digest.json` with a readable `tasks.jsonl` beside it; no `--digest` and no `digest:` in the row | Every verb takes `--root ` (default: this checkout) and every dated verb `--today YYYY-MM-DD`, so the tests run against a copy of the fixture with a @@ -123,7 +124,9 @@ scans the raw column-0 keys and reports every duplicate with its line numbers: | `witness_file` | glob relative to the mirror | kept for the projects' own tooling; the Cortex no longer scores it | | `partition` | `gpu` \| `ral` \| `both` | which SLURM partition(s) the project runs on; `run` needs `--partition` on `both` | | `status` | `active` \| `dormant` \| `planned` \| `retired` | `retire` writes the last | -| `note` | free text, **optional** | the only optional field; an empty `note:` is drift | +| `note` | free text, **optional** | an empty `note:` is drift | +| `campaign_file` | path relative to `local_path` (or absolute), **optional** | the project's PyAutoFit `campaign.yaml`; must end `.yaml`/`.yml`, no `..` | +| `digest` | path relative to `local_path` (or absolute), **optional** | the pulled campaign `digest.json` `sync-runs` reads by default; must name a `digest.json`, no `..` | **Partition rule (human, 2026-09-30).** `both` means a project has GPU jobs *and* CPU jobs, each on its own partition — never a CPU-only array submitted as @@ -135,6 +138,34 @@ CPUs/task, throttle ≤`%2` and no pending gres/gpu jobs (`squeue -p gpu -t PD`) Record a run as GPU only after `scontrol show job` shows `gres/gpu` in its TRES. Never reorder, hold or cancel another campaign's jobs without the human's OK. +## Run status from a campaign digest (`sync-runs`) + +A project whose runs are PyAutoFit campaigns (`autofit.campaign`) has a +`digest.json` with a `tasks.jsonl` beside it, pulled from the cluster by the +project's sync CLI. `sync-runs` reads them and brings `## Runs` in line: + +- **The key is the SLURM job id.** A run line `3002_[0-3]` is matched to the + `tasks.jsonl` rows whose `job_id` is `3002` and whose `key` + (`_`) falls in `0–3`; a bare `3002` takes every task of + the array. The digest's `campaign` name is reported, not matched on. +- An `open` run with a task in `running`/`timeout_pending` or already ended + becomes `running`, with one `run` entry giving the task counts. +- A run whose matched tasks have all ended (`completed`, `failed`, `timeout`, + `oom`, `node_fail`, `cancelled`, `preempted`) goes through `done`: the counts + are its note, the wall is first start → last end, and it is `failed` when no + task completed. +- A run with no tasks in the digest, and a digest job with no run line, are + printed and left alone — `sync-runs` never adds a run (`run` records a + submission) and never removes one it cannot account for. +- It is idempotent, `--dry-run` writes nothing, and the first line printed is + `sync-runs: : changed`. Exit 0 on success (changes or not), + 1 on a malformed digest. + +**The boundary.** Run status is a cluster fact, so a script may record it +from the scheduler's accounting. What the runs *mean* is the human's: a +`result` or `lesson` entry and `## Now` are only ever written on their ask, and +`sync-runs` writes neither (schema decision 64). + ## Repository layout ``` diff --git a/docs/schema_decisions.md b/docs/schema_decisions.md index 86147b5..663f2b0 100644 --- a/docs/schema_decisions.md +++ b/docs/schema_decisions.md @@ -506,3 +506,19 @@ Choices made in phase 2 — the conductor, the board and the workflows Why: projects were started by hand-copying an older one and never got a row; `drift` (decision 62) now reports such a folder as unregistered, and birth is the one step that prevents it. + +64. **Run status may be synced from a campaign digest; meaning may not** + (2026-10-30, PyAutoCortex#82; hpc-campaign ruling F3 — named "decision 62" + in the hpc-campaign ledger before 62 and 63 were taken). `cortex.py + sync-runs --digest ` reads a PyAutoFit + `autofit.campaign` digest and its `tasks.jsonl`, matches run lines by + SLURM job id (narrowed to the array indices the ident names), flips + `open → running` and calls `done` with the task counts when every matched + task has ended. The two run states stay (`open`, `running`); a finished run + is still a log entry. It writes `run` entries only — never `result`, + `lesson` or `## Now` — and never adds or drops a run it cannot account for. + `projects.yaml` gains two optional fields, `campaign_file` and `digest`. + Why: a status check of a campaign was 5–10 ssh round trips and a + hand-read `squeue`; the digest is one small file the collector already + writes. Decision 60 still holds: nothing in a ledger is inferred from + results, and a run's start and end are cluster facts. diff --git a/scripts/cortex.py b/scripts/cortex.py index 53ff1f2..966fe4f 100644 --- a/scripts/cortex.py +++ b/scripts/cortex.py @@ -33,6 +33,7 @@ python3 scripts/cortex.py issue [-n 5] # the issue-top block python3 scripts/cortex.py link # set `Issue:` python3 scripts/cortex.py retire --why "" # a project's row + python3 scripts/cortex.py sync-runs --digest [--dry-run] Exit codes: 0 = done · 1 = drift or a refused edit · 2 = bad arguments. """ @@ -40,9 +41,10 @@ from __future__ import annotations import argparse +import json import re import sys -from datetime import date +from datetime import date, datetime from pathlib import Path import yaml @@ -69,7 +71,9 @@ PROJECT_FIELDS = ("remote", "local_path", "ral_root", "mirror", "sync_cli", "sync_verbs", "ledger", "assistant", "witness_file", "partition", "status") -PROJECT_OPTIONAL_FIELDS = ("note",) +PROJECT_OPTIONAL_FIELDS = ("note", "campaign_file", "digest") +#: the optional campaign fields: paths relative to `local_path` (or absolute). +CAMPAIGN_FIELDS = ("campaign_file", "digest") PARTITIONS = ("gpu", "ral", "both") PROJECT_STATUSES = ("active", "dormant", "planned", "retired") @@ -278,6 +282,16 @@ def _finish_row(key: str, row: dict, problems: "list[str]") -> None: if assistant and assistant != "none" and not re.match(r"^[A-Za-z0-9_.-]+$", assistant): problems.append(f"projects.yaml: {key}.assistant must be 'none' or a bare " f"workspace-relative directory name (got {assistant!r})") + for f in CAMPAIGN_FIELDS: + value = row.get(f) + if not value: + continue + if ".." in Path(value).parts: + problems.append(f"projects.yaml: {key}.{f} must not climb out with `..`") + if f == "campaign_file" and not value.endswith((".yaml", ".yml")): + problems.append(f"projects.yaml: {key}.campaign_file must be a .yaml file") + if f == "digest" and Path(value).name != "digest.json": + problems.append(f"projects.yaml: {key}.digest must name a digest.json") def load_projects(root: Path) -> "tuple[dict[str, dict], list[str]]": @@ -906,6 +920,163 @@ def birth_project(root: Path, key: str, row: dict, summary: str, today: date, *, return f"born {key}: projects.yaml row + {LEDGER_DIR}/{key}.md" +# --------------------------------------------------------------------------- # +# sync-runs: run status from a campaign digest (schema decision 64) +# --------------------------------------------------------------------------- # +DIGEST_SCHEMA = 1 +#: a task's state in `tasks.jsonl` (PyAutoFit `autofit.campaign`) that ends it. +TERMINAL_TASK_STATES = ("completed", "failed", "timeout", "oom", "node_fail", + "cancelled", "preempted") +#: states that say a task has begun on the cluster (or already ended). +STARTED_TASK_STATES = ("running", "timeout_pending") + TERMINAL_TASK_STATES + + +def _task_set(ident: str) -> "tuple[str, set[int] | None]": + """The job id and the array indices a run ident names (None = every task).""" + m = re.match(r"^(\d+)(?:_(\d+)|_\[([\d,\-]+)\])?$", ident) + if not m: + raise CortexError(f"run ident does not parse: {ident}") + job, one, many = m.groups() + if one is not None: + return job, {int(one)} + if many is None: + return job, None + out: "set[int]" = set() + for part in many.split(","): + lo, _, hi = part.partition("-") + out.update(range(int(lo), int(hi or lo) + 1)) + return job, out + + +def read_digest(path: Path) -> "tuple[dict, list[dict]]": + """`digest.json` and the `tasks.jsonl` beside it; a CortexError if either is + missing or malformed. `path` may be the file or its folder.""" + path = Path(path) + if path.is_dir(): + path = path / "digest.json" + try: + digest = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as e: + raise CortexError(f"{path}: not a readable digest ({e})") + if not isinstance(digest, dict) or digest.get("schema") != DIGEST_SCHEMA: + raise CortexError(f"{path}: not a schema-{DIGEST_SCHEMA} campaign digest") + tasks_path = path.with_name("tasks.jsonl") + tasks: "list[dict]" = [] + try: + lines = tasks_path.read_text(encoding="utf-8").splitlines() + except OSError as e: + raise CortexError(f"{tasks_path}: missing beside the digest ({e})") + for n, line in enumerate(lines, 1): + if not line.strip(): + continue + try: + row = json.loads(line) + except ValueError: + raise CortexError(f"{tasks_path}:{n}: not JSON") + if not isinstance(row, dict) or "state" not in row: + raise CortexError(f"{tasks_path}:{n}: a task row needs a state") + tasks.append(row) + return digest, tasks + + +def _tasks_of(ident: str, tasks: "list[dict]") -> "list[dict]": + job, wanted = _task_set(ident) + out = [] + for t in tasks: + if str(t.get("job_id")) != job: + continue + if wanted is not None: + key = str(t.get("key", "")) + suffix = key.rpartition("_")[2] + if not suffix.isdigit() or int(suffix) not in wanted: + continue + out.append(t) + return out + + +def _wall(tasks: "list[dict]") -> "str | None": + """H:MM from the first task start to the last task end, when both are known.""" + try: + starts = [datetime.fromisoformat(t["start"]) for t in tasks if t.get("start")] + ends = [datetime.fromisoformat(t["end"]) for t in tasks if t.get("end")] + except (TypeError, ValueError): + return None + if not starts or not ends: + return None + seconds = max(0, int((max(ends) - min(starts)).total_seconds())) + return f"{seconds // 3600}:{seconds % 3600 // 60:02d}" + + +def _counts(tasks: "list[dict]") -> str: + from collections import Counter + c = Counter(str(t["state"]) for t in tasks) + return ", ".join(f"{n} {s}" for s, n in sorted(c.items(), key=lambda x: (-x[1], x[0]))) + + +def _digest_path(root: Path, key: str, given: "str | None") -> Path: + if given: + return Path(given) + rows, _ = load_projects(root) + row = rows.get(key, {}) + if not row.get("digest"): + raise CortexError(f"give --digest, or set {key}.digest in projects.yaml") + p = Path(row["digest"]) + return p if p.is_absolute() else Path(row.get("local_path", "")) / p + + +def sync_runs(root: Path, key: str, digest_path: Path, today: date, *, + dry_run: bool = False) -> "list[str]": + """Bring `## Runs` in line with a campaign digest — cluster facts only. + + A run line is matched to the digest's tasks by its SLURM job id (the array + job), narrowed to the array indices its ident names (`3002_[0-3]`). An `open` + run with a task that has started becomes `running` (with a `run` entry); a run + whose matched tasks have all ended goes through `done`, with the task counts + as its note and `--failed` when none completed. Nothing else is written: + never a `result` or `lesson`, never `## Now`. Returns the lines to print, the + first `sync-runs: : changed`.""" + digest, tasks = read_digest(digest_path) + led = _ledger_at(root, key) + campaign = digest.get("campaign") or "?" + source = f"digest {campaign} updated {digest.get('updated') or '?'}" + changes: "list[str]" = [] + notes: "list[str]" = [] + seen_jobs: "set[str]" = set() + for run in list(led.runs): + matched = _tasks_of(run.ident, tasks) + seen_jobs.add(_task_set(run.ident)[0]) + if not matched: + notes.append(f" {run.ident}: not in the digest — unchanged") + continue + states = [str(t["state"]) for t in matched] + if all(s in TERMINAL_TASK_STATES for s in states): + completed = states.count("completed") + failed = completed == 0 + note = f"{_counts(matched)} ({source})" + changes.append(f" {run.ident}: done — {'failed' if failed else 'finished'}: " + f"{_counts(matched)}") + if not dry_run: + finish_run(root, key, run.ident, today, failed=failed, + wall=_wall(matched), note=note) + elif run.state == "open" and any(s in STARTED_TASK_STATES for s in states): + changes.append(f" {run.ident}: open → running ({_counts(matched)})") + if not dry_run: + set_running(root, key, run.ident) + current = _ledger_at(root, key) + raw = current.path.read_text(encoding="utf-8") + _write(current, _prepend_entry( + raw, f"- {today.isoformat()} — run — {run.ident} running: {run.what} " + f"— {_counts(matched)} ({source})")) + unknown = sorted({str(t.get("job_id")) for t in tasks + if t.get("job_id") and str(t.get("job_id")) not in seen_jobs}) + if unknown: + shown = ", ".join(unknown[:8]) + (f" … (+{len(unknown) - 8})" if len(unknown) > 8 else "") + notes.append(f" {len(unknown)} digest job(s) are not runs in {led.rel} " + f"and were ignored: {shown}") + head = f"sync-runs: {key}: {len(changes)} changed" + (" (dry run)" if dry_run else "") + return [head] + changes + notes + + # --------------------------------------------------------------------------- # # CLI # --------------------------------------------------------------------------- # @@ -958,6 +1129,13 @@ def cmd_birth(args) -> int: return 0 +def cmd_sync_runs(args) -> int: + path = _digest_path(args.root, args.project, args.digest) + for line in sync_runs(args.root, args.project, path, _today(args), dry_run=args.dry_run): + print(line) + return 0 + + def cmd_retire(args) -> int: print(retire_project(args.root, args.project, args.why, _today(args))) return 0 @@ -1055,6 +1233,14 @@ def build_parser() -> argparse.ArgumentParser: _common(p, dated=True) p.set_defaults(func=cmd_birth) + p = sub.add_parser("sync-runs", help="flip run lines from a campaign digest " + "(open → running, done with counts)") + p.add_argument("project") + p.add_argument("--digest", help="digest.json (or its folder); default the row's `digest`") + p.add_argument("--dry-run", action="store_true", help="print what would change; write nothing") + _common(p, dated=True) + p.set_defaults(func=cmd_sync_runs) + p = sub.add_parser("retire", help="retire a project's row (with `birth`, the only verbs " "that write projects.yaml)") p.add_argument("project") diff --git a/tests/test_cortex.py b/tests/test_cortex.py index 9fdc19a..e27fa02 100644 --- a/tests/test_cortex.py +++ b/tests/test_cortex.py @@ -419,7 +419,7 @@ def test_retire_preserves_every_other_byte_of_projects_yaml(tmp_path): def test_every_verb_is_registered(): parser = cortex.build_parser() verbs = set(parser._subparsers._group_actions[0].choices) - assert verbs == {"check", "new", "run", "running", "done", "log", "now", "issue", "link", "retire", "birth"} + assert verbs == {"check", "new", "run", "running", "done", "log", "now", "issue", "link", "retire", "birth", "sync-runs"} def test_a_bad_root_is_a_usage_error(tmp_path): diff --git a/tests/test_sync_runs.py b/tests/test_sync_runs.py new file mode 100644 index 0000000..b0b3463 --- /dev/null +++ b/tests/test_sync_runs.py @@ -0,0 +1,216 @@ +"""Contract tests for `cortex.py sync-runs` (schema decision 64): run status from +a PyAutoFit campaign digest — open → running, done with counts, nothing else. + +The digest fixtures are written per test in the shape `autofit.campaign.digest` +writes them: `digest.json` (schema 1) with `tasks.jsonl` beside it, one row per +task with `job_id`, `key` (`_`), `state`, and optionally +`start`/`end`. +""" + +import json +import shutil +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +import cortex # noqa: E402 + +REPO = Path(__file__).resolve().parents[1] +SCRIPT = REPO / "scripts" / "cortex.py" +SKELETON = REPO / "tests" / "fixtures" / "skeleton" +TODAY = "2026-09-03" +EX = "projects/example.md" + + +def _copy(tmp_path: Path) -> Path: + root = tmp_path / "tree" + shutil.copytree(SKELETON, root) + return root + + +def _digest(tmp_path: Path, tasks, campaign="example_campaign", schema=1) -> Path: + d = tmp_path / "campaign" + d.mkdir(exist_ok=True) + (d / "digest.json").write_text(json.dumps( + {"schema": schema, "campaign": campaign, "updated": "2026-09-03T08:00:00Z", + "totals": {}, "stages": {}})) + (d / "tasks.jsonl").write_text("".join(json.dumps(t) + "\n" for t in tasks)) + return d / "digest.json" + + +def _task(job, index, state, start=None, end=None): + t = {"stage": "s", "key": f"{job}_{index}", "job_id": str(job), "state": state} + if start: + t["start"] = start + if end: + t["end"] = end + return t + + +def _run(*args, root): + return subprocess.run([sys.executable, str(SCRIPT), *args, "--root", str(root)], + capture_output=True, text=True) + + +def _runs(root): + led = next(l for l in cortex.load_ledgers(root)[0] if l.key == "example") + return {r.ident: r.state for r in led.runs}, led + + +def test_open_run_with_a_started_task_becomes_running(tmp_path): + root = _copy(tmp_path) + digest = _digest(tmp_path, [_task(3003, 0, "running"), _task(3002, 0, "pending")]) + r = _run("sync-runs", "example", "--digest", str(digest), "--today", TODAY, root=root) + assert r.returncode == 0, r.stderr + assert r.stdout.splitlines()[0] == "sync-runs: example: 1 changed" + runs, led = _runs(root) + assert runs["3003"] == "running" + assert led.log[0].kind == "run" and "3003 running" in led.log[0].text() + assert cortex.check_problems(root) == [] + + +def test_all_terminal_tasks_call_done_with_counts_and_wall(tmp_path): + root = _copy(tmp_path) + tasks = [_task(3002, i, "completed", "2026-09-01T10:00:00", "2026-09-01T12:30:00") + for i in range(3)] + tasks.append(_task(3002, 3, "timeout", "2026-09-01T10:00:00", "2026-09-01T13:05:00")) + digest = _digest(tmp_path, tasks) + out = cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + assert out[0] == "sync-runs: example: 1 changed" + runs, led = _runs(root) + assert "3002_[0-3]" not in runs + head = led.log[0] + assert head.kind == "run" + assert head.text().startswith("3002_[0-3] finished — wall 3:05:") + assert "3 completed, 1 timeout" in head.text() + assert cortex.check_problems(root) == [] + + +def test_no_completed_task_is_a_failed_run(tmp_path): + root = _copy(tmp_path) + digest = _digest(tmp_path, [_task(3003, 0, "failed")]) + cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + _, led = _runs(root) + assert led.log[0].text().startswith("3003 failed:") + + +def test_array_ident_only_counts_its_own_indices(tmp_path): + root = _copy(tmp_path) + # index 7 is outside 3002_[0-3]: it is still running but must not hold the run open + tasks = [_task(3002, i, "completed") for i in range(4)] + [_task(3002, 7, "running")] + cortex.sync_runs(root, "example", _digest(tmp_path, tasks), + cortex.date.fromisoformat(TODAY)) + runs, _ = _runs(root) + assert "3002_[0-3]" not in runs + + +def test_rerun_is_idempotent(tmp_path): + root = _copy(tmp_path) + digest = _digest(tmp_path, [_task(3002, i, "completed") for i in range(4)] + + [_task(3003, 0, "running")]) + first = cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + text = (root / EX).read_text() + second = cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + assert first[0] == "sync-runs: example: 2 changed" + assert second[0] == "sync-runs: example: 0 changed" + assert (root / EX).read_text() == text + + +def test_dry_run_changes_nothing(tmp_path): + root = _copy(tmp_path) + before = (root / EX).read_text() + digest = _digest(tmp_path, [_task(3003, 0, "completed")]) + r = _run("sync-runs", "example", "--digest", str(digest), "--dry-run", root=root) + assert r.returncode == 0 + assert r.stdout.splitlines()[0] == "sync-runs: example: 1 changed (dry run)" + assert "3003: done" in r.stdout + assert (root / EX).read_text() == before + + +def test_unknown_digest_job_and_absent_run_are_reported_not_acted_on(tmp_path): + root = _copy(tmp_path) + before = (root / EX).read_text() + digest = _digest(tmp_path, [_task(9999, 0, "completed")]) + r = _run("sync-runs", "example", "--digest", str(digest), root=root) + assert r.returncode == 0 + assert r.stdout.splitlines()[0] == "sync-runs: example: 0 changed" + assert "not in the digest" in r.stdout + assert "1 digest job(s) are not runs" in r.stdout and "9999" in r.stdout + assert (root / EX).read_text() == before + + +def test_never_writes_result_lesson_or_now(tmp_path): + root = _copy(tmp_path) + _, before = _runs(root) + digest = _digest(tmp_path, [_task(3002, i, "completed") for i in range(4)] + + [_task(3003, 0, "running")]) + cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + _, after = _runs(root) + new = after.log[: len(after.log) - len(before.log)] + assert new and all(e.kind == "run" for e in new) + keep = [(e.date, e.kind, e.text()) for e in before.log if e.kind in ("result", "lesson")] + assert [(e.date, e.kind, e.text()) for e in after.log + if e.kind in ("result", "lesson")] == keep + assert after.now == before.now + + +@pytest.mark.parametrize("bad", ["not json", json.dumps({"schema": 2})]) +def test_malformed_digest_exits_nonzero(tmp_path, bad): + root = _copy(tmp_path) + d = tmp_path / "bad" + d.mkdir() + (d / "digest.json").write_text(bad) + (d / "tasks.jsonl").write_text("") + r = _run("sync-runs", "example", "--digest", str(d / "digest.json"), root=root) + assert r.returncode != 0 + assert r.stdout == "" + + +def test_missing_tasks_or_bad_row_exits_nonzero(tmp_path): + root = _copy(tmp_path) + digest = _digest(tmp_path, []) + (digest.parent / "tasks.jsonl").unlink() + assert _run("sync-runs", "example", "--digest", str(digest), root=root).returncode != 0 + (digest.parent / "tasks.jsonl").write_text('{"job_id": "1"}\n') + assert _run("sync-runs", "example", "--digest", str(digest), root=root).returncode != 0 + + +def test_digest_folder_and_projects_yaml_default(tmp_path): + root = _copy(tmp_path) + digest = _digest(tmp_path, [_task(3003, 0, "running")]) + yml = root / "projects.yaml" + yml.write_text(yml.read_text().replace( + ' note: "the fixture project"', + f' note: "the fixture project"\n digest: {digest}\n campaign_file: campaign.yaml', 1)) + assert cortex.check_problems(root) == [] + r = _run("sync-runs", "example", root=root) + assert r.returncode == 0, r.stderr + assert r.stdout.splitlines()[0] == "sync-runs: example: 1 changed" + # the folder works as well as the file + r = _run("sync-runs", "example", "--digest", str(digest.parent), root=root) + assert r.stdout.splitlines()[0] == "sync-runs: example: 0 changed" + + +def test_no_digest_anywhere_is_refused(tmp_path): + root = _copy(tmp_path) + r = _run("sync-runs", "example", root=root) + assert r.returncode == 1 + assert "--digest" in r.stderr + + +@pytest.mark.parametrize("field,value,message", [ + ("campaign_file", "campaign.txt", "campaign_file must be a .yaml file"), + ("digest", "out/summary.json", "digest must name a digest.json"), + ("digest", "../other/digest.json", "must not climb out"), + ("campaign_file", '""', "campaign_file is empty"), +]) +def test_check_validates_the_campaign_fields(tmp_path, field, value, message): + root = _copy(tmp_path) + yml = root / "projects.yaml" + yml.write_text(yml.read_text().replace( + ' note: "the fixture project"', f' note: "the fixture project"\n {field}: {value}', 1)) + assert any(message in p for p in cortex.check_problems(root)), cortex.check_problems(root) From d590eac9a93eb54d8a01a5f8c0b442f9f32befca Mon Sep 17 00:00:00 2001 From: Jammy2211 Date: Fri, 30 Oct 2026 16:28:47 +0000 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20sync-runs=20review=20findings=20?= =?UTF-8?q?=E2=80=94=20pending=20slices,=20one=20write,=20newest-first?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - A pending array slice the scheduler reports as one row (job_id/key 3002_[2-3], optionally %throttle) now stands for each of its indices; a run is done only when every index its ident names has a terminal row, and for a bare job id any unfinished row (compound stubs included) keeps it on the ledger. Counts weigh a compound row by its indices, so cancelled stubs count per task. - Every edit is computed in memory, read back and written once: a refusal or a failed read-back writes nothing. - The log's newest-first rule is checked before any entry is written. Refs #82 Co-Authored-By: Claude Opus 5.5 --- scripts/cortex.py | 144 ++++++++++++++++++++++++++++++---------- tests/test_sync_runs.py | 71 ++++++++++++++++++++ 2 files changed, 181 insertions(+), 34 deletions(-) diff --git a/scripts/cortex.py b/scripts/cortex.py index 966fe4f..fb51b58 100644 --- a/scripts/cortex.py +++ b/scripts/cortex.py @@ -979,21 +979,75 @@ def read_digest(path: Path) -> "tuple[dict, list[dict]]": return digest, tasks +def _indices(text: str) -> "set[int] | None": + """The array indices a bracket like `[2-3]`, `[586,588-594]` or `[0-9%4]` + names; None when there is no bracket. A part that does not parse is skipped.""" + m = re.search(r"\[([^\]]*)\]", text or "") + if not m: + return None + out: "set[int]" = set() + for part in m.group(1).split("%")[0].split(","): + lo, _, hi = part.strip().partition("-") + if lo.isdigit() and (not hi or hi.isdigit()): + out.update(range(int(lo), int(hi or lo) + 1)) + return out + + +def _task_indices(t: dict) -> "set[int] | None": + """The array indices one `tasks.jsonl` row stands for: its `key` suffix + (`3002_1`), or — for a pending array slice the scheduler reports as one row + (`job_id` `3002_[2-3]`) — every index in its bracket. None when the row names + no index (a plain job).""" + for field in ("key", "job_id"): + found = _indices(str(t.get(field, ""))) + if found is not None: + return found + suffix = str(t.get("key", "")).rpartition("_")[2] + return {int(suffix)} if suffix.isdigit() else None + + def _tasks_of(ident: str, tasks: "list[dict]") -> "list[dict]": + """The rows of the run's job (base id of `job_id`, so a compound pending stub + `3002_[2-3]` belongs to job 3002) that touch the indices the ident names.""" job, wanted = _task_set(ident) out = [] for t in tasks: - if str(t.get("job_id")) != job: + if str(t.get("job_id", "")).split("_")[0] != job: continue if wanted is not None: - key = str(t.get("key", "")) - suffix = key.rpartition("_")[2] - if not suffix.isdigit() or int(suffix) not in wanted: + idx = _task_indices(t) + if idx is None or not (idx & wanted): continue out.append(t) return out +def _weight(t: dict, wanted: "set[int] | None") -> int: + """How many tasks a row counts for: one, or the indices of a compound slice + (inside the ident's set when it names one).""" + idx = _task_indices(t) + if idx is None or len(idx) <= 1: + return 1 + return len(idx & wanted) if wanted is not None else len(idx) + + +def _all_ended(ident: str, matched: "list[dict]") -> bool: + """Every task the run stands for has ended. For an ident naming indices, + each index needs a row and all its rows must be terminal; for a bare job id, + no row (compound pending stubs included) may be unfinished.""" + if not matched: + return False + if any(str(t["state"]) not in TERMINAL_TASK_STATES for t in matched): + return False + _, wanted = _task_set(ident) + if wanted is None: + return True + covered: "set[int]" = set() + for t in matched: + covered |= _task_indices(t) or set() + return wanted <= covered + + def _wall(tasks: "list[dict]") -> "str | None": """H:MM from the first task start to the last task end, when both are known.""" try: @@ -1007,9 +1061,11 @@ def _wall(tasks: "list[dict]") -> "str | None": return f"{seconds // 3600}:{seconds % 3600 // 60:02d}" -def _counts(tasks: "list[dict]") -> str: +def _counts(tasks: "list[dict]", wanted: "set[int] | None" = None) -> str: from collections import Counter - c = Counter(str(t["state"]) for t in tasks) + c: "Counter[str]" = Counter() + for t in tasks: + c[str(t["state"])] += _weight(t, wanted) return ", ".join(f"{n} {s}" for s, n in sorted(c.items(), key=lambda x: (-x[1], x[0]))) @@ -1029,50 +1085,70 @@ def sync_runs(root: Path, key: str, digest_path: Path, today: date, *, """Bring `## Runs` in line with a campaign digest — cluster facts only. A run line is matched to the digest's tasks by its SLURM job id (the array - job), narrowed to the array indices its ident names (`3002_[0-3]`). An `open` - run with a task that has started becomes `running` (with a `run` entry); a run - whose matched tasks have all ended goes through `done`, with the task counts - as its note and `--failed` when none completed. Nothing else is written: - never a `result` or `lesson`, never `## Now`. Returns the lines to print, the - first `sync-runs: : changed`.""" + job), narrowed to the array indices its ident names (`3002_[0-3]`); a pending + array slice reported as one row (`3002_[2-3]`) stands for each of its + indices. An `open` run with a task that has started becomes `running` (with a + `run` entry); a run all of whose tasks have ended — every index it names + covered by a terminal row — leaves `## Runs` as `done` would, with the task + counts as its note and `failed` when none completed. Every edit is made in + memory, read back, and written once: a refusal writes nothing. Never a + `result` or `lesson`, never `## Now`. Returns the lines to print, the first + `sync-runs: : changed`.""" digest, tasks = read_digest(digest_path) led = _ledger_at(root, key) campaign = digest.get("campaign") or "?" source = f"digest {campaign} updated {digest.get('updated') or '?'}" changes: "list[str]" = [] notes: "list[str]" = [] + entries: "list[str]" = [] + remaining: "list[Run]" = [] seen_jobs: "set[str]" = set() - for run in list(led.runs): + stamp = today.isoformat() + for run in led.runs: + job, wanted = _task_set(run.ident) + seen_jobs.add(job) matched = _tasks_of(run.ident, tasks) - seen_jobs.add(_task_set(run.ident)[0]) if not matched: notes.append(f" {run.ident}: not in the digest — unchanged") + remaining.append(run) continue + counts = _counts(matched, wanted) states = [str(t["state"]) for t in matched] - if all(s in TERMINAL_TASK_STATES for s in states): - completed = states.count("completed") - failed = completed == 0 - note = f"{_counts(matched)} ({source})" - changes.append(f" {run.ident}: done — {'failed' if failed else 'finished'}: " - f"{_counts(matched)}") - if not dry_run: - finish_run(root, key, run.ident, today, failed=failed, - wall=_wall(matched), note=note) - elif run.state == "open" and any(s in STARTED_TASK_STATES for s in states): - changes.append(f" {run.ident}: open → running ({_counts(matched)})") - if not dry_run: - set_running(root, key, run.ident) - current = _ledger_at(root, key) - raw = current.path.read_text(encoding="utf-8") - _write(current, _prepend_entry( - raw, f"- {today.isoformat()} — run — {run.ident} running: {run.what} " - f"— {_counts(matched)} ({source})")) - unknown = sorted({str(t.get("job_id")) for t in tasks - if t.get("job_id") and str(t.get("job_id")) not in seen_jobs}) + if _all_ended(run.ident, matched): + completed = sum(_weight(t, wanted) for t in matched if t["state"] == "completed") + verb = "failed" if completed == 0 else "finished" + wall = _wall(matched) + head = " — ".join([f"{run.ident} {verb}"] + ([f"wall {wall}"] if wall else [])) + entries.append(f"- {stamp} — run — {head}: {run.what} — {counts} ({source})") + changes.append(f" {run.ident}: done — {verb}: {counts}") + continue + if run.state == "open" and any(s in STARTED_TASK_STATES for s in states): + run.state = "running" + entries.append(f"- {stamp} — run — {run.ident} running: {run.what} — {counts} " + f"({source})") + changes.append(f" {run.ident}: open → running ({counts})") + remaining.append(run) + unknown = sorted({str(t.get("job_id", "")).split("_")[0] for t in tasks + if t.get("job_id")} - seen_jobs) if unknown: shown = ", ".join(unknown[:8]) + (f" … (+{len(unknown) - 8})" if len(unknown) > 8 else "") notes.append(f" {len(unknown)} digest job(s) are not runs in {led.rel} " f"and were ignored: {shown}") + + if entries and not dry_run: + if led.log and stamp < led.log[0].date: + raise CortexError(f"the log's head is {led.log[0].date}; entries dated {stamp} " + f"would break newest-first — nothing written") + text = _replace_section(led.path.read_text(encoding="utf-8"), "Runs", + _runs_body(remaining)) + for entry in entries: + text = _prepend_entry(text, entry) + after = parse_ledger(text, led.key, led.path, led.rel) + if after.problems: + raise CortexError("the edit would not read back — nothing written: " + + "; ".join(after.problems)) + led.path.write_text(text, encoding="utf-8") + head = f"sync-runs: {key}: {len(changes)} changed" + (" (dry run)" if dry_run else "") return [head] + changes + notes diff --git a/tests/test_sync_runs.py b/tests/test_sync_runs.py index b0b3463..0ffc934 100644 --- a/tests/test_sync_runs.py +++ b/tests/test_sync_runs.py @@ -214,3 +214,74 @@ def test_check_validates_the_campaign_fields(tmp_path, field, value, message): yml.write_text(yml.read_text().replace( ' note: "the fixture project"', f' note: "the fixture project"\n {field}: {value}', 1)) assert any(message in p for p in cortex.check_problems(root)), cortex.check_problems(root) + + +def test_pending_array_slice_keeps_the_run_on_the_ledger(tmp_path): + # review finding 1: the scheduler reports a pending slice as one compound row + root = _copy(tmp_path) + tasks = [_task(3002, 0, "completed"), _task(3002, 1, "completed"), + {"key": "3002_[2-3]", "job_id": "3002_[2-3]", "state": "pending"}] + out = cortex.sync_runs(root, "example", _digest(tmp_path, tasks), + cortex.date.fromisoformat(TODAY)) + runs, _ = _runs(root) + assert runs["3002_[0-3]"] == "running" + assert out[0] == "sync-runs: example: 0 changed" + # an index with no row at all also keeps it on the ledger + root2 = _copy(tmp_path / "b") + tasks = [_task(3002, i, "completed") for i in range(3)] + cortex.sync_runs(root2, "example", _digest(tmp_path / "b", tasks), + cortex.date.fromisoformat(TODAY)) + assert "3002_[0-3]" in _runs(root2)[0] + # cancelled stubs count per index and do end the run + root3 = _copy(tmp_path / "c") + tasks = [_task(3002, 0, "completed"), + {"key": "3002_[1-3]", "job_id": "3002_[1-3]", "state": "cancelled"}] + out = cortex.sync_runs(root3, "example", _digest(tmp_path / "c", tasks), + cortex.date.fromisoformat(TODAY)) + assert "3 cancelled, 1 completed" in out[1] + assert "3002_[0-3]" not in _runs(root3)[0] + + +def test_bare_job_with_a_pending_stub_is_not_done(tmp_path): + root = _copy(tmp_path) + tasks = [_task(3003, 0, "completed"), + {"key": "3003_[1-9%2]", "job_id": "3003_[1-9%2]", "state": "pending"}] + cortex.sync_runs(root, "example", _digest(tmp_path, tasks), + cortex.date.fromisoformat(TODAY)) + assert _runs(root)[0]["3003"] == "running" + + +def test_a_refused_sync_writes_nothing(tmp_path): + # review finding 2: every edit is made in memory and written once + root = _copy(tmp_path) + before = (root / EX).read_text() + digest = _digest(tmp_path, [_task(3002, i, "completed") for i in range(4)] + + [_task(3003, 0, "running")]) + orig = cortex.parse_ledger + calls = {"n": 0} + + def failing(text, key, path, rel): + led = orig(text, key, path, rel) + calls["n"] += 1 + if calls["n"] > 1: # the read-back of the combined edit + led.problems.append("injected") + return led + + cortex.parse_ledger = failing + try: + with pytest.raises(cortex.CortexError): + cortex.sync_runs(root, "example", digest, cortex.date.fromisoformat(TODAY)) + finally: + cortex.parse_ledger = orig + assert (root / EX).read_text() == before + + +def test_entry_older_than_the_log_head_is_refused(tmp_path): + # review finding 3: newest-first holds for the running entry too + root = _copy(tmp_path) + before = (root / EX).read_text() + digest = _digest(tmp_path, [_task(3003, 0, "running")]) + r = _run("sync-runs", "example", "--digest", str(digest), "--today", "2026-08-01", root=root) + assert r.returncode == 1 + assert "newest-first" in r.stderr + assert (root / EX).read_text() == before