diff --git a/docs/agent.md b/docs/agent.md index 0588591..cc87b82 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -194,6 +194,8 @@ the fetched downloads, and the build outputs **all co-locate** under it: ```text .tablassert/agent/ # = state_dir (the workspace root) state.json # supervisor checkpoint (atomic; unchanged location) + graph.yaml # SHARED aggregate graph registry (flock-serialized, atomic) + graph.yaml.lock # sidecar lock file for graph.yaml (exclusive flock) configs/.yaml # best / accepted config (ALL configs in ONE folder) configs/.derived.yaml # initial agent-derived config downloads///... # fetched PMC payload (main text + metadata + tables) — stable, persists @@ -203,6 +205,8 @@ the fetched downloads, and the build outputs **all co-locate** under it: | Path | Contents | Lifecycle | | --- | --- | --- | | `state.json` | supervisor checkpoint: `{pmc_id, status, config_path, coverage_history[], qc_pass_rate, attempts, last_edits, best_coverage, best_config_path}` per record | written **atomically** (tmp write + `os.replace`) after each config and each improve iteration; git-ignored | +| `graph.yaml` | SHARED aggregate graph registry: one `tables` entry per successful (`MAPPED` / `BUILT_UNMEASURED`) build | maintained under an exclusive `graph.yaml.lock` flock; atomic writes; see [Parallel agents and the shared graph registry](#parallel-agents-and-the-shared-graph-registry) | +| `graph.yaml.lock` | sidecar lock file serializing registry read-modify-write | created on first registration; never deleted | | `configs/.yaml` | the best / accepted config for the article | the reuse entry point (below) | | `configs/.derived.yaml` | the agent's initial derived config | kept for provenance | | `downloads///` | fetched PMC payload (main text + metadata + tables) | **stable** — persists across runs | @@ -219,7 +223,7 @@ ready-to-build `graph.yaml` (wrapping `table.yaml` with the resolved fullmap) in ```bash cd .tablassert/agent/builds/PMC11708054 -tablassert build-kg graph.yaml +tablassert build-kg -f graph.yaml ``` !!! warning "Not relocatable" @@ -227,6 +231,47 @@ tablassert build-kg graph.yaml therefore **not relocatable** — moving or renaming the `.tablassert/agent` folder breaks that reference (re-run the agent, or fix `source.local`, after any move). +### Parallel agents and the shared graph registry + +Several `tablassert agent` processes can run CONCURRENTLY against the SAME shared `--state-dir` and +each successful build self-registers into ONE aggregate graph config that a single `build-kg` then +builds as a whole: + +```bash +# fan out over DISJOINT pmc sets, all pointed at one shared state dir +tablassert agent PMC1 PMC2 --fullmap ./fullmap --state-dir ./shared & +tablassert agent PMC3 PMC4 --fullmap ./fullmap --state-dir ./shared & +wait + +# one build of the whole registered graph +tablassert build-kg -f ./shared/graph.yaml +``` + +Use **disjoint pmc sets**: each process owns its own ids. The shared registry itself is fully +cross-process safe, but the per-process checkpoint (`state.json`) read-modify cycle is not +cross-process locked, so two processes must not own the same pmc id. + +**How the registry works.** Every build that ends `MAPPED` or `BUILT_UNMEASURED` (both are +successful builds) UPSERTS its best config into `/graph.yaml`: + +- **Concurrency-safe** — each registration takes an EXCLUSIVE `flock` on the + `/graph.yaml.lock` sidecar around the read-modify-write, then persists atomically + (tmp file + `os.replace`, the same pattern as `state.json`). No registration can lose or tear + another process's entry. +- **Upsert by pmc id** — a re-run REPLACES the prior entry for the same pmc id (matched by config + basename stem); other entries keep their insertion order. Entries are ABSOLUTE paths, so + `build-kg` works from any CWD. +- **`fullmap` is first-wins** — the first fullmap recorded stays; a later run passing a different + fullmap keeps the existing value and logs a warning. +- **Self-healing** — a corrupt registry (bad YAML, not a mapping, or failing `Graph.model_validate`) + is renamed `graph.yaml.corrupt-` and rebuilt fresh with a warning, so unattended + parallel runs never wedge on a damaged file. +- **Registered statuses** — only `MAPPED` and `BUILT_UNMEASURED` register; `SKIPPED` never does. A + re-run that SKIPS an already-MAPPED pmc keeps the existing entry (resume skips terminal records + entirely, so nothing rewrites them). `tablassert rebuild-agent-graph --state-dir ./shared + --fullmap ./fullmap` reconstructs the registry from `state.json` and prunes stale entries + (deleted configs, non-registered statuses). + ## The tools | Tool | Kind | Purpose | diff --git a/docs/cli.md b/docs/cli.md index 454c09d..354d9e4 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -1,8 +1,9 @@ # CLI Reference Tablassert extracts knowledge assertions from tabular data into KGX NDJSON. The `tablassert` app -exposes **four subcommands** — `agent`, `build-fullmap`, `build-kg`, `validate` — plus an app-level -`--version` flag. Run `tablassert --help` (or ` --help`) for the live surface. +exposes **five subcommands** — `agent`, `build-fullmap`, `build-kg`, `rebuild-agent-graph`, +`validate` — plus an app-level `--version` flag. Run `tablassert --help` (or ` --help`) +for the live surface. ## Command index @@ -11,6 +12,7 @@ exposes **four subcommands** — `agent`, `build-fullmap`, `build-kg`, `validate | [`agent`](#agent) | Autonomously derive, build, audit, and improve KG configs from PMC articles | | [`build-fullmap`](#build-fullmap) | Build the embedded fullmap redb used for entity resolution | | [`build-kg`](#build-kg) | Build a KGX NDJSON knowledge graph from a YAML configuration | +| [`rebuild-agent-graph`](#rebuild-agent-graph) | Rebuild the shared agent graph registry from the supervisor checkpoint | | [`validate`](#validate) | Validate a graph or table configuration without executing it | ## App flags @@ -137,6 +139,37 @@ Output is written to the current directory as `{name}_{version}.nodes.ndjson`, --- +## rebuild-agent-graph + +Use this to rebuild the SHARED agent graph registry (`/graph.yaml`) from the supervisor +checkpoint (`/state.json`) — e.g. to prune stale entries after deleting configs, or to +recover a hand-edited/damaged registry. Parallel `tablassert agent` runs maintain the registry +incrementally (see [Agent — Parallel agents and the shared graph registry](agent.md#parallel-agents-and-the-shared-graph-registry)); +this command reconstructs it deterministically from `state.json`. + +```bash +tablassert rebuild-agent-graph [ARGS] +``` + +| Option | Type | Required | Default | Description | +| --- | --- | --- | --- | --- | +| `--state-dir`, `-sd` | Path | No | `.tablassert/agent` | Agent state directory holding `state.json` + `configs/` | +| `--fullmap`, `-f` | Path | Yes | — | Fullmap redb file or base directory recorded in the registry | + +Every `MAPPED` / `BUILT_UNMEASURED` record whose best config still exists on disk becomes a +`tables` entry (absolute path, sorted by pmc id); every other entry — `SKIPPED` records, deleted +configs, stale leftovers — is pruned. The registry `fullmap` is **first-wins**: an existing value +that differs from `--fullmap` is kept (with a warning). The write is concurrency-safe (exclusive +`graph.yaml.lock` flock + atomic replace), so the command never corrupts the registry; run it +while agents are quiescent for a complete snapshot. + +```bash +tablassert rebuild-agent-graph --state-dir .tablassert/agent --fullmap ./fullmap +tablassert build-kg -f .tablassert/agent/graph.yaml +``` + +--- + ## validate Use this to validate a configuration against a schema without running the build — ideal for CI and diff --git a/src/tablassert/agent.py b/src/tablassert/agent.py index da84d74..cabe277 100644 --- a/src/tablassert/agent.py +++ b/src/tablassert/agent.py @@ -33,6 +33,7 @@ from tablassert.enums import EncodingMethods from tablassert.errors import GraphValidationError, QcRuntimeMissingError, SectionValidationError, TablassertValidationError from tablassert.fullmap import distinct, fullmap_db_path, is_lock_contention, lookup_rows +from tablassert.graph_registry import REGISTERED_STATUSES, register_build from tablassert.lib import Tcode from tablassert.log import cat from tablassert.models import NodeEncoding, Section @@ -2768,7 +2769,9 @@ def run_supervisor( best_path: Path = best_config_path(state_dir, pmc_id) best_path.write_text(current_config) - rec.best_config_path = str(best_path) + # Persist the ABSOLUTE path: a relative --state-dir would otherwise store a CWD-relative + # entry that rebuild_graph (which may run from a different CWD) could not locate. + rec.best_config_path = str(best_path.resolve()) rec.config_path = str(best_path) if current_cov >= map_threshold: # Optional semantic gate (W1): when a real judge model is configured, MAPPED additionally @@ -2804,6 +2807,18 @@ def run_supervisor( f"SKIPPED: could not reach map_threshold={map_threshold} after {max_improve_iters} " f"improve iters (best coverage {current_cov:.3f})" ) + # Shared graph registry: successful builds upsert into /graph.yaml so concurrent + # agents over one --state-dir converge on a single aggregate config. SKIPPED never registers, + # and registration must NEVER flip a successful status: on any error log + note and keep the + # status. A re-run that SKIPS an already-MAPPED pmc keeps its registry entry: resume skips + # terminal records entirely, and rebuild-agent-graph prunes stale entries from state.json. + if rec.status in REGISTERED_STATUSES: + try: + register_build(state_dir, pmc_id, best_path, fullmap) + except Exception as reg_exc: # a registry failure is a note, never a status change + logger.error("graph registry update failed for {pmc}: {error}", pmc=pmc_id, error=reg_exc) + note: str = f"graph registry update failed (status kept {rec.status}): {reg_exc}" + rec.notes = f"{rec.notes}; {note}" if rec.notes else note save_state(state_dir, state) except Exception as exc: # one bad pmc never aborts the batch rec.status = "SKIPPED" diff --git a/src/tablassert/cli.py b/src/tablassert/cli.py index 730c332..74732ab 100644 --- a/src/tablassert/cli.py +++ b/src/tablassert/cli.py @@ -717,6 +717,37 @@ def metric(key: str, default: float) -> float: ) +@APP.command(name="rebuild-agent-graph") +def rebuild_agent_graph( + state_dir: Annotated[Path, cyclopts.Parameter(name=["--state-dir", "-sd"])] = Path(".tablassert") / "agent", + *, + fullmap: Annotated[Path, cyclopts.Parameter(name=["--fullmap", "-f"])], +) -> None: + """Rebuild the shared agent graph registry from the supervisor checkpoint. + + Reconstructs ``/graph.yaml`` from ``/state.json``: every MAPPED / + BUILT_UNMEASURED record whose best config still exists on disk becomes a ``tables`` entry + (sorted by pmc id); stale entries (deleted configs, non-registered statuses) are pruned. + Parallel ``tablassert agent`` runs maintain the registry incrementally; this command + reconstructs it deterministically. Concurrency-safe: the same exclusive ``graph.yaml.lock`` + flock + atomic write the agent registration uses. + + Args: + state_dir: Agent state directory holding ``state.json`` + ``configs/``. + fullmap: Fullmap redb file or base directory recorded in the registry (first-wins: an + existing registry fullmap that differs is kept with a warning). + """ + import yaml + + from tablassert.graph_registry import rebuild_graph + + graph_path: Path = rebuild_graph(state_dir, fullmap) + data: object = yaml.safe_load(graph_path.read_text(encoding="utf-8")) + tables: object = data.get("tables", []) if isinstance(data, dict) else [] + count: int = len(tables) if isinstance(tables, list) else 0 + print(f"tablassert rebuild-agent-graph: wrote {graph_path} with {count} table config(s).") + + def build_fullmap_pipeline( output: Path, progress: PipelineProgress, cache: Path = Path("./fullmap/downloads"), version: str = BABEL_VERSION, threads: int | None = None ) -> None: diff --git a/src/tablassert/graph_registry.py b/src/tablassert/graph_registry.py new file mode 100644 index 0000000..940ab5e --- /dev/null +++ b/src/tablassert/graph_registry.py @@ -0,0 +1,184 @@ +"""Concurrency-safe SHARED graph registry for parallel ``tablassert agent`` runs. + +Several agent processes pointed at the SAME ``--state-dir`` each self-register their successful +builds (status ``MAPPED`` or ``BUILT_UNMEASURED``) into ONE aggregate ``/graph.yaml`` +that ``tablassert build-kg -f /graph.yaml`` then builds as a whole. + +Concurrency safety: an EXCLUSIVE ``fcntl.flock`` on the ``/graph.yaml.lock`` sidecar +serializes every read-modify-write, and the write itself is atomic (a tmp file in the same +directory + ``os.replace`` — the same pattern as ``save_state``). A corrupt registry (YAML parse +error, not a mapping, or failing ``Graph.model_validate``) is quarantined to +``graph.yaml.corrupt-`` and rebuilt fresh, so unattended parallel runs self-heal +instead of wedging. Stdlib only — no new dependencies. +""" + +from __future__ import annotations + +import contextlib +import fcntl +import os +from collections.abc import Iterator +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import pydantic +import yaml + +from tablassert.log import cat +from tablassert.models import Graph + +logger = cat("REGISTRY") + +GRAPH_YAML: str = "graph.yaml" +LOCK_NAME: str = f"{GRAPH_YAML}.lock" +TMP_NAME: str = f"{GRAPH_YAML}.tmp" +CORRUPT_PREFIX: str = f"{GRAPH_YAML}.corrupt-" +GRAPH_NAME: str = "tablassert-agent" +GRAPH_VERSION: str = "1" +GRAPH_DESCRIPTION: str = "Aggregate graph of agent-built PMC table configs" +#: Record statuses whose best config self-registers (both are SUCCESSFUL builds). +REGISTERED_STATUSES: frozenset[str] = frozenset({"MAPPED", "BUILT_UNMEASURED"}) + + +@contextlib.contextmanager +def _registry_lock(state_dir: Path) -> Iterator[None]: + """Hold an EXCLUSIVE ``flock`` on ``/graph.yaml.lock`` (created if missing). + + The sidecar lock file is never deleted, so concurrent processes always contend on the same + inode even while ``graph.yaml`` itself is atomically replaced underneath them. + """ + state_dir.mkdir(parents=True, exist_ok=True) + lock_path: Path = state_dir / LOCK_NAME + with lock_path.open("a") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + +def _fresh_doc() -> dict[str, Any]: + """A brand-new registry document (``_apply_fullmap`` fills the first-wins ``fullmap``).""" + return {"name": GRAPH_NAME, "version": GRAPH_VERSION, "description": GRAPH_DESCRIPTION, "tables": []} + + +def _quarantine(state_dir: Path, reason: str) -> Path: + """Rename a corrupt ``graph.yaml`` aside to ``graph.yaml.corrupt-`` and warn. + + The corrupt bytes are preserved for forensics; the caller continues against a fresh document, + which is what lets unattended parallel runs self-heal. + """ + stamp: str = datetime.now(UTC).strftime("%Y%m%dT%H%M%S%fZ") + quarantined: Path = state_dir / f"{CORRUPT_PREFIX}{stamp}" + os.replace(state_dir / GRAPH_YAML, quarantined) + logger.warning("graph registry: quarantined corrupt {name} -> {quarantined} ({reason})", name=GRAPH_YAML, quarantined=quarantined, reason=reason) + return quarantined + + +def _load_registry(state_dir: Path) -> dict[str, Any]: + """Load the existing registry document; absent -> fresh, corrupt -> quarantine + fresh. + + Corrupt means: a YAML parse error, a top level that is not a mapping, or a document failing + ``Graph.model_validate``. Validation on load guarantees every mutation below starts from a + document the pipelines would accept. + """ + path: Path = state_dir / GRAPH_YAML + if not path.is_file(): + return _fresh_doc() + try: + data: object = yaml.safe_load(path.read_text(encoding="utf-8")) + except yaml.YAMLError as exc: + _quarantine(state_dir, f"YAML parse error: {exc}") + return _fresh_doc() + if not isinstance(data, dict): + _quarantine(state_dir, f"top level is not a mapping (got {type(data).__name__})") + return _fresh_doc() + try: + Graph.model_validate(data) + except pydantic.ValidationError as exc: + _quarantine(state_dir, f"fails Graph.model_validate ({len(exc.errors())} error(s))") + return _fresh_doc() + return data + + +def _apply_fullmap(doc: dict[str, Any], fullmap: Path) -> None: + """Apply the FIRST-WINS fullmap rule: set when absent; keep + warn when present and different.""" + resolved: str = str(fullmap.resolve()) + existing: object = doc.get("fullmap") + if existing is None: + doc["fullmap"] = resolved + return + if str(existing) != resolved: + logger.warning( + "graph registry: keeping existing fullmap {existing}; caller requested {requested} (fullmap is first-wins)", + existing=str(existing), + requested=resolved, + ) + + +def _write_registry(state_dir: Path, doc: dict[str, Any]) -> Path: + """Validate and atomically persist the registry (tmp write + ``os.replace`` under the held lock).""" + Graph.model_validate(doc) # final gate: never persist a document the pipelines would reject + target: Path = state_dir / GRAPH_YAML + tmp: Path = state_dir / TMP_NAME + tmp.write_text(yaml.safe_dump(doc, sort_keys=False), encoding="utf-8") + os.replace(tmp, target) + return target + + +def register_build(state_dir: Path, pmc_id: str, config_path: Path, fullmap: Path) -> Path: + """Upsert one successful agent build into the shared ``/graph.yaml`` registry. + + Drops any existing ``tables`` entry whose basename stem equals ``pmc_id`` (a re-run REPLACES + the prior entry), then appends the ABSOLUTE config path — the pipelines resolve ``Graph.tables`` + against the CWD, and saved configs already carry absolute ``source.local``, so absolute entries + make ``build-kg`` work from any CWD. The fullmap is first-wins and the write is atomic under + the exclusive sidecar lock, so concurrent registrations serialize and never lose entries. + + Returns the registry path. + """ + absolute: str = str(config_path.resolve()) + with _registry_lock(state_dir): + doc: dict[str, Any] = _load_registry(state_dir) + tables: list[Any] = doc.get("tables", []) + doc["tables"] = [entry for entry in tables if Path(str(entry)).stem != pmc_id] + [absolute] + _apply_fullmap(doc, fullmap) + target: Path = _write_registry(state_dir, doc) + logger.info("graph registry: registered {pmc} -> {config} in {target}", pmc=pmc_id, config=absolute, target=target) + return target + + +def rebuild_graph(state_dir: Path, fullmap: Path) -> Path: + """Reconstruct the registry ``tables`` from ``state.json`` records, pruning stale entries. + + Keeps every record whose status is in ``REGISTERED_STATUSES`` and whose ``best_config_path`` + still exists on disk (sorted by pmc id); every other entry — SKIPPED records, deleted configs, + stale leftovers — is pruned. Same exclusive lock + atomic write as ``register_build``; the + fullmap is first-wins. Returns the registry path. + """ + from tablassert.agent import load_state # deferred: tablassert.agent imports this module at top level + + with _registry_lock(state_dir): + doc: dict[str, Any] = _load_registry(state_dir) + try: + state = load_state(state_dir) + except (OSError, ValueError) as exc: # ValueError covers json.JSONDecodeError + UnicodeDecodeError + # The recovery command must not die on the very damage it is meant to fix: warn and + # rebuild an empty registry instead of raising a raw traceback. + logger.warning("graph registry: unreadable state.json ({error}); rebuilding an empty registry", error=exc) + state = None + tables: list[str] = [] + if state is not None: + for pmc_id in sorted(state.records): + record = state.records[pmc_id] + if record.status not in REGISTERED_STATUSES or record.best_config_path is None: + continue + best: Path = Path(record.best_config_path) + if best.is_file(): + tables.append(str(best.resolve())) + doc["tables"] = tables + _apply_fullmap(doc, fullmap) + target: Path = _write_registry(state_dir, doc) + logger.info("graph registry: rebuilt {target} from state.json ({count} table config(s))", target=target, count=len(tables)) + return target diff --git a/tests/test_agent_cli.py b/tests/test_agent_cli.py index 2136526..86b800c 100644 --- a/tests/test_agent_cli.py +++ b/tests/test_agent_cli.py @@ -13,9 +13,11 @@ from pathlib import Path import pytest +import yaml +from cyclopts.exceptions import MissingArgumentError # pyright: ignore[reportMissingImports] from tablassert.agent import ENV_API_BASE, ENV_API_KEY, ENV_MODEL_ID, load_optimized_instructions, save_optimized_instructions -from tablassert.cli import APP, agent +from tablassert.cli import APP, agent, rebuild_agent_graph def test_agent_command_registered() -> None: @@ -361,3 +363,60 @@ def __init__( assert captured[0]["max_tokens"] == 16000 # default timeout bounds each request so a stalled connection cannot hang the optimizer assert captured[0]["timeout"] == 600 + + +# --------------------------------------------------------------------------- # +# rebuild-agent-graph: the shared-registry reconstruction command (wiring only) +# --------------------------------------------------------------------------- # + + +def test_rebuild_agent_graph_command_registered() -> None: + """The ``rebuild-agent-graph`` subcommand is registered as a flat peer of ``build-kg``.""" + assert "rebuild-agent-graph" in APP.resolved_commands() + + +def test_rebuild_agent_graph_flags_parse(tmp_path: Path) -> None: + """``--state-dir``/``-sd`` + required ``--fullmap``/``-f`` bind; the state-dir default is pinned.""" + + def parse(argv: list[str]) -> dict[str, object]: + fn, bound, _ = APP.parse_args(argv, exit_on_error=False) + assert fn is rebuild_agent_graph + bound.apply_defaults() # bound.arguments only carries explicitly-parsed tokens + return dict(bound.arguments) + + arguments = parse(["rebuild-agent-graph", "--state-dir", str(tmp_path), "--fullmap", "/tmp/fm.redb"]) + assert arguments["state_dir"] == tmp_path + assert arguments["fullmap"] == Path("/tmp/fm.redb") + + alias_arguments = parse(["rebuild-agent-graph", "-sd", str(tmp_path), "-f", "/tmp/fm.redb"]) + assert alias_arguments["state_dir"] == tmp_path + assert alias_arguments["fullmap"] == Path("/tmp/fm.redb") + + default_arguments = parse(["rebuild-agent-graph", "-f", "/tmp/fm.redb"]) + assert default_arguments["state_dir"] == Path(".tablassert") / "agent" + + with pytest.raises(MissingArgumentError): + APP.parse_args(["rebuild-agent-graph", "--state-dir", str(tmp_path)], exit_on_error=False) + + +def test_rebuild_agent_graph_rebuilds_and_reports(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + """The command rebuilds the registry from ``state.json`` and prints the path + entry count.""" + from tablassert.agent import ConfigRecord, SupervisorState, save_state + + config: Path = tmp_path / "configs" / "PMC1.yaml" + config.parent.mkdir() + config.write_text("sections: []\n") + save_state( + tmp_path, SupervisorState(pmc_ids=["PMC1"], records={"PMC1": ConfigRecord(pmc_id="PMC1", status="MAPPED", best_config_path=str(config))}) + ) + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + rebuild_agent_graph(state_dir=tmp_path, fullmap=fullmap) + + out: str = capsys.readouterr().out + assert str(tmp_path / "graph.yaml") in out + assert "1 table config" in out + data: object = yaml.safe_load((tmp_path / "graph.yaml").read_text()) + assert isinstance(data, dict) + assert data["tables"] == [str(config.resolve())] diff --git a/tests/test_agent_storage.py b/tests/test_agent_storage.py index 1eb2c61..8337a02 100644 --- a/tests/test_agent_storage.py +++ b/tests/test_agent_storage.py @@ -277,7 +277,7 @@ def fake_fetch(pmc_id: str, outdir: Path, *, timeout: int = 120) -> list[Path]: # BOTH configs live in the ONE dedicated configs/ folder, and the record points there. assert best.is_file(), "the BEST config must be written to configs/.yaml" assert derived.is_file(), "the derived config must be written to configs/.derived.yaml" - assert rec.best_config_path == str(best), "best_config_path must point into configs/" + assert Path(str(rec.best_config_path)).resolve() == best.resolve(), "best_config_path must point into configs/" assert rec.config_path == str(best), "config_path must point at the BEST config in configs/" # state.json stays at the state-dir ROOT, never inside configs/. assert (state_dir / "state.json").is_file(), "state.json must persist at the state-dir root" diff --git a/tests/test_agent_supervisor.py b/tests/test_agent_supervisor.py index cc92d9c..52937f3 100644 --- a/tests/test_agent_supervisor.py +++ b/tests/test_agent_supervisor.py @@ -21,6 +21,7 @@ from tablassert import rs from tablassert.agent import ConfigRecord, SupervisorState, load_state, make_fake_model, run_supervisor, save_state +from tablassert.models import Graph pytest.importorskip("smolagents") @@ -995,3 +996,195 @@ def test_supervisor_local_payload_no_table_skipped(tmp_path: Path, fullmap_db: P ) rec: ConfigRecord = result["records"]["PMC1"] # pyright: ignore[reportIndexIssue] assert rec.status == "SKIPPED" + + +# --------------------------------------------------------------------------- # +# Shared graph registry wiring: successful builds self-register into graph.yaml +# --------------------------------------------------------------------------- # + + +def _read_registered(state_dir: Path) -> Graph: + """Load ``/graph.yaml`` and validate it as a Graph (asserts it exists).""" + graph_path: Path = state_dir / "graph.yaml" + assert graph_path.is_file(), "the shared registry must exist after a successful build" + data: object = yaml.safe_load(graph_path.read_text()) + assert isinstance(data, dict) + return Graph.model_validate(data) + + +def test_supervisor_mapped_registers_in_graph_yaml(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A MAPPED build self-registers: one ABSOLUTE tables entry + the resolved fullmap, first-wins.""" + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\nbrca1\tmapk1\n") + _patch_fetch(monkeypatch, table) + good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + state_dir: Path = tmp_path / "state" + + result = run_supervisor( + ["PMC1"], + fullmap=fullmap_db, + build_model_factory=lambda: make_fake_model(final_yaml=good_yaml), + map_threshold=0.8, + state_dir=state_dir, + workdir=tmp_path / "w", + ) + rec: ConfigRecord = result["records"]["PMC1"] # pyright: ignore[reportIndexIssue] + assert rec.status == "MAPPED" + graph: Graph = _read_registered(state_dir) + assert len(graph.tables) == 1 + assert graph.tables[0] == Path(str(rec.best_config_path)).resolve() + assert graph.tables[0].is_absolute() + assert graph.fullmap == fullmap_db.resolve() + + +def test_supervisor_built_unmeasured_registers(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """BUILT_UNMEASURED is a successful build, so it self-registers exactly like MAPPED.""" + import tablassert.agent as agent_mod + + table: Path = _write_table(tmp_path, "d.tsv", "brca1\tmapk1\n") + _patch_fetch(monkeypatch, table) + good_yaml: str = yaml.safe_dump(_column_cfg(table)) + state_dir: Path = tmp_path / "state" + + monkeypatch.setattr( + agent_mod, + "build_and_audit", + lambda *a, **k: { + "ok": True, + "coverage_pct": 0.0, + "measured": False, + "qc_pass_rate": None, + "errors": [], + "error_codes": [], + "kgx_path": None, + "edges_path": None, + "node_count": 1, + "edge_count": 1, + "unresolved": [], + }, + ) + monkeypatch.setattr(agent_mod, "map_coverage", lambda *a, **k: {"overall": 0.0, "measured": False, "per_column": {}, "unresolved": []}) + monkeypatch.setattr(agent_mod, "propose_config_edit", lambda cfg, rep: (good_yaml, "no safe edit")) + + result = run_supervisor( + ["PMC1"], + fullmap=fullmap_db, + build_model_factory=lambda: make_fake_model(final_yaml=good_yaml), + map_threshold=0.8, + state_dir=state_dir, + workdir=tmp_path / "w", + ) + assert result["records"]["PMC1"].status == "BUILT_UNMEASURED" # pyright: ignore[reportIndexIssue] + graph: Graph = _read_registered(state_dir) + assert len(graph.tables) == 1 + assert graph.tables[0].name == "PMC1.yaml" + + +def test_supervisor_skipped_does_not_register(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A SKIPPED build never touches the registry: no ``graph.yaml`` is created.""" + table: Path = _write_table(tmp_path, "bad.tsv", "brca1\tzzznotreal\nbrca1\tzzznotreal\n") + _patch_fetch(monkeypatch, table) + bad_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + state_dir: Path = tmp_path / "state" + + result = run_supervisor( + ["PMC1"], + fullmap=fullmap_db, + build_model_factory=lambda: make_fake_model(final_yaml=bad_yaml), + map_threshold=1.0, + max_improve_iters=0, + state_dir=state_dir, + workdir=tmp_path / "w", + ) + assert result["records"]["PMC1"].status == "SKIPPED" # pyright: ignore[reportIndexIssue] + assert not (state_dir / "graph.yaml").exists(), "SKIPPED must never register" + + +def test_supervisor_registration_failure_keeps_status(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A failing registry write NEVER flips a successful status: MAPPED is kept + noted.""" + import tablassert.agent as agent_mod + + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\nbrca1\tmapk1\n") + _patch_fetch(monkeypatch, table) + good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + state_dir: Path = tmp_path / "state" + + def boom(state_dir_: Path, pmc_id: str, config_path: Path, fullmap_: Path) -> Path: # pyright: ignore[reportUnusedParameter] + raise RuntimeError("registry lock poisoned") + + monkeypatch.setattr(agent_mod, "register_build", boom) + + result = run_supervisor( + ["PMC1"], + fullmap=fullmap_db, + build_model_factory=lambda: make_fake_model(final_yaml=good_yaml), + map_threshold=0.8, + state_dir=state_dir, + workdir=tmp_path / "w", + ) + rec: ConfigRecord = result["records"]["PMC1"] # pyright: ignore[reportIndexIssue] + assert rec.status == "MAPPED", "registration failure must never flip a successful status" + assert "graph registry update failed" in rec.notes + assert "registry lock poisoned" in rec.notes + # The failure is persisted in the checkpoint too (the note survives a reload). + reloaded: SupervisorState | None = load_state(state_dir) + assert reloaded is not None + assert "graph registry update failed" in reloaded.records["PMC1"].notes + + +def test_supervisor_relative_state_dir_rebuilds_from_any_cwd(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A RELATIVE --state-dir stores an ABSOLUTE best_config_path, so rebuild works from any CWD. + + Regression: ``rebuild_graph`` checks ``Path(best_config_path).is_file()`` against ITS OWN cwd; + a CWD-relative checkpoint entry would silently prune every entry when the rebuild runs elsewhere. + """ + import contextlib + + from tablassert.graph_registry import rebuild_graph + + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\nbrca1\tmapk1\n") + _patch_fetch(monkeypatch, table) + good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + + with contextlib.chdir(tmp_path): + result = run_supervisor( + ["PMC1"], + fullmap=fullmap_db, + build_model_factory=lambda: make_fake_model(final_yaml=good_yaml), + map_threshold=0.8, + state_dir=Path("rel-state"), # RELATIVE state dir (the CLI default is relative too) + workdir=tmp_path / "w", + ) + rec: ConfigRecord = result["records"]["PMC1"] # pyright: ignore[reportIndexIssue] + assert rec.status == "MAPPED" + assert Path(str(rec.best_config_path)).is_absolute(), "the checkpoint must persist an absolute config path" + + elsewhere: Path = tmp_path / "elsewhere" + elsewhere.mkdir() + with contextlib.chdir(elsewhere): + rebuild_graph(tmp_path / "rel-state", fullmap_db) + + data: object = yaml.safe_load((tmp_path / "rel-state" / "graph.yaml").read_text()) + assert isinstance(data, dict) + tables: object = data["tables"] + assert isinstance(tables, list), "the entry must survive a cross-CWD rebuild" + assert len(tables) == 1, "the entry must survive a cross-CWD rebuild" + + +def test_supervisor_resume_skip_keeps_registry_entry(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A re-run that SKIPS an already-MAPPED pmc (resume skips terminal records) keeps its entry.""" + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\nbrca1\tmapk1\n") + _patch_fetch(monkeypatch, table) + good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + state_dir: Path = tmp_path / "state" + + def factory() -> object: + return make_fake_model(final_yaml=good_yaml) + + first = run_supervisor(["PMC1"], fullmap=fullmap_db, build_model_factory=factory, map_threshold=0.8, state_dir=state_dir, workdir=tmp_path / "w") + assert first["records"]["PMC1"].status == "MAPPED" # pyright: ignore[reportIndexIssue] + before: Graph = _read_registered(state_dir) + + second = run_supervisor(["PMC1"], fullmap=fullmap_db, build_model_factory=factory, map_threshold=0.8, state_dir=state_dir, workdir=tmp_path / "w") + assert second["records"]["PMC1"].status == "MAPPED" # pyright: ignore[reportIndexIssue] + after: Graph = _read_registered(state_dir) + assert after.tables == before.tables, "the resume-skipped record keeps its existing registry entry" diff --git a/tests/test_docs_cli_coverage.py b/tests/test_docs_cli_coverage.py index adbd56a..cae2284 100644 --- a/tests/test_docs_cli_coverage.py +++ b/tests/test_docs_cli_coverage.py @@ -25,12 +25,14 @@ "agent": ("agent.md",), "build-fullmap": ("cli.md", "fullmap.md"), "build-kg": ("cli.md",), + "rebuild-agent-graph": ("cli.md",), "validate": ("cli.md",), } COMMAND_FLAG_DOCS: dict[str, tuple[str, ...]] = { "agent": ("agent.md", "cli.md"), "build-fullmap": ("cli.md", "fullmap.md"), "build-kg": ("cli.md",), + "rebuild-agent-graph": ("cli.md",), "validate": ("cli.md",), } diff --git a/tests/test_graph_registry.py b/tests/test_graph_registry.py new file mode 100644 index 0000000..ed68388 --- /dev/null +++ b/tests/test_graph_registry.py @@ -0,0 +1,313 @@ +"""Tests for the concurrency-safe SHARED graph registry (``/graph.yaml``). + +Several ``tablassert agent`` processes pointed at the same ``--state-dir`` self-register their +successful builds into ONE aggregate graph config. Covered here: upsert/dedupe/replace-by-pmc +semantics, fresh creation, corrupt-file quarantine + self-healing rebuild, the first-wins fullmap +rule (with a warning on mismatch), TRUE multi-process concurrency over one registry path, and +``rebuild_graph`` reconstruction/pruning from ``state.json``. The registry module needs no extras; +``rebuild_graph`` reaches ``tablassert.agent.load_state`` lazily, which is lazy-import safe in the +base environment too. +""" + +from __future__ import annotations + +import multiprocessing +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tablassert.agent import ConfigRecord, SupervisorState, save_state +from tablassert.graph_registry import GRAPH_DESCRIPTION, GRAPH_NAME, GRAPH_VERSION, rebuild_graph, register_build +from tablassert.models import Graph + + +def _make_config(tmp_path: Path, pmc_id: str, body: str = "sections: []\n") -> Path: + """Write a stand-in best-config file for ``pmc_id`` (content is irrelevant to the registry).""" + config: Path = tmp_path / "configs" / f"{pmc_id}.yaml" + config.parent.mkdir(parents=True, exist_ok=True) + config.write_text(body) + return config + + +def _read_registry(state_dir: Path) -> dict[str, Any]: + """Parse the registry YAML and assert it satisfies ``Graph.model_validate``.""" + path: Path = state_dir / "graph.yaml" + assert path.is_file(), "the registry must exist" + data: object = yaml.safe_load(path.read_text()) + assert isinstance(data, dict) + Graph.model_validate(data) + return data + + +def test_register_build_creates_fresh_registry(tmp_path: Path) -> None: + """First registration creates the template registry with one ABSOLUTE entry + resolved fullmap.""" + state_dir: Path = tmp_path / "state" + config: Path = _make_config(tmp_path, "PMC1") + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + path: Path = register_build(state_dir, "PMC1", config, fullmap) + + assert path == state_dir / "graph.yaml" + data: dict[str, Any] = _read_registry(state_dir) + assert data["name"] == GRAPH_NAME + assert data["version"] == GRAPH_VERSION + assert data["description"] == GRAPH_DESCRIPTION + assert data["tables"] == [str(config.resolve())] + assert Path(data["tables"][0]).is_absolute() + assert data["fullmap"] == str(fullmap.resolve()) + assert not (state_dir / "graph.yaml.tmp").exists(), "the atomic write must not leave a tmp behind" + + +def test_register_build_replaces_same_pmc_and_preserves_order(tmp_path: Path) -> None: + """Re-registering a pmc REPLACES its entry (matched by basename stem); others keep their order.""" + state_dir: Path = tmp_path / "state" + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + first: Path = _make_config(tmp_path, "PMC1", "v: 1\n") + other: Path = _make_config(tmp_path, "PMC2") + + register_build(state_dir, "PMC1", first, fullmap) + register_build(state_dir, "PMC2", other, fullmap) + + # Re-run of PMC1 with a NEW config path: the old entry is dropped, the new one appended last. + rerun: Path = tmp_path / "configs-rerun" / "PMC1.yaml" + rerun.parent.mkdir() + rerun.write_text("v: 2\n") + register_build(state_dir, "PMC1", rerun, fullmap) + + data: dict[str, Any] = _read_registry(state_dir) + assert data["tables"] == [str(other.resolve()), str(rerun.resolve())] + assert [Path(str(t)).stem for t in data["tables"]].count("PMC1") == 1, "exactly one entry per pmc id" + assert str(first.resolve()) not in data["tables"], "the replaced entry must be gone" + + +def test_register_build_preserves_unrelated_entries(tmp_path: Path) -> None: + """Entries whose basename stem differs from the pmc id survive the upsert untouched.""" + state_dir: Path = tmp_path / "state" + state_dir.mkdir() + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + foreign: Path = _make_config(tmp_path, "hand-authored") + seed: dict[str, Any] = { + "name": GRAPH_NAME, + "version": GRAPH_VERSION, + "description": GRAPH_DESCRIPTION, + "tables": [str(foreign.resolve())], + "fullmap": str(fullmap.resolve()), + } + (state_dir / "graph.yaml").write_text(yaml.safe_dump(seed, sort_keys=False)) + + config: Path = _make_config(tmp_path, "PMC9") + register_build(state_dir, "PMC9", config, fullmap) + + data: dict[str, Any] = _read_registry(state_dir) + assert data["tables"] == [str(foreign.resolve()), str(config.resolve())] + + +@pytest.mark.parametrize( + "payload", + [ + "name: [unclosed", # YAML parse error + "- just\n- a list\n", # top level is not a mapping + yaml.safe_dump({"name": "not-a-graph"}), # valid mapping, fails Graph.model_validate + ], + ids=["yaml-error", "not-a-mapping", "schema-invalid"], +) +def test_corrupt_registry_quarantined_and_rebuilt(payload: str, tmp_path: Path) -> None: + """A corrupt registry is renamed ``graph.yaml.corrupt-`` and rebuilt fresh (self-healing).""" + state_dir: Path = tmp_path / "state" + state_dir.mkdir() + (state_dir / "graph.yaml").write_text(payload) + config: Path = _make_config(tmp_path, "PMC1") + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + path: Path = register_build(state_dir, "PMC1", config, fullmap) + + quarantined: list[Path] = sorted(state_dir.glob("graph.yaml.corrupt-*")) + assert len(quarantined) == 1, "exactly one quarantine file" + assert quarantined[0].read_text() == payload, "the corrupt bytes are preserved" + data: dict[str, Any] = _read_registry(state_dir) + assert path == state_dir / "graph.yaml" + assert data["tables"] == [str(config.resolve())], "the rebuild starts fresh" + + +def test_corrupt_registry_quarantine_on_rebuild_graph(tmp_path: Path) -> None: + """``rebuild_graph`` self-heals the same way before reconstructing from ``state.json``.""" + state_dir: Path = tmp_path / "state" + state_dir.mkdir() + (state_dir / "graph.yaml").write_text("{{{::: not yaml") + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + rebuild_graph(state_dir, fullmap) + + assert len(list(state_dir.glob("graph.yaml.corrupt-*"))) == 1 + data: dict[str, Any] = _read_registry(state_dir) + assert data["tables"] == [], "no state.json -> empty tables" + assert data["fullmap"] == str(fullmap.resolve()) + + +def test_fullmap_first_wins_and_warns_on_mismatch(tmp_path: Path) -> None: + """The first fullmap recorded stays; a differing later fullmap is rejected with a warning.""" + from tablassert.log import logger + + state_dir: Path = tmp_path / "state" + fm1: Path = tmp_path / "fm1.redb" + fm2: Path = tmp_path / "fm2.redb" + fm1.touch() + fm2.touch() + register_build(state_dir, "PMC1", _make_config(tmp_path, "PMC1"), fm1) + + records: list[str] = [] + sink_id: int = logger.add(lambda message: records.append(str(message)), level="WARNING") + try: + register_build(state_dir, "PMC2", _make_config(tmp_path, "PMC2"), fm2) + finally: + logger.remove(sink_id) + + data: dict[str, Any] = _read_registry(state_dir) + assert data["fullmap"] == str(fm1.resolve()), "fullmap is FIRST-WINS" + assert any("first-wins" in record for record in records), f"expected a fullmap-mismatch warning, got {records}" + + # A repeat of the SAME fullmap never warns. + records.clear() + sink_id = logger.add(lambda message: records.append(str(message)), level="WARNING") + try: + register_build(state_dir, "PMC3", _make_config(tmp_path, "PMC3"), fm1) + finally: + logger.remove(sink_id) + assert not any("first-wins" in record for record in records) + + +def _concurrent_register(args: tuple[str, str, str, str]) -> str: + """Worker: register ONE pmc config from a child process (module-level => picklable).""" + state_dir, pmc_id, config_path, fullmap = args + from tablassert.graph_registry import register_build as worker_register + + worker_register(Path(state_dir), pmc_id, Path(config_path), Path(fullmap)) + return pmc_id + + +def test_concurrent_registrations_converge(tmp_path: Path) -> None: + """TRUE concurrency: >=8 processes registering DISTINCT pmc ids at ONE path -> exactly N entries. + + Every process contends on the same ``graph.yaml.lock``; the flock + atomic write must serialize + them so no registration is lost, duplicated, or torn (the result parses and validates). + """ + state_dir: Path = tmp_path / "shared" + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + n: int = 10 + tasks: list[tuple[str, str, str, str]] = [] + for i in range(n): + pmc_id: str = f"PMC{i}" + config: Path = _make_config(tmp_path, pmc_id) + tasks.append((str(state_dir), pmc_id, str(config), str(fullmap))) + + with multiprocessing.Pool(processes=n) as pool: + done: list[str] = pool.map(_concurrent_register, tasks) + + assert sorted(done) == sorted(f"PMC{i}" for i in range(n)) + data: dict[str, Any] = _read_registry(state_dir) # parses + Graph.model_validate passes + tables: list[Any] = data["tables"] + assert len(tables) == n, "no lost updates" + assert len(set(tables)) == n, "no duplicates" + assert {Path(str(t)).stem for t in tables} == {f"PMC{i}" for i in range(n)} + assert all(Path(str(t)).is_absolute() for t in tables) + assert data["fullmap"] == str(fullmap.resolve()) + + +def _seed_state(state_dir: Path, records: dict[str, ConfigRecord]) -> None: + """Persist a supervisor checkpoint with the given records.""" + save_state(state_dir, SupervisorState(pmc_ids=sorted(records), records=records)) + + +def test_rebuild_graph_prunes_and_excludes(tmp_path: Path) -> None: + """``rebuild_graph`` keeps registered statuses with existing configs, sorted; prunes everything else.""" + state_dir: Path = tmp_path / "state" + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + pmc1: Path = _make_config(tmp_path, "PMC1") # MAPPED, exists + pmc2: Path = _make_config(tmp_path, "PMC2") # BUILT_UNMEASURED, exists + _make_config(tmp_path, "PMC3") # MAPPED, but deleted before rebuild + (tmp_path / "configs" / "PMC3.yaml").unlink() + _make_config(tmp_path, "PMC4") # SKIPPED, exists on disk but must NOT register + stale: Path = _make_config(tmp_path, "stale-leftover") + + # Pre-seed the registry with a stale entry that no record supports (must be pruned). + state_dir.mkdir() + seed: dict[str, Any] = { + "name": GRAPH_NAME, + "version": GRAPH_VERSION, + "description": GRAPH_DESCRIPTION, + "tables": [str(stale.resolve())], + "fullmap": str(fullmap.resolve()), + } + (state_dir / "graph.yaml").write_text(yaml.safe_dump(seed, sort_keys=False)) + + _seed_state( + state_dir, + { + "PMC1": ConfigRecord(pmc_id="PMC1", status="MAPPED", best_config_path=str(pmc1)), + "PMC2": ConfigRecord(pmc_id="PMC2", status="BUILT_UNMEASURED", best_config_path=str(pmc2)), + "PMC3": ConfigRecord(pmc_id="PMC3", status="MAPPED", best_config_path=str(tmp_path / "configs" / "PMC3.yaml")), + "PMC4": ConfigRecord(pmc_id="PMC4", status="SKIPPED", best_config_path=str(tmp_path / "configs" / "PMC4.yaml")), + }, + ) + + path: Path = rebuild_graph(state_dir, fullmap) + + assert path == state_dir / "graph.yaml" + data: dict[str, Any] = _read_registry(state_dir) + assert data["tables"] == [str(pmc1.resolve()), str(pmc2.resolve())], "sorted by pmc id; stale/missing/SKIPPED pruned" + + +def test_rebuild_graph_tolerates_corrupt_state_json(tmp_path: Path) -> None: + """A damaged ``state.json`` warns + rebuilds an empty registry instead of raising (recovery path).""" + state_dir: Path = tmp_path / "state" + state_dir.mkdir() + (state_dir / "state.json").write_text("{not json at all") + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + path: Path = rebuild_graph(state_dir, fullmap) # must NOT raise + + data: dict[str, Any] = _read_registry(state_dir) + assert path == state_dir / "graph.yaml" + assert data["tables"] == [] + assert data["fullmap"] == str(fullmap.resolve()) + + +def test_rebuild_graph_without_state_creates_empty_registry(tmp_path: Path) -> None: + """No ``state.json`` at all -> an empty-but-valid registry carrying the fullmap.""" + state_dir: Path = tmp_path / "state" + fullmap: Path = tmp_path / "fullmap.redb" + fullmap.touch() + + path: Path = rebuild_graph(state_dir, fullmap) + + assert path.is_file() + data: dict[str, Any] = _read_registry(state_dir) + assert data["tables"] == [] + assert data["fullmap"] == str(fullmap.resolve()) + + +def test_rebuild_graph_is_first_wins_for_fullmap(tmp_path: Path) -> None: + """An existing registry fullmap survives a ``rebuild_graph`` that passes a different one.""" + state_dir: Path = tmp_path / "state" + fm1: Path = tmp_path / "fm1.redb" + fm2: Path = tmp_path / "fm2.redb" + fm1.touch() + fm2.touch() + config: Path = _make_config(tmp_path, "PMC1") + register_build(state_dir, "PMC1", config, fm1) + _seed_state(state_dir, {"PMC1": ConfigRecord(pmc_id="PMC1", status="MAPPED", best_config_path=str(config))}) + + rebuild_graph(state_dir, fm2) + + data: dict[str, Any] = _read_registry(state_dir) + assert data["fullmap"] == str(fm1.resolve()), "rebuild honors the first-wins fullmap"