From dc0b093850d4278f103198f6fe1a8828fc1173e6 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 14 Aug 2026 12:11:00 -0700 Subject: [PATCH] feat(agent)!: build into a caller-owned graph config instead of the internal registry --- docs/agent.md | 128 +++++------ docs/cli.md | 49 +---- examples/agent/README.md | 4 +- examples/agent/qc/qc_report.py | 21 +- examples/agent/qc/qc_reviewer.py | 6 +- src/tablassert/agent.py | 360 ++++++++++++++++++++++--------- src/tablassert/cli.py | 85 ++++---- src/tablassert/graph_registry.py | 215 ------------------ src/tablassert/graph_target.py | 135 ++++++++++++ tests/test_agent_cli.py | 150 ++++++------- tests/test_agent_docs.py | 2 +- tests/test_agent_supervisor.py | 252 +++++++++------------- tests/test_docs_cli_coverage.py | 2 - tests/test_graph_registry.py | 313 --------------------------- tests/test_graph_target.py | 112 ++++++++++ 15 files changed, 780 insertions(+), 1054 deletions(-) delete mode 100644 src/tablassert/graph_registry.py create mode 100644 src/tablassert/graph_target.py delete mode 100644 tests/test_graph_registry.py create mode 100644 tests/test_graph_target.py diff --git a/docs/agent.md b/docs/agent.md index 739a643..5442ade 100644 --- a/docs/agent.md +++ b/docs/agent.md @@ -88,10 +88,10 @@ pipeline on an article you already hold locally (e.g. a non-open-access paper), ```bash # one directory used for every PMC id -tablassert agent PMC11708054 --fullmap ./fullmap --local ./payloads/PMC11708054 +tablassert agent PMC11708054 --configuration-file ./graph.yaml --local ./payloads/PMC11708054 # per-article directories -tablassert agent PMC1 PMC2 --fullmap ./fullmap --local PMC1=./payloads/p1 PMC2=./payloads/p2 +tablassert agent PMC1 PMC2 --configuration-file ./graph.yaml --local PMC1=./payloads/p1 PMC2=./payloads/p2 ``` A local payload directory holds the table(s) and (optionally) the article main text. When `--local` is @@ -127,15 +127,16 @@ With network access and a configured endpoint: ```bash tablassert agent PMC11708054 PMC12345678 \ - --fullmap ./fullmap \ + --configuration-file ./graph.yaml \ --map-threshold 0.25 \ --max-improve-iters 3 \ --max-steps 20 \ --state-dir .tablassert/agent ``` -Flags: `--max-steps`/`-ms`, `--map-threshold`/`-mt`, `--max-improve-iters`/`-mi`, -`--state-dir`/`-sd`, `--backend {openai,litellm}`/`-b`, plus `--local`/`-l`, `--reflexion`, +The required target is `--configuration-file`/`-f`; it supplies the fullmap, graph identity, RIG, +artifact metadata, and existing table list. Flags: `--max-steps`/`-ms`, `--map-threshold`/`-mt`, +`--max-improve-iters`/`-mi`, `--state-dir`/`-sd`, `--backend {openai,litellm}`/`-b`, plus `--local`/`-l`, `--reflexion`, `--judge-model`, `--judge-threshold`, `--biolink-threshold`, and the `--optimize`/`-o` prompt-optimization flags (`--instructions-file`, `--instructions-out`, `--max-metric-calls`, `--dataset`). The [CLI reference — `agent`](cli.md#agent) is the authoritative flag table; the list here is a compact @@ -254,92 +255,59 @@ each section independently from its own coverage entry. A single-table paper is one section. State and storage stay **per-paper**: one best config (`configs/.yaml`) holding all sections, with `section_coverages` recorded for visibility. -### Workspace layout & checkpoint / resume +### Workspace layout, target graph, and checkpoint / rerun -`tablassert agent` uses a **single stable workspace root** — `state_dir` (default `.tablassert/agent`, -override with `--state-dir`). The CLI never sets a separate artifact root, so the checkpoint, the configs, -the fetched downloads, and the build outputs **all co-locate** under it: +`--configuration-file` is the caller-owned Graph YAML that the agent updates in place. Its `fullmap`, +`name`, `version`, complete `rig:`, and artifact metadata drive every one-table audit. The agent does +**not** create an aggregate graph under `state_dir`; `state_dir` remains only the checkpoint and working +artifact directory (default `.tablassert/agent`, override with `--state-dir`): ```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 - builds// # KGX agent_0.0.1.{nodes,edges}.ndjson + table.yaml + graph.yaml + .tablassert/store — stable +project/graph.yaml # caller-owned aggregate graph, updated in place +.tablassert/agent/ # checkpoint/artifact workspace + state.json # supervisor checkpoint (atomic) + configs/.yaml # accepted generated table config (absolute source.local) + configs/.derived.yaml # initial generated config + downloads///... # fetched PMC payload; stable across runs + builds//table.yaml # temporary one-table audit input + builds//artifacts/ # _.{nodes,edges}.ndjson + RIG + builds//.tablassert/store/ # temporary parquet cache ``` -| 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, biolink_valid_pct, demoted_edge_pct}` 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 | -| `builds//` | KGX artifacts: `agent_0.0.1.{nodes,edges}.ndjson`, `table.yaml`, `graph.yaml`, `.tablassert/store` | **stable** — the built graph for the article | - -Re-running the same command **resumes** from the checkpoint: records already `MAPPED`/`SKIPPED` are -skipped. The `downloads/` payload persists on disk across runs. +Only newly generated agent table configs are normalized: every section's `source.local` is written as +an absolute local/data-lake path, and the graph's new `tables` entry is an absolute path. Existing +user-authored table YAMLs and their source paths are not rewritten. The target graph's existing metadata +and unrelated table entries are preserved. -### Reusing agent outputs with the full pipeline +A result is appended to the target graph only when it is `MAPPED` or `BUILT_UNMEASURED`. `SKIPPED` +articles never append. If the same PMC is processed again, its old table entry is replaced and the new +absolute config path is appended. Requested PMCs are deliberately processed again even when `state.json` +contains a terminal record; this makes reruns effective while retaining attempts, coverage history, and +metrics. A failed rerun does not replace the prior successful config. -The best config's `source.local` points at the downloaded table under `downloads//`, so the full -(non-agent) pipeline can reuse the agent's output **without re-fetching**. The agent already writes a -ready-to-build `graph.yaml` (wrapping `table.yaml` with the resolved fullmap) into `builds//`: +The agent audits each candidate with a one-table in-process graph, so it does not rebuild every table +already present in the target graph. The temporary audit inherits the target graph's semantic metadata +and graph identity but writes physical artifacts to an isolated per-article workspace. Build the complete +aggregate explicitly after the agent finishes: ```bash -cd .tablassert/agent/builds/PMC11708054 -tablassert build-kg -f graph.yaml +tablassert agent PMC11708054 --configuration-file ./graph.yaml --state-dir .tablassert/agent +# inspect graph.yaml, then build every existing + generated table together +tablassert build-kg -f ./graph.yaml ``` -!!! warning "Not relocatable" - `source.local` in the best config is an **absolute** path into `downloads//`. The workspace is - 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: +!!! warning "Absolute paths are intentional" + Generated `tables` entries and generated `source.local` values are absolute so the target graph can + be built from any current working directory. Moving the data lake, downloaded payload, or workspace + requires updating those generated paths or rerunning the agent. -```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 -``` +### Concurrent agents targeting one graph -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). +Several agent processes may target the same caller-owned graph. Each successful append takes an exclusive +`.lock` sidecar lock and atomically replaces the graph YAML, so distinct PMCs do not lose one +another's entries and a same-PMC rerun has deterministic last-writer-wins replacement. The checkpoint +`state.json` read-modify-write is still per-workspace and is not cross-process locked; use separate +`state_dir` values for concurrent processes unless they intentionally coordinate their article ids. ## The tools @@ -429,13 +397,13 @@ process-global), so a higher thread count does not speed up the expensive build/ ```bash # optimize the agent prompt over a dataset of examples, writing the result to a file -tablassert agent PMC11708054 --fullmap ./fullmap --optimize \ +tablassert agent PMC11708054 --configuration-file ./graph.yaml --optimize \ --dataset examples/gepa-dataset.yaml --task-model qwen-flash \ --max-metric-calls 30 --gepa-threads 4 \ --instructions-out .tablassert/agent/optimized_instructions.yaml # later, run the supervisor with the optimized prompt -tablassert agent PMC11708054 --fullmap ./fullmap \ +tablassert agent PMC11708054 --configuration-file ./graph.yaml \ --instructions-file .tablassert/agent/optimized_instructions.yaml ``` diff --git a/docs/cli.md b/docs/cli.md index c6e75a2..66341b9 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -1,8 +1,8 @@ # CLI Reference Tablassert extracts knowledge assertions from tabular data into KGX NDJSON. The `tablassert` app -exposes **five subcommands** — `agent`, `build-fullmap`, `build-kg`, `rebuild-agent-graph`, -`validate` — plus an app-level `--version` flag. Run `tablassert --help` (or ` --help`) +exposes **five subcommands** — `agent`, `build-fullmap`, `build-kg`, `validate`, and `validate-kgx` — +plus an app-level `--version` flag. Run `tablassert --help` (or ` --help`) for the live surface. ## Command index @@ -12,7 +12,6 @@ for the live surface. | [`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 | | [`validate-kgx`](#validate-kgx) | Validate built KGX NDJSON against the Biolink Model | @@ -41,16 +40,19 @@ before any model is built or article fetched, so a missing extra is reported wit command instead of surfacing mid-run — see [When an extra is missing](installation.md#when-an-extra-is-missing). ```bash -tablassert agent --fullmap PATH [OPTIONS] PMC-IDS... +tablassert agent PMC-IDS... --configuration-file GRAPH.yaml [OPTIONS] ``` -PMC ids are passed positionally (also accepted as `--pmc-ids`). This page lists the flags; see +PMC ids are passed positionally (also accepted as `--pmc-ids`). The required graph target is accepted +as `--configuration-file` or `-f`; it is modified in place after successful article builds. The graph's +`fullmap`, name, version, RIG, and artifact metadata replace the old standalone fullmap argument. This +page lists the flags; see [Agent](agent.md) for the full pipeline, workspace layout, checkpoint/resume, and tooling. | Option | Type | Required | Default | Description | | --- | --- | --- | --- | --- | | `PMC-IDS` (`--pmc-ids`) | list[str] | Yes | — | One or more PMC article ids (positional) | -| `--fullmap`, `-f` | Path | Yes | — | Fullmap redb file or base directory | +| `--configuration-file`, `-f` | Path | Yes | — | Caller-owned Graph YAML; supplies build metadata/fullmap and receives successful absolute table entries | | `--model-id`, `-m` | str | No | `None` | Model id (env `TABLASSERT_AGENT_MODEL_ID`) | | `--api-base`, `-ab` | str | No | `None` | OpenAI-compatible base URL (env `TABLASSERT_AGENT_API_BASE`) | | `--api-key`, `-ak` | str | No | `None` | API key secret (env `TABLASSERT_AGENT_API_KEY`) | @@ -73,7 +75,9 @@ PMC ids are passed positionally (also accepted as `--pmc-ids`). This page lists | `--gepa-threads` | int | No | `None` | Thread count for GEPA's evaluation pool (`--optimize`) — parallelizes candidate LM forward passes only; coverage-scoring builds stay serialized on `_GEPA_BUILD_LOCK` | ```bash -tablassert agent PMC11708054 --fullmap ./fullmap +tablassert agent PMC11708054 --configuration-file ./graph.yaml +# equivalent short form: +tablassert agent PMC11708054 -f ./graph.yaml ``` !!! warning "Secrets" @@ -162,37 +166,6 @@ or incomplete RIG fails the build with `[rig-validation-failed]` and nothing is --- -## 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/examples/agent/README.md b/examples/agent/README.md index 7ed60b5..bcf8edd 100644 --- a/examples/agent/README.md +++ b/examples/agent/README.md @@ -10,7 +10,7 @@ These artifacts come from running the Tablassert `[agent]` GEPA prompt-optimizat optimization cost in production: ```bash - tablassert agent PMC11947420 --fullmap /path/to/fullmap \ + tablassert agent PMC11947420 --configuration-file /path/to/graph.yaml \ --instructions-file examples/agent/optimized_instructions.yaml ``` @@ -41,7 +41,7 @@ export TABLASSERT_AGENT_MODEL_ID="qwen3.8-max-preview" # strong reflection L export TABLASSERT_AGENT_API_BASE="https://YOUR-ENDPOINT/v1" export TABLASSERT_AGENT_API_KEY="sk-***" -tablassert agent PMC11947420 --fullmap /path/to/fullmap --optimize \ +tablassert agent PMC11947420 --configuration-file /path/to/graph.yaml --optimize \ --dataset examples/agent/gepa-dataset.yaml \ --task-model qwen3.6-flash \ --max-metric-calls 30 --gepa-threads 4 \ diff --git a/examples/agent/qc/qc_report.py b/examples/agent/qc/qc_report.py index 35ce97c..d7449d3 100644 --- a/examples/agent/qc/qc_report.py +++ b/examples/agent/qc/qc_report.py @@ -73,20 +73,23 @@ def load_config(pmc: str) -> tuple[str, dict] | None: return None -def kg_counts(pmc: str) -> tuple[int, int]: +def artifact_path(pmc: str, suffix: str) -> Path | None: + """Find a generated artifact, preferring the target-identity artifacts directory.""" bdir = STATE_DIR / "builds" / pmc - n = e = 0 - np_, ep = bdir / "agent_0.0.1.nodes.ndjson", bdir / "agent_0.0.1.edges.ndjson" - if np_.is_file(): - n = sum(1 for line in np_.open() if line.strip()) - if ep.is_file(): - e = sum(1 for line in ep.open() if line.strip()) + candidates = sorted((bdir / "artifacts").glob(f"*.{suffix}")) + sorted(bdir.glob(f"*.{suffix}")) + return next((path for path in candidates if path.is_file()), None) + + +def kg_counts(pmc: str) -> tuple[int, int]: + np_, ep = artifact_path(pmc, "nodes.ndjson"), artifact_path(pmc, "edges.ndjson") + n = sum(1 for line in np_.open() if line.strip()) if np_ else 0 + e = sum(1 for line in ep.open() if line.strip()) if ep else 0 return n, e def sample_edges(pmc: str, k: int = 5) -> list[dict]: - ep = STATE_DIR / "builds" / pmc / "agent_0.0.1.edges.ndjson" - if not ep.is_file(): + ep = artifact_path(pmc, "edges.ndjson") + if ep is None: return [] out = [] with ep.open() as fh: diff --git a/examples/agent/qc/qc_reviewer.py b/examples/agent/qc/qc_reviewer.py index f3a7003..5a1db7e 100644 --- a/examples/agent/qc/qc_reviewer.py +++ b/examples/agent/qc/qc_reviewer.py @@ -173,8 +173,10 @@ def get_table_summary(config: dict) -> str: def get_edges(pmc: str, k: int = 8) -> str: - ep = STATE_DIR / "builds" / pmc / "agent_0.0.1.edges.ndjson" - if not ep.is_file(): + bdir = STATE_DIR / "builds" / pmc + candidates = sorted((bdir / "artifacts").glob("*.edges.ndjson")) + sorted(bdir.glob("*.edges.ndjson")) + ep = next((path for path in candidates if path.is_file()), None) + if ep is None: return "(no edges built)" rows = [] with ep.open() as fh: diff --git a/src/tablassert/agent.py b/src/tablassert/agent.py index 2d5f993..b412a8e 100644 --- a/src/tablassert/agent.py +++ b/src/tablassert/agent.py @@ -39,10 +39,10 @@ from tablassert.errors import GraphValidationError, QcRuntimeMissingError, SectionValidationError, TablassertValidationError from tablassert.extras import install_command, require_module from tablassert.fullmap import distinct, fullmap_db_path, is_lock_contention, lookup_rows -from tablassert.graph_registry import REGISTERED_STATUSES, register_build +from tablassert.graph_target import append_successful_config from tablassert.lib import Tcode from tablassert.log import cat -from tablassert.models import NodeEncoding, Section +from tablassert.models import Graph, NodeEncoding, Section from tablassert.progress import flatten_pydantic_error if TYPE_CHECKING: @@ -62,6 +62,8 @@ logger = cat("AGENT") +SUCCESSFUL_STATUSES: frozenset[str] = frozenset({"MAPPED", "BUILT_UNMEASURED"}) + def _require(name: str) -> None: """Import an optional dependency or raise a loud, actionable ImportError. @@ -794,6 +796,45 @@ def validate_table_config(cfg: str, agent_memory: object = None, agent: object = return error is None +def normalize_agent_table_config(config_yaml: str, *, base_dirs: Sequence[Path] = ()) -> str: + """Return an agent-generated table config with absolute ``source.local`` values. + + Only the newly generated config is rewritten. Existing table YAMLs referenced by + the caller's graph are never opened or modified. Relative source paths are resolved + against the supplied bases in order; the first existing candidate wins, otherwise + the first base still produces a deterministic absolute path and the normal build + validation reports a missing source. + """ + data: object = yaml.safe_load(config_yaml) + if not isinstance(data, dict): + raise ValueError("config is not a YAML mapping") + + bases: tuple[Path, ...] = tuple(Path(base).expanduser().resolve() for base in base_dirs) + default_base: Path = bases[0] if bases else Path.cwd() + + def absolute_local(value: object) -> str: + candidate: Path = Path(str(value)).expanduser() + if candidate.is_absolute(): + return str(candidate.resolve()) + possibilities: list[Path] = [(base / candidate).resolve() for base in bases] + existing = next((path for path in possibilities if path.is_file()), None) + return str(existing or (default_base / candidate).resolve()) + + def visit(value: object) -> None: + if isinstance(value, dict): + source: object = value.get("source") + if isinstance(source, dict) and "local" in source: + source["local"] = absolute_local(source["local"]) + for nested in value.values(): + visit(nested) + elif isinstance(value, list): + for nested in value: + visit(nested) + + visit(data) + return yaml.safe_dump(data, sort_keys=False) + + def make_derive_config_tool() -> Tool: """Build the ``derive_config`` smolagents Tool lazily (imports smolagents on first call). @@ -1251,23 +1292,32 @@ def _biolink_report(nodes: Path, edges: Path) -> dict[str, object]: def build_and_audit( - config_yaml: str, *, fullmap: Path, name: str = "agent", version: str = "0.0.1", qc: bool = False, head: bool = False, workdir: Path | None = None + config_yaml: str, + *, + graph: Graph | None = None, + fullmap: Path | None = None, + name: str = "agent", + version: str = "0.0.1", + qc: bool = False, + head: bool = False, + workdir: Path | None = None, ) -> dict[str, object]: """Validate, build, (QC), and score a config in ONE deterministic call. - Runs the REAL ``validate_pipeline`` + ``build_pipeline`` (headless ``_NullProgress``) - inside an isolated ``workdir`` (``contextlib.chdir``), then measures fullmap - coverage via :func:`map_coverage`. The build writes ``_.nodes.ndjson`` - / ``.edges.ndjson`` to the CWD, so the pipelines run inside ``workdir`` (after - ``mkdir -p workdir/.tablassert/store``, mirroring the e2e recipe) and the artifacts - land there. + Runs the REAL validate/build stages (headless ``_NullProgress``) inside an isolated + ``workdir`` (``contextlib.chdir``), then measures fullmap coverage via + :func:`map_coverage`. When ``graph`` is supplied, the build uses a one-table copy of + that graph, retaining its name/version/full RIG/fullmap while redirecting physical + artifacts to the isolated workdir. The legacy ``fullmap``/``name``/``version`` inputs + remain available for direct callers that do not yet supply a graph. Args: config_yaml: A Tablassert Section/table config YAML; a bare merged section is auto-wrapped as ``{template:
}``. - fullmap: Fullmap redb file or base directory (see ``fullmap_db_path``). - name: Graph name (drives the output artifact prefix). - version: Graph version label (drives the output artifact prefix). + graph: Prepared target graph whose metadata drives a one-table temporary build. + fullmap: Legacy fullmap redb file or base directory when ``graph`` is omitted. + name: Legacy graph name when ``graph`` is omitted. + version: Legacy graph version when ``graph`` is omitted. qc: When True, run the build's quality-control audit. head: When True, preview-build a random sample of up to 5 rows per section (fast; the ``--head`` lever) for intermediate improve-loop scoring. Coverage is still measured on @@ -1312,52 +1362,77 @@ def build_and_audit( # to_sections (used by the pipelines) sees the table-config shape it expects. table_cfg: dict[str, object] = data if ("template" in data or "sections" in data) else {"template": data} - (root / "table.yaml").write_text(yaml.safe_dump(table_cfg, sort_keys=False)) - # The measurement build still emits a RIG (every build does), so it carries an - # honest minimal rig block: the agent only mines PMC open-access tables, and the - # artifacts live in this throwaway workdir (file:// base = unpublished). - resolved_root: str = str(root.resolve()) - graph_cfg: dict[str, object] = { - "name": name, - "version": version, - "tables": ["table.yaml"], # relative to workdir (the pipelines chdir there) - "fullmap": str(fullmap), - "rig": { - "source_info": { - "infores_id": f"infores:{name.lower().replace('_', '-')}", - "name": f"Agent-built measurement graph {name}", - "terms_of_use_info": { - "terms_of_use_url": "https://pmc.ncbi.nlm.nih.gov/about/copyright/", - "terms_of_use_description": "PubMed Central open-access supplementary table; individual article licenses apply.", + table_path: Path = root / "table.yaml" + table_path.write_text(yaml.safe_dump(table_cfg, sort_keys=False)) + + # Direct callers from the pre-target-graph API can still score a config with a + # standalone fullmap. Normal agent runs always pass the prepared target Graph and + # therefore never synthesize metadata here. + if graph is None: + if fullmap is None: + return _fail(["build_and_audit requires a graph or fullmap"]) + resolved_root: str = str(root.resolve()) + graph_cfg: dict[str, object] = { + "name": name, + "version": version, + "tables": [str(table_path)], + "fullmap": str(fullmap), + "rig": { + "source_info": { + "infores_id": f"infores:{name.lower().replace('_', '-')}", + "name": f"Agent-built measurement graph {name}", + "terms_of_use_info": { + "terms_of_use_url": "https://pmc.ncbi.nlm.nih.gov/about/copyright/", + "terms_of_use_description": "PubMed Central open-access supplementary table; individual article licenses apply.", + }, + "data_access_locations": ["PubMed Central - https://pmc.ncbi.nlm.nih.gov/"], + "source_status": "unknown", }, - "data_access_locations": ["PubMed Central - https://pmc.ncbi.nlm.nih.gov/"], - "source_status": "unknown", - }, - "ingest_info": { - "utility": f"Transient measurement graph used to score agent-derived table configs for {name}.", - "scope": "Associations mined from one PMC supplementary table config under audit.", + "ingest_info": { + "utility": f"Transient measurement graph used to score agent-derived table configs for {name}.", + "scope": "Associations mined from one PMC supplementary table config under audit.", + }, + "provenance_info": {"contributions": ["Tablassert agent: automated config derivation and measurement build"]}, + "artifact_base_url": f"file://{resolved_root}", + "artifact_base_path": resolved_root, }, - "provenance_info": {"contributions": ["Tablassert agent: automated config derivation and measurement build"]}, - "artifact_base_url": f"file://{resolved_root}", - "artifact_base_path": resolved_root, - }, - } - (root / "graph.yaml").write_text(yaml.safe_dump(graph_cfg, sort_keys=False)) + } + build_graph: Graph = Graph.model_validate(graph_cfg) + else: + build_graph = graph.model_copy(deep=True) + # The target RIG describes the aggregate graph. Its semantic metadata is + # retained, while only the physical artifact directory is isolated so each + # one-table audit cannot overwrite another article's output. + build_graph.rig.artifact_base_path = root / "artifacts" + + build_graph.tables = [table_path] + build_graph_path: Path = root / "agent-graph.yaml" - from tablassert.cli import build_pipeline, validate_pipeline # deferred: keeps the cli APP off the module top + from tablassert.cli import build_graph_pipeline, validate_pipeline # deferred: keeps the cli APP off the module top try: with contextlib.chdir(root): (root / ".tablassert" / "store").mkdir(parents=True, exist_ok=True) validate_pipeline(Path("table.yaml"), _NullProgress()) # pyright: ignore[reportArgumentType] - build_pipeline(Path("graph.yaml"), _NullProgress(), qc=qc, head=head) # pyright: ignore[reportArgumentType] + build_graph_pipeline( + build_graph, + build_graph_path, + _NullProgress(), # pyright: ignore[reportArgumentType] + qc=qc, + head=head, + audit_sources=False, + ) except (GraphValidationError, SectionValidationError, TablassertValidationError, QcRuntimeMissingError) as exc: return _err(exc) except pydantic.ValidationError as exc: return _err(exc) - nodes: Path = root / f"{name}_{version}.nodes.ndjson" - edges: Path = root / f"{name}_{version}.edges.ndjson" + artifact_dir: Path = Path(build_graph.rig.artifact_base_path) + output_name: str = build_graph.name + output_version: str = build_graph.version + build_fullmap: Path = build_graph.fullmap + nodes: Path = artifact_dir / f"{output_name}_{output_version}.nodes.ndjson" + edges: Path = artifact_dir / f"{output_name}_{output_version}.edges.ndjson" # Coverage is NON-fatal: the KG already built, so a bad fullmap (or any coverage # failure) keeps ok=True with coverage_pct=0.0 and a note, never masking success. @@ -1375,7 +1450,7 @@ def build_and_audit( # resolves against root (the build's CWD) — measuring from the original CWD would fail # the frame reproduction and report a false/unmeasurable coverage. with contextlib.chdir(root): - cov: dict[str, object] = map_coverage(table_cfg, fullmap=fullmap, workdir=root) + cov: dict[str, object] = map_coverage(table_cfg, fullmap=build_fullmap, workdir=root) overall: object = cov.get("overall") coverage_pct = float(overall) if isinstance(overall, (int, float)) else 0.0 measured = bool(cov.get("measured")) @@ -1423,15 +1498,19 @@ def build_and_audit( def make_build_and_audit_tool( - get_fullmap: Callable[[], Path], *, name: str = "agent", version: str = "0.0.1", qc: bool = False, head: bool = False + get_fullmap: Callable[[], Path] | None = None, + *, + graph: Graph | None = None, + name: str = "agent", + version: str = "0.0.1", + qc: bool = False, + head: bool = False, ) -> Tool: - """Build the ``build_and_audit`` smolagents Tool lazily, binding the fullmap via closure. + """Build the ``build_and_audit`` tool with a target graph or legacy fullmap closure. - ``get_fullmap`` is a zero-arg callable returning the fullmap redb path (the - supervisor supplies it when assembling tools); ``forward(config_yaml)`` returns the - JSON-encoded audit report so the agent validates, builds, (QC), and scores a config - in a single call. The subclass is defined INSIDE this factory so the module top - never forces the optional smolagents import. + Normal agent runs pass ``graph`` so the one-table audit inherits the caller's complete + graph metadata. ``get_fullmap`` remains for direct/tool API compatibility outside the + target-graph supervisor. """ _require("smolagents") from smolagents import Tool # local import keeps module import lazy # pyright: ignore[reportMissingImports] @@ -1451,7 +1530,13 @@ class BuildAndAuditTool(Tool): # pyright: ignore[reportMissingImports] output_type = "string" def forward(self, config_yaml: str) -> str: - return json.dumps(build_and_audit(config_yaml, fullmap=get_fullmap(), name=name, version=version, qc=qc, head=head), default=str) + if graph is not None: + report = build_and_audit(config_yaml, graph=graph, qc=qc, head=head) + else: + if get_fullmap is None: + raise ValueError("make_build_and_audit_tool requires graph or get_fullmap") + report = build_and_audit(config_yaml, fullmap=get_fullmap(), name=name, version=version, qc=qc, head=head) + return json.dumps(report, default=str) return BuildAndAuditTool() @@ -2169,8 +2254,10 @@ def predicate_cheatsheet(pairs: Sequence[tuple[str, str]] = CHEATSHEET_PAIRS) -> Emit exactly ONE table config as YAML shaped as {template: {...}, sections: [...]}. The `template` carries the shared per-article PROVENANCE (repo + publication id) and NOTHING else — in particular NO `source` (each section owns its source). The `sections` list has ONE entry per -mappable table/worksheet; each section supplies its OWN `source` (the table's local path + that -file's source.url, plus sheet/row_slice/delimiter as needed) and its OWN `statement`. Within each +mappable table/worksheet; each section supplies its OWN `source` (the table's ABSOLUTE local/data-lake path + that +file's source.url, plus sheet/row_slice/delimiter as needed) and its OWN `statement`. Copy the exact +absolute candidate path shown by the task into every `source.local`; never emit a relative local path. +Within each section choose column-letter encodings for entity columns and literal CURIEs for fixed values; pick a predicate the subject/object pair actually permits (see BIOLINK MODELING below); add statistical annotations (p_value / effect_size / effect_type) when that table has them — @@ -2502,7 +2589,7 @@ def generate( # config (agent.run -> final_answer, gated by validate_section); the improve loop is # deterministic Python (propose_config_edit -> build_and_audit -> accept IFF strictly # better, so coverage_history is monotonic non-decreasing). State checkpoints atomically -# to /state.json so a crashed batch resumes, skipping terminal records +# to /state.json so a crashed batch resumes while requested terminal records are rerunnable # (DONE/MAPPED/SKIPPED). One bad pmc never aborts the batch: the whole per-pmc body is # wrapped in try/except -> status=SKIPPED with the reason. Only the inner agent.run needs # the [agent] extra; the state dataclasses + load/save are pure stdlib. @@ -2579,17 +2666,20 @@ def forward(self, source: str) -> str: def make_tools( *, - fullmap: Path, + graph: Graph | None = None, + fullmap: Path | None = None, table_path: Path | None = None, # pyright: ignore[reportUnusedParameter] # reserved for future table-bound tools; read_table takes source from the LLM name: str = "agent", version: str = "0.0.1", qc: bool = False, derive_mode: DeriveMode = "full", ) -> list[object]: - """Assemble the fullmap-bound smolagents tools the supervisor hands to the inner agent. + """Assemble the graph/fullmap-bound tools the supervisor hands to the inner agent. - ``get_fullmap = lambda: fullmap`` binds the redb path via closure so each tool's ``forward`` - needs only the LLM-provided args. Returns ``[read_table, pmc_article_context, derive_config, + A supplied ``graph`` binds the complete target metadata to ``build_and_audit`` while + its resolved fullmap remains available to coverage tools. The legacy ``fullmap`` path + is accepted for direct callers outside the target-graph supervisor. Returns + ``[read_table, pmc_article_context, derive_config, build_and_audit, map_coverage, propose_config_edit]``. All construction is offline-safe (no network, no model I/O); the smolagents import happens lazily inside each factory. ``table_path`` is accepted for API symmetry with the supervisor call site (the read_table tool reads whatever ``source`` the @@ -2609,9 +2699,14 @@ def make_tools( rebuild on the next lookup. """ + if graph is None and fullmap is None: + raise ValueError("make_tools requires graph or fullmap") + + bound_fullmap: Path = graph.fullmap if graph is not None else cast(Path, fullmap) + def get_fullmap() -> Path: """Return the bound fullmap redb path the tools read.""" - return fullmap + return bound_fullmap if derive_mode == "derive_only": return [make_read_table_tool(), make_pmc_article_context_tool(), make_derive_config_tool()] @@ -2621,7 +2716,7 @@ def get_fullmap() -> Path: make_read_table_tool(), make_pmc_article_context_tool(), make_derive_config_tool(), - make_build_and_audit_tool(get_fullmap, name=name, version=version, qc=qc), + make_build_and_audit_tool(graph=graph, get_fullmap=None if graph is not None else get_fullmap, name=name, version=version, qc=qc), make_map_coverage_tool(get_fullmap), make_propose_config_edit_tool(), ] @@ -2825,7 +2920,9 @@ def _is_improvement(current_cov: float, current_report: dict[str, object], new_c def run_supervisor( pmc_ids: list[str] | str, *, - fullmap: Path, + graph: Graph | None = None, + graph_path: Path | None = None, + fullmap: Path | None = None, build_model_factory: Callable[[], object], map_threshold: float = 0.25, max_improve_iters: int = 3, @@ -2844,7 +2941,7 @@ def run_supervisor( ) -> dict[str, object]: """Run the deterministic supervisor over a batch of PMC ids with checkpoint/resume. - For each pmc id (resume-aware: terminal DONE/MAPPED/SKIPPED records are skipped): + For each requested pmc id (including ids with terminal records from an earlier invocation): 1. mark RUNNING + checkpoint; fetch the latest-version article payload (``fetch_pmc_article``, the single seam tests monkeypatch) and present ALL candidate tables + the main-text path to the agent; 2. run the INNER agent (``build_agent`` + ``build_model_factory()``) whose schema-gated @@ -2859,6 +2956,11 @@ def run_supervisor( ``judge_model`` is configured — judge score ≥ ``judge_threshold``), BUILT_UNMEASURED (built but coverage unmeasurable), or SKIPPED. + ``graph`` is the prepared caller-owned target graph for normal agent runs. The legacy + ``fullmap``/``name``/``version`` arguments remain available for direct callers while the + target-graph migration settles. ``graph_path`` identifies the YAML to update after a + successful result. + ``biolink_threshold`` defaults to 0.0 (report-only): every record carries its ``biolink_valid_pct`` / ``demoted_edge_pct`` regardless, and raising the threshold turns that measurement into a terminal gate. @@ -2876,12 +2978,45 @@ def run_supervisor( verbosity = None ids: list[str] = [pmc_ids] if isinstance(pmc_ids, str) else list(pmc_ids) + target_graph: Graph | None = graph + if target_graph is not None and graph_path is None: + raise ValueError("run_supervisor requires graph_path when graph is supplied") + if target_graph is None: + if fullmap is None: + raise ValueError("run_supervisor requires graph or fullmap") + effective_fullmap: Path = fullmap + effective_name: str = name + effective_version: str = version + else: + effective_fullmap = target_graph.fullmap + effective_name = target_graph.name + effective_version = target_graph.version + assert effective_fullmap is not None + art_root: Path = artifact_root(state_dir, workdir) + source_bases: tuple[Path, ...] = tuple( + dict.fromkeys( + path.expanduser().resolve() for path in (Path.cwd(), state_dir, art_root, graph_path.parent if graph_path is not None else state_dir) + ) + ) + + def normalize_config(config_yaml: str) -> str: + """Normalize only the generated candidate, never target graph table files.""" + return normalize_agent_table_config(config_yaml, base_dirs=source_bases) + + def audit_config(config_yaml: str, **kwargs: Any) -> dict[str, object]: + """Build one candidate with the target graph, or legacy scalar metadata.""" + normalized: str = normalize_config(config_yaml) + if target_graph is not None: + return build_and_audit(normalized, graph=target_graph, **kwargs) # pyright: ignore[reportArgumentType] + return build_and_audit(normalized, fullmap=effective_fullmap, name=effective_name, version=effective_version, **kwargs) # pyright: ignore[reportArgumentType] + art_root.mkdir(parents=True, exist_ok=True) loaded: SupervisorState | None = load_state(state_dir) state: SupervisorState = loaded if loaded is not None else SupervisorState(pmc_ids=list(ids)) - # Resume merge: keep existing records (so terminal statuses are skipped) and add any new ids. + # Merge checkpoint records and add any new ids. Terminal records are deliberately retained + # for history but are processed again below on every requested invocation. for pid in ids: if pid not in state.records: state.records[pid] = ConfigRecord(pmc_id=pid) @@ -2892,8 +3027,6 @@ def run_supervisor( all_metrics: list[dict[str, object]] = [] for pmc_id in ids: rec: ConfigRecord = state.records[pmc_id] - if rec.status in {"DONE", "MAPPED", "SKIPPED", "BUILT_UNMEASURED", "DERIVED"}: - continue # resume: already terminal try: rec.status = "RUNNING" rec.attempts += 1 @@ -2931,7 +3064,14 @@ def run_supervisor( metrics: dict[str, object] = {} agent: object = build_agent( model=build_model_factory(), - tools=make_tools(fullmap=fullmap, table_path=tables[0], name=name, version=version, derive_mode=derive_mode), + tools=make_tools( + graph=target_graph, + fullmap=effective_fullmap, + table_path=tables[0], + name=effective_name, + version=effective_version, + derive_mode=derive_mode, + ), max_steps=max_steps, step_callbacks=[make_step_callback(metrics)], verbosity_level=verbosity, @@ -2949,20 +3089,32 @@ def run_supervisor( f"Candidate tables:\n{table_list}\n" "Inspect candidates with read_table(path): for an Excel file it lists ALL worksheets (pass sheet='' to " "read one, and set source.sheet in the config). Choose the table and worksheet that yield the cleanest " - "subject-predicate-object mapping, then author and build the config. Maximize fullmap mapping coverage; return the config YAML." + "subject-predicate-object mapping, then author and build the config. Copy the exact ABSOLUTE candidate " + "path into every source.local; never emit a relative local/data-lake path. Maximize fullmap mapping coverage; " + "return the config YAML." ) result: object = agent.run(task) # pyright: ignore[reportAttributeAccessIssue] - config: str = str(result) + raw_config: str = str(result) all_metrics.append(metrics) - if not validate_table_config(config): # the final-answer gate should prevent this; be safe + # Check the raw model response before path normalization so malformed YAML gets the same + # actionable final-answer-gate status as a schema-invalid mapping. Normalization is only + # applied after that gate and is then checked once more because it mutates newly generated + # source.local values. + if not validate_table_config(raw_config): # the final-answer gate should prevent this; be safe rec.status = "SKIPPED" rec.notes = "SKIPPED: agent final answer failed the validate_table_config gate." save_state(state_dir, state) continue + config: str = normalize_config(raw_config) + if not validate_table_config(config): + rec.status = "SKIPPED" + rec.notes = "SKIPPED: normalized agent answer failed the validate_table_config gate." + save_state(state_dir, state) + continue configs_dir(state_dir).mkdir(parents=True, exist_ok=True) - derived_path: Path = derived_config_path(state_dir, pmc_id) + derived_path: Path = derived_config_path(state_dir, pmc_id).resolve() derived_path.write_text(config) rec.config_path = str(derived_path) @@ -2974,7 +3126,7 @@ def run_supervisor( save_state(state_dir, state) continue - report: dict[str, object] = build_and_audit(config, fullmap=fullmap, name=name, version=version, workdir=pmc_build_dir(art_root, pmc_id)) + report: dict[str, object] = audit_config(config, workdir=pmc_build_dir(art_root, pmc_id)) raw_cov: object = report.get("coverage_pct") coverage: float = float(raw_cov) if isinstance(raw_cov, (int, float)) else 0.0 rec.coverage_history.append(coverage) @@ -3000,7 +3152,7 @@ def run_supervisor( improve_tmp: Path = pmc_build_dir(art_root, pmc_id) / ".improve-tmp" while current_cov < map_threshold and iters < max_improve_iters: try: - cov_report: dict[str, object] = map_coverage(current_config, fullmap=fullmap, workdir=pmc_build_dir(art_root, pmc_id)) + cov_report: dict[str, object] = map_coverage(current_config, fullmap=effective_fullmap, workdir=pmc_build_dir(art_root, pmc_id)) except Exception: # a coverage failure must not abort the improve attempt cov_report = {"per_column": {}, "unresolved": []} @@ -3008,15 +3160,12 @@ def run_supervisor( # Tier 1: deterministic ranked candidates (distinct edits), best-first. for edited, rationale in propose_config_candidates(current_config, cov_report): - head_report: dict[str, object] = build_and_audit( - edited, fullmap=fullmap, name=name, version=version, head=True, workdir=improve_tmp - ) + edited = normalize_config(edited) + head_report: dict[str, object] = audit_config(edited, head=True, workdir=improve_tmp) raw_cov2: object = head_report.get("coverage_pct") cov2: float = float(raw_cov2) if isinstance(raw_cov2, (int, float)) else 0.0 if _is_improvement(current_cov, current_report, cov2, head_report): # head looks better -> confirm with a FULL build - full_report: dict[str, object] = build_and_audit( - edited, fullmap=fullmap, name=name, version=version, workdir=pmc_build_dir(art_root, pmc_id) - ) + full_report: dict[str, object] = audit_config(edited, workdir=pmc_build_dir(art_root, pmc_id)) full_cov: object = full_report.get("coverage_pct") full_cov_f: float = float(full_cov) if isinstance(full_cov, (int, float)) else 0.0 # Commit IFF the full build actually succeeded AND beat the prior best. A failing or @@ -3041,15 +3190,12 @@ def run_supervisor( if not improved and reflexion_model_factory is not None: revised: str | None = llm_propose_config_edit(current_config, cov_report, task, model=reflexion_model_factory()) if revised is not None: - head_report3: dict[str, object] = build_and_audit( - revised, fullmap=fullmap, name=name, version=version, head=True, workdir=improve_tmp - ) + revised = normalize_config(revised) + head_report3: dict[str, object] = audit_config(revised, head=True, workdir=improve_tmp) raw_cov3: object = head_report3.get("coverage_pct") cov3: float = float(raw_cov3) if isinstance(raw_cov3, (int, float)) else 0.0 if _is_improvement(current_cov, current_report, cov3, head_report3): # head looks better -> confirm with a FULL build - full_report3: dict[str, object] = build_and_audit( - revised, fullmap=fullmap, name=name, version=version, workdir=pmc_build_dir(art_root, pmc_id) - ) + full_report3: dict[str, object] = audit_config(revised, workdir=pmc_build_dir(art_root, pmc_id)) full_cov3: object = full_report3.get("coverage_pct") full_cov3_f: float = float(full_cov3) if isinstance(full_cov3, (int, float)) else 0.0 # Same guard as tier 1: commit IFF the full build succeeded AND beat the prior best; @@ -3077,7 +3223,7 @@ def run_supervisor( # Record per-section coverages for visibility (W3 multi-section; best-effort, never aborts). try: - final_cov: dict[str, object] = map_coverage(current_config, fullmap=fullmap, workdir=pmc_build_dir(art_root, pmc_id)) + final_cov: dict[str, object] = map_coverage(current_config, fullmap=effective_fullmap, workdir=pmc_build_dir(art_root, pmc_id)) raw_sections: object = final_cov.get("sections") if isinstance(raw_sections, list): per_section: list[float] = [] @@ -3089,12 +3235,7 @@ def run_supervisor( except Exception: # visibility-only; a measurement failure must not abort the run pass - best_path: Path = best_config_path(state_dir, pmc_id) - best_path.write_text(current_config) - # 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) + current_config = normalize_config(current_config) # Record the best build's Biolink compliance whether or not it gates, so state.json # always shows whether this paper's KGX is actually consumable downstream. rec.biolink_valid_pct = biolink_validity_metric(current_report) @@ -3139,17 +3280,28 @@ 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: + # Only a successful terminal result may replace the stable best config. A failed + # rerun therefore leaves both the previous file and its target-graph entry intact. + best_path: Path | None = None + if rec.status in SUCCESSFUL_STATUSES: + best_path = best_config_path(state_dir, pmc_id).resolve() + best_path.parent.mkdir(parents=True, exist_ok=True) + config_tmp: Path = best_path.with_name(f".{best_path.name}.tmp") + config_tmp.write_text(current_config) + os.replace(config_tmp, best_path) + rec.best_config_path = str(best_path) + rec.config_path = str(best_path) + + # Normal runs update the caller-owned graph. Legacy direct callers retain + # their scalar build behavior but do not have an aggregate target to mutate. 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}" + if target_graph is not None: + if graph_path is None: + raise ValueError("graph_path is required when graph is supplied") + append_successful_config(graph_path, pmc_id, best_path) + except Exception as reg_exc: # a graph update failure is a note, never a status change + logger.error("target graph update failed for {pmc}: {error}", pmc=pmc_id, error=reg_exc) + note: str = f"target graph 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 diff --git a/src/tablassert/cli.py b/src/tablassert/cli.py index 4dcc706..06c6a62 100644 --- a/src/tablassert/cli.py +++ b/src/tablassert/cli.py @@ -118,24 +118,36 @@ def _load_graph(configuration_file: Path) -> Graph: def build_pipeline( configuration_file: Path, progress: PipelineProgress, release: bool = False, qc: bool = False, log: bool = False, head: bool = False ) -> None: - """Build a knowledge graph from a YAML configuration file. + """Load a graph YAML and build it through the shared in-process core.""" + graph: Graph = _load_graph(configuration_file) + build_graph_pipeline(graph, configuration_file, progress, release=release, qc=qc, log=log, head=head) + + +def build_graph_pipeline( + graph: Graph, + configuration_file: Path, + progress: PipelineProgress, + release: bool = False, + qc: bool = False, + log: bool = False, + head: bool = False, + audit_sources: bool = True, +) -> None: + """Build a validated :class:`Graph` without loading another graph YAML. - Runs the six-stage build pipeline: load tables → extract sections → build - Tcodes → collect instructions → build subgraphs → compile graph. With ``qc`` - enabled a seventh stage studies the final NDJSON files. + The public ``build-kg`` command uses :func:`build_pipeline`, while agent audits pass + a one-table temporary ``Graph`` here. Keeping the core in-process lets those audits + reuse the production stages and metadata without building every table in the caller's + target graph. Args: - configuration_file: Path to the graph YAML file. + graph: Validated graph model controlling tables, fullmap, identity, and RIG. + configuration_file: Logical graph path used in validation error messages. progress: Pipeline progress reporter. release: When ``True``, emit release-mode artifacts. - qc: When ``True``, run quality-control audits on each section and assert - over the final NDJSON files (failing the build on any violation). + qc: When ``True``, run quality-control audits and final study assertions. log: When ``True``, enable per-section verbose logging. - head: When ``True``, preview a random sample of up to 5 rows per section (fast schema/shape check). - - Raises: - GraphValidationError: If the graph YAML fails Pydantic validation. - SectionValidationError: If any section fails Pydantic validation. + head: When ``True``, build a random sample of up to five rows per section. """ from tablassert.fullmap import fullmap_db_path from tablassert.lib import Tcode, compile_graph, compile_subgraph @@ -144,7 +156,7 @@ def build_pipeline( # Stage 1/6: load tables. progress.stage("Loading Tables") - g: Graph = _load_graph(configuration_file) + g: Graph = graph # imap_unordered yields in completion order, so each worker carries its input # index and we reassemble by index to keep raw[i] aligned with g.tables[i]. start, advance, _ = progress.section_loop(len(g.tables), "Load") @@ -231,7 +243,7 @@ def build_pipeline( start(f"{g.name} · v{g.version}") # on_phase drives the phase tag (scan → normalize → write-nodes → write-edges → dedup → rig); # on_subgraph ticks the bar once per subgraph, so the total is len(subgraphs). - compile_graph(subgraphs, g.name, g.version, g.rig, section_sources, on_phase=sub_step, on_subgraph=advance) + compile_graph(subgraphs, g.name, g.version, g.rig, section_sources if audit_sources else None, on_phase=sub_step, on_subgraph=advance) # Stage 7/7 (only with --qc): assert over the final NDJSON files. if qc: @@ -711,7 +723,7 @@ def validate_kgx_command( def agent( pmc_ids: Annotated[list[str], cyclopts.Parameter(allow_leading_hyphen=False)], *, - fullmap: Annotated[Path, cyclopts.Parameter(name=["--fullmap", "-f"])], + graph_configuration_file: Annotated[Path, cyclopts.Parameter(name=["--configuration-file", "-f"])], model_id: Annotated[str | None, cyclopts.Parameter(name=["--model-id", "-m"])] = None, api_base: Annotated[str | None, cyclopts.Parameter(name=["--api-base", "-ab"])] = None, api_key: Annotated[str | None, cyclopts.Parameter(name=["--api-key", "-ak"])] = None, @@ -740,7 +752,8 @@ def agent( supplementary tables -> an inner LLM agent derives a schema-gated Section config -> build_and_audit scores it -> a deterministic improve loop proposes/accepts edits IFF strictly better -> the config is accepted when coverage reaches ``--map-threshold`` or SKIPPED when the improve budget is exhausted. - State checkpoints to ``--state-dir`` so an interrupted batch resumes, skipping finished articles. + State checkpoints to ``--state-dir`` for downloads, artifacts, and run history; requested articles + are processed again on later invocations so a successful rerun can replace its target-graph entry. Model config comes from ``--model-id``/``--api-base``/``--api-key`` OR the ``TABLASSERT_AGENT_MODEL_ID`` / ``TABLASSERT_AGENT_API_BASE`` / ``TABLASSERT_AGENT_API_KEY`` environment variables (explicit flags win). @@ -749,7 +762,8 @@ def agent( Args: pmc_ids: One or more PMC article ids (positional). - fullmap: Fullmap redb file or base directory (required). + graph_configuration_file: Caller-owned Graph YAML to validate, use for metadata/fullmap, and + update in place after successful article builds. model_id: Model id (falls back to ``TABLASSERT_AGENT_MODEL_ID``). api_base: API base URL (falls back to ``TABLASSERT_AGENT_API_BASE``). api_key: API key (falls back to ``TABLASSERT_AGENT_API_KEY``). @@ -786,6 +800,9 @@ def agent( ``_GEPA_BUILD_LOCK`` (``os.chdir`` is process-global), so more threads do not speed up builds. """ from tablassert import agent as agent_mod + from tablassert.graph_target import prepare_graph + + prepared_graph = prepare_graph(graph_configuration_file) resolved_id, resolved_base, resolved_key = agent_mod.resolve_model_config(model_id, api_base, api_key) # Fail loud on any missing secret BEFORE building a model (so this path never touches smolagents). @@ -915,7 +932,8 @@ def parse_local(specs: list[str] | None) -> dict[str, Path] | Path | None: result: dict[str, object] = agent_mod.run_supervisor( list(pmc_ids), - fullmap=fullmap, + graph=prepared_graph.graph, + graph_path=prepared_graph.path, build_model_factory=build_model_factory, map_threshold=map_threshold, max_improve_iters=max_improve_iters, @@ -949,37 +967,6 @@ 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).") - - class PrebuiltFullmapUnavailable(Exception): """A prebuilt fullmap could not be fetched or extracted. diff --git a/src/tablassert/graph_registry.py b/src/tablassert/graph_registry.py deleted file mode 100644 index b28f803..0000000 --- a/src/tablassert/graph_registry.py +++ /dev/null @@ -1,215 +0,0 @@ -"""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" -#: Record statuses whose best config self-registers (both are SUCCESSFUL builds). -REGISTERED_STATUSES: frozenset[str] = frozenset({"MAPPED", "BUILT_UNMEASURED"}) - - -def _registry_rig(state_dir: Path) -> dict[str, Any]: - """The honest ``rig:`` block for the aggregate agent registry graph. - - Every fact here is mechanical: the agent only mines PubMed Central - open-access supplementary tables, so source terms/access describe PMC, and - the artifact bases point at the state directory itself (a ``file://`` base - is a valid unpublished URI; swap it for a public https base before sending - the generated RIG anywhere). - """ - resolved: str = str(state_dir.resolve()) - return { - "source_info": { - "infores_id": "infores:tablassert-agent", - "name": "PubMed Central open-access supplementary tables", - "description": "Aggregate of tabular associations mined from PubMed Central open-access supplementary files by the Tablassert agent.", - "terms_of_use_info": { - "terms_of_use_url": "https://pmc.ncbi.nlm.nih.gov/about/copyright/", - "terms_of_use_description": "PubMed Central open-access subset; individual article licenses apply.", - }, - "data_access_locations": ["PubMed Central - https://pmc.ncbi.nlm.nih.gov/"], - "source_status": "unknown", - }, - "ingest_info": { - "utility": "Aggregates agent-derived tabular knowledge assertions for Translator-style querying.", - "scope": "All agent-built table configs registered under this state directory.", - }, - "provenance_info": {"contributions": ["Tablassert agent: automated config derivation and build"]}, - "artifact_base_url": f"file://{resolved}", - "artifact_base_path": resolved, - } - - -@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(state_dir: Path) -> dict[str, Any]: - """A brand-new registry document (``_apply_fullmap`` fills the first-wins ``fullmap``).""" - return {"name": GRAPH_NAME, "version": GRAPH_VERSION, "tables": [], "rig": _registry_rig(state_dir)} - - -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(state_dir) - 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(state_dir) - if not isinstance(data, dict): - _quarantine(state_dir, f"top level is not a mapping (got {type(data).__name__})") - return _fresh_doc(state_dir) - 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(state_dir) - 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/src/tablassert/graph_target.py b/src/tablassert/graph_target.py new file mode 100644 index 0000000..b425719 --- /dev/null +++ b/src/tablassert/graph_target.py @@ -0,0 +1,135 @@ +"""Caller-owned graph configuration preparation and mutation helpers. + +The agent deliberately targets a graph YAML supplied by the user instead of creating an +agent-owned aggregate graph. This module keeps the path/validation boundary separate from +the agent loop so the CLI can fail before it builds a model or fetches a PMC payload. +""" + +from __future__ import annotations + +import contextlib +import fcntl +import os +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pydantic +import yaml + +from tablassert.errors import GraphValidationError +from tablassert.ingests import from_yaml +from tablassert.models import Graph +from tablassert.progress import flatten_pydantic_error + + +@dataclass(frozen=True) +class PreparedGraph: + """A validated target graph plus paths resolved for the in-process agent build. + + ``path`` is the absolute YAML file that will be modified in place. ``graph`` is a + validated copy used by agent execution: the target graph's semantic metadata is + preserved, while path-valued fields that the temporary build must access are resolved + relative to the graph file. The source YAML is never rewritten by preparation. + """ + + path: Path + graph: Graph + + +def _absolute_from_graph(path: Path, value: Path) -> Path: + """Resolve a graph-owned path relative to the graph YAML's directory.""" + candidate: Path = value.expanduser() + return (candidate if candidate.is_absolute() else path.parent / candidate).resolve() + + +def _read_valid_target(path: Path) -> dict[str, Any]: + """Read and validate a caller-owned Graph YAML without changing it.""" + raw: object = from_yaml(path) + if not isinstance(raw, dict): + raise GraphValidationError(path, f"expected a YAML mapping, got {type(raw).__name__}") + try: + Graph.model_validate(raw) + except pydantic.ValidationError as exc: + raise GraphValidationError(path, flatten_pydantic_error(exc)) from exc + return raw + + +def prepare_graph(configuration_file: Path) -> PreparedGraph: + """Resolve and validate an agent target graph before any model or article work. + + Relative ``fullmap`` and ``rig.artifact_base_path`` values are resolved against the + graph YAML's directory in the in-memory copy. Existing table YAMLs and their contents + are intentionally left untouched; newly generated agent configs are normalized by the + agent before persistence. + + Args: + configuration_file: User-supplied graph YAML path. + + Returns: + A :class:`PreparedGraph` with an absolute target path and validated build graph. + + Raises: + GraphValidationError: If the YAML does not satisfy the Graph model. + OSError: If the target graph cannot be read. + """ + target: Path = configuration_file.expanduser().resolve() + raw: dict[str, Any] = _read_valid_target(target) + graph: Graph = Graph.model_validate(raw) + + prepared: Graph = graph.model_copy(deep=True) + prepared.fullmap = _absolute_from_graph(target, prepared.fullmap) + prepared.rig.artifact_base_path = _absolute_from_graph(target, prepared.rig.artifact_base_path) + return PreparedGraph(path=target, graph=prepared) + + +@contextlib.contextmanager +def _target_lock(target: Path) -> Iterator[None]: + """Serialize target-graph read/modify/write operations with a sidecar ``flock``.""" + target.parent.mkdir(parents=True, exist_ok=True) + lock_path: Path = Path(f"{target}.lock") + 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 append_successful_config(target_graph: Path, pmc_id: str, config_path: Path) -> Path: + """Replace one PMC entry and append its new absolute table config in place. + + The target is reread while its sidecar lock is held, so concurrent agents cannot lose + one another's table entries. Existing graph metadata and unrelated table entries are + copied verbatim. A malformed target raises without quarantining or overwriting the + caller-owned file. + """ + target: Path = target_graph.expanduser().resolve() + config: Path = config_path.expanduser().resolve() + if not target.is_file(): + raise FileNotFoundError(f"Target graph configuration does not exist: {target}") + if not config.is_file(): + raise FileNotFoundError(f"Successful table configuration does not exist: {config}") + + with _target_lock(target): + document: dict[str, Any] = _read_valid_target(target) + tables: object = document.get("tables") + if not isinstance(tables, list): # Graph validation above makes this defensive only. + raise GraphValidationError(target, "tables field must be a list") + document["tables"] = [entry for entry in tables if Path(str(entry)).stem != pmc_id] + [str(config)] + try: + Graph.model_validate(document) + except pydantic.ValidationError as exc: + raise GraphValidationError(target, flatten_pydantic_error(exc)) from exc + + tmp: Path = target.with_name(f".{target.name}.tmp") + try: + tmp.write_text(yaml.safe_dump(document, sort_keys=False), encoding="utf-8") + os.replace(tmp, target) + finally: + tmp.unlink(missing_ok=True) + return target + + +__all__: tuple[str, ...] = ("PreparedGraph", "append_successful_config", "prepare_graph") diff --git a/tests/test_agent_cli.py b/tests/test_agent_cli.py index 79a08f4..d12175a 100644 --- a/tests/test_agent_cli.py +++ b/tests/test_agent_cli.py @@ -8,20 +8,51 @@ from __future__ import annotations +import os import sys +import tempfile import types from pathlib import Path import pytest import yaml -from cyclopts.exceptions import MissingArgumentError # pyright: ignore[reportMissingImports] +from cyclopts.exceptions import UnknownOptionError # pyright: ignore[reportMissingImports] from tablassert import extras 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, rebuild_agent_graph +from tablassert.cli import APP, agent from tablassert.errors import MissingExtraError +def _graph_path() -> Path: + """Return a valid caller-owned target graph for CLI wiring tests.""" + path: Path = Path(tempfile.gettempdir()) / f"tablassert-agent-cli-target-{os.getpid()}.yaml" + path.write_text( + yaml.safe_dump( + { + "name": "CLI_TARGET", + "version": "1.0.0", + "tables": [], + "fullmap": "/tmp/fm", + "rig": { + "source_info": { + "infores_id": "infores:cli-target", + "terms_of_use_info": {"license_name": "CC0"}, + "data_access_locations": ["Test - https://example.org/data"], + "source_status": "unknown", + }, + "ingest_info": {"utility": "CLI test.", "scope": "CLI test."}, + "provenance_info": {"contributions": ["Test"]}, + "artifact_base_url": "https://example.org/cli-target", + "artifact_base_path": "/tmp/cli-target-output", + }, + }, + sort_keys=False, + ) + ) + return path + + @pytest.fixture(autouse=True) def _extras_present(monkeypatch: pytest.MonkeyPatch) -> None: """Report every optional extra as installed for this module's wiring tests. @@ -56,7 +87,7 @@ def test_agent_no_secret_fails_loud(monkeypatch: pytest.MonkeyPatch, capsys: pyt monkeypatch.delenv(ENV_API_BASE, raising=False) monkeypatch.delenv(ENV_API_KEY, raising=False) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm")) + agent(["PMC1"], graph_configuration_file=_graph_path()) assert exc_info.value.code == 2 captured = capsys.readouterr() assert ENV_MODEL_ID in captured.err @@ -83,7 +114,7 @@ def test_agent_without_extra_names_the_install_command(monkeypatch: pytest.Monke monkeypatch.setattr("tablassert.agent.run_supervisor", lambda *a, **k: pytest.fail("supervisor ran without the extra")) with pytest.raises(MissingExtraError) as excinfo: - agent(["PMC1"], fullmap=Path("/tmp/fm")) + agent(["PMC1"], graph_configuration_file=_graph_path()) message: str = str(excinfo.value) assert "smolagents" in message @@ -102,7 +133,7 @@ def test_agent_optimize_without_optimize_extra_points_at_optimize(monkeypatch: p monkeypatch.setattr("tablassert.agent.run_gepa", lambda *a, **k: pytest.fail("GEPA ran without the extra")) with pytest.raises(MissingExtraError) as excinfo: - agent(["PMC1"], fullmap=Path("/tmp/fm"), optimize=True) + agent(["PMC1"], graph_configuration_file=_graph_path(), optimize=True) message: str = str(excinfo.value) assert 'pip install "tablassert[optimize]"' in message @@ -121,7 +152,7 @@ def test_agent_missing_secret_is_reported_before_missing_extra(monkeypatch: pyte monkeypatch.setattr(extras, "missing", lambda extra: ("smolagents",)) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm")) + agent(["PMC1"], graph_configuration_file=_graph_path()) assert exc_info.value.code == 2 assert ENV_MODEL_ID in capsys.readouterr().err @@ -132,7 +163,7 @@ def test_agent_env_fallback_and_forwarding(monkeypatch: pytest.MonkeyPatch, caps Why: the CLI is thin glue over ``run_supervisor``. With the three env vars set (and no flags), ``resolve_model_config`` must fill the model config from the environment, and the thresholds / - fullmap must reach the supervisor unchanged. ``run_supervisor`` and + target graph/path must reach the supervisor unchanged. ``run_supervisor`` and ``build_model`` are monkeypatched (module attributes the command looks up at call time) so no real agent runs; invoking the forwarded ``build_model_factory`` then proves the factory resolved the env config and handed it to ``build_model``. @@ -157,10 +188,12 @@ def fake_build_model(*args: object, **kwargs: object) -> object: monkeypatch.setattr("tablassert.agent.run_supervisor", fake_run_supervisor) monkeypatch.setattr("tablassert.agent.build_model", fake_build_model) - agent(["PMC1", "PMC2"], fullmap=Path("/tmp/fm"), map_threshold=0.7, max_improve_iters=5) + agent(["PMC1", "PMC2"], graph_configuration_file=_graph_path(), map_threshold=0.7, max_improve_iters=5) assert captured["pmc_ids"] == ["PMC1", "PMC2"] - assert captured["fullmap"] == Path("/tmp/fm") + target_graph = captured["graph"] + assert target_graph.fullmap == Path("/tmp/fm").resolve() # pyright: ignore[reportAttributeAccessIssue] + assert captured["graph_path"] == _graph_path().resolve() assert captured["map_threshold"] == 0.7 assert captured["max_improve_iters"] == 5 @@ -179,20 +212,24 @@ def fake_build_model(*args: object, **kwargs: object) -> object: def test_agent_cli_flag_parsing() -> None: """A full argv parses into the command's bound args WITHOUT executing the body. - Why: the documented UX is positional PMC ids plus flags. cyclopts' ``parse_args`` binds tokens to - the signature without running the function, so this proves ``agent PMC9 --fullmap ... --map-threshold`` - parses (positional list + required ``--fullmap`` + typed flags) with no model/network run. + Why: the documented UX is positional PMC ids plus a required target graph flag. cyclopts' ``parse_args`` + binds tokens to the signature without running the function, so this proves both target graph forms + parse and the removed fullmap flag is rejected. """ - fn, bound, _ = APP.parse_args(["agent", "PMC9", "--fullmap", "/tmp/fm", "--map-threshold", "0.5"], exit_on_error=False) + fn, bound, _ = APP.parse_args(["agent", "PMC9", "--configuration-file", "/tmp/graph.yaml", "--map-threshold", "0.5"], exit_on_error=False) assert fn is agent assert bound.args == (["PMC9"],) - assert bound.kwargs["fullmap"] == Path("/tmp/fm") + assert bound.kwargs["graph_configuration_file"] == Path("/tmp/graph.yaml") assert bound.kwargs["map_threshold"] == 0.5 + _, alias_bound, _ = APP.parse_args(["agent", "PMC9", "-f", "/tmp/graph.yaml"], exit_on_error=False) + assert alias_bound.kwargs["graph_configuration_file"] == Path("/tmp/graph.yaml") + with pytest.raises(UnknownOptionError): + APP.parse_args(["agent", "PMC9", "--fullmap", "/tmp/fm"], exit_on_error=False) def test_agent_optimize_flag_parses() -> None: """``-o``/``--optimize`` parses to optimize=True without executing the body.""" - fn, bound, _ = APP.parse_args(["agent", "PMC9", "--fullmap", "/tmp/fm", "-o"], exit_on_error=False) + fn, bound, _ = APP.parse_args(["agent", "PMC9", "--configuration-file", str(_graph_path()), "-o"], exit_on_error=False) assert fn is agent assert bound.kwargs["optimize"] is True @@ -217,7 +254,7 @@ def fail_supervisor(*a: object, **k: object) -> object: monkeypatch.setattr("tablassert.agent.run_supervisor", fail_supervisor) out: Path = tmp_path / "opt.yaml" - agent(["PMC1"], fullmap=Path("/tmp/fm"), optimize=True, instructions_out=out) + agent(["PMC1"], graph_configuration_file=_graph_path(), optimize=True, instructions_out=out) assert out.is_file() assert load_optimized_instructions(out) == "OPTIMIZED PROMPT" @@ -241,7 +278,7 @@ def fake_run_supervisor(pmc_ids: list[str], **kwargs: object) -> dict[str, objec instr_file: Path = tmp_path / "instr.yaml" save_optimized_instructions(instr_file, "CUSTOM PROMPT") - agent(["PMC1"], fullmap=Path("/tmp/fm"), instructions_file=instr_file) + agent(["PMC1"], graph_configuration_file=_graph_path(), instructions_file=instr_file) assert captured["instructions"] == "CUSTOM PROMPT" @@ -259,7 +296,7 @@ def fake_run_supervisor(pmc_ids: list[str], **kwargs: object) -> dict[str, objec monkeypatch.setattr("tablassert.agent.run_supervisor", fake_run_supervisor) - agent(["PMC1"], fullmap=Path("/tmp/fm")) + agent(["PMC1"], graph_configuration_file=_graph_path()) assert captured["instructions"] is None @@ -278,7 +315,7 @@ def fail_supervisor(*a: object, **k: object) -> object: monkeypatch.setattr("tablassert.agent.run_supervisor", fail_supervisor) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm"), judge_threshold=bad_threshold) + agent(["PMC1"], graph_configuration_file=_graph_path(), judge_threshold=bad_threshold) assert exc_info.value.code == 2 assert "judge-threshold" in capsys.readouterr().err @@ -297,7 +334,7 @@ def fake_run_supervisor(pmc_ids: list[str], **kwargs: object) -> dict[str, objec monkeypatch.setattr("tablassert.agent.run_supervisor", fake_run_supervisor) - agent(["PMC1"], fullmap=Path("/tmp/fm"), judge_threshold=0.7) + agent(["PMC1"], graph_configuration_file=_graph_path(), judge_threshold=0.7) assert captured["judge_threshold"] == 0.7 @@ -319,7 +356,7 @@ def fail_model_init(*a: object, **k: object) -> object: monkeypatch.setattr("tablassert.agent.make_dspy_lm", fail_model_init) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm"), gepa_threads=bad_threads) + agent(["PMC1"], graph_configuration_file=_graph_path(), gepa_threads=bad_threads) assert exc_info.value.code == 2 assert "gepa-threads" in capsys.readouterr().err @@ -337,7 +374,7 @@ def fail_supervisor(*a: object, **k: object) -> object: monkeypatch.setattr("tablassert.agent.run_supervisor", fail_supervisor) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm"), local=[bad_spec]) + agent(["PMC1"], graph_configuration_file=_graph_path(), local=[bad_spec]) assert exc_info.value.code == 2 assert "--local" in capsys.readouterr().err @@ -356,7 +393,7 @@ def fake_run_supervisor(pmc_ids: list[str], **kwargs: object) -> dict[str, objec monkeypatch.setattr("tablassert.agent.run_supervisor", fake_run_supervisor) - agent(["PMC1"], fullmap=Path("/tmp/fm"), local=[f"PMC1={tmp_path}"]) + agent(["PMC1"], graph_configuration_file=_graph_path(), local=[f"PMC1={tmp_path}"]) assert captured["local"] == {"PMC1": tmp_path} @@ -378,7 +415,7 @@ def fake_make_dspy_lm(*args: object, **kwargs: object) -> object: ) out: Path = tmp_path / "opt.yaml" - agent(["PMC1"], fullmap=Path("/tmp/fm"), optimize=True, backend="litellm", instructions_out=out) + agent(["PMC1"], graph_configuration_file=_graph_path(), optimize=True, backend="litellm", instructions_out=out) # Without --task-model the CLI builds EXACTLY ONE LM (the reflection LM) — assert the count so a # regression that reorders/adds LM constructions cannot hide behind lm_calls[0]. @@ -403,7 +440,7 @@ def test_agent_optimize_gepa_error_exits_nonzero(monkeypatch: pytest.MonkeyPatch out: Path = tmp_path / "opt.yaml" with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm"), optimize=True, instructions_out=out) + agent(["PMC1"], graph_configuration_file=_graph_path(), optimize=True, instructions_out=out) assert exc_info.value.code == 1 assert not out.is_file() # the unoptimized seed is NOT persisted assert "GEPA optimization failed" in capsys.readouterr().err @@ -443,63 +480,6 @@ def __init__( 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())] - - @pytest.mark.parametrize("bad_threshold", [-1.0, 2.0, float("nan"), float("inf")]) def test_agent_biolink_threshold_out_of_range_exits_2( bad_threshold: float, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] @@ -515,7 +495,7 @@ def fail_supervisor(*a: object, **k: object) -> object: monkeypatch.setattr("tablassert.agent.run_supervisor", fail_supervisor) with pytest.raises(SystemExit) as exc_info: - agent(["PMC1"], fullmap=Path("/tmp/fm"), biolink_threshold=bad_threshold) + agent(["PMC1"], graph_configuration_file=_graph_path(), biolink_threshold=bad_threshold) assert exc_info.value.code == 2 assert "biolink-threshold" in capsys.readouterr().err @@ -534,8 +514,8 @@ def fake_run_supervisor(pmc_ids: list[str], **kwargs: object) -> dict[str, objec monkeypatch.setattr("tablassert.agent.run_supervisor", fake_run_supervisor) - agent(["PMC1"], fullmap=Path("/tmp/fm")) + agent(["PMC1"], graph_configuration_file=_graph_path()) assert captured["biolink_threshold"] == 0.0 - agent(["PMC1"], fullmap=Path("/tmp/fm"), biolink_threshold=0.95) + agent(["PMC1"], graph_configuration_file=_graph_path(), biolink_threshold=0.95) assert captured["biolink_threshold"] == 0.95 diff --git a/tests/test_agent_docs.py b/tests/test_agent_docs.py index 57f7521..a5cb787 100644 --- a/tests/test_agent_docs.py +++ b/tests/test_agent_docs.py @@ -34,7 +34,7 @@ def test_agent_doc_references_cli_surface() -> None: """The documented command + flags match the real CLI (tablassert agent + key flags + env vars).""" text: str = DOC.read_text() assert "tablassert agent" in text - for flag in ("--fullmap", "--map-threshold", "--max-improve-iters", "--state-dir"): + for flag in ("--configuration-file", "--map-threshold", "--max-improve-iters", "--state-dir"): assert flag in text, f"docs missing CLI flag {flag}" for env in ("TABLASSERT_AGENT_MODEL_ID", "TABLASSERT_AGENT_API_BASE", "TABLASSERT_AGENT_API_KEY"): assert env in text, f"docs missing env var {env}" diff --git a/tests/test_agent_supervisor.py b/tests/test_agent_supervisor.py index 34560aa..68cc231 100644 --- a/tests/test_agent_supervisor.py +++ b/tests/test_agent_supervisor.py @@ -21,7 +21,6 @@ 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") @@ -243,8 +242,8 @@ def fake_fetch(pmc_id: str, outdir: Path, *, timeout: int = 120) -> list[Path]: assert metrics["skipped"] == 1 -def test_supervisor_resume_skips_done(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """A second run over [A,B] with the same state_dir skips the already-MAPPED A and processes B.""" +def test_supervisor_reruns_terminal_records(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A later invocation reprocesses an already-MAPPED A and also processes new B.""" table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\nbrca1\tmapk1\n") calls: list[str] = _patch_fetch(monkeypatch, table) good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) @@ -262,9 +261,9 @@ def factory() -> object: ) records: dict[str, ConfigRecord] = second["records"] # pyright: ignore[reportAssignmentType] assert records["PMCA"].status == "MAPPED" - assert records["PMCA"].attempts == first_attempts, "PMCA must NOT be reprocessed on resume" + assert records["PMCA"].attempts > first_attempts, "PMCA must be reprocessed on a later invocation" assert records["PMCB"].status == "MAPPED" - assert calls.count("PMCA") == 1, "fetch must not be called again for the already-MAPPED PMCA" + assert calls.count("PMCA") == 2 assert calls.count("PMCB") == 1 @@ -490,8 +489,8 @@ def fake_build( assert metrics["skipped"] == 0 -def test_supervisor_resume_skips_built_unmeasured(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: - """W5: BUILT_UNMEASURED is TERMINAL — a resume over the same id does not reprocess it (no re-fetch).""" +def test_supervisor_reruns_built_unmeasured(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """W5: BUILT_UNMEASURED remains non-failing but is processed again on rerun.""" import tablassert.agent as agent_mod table: Path = _write_table(tmp_path, "d.tsv", "brca1\tmapk1\n") @@ -527,7 +526,7 @@ def factory() -> object: 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 == "BUILT_UNMEASURED" # pyright: ignore[reportIndexIssue] - assert calls.count("PMC1") == 1, "fetch must not be called again for the terminal BUILT_UNMEASURED record" + assert calls.count("PMC1") == 2 # --------------------------------------------------------------------------- # @@ -999,192 +998,137 @@ def test_supervisor_local_payload_no_table_skipped(tmp_path: Path, fullmap_db: P # --------------------------------------------------------------------------- # -# Shared graph registry wiring: successful builds self-register into graph.yaml +# Caller-owned target graph wiring # --------------------------------------------------------------------------- # -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 _target_graph(tmp_path: Path, fullmap: Path, tables: list[Path] | None = None) -> Path: + """Write a valid target graph with distinctive metadata for supervisor tests.""" + path = tmp_path / "target-graph.yaml" + data: dict[str, Any] = { + "name": "TARGET_KG", + "version": "9.0.0", + "tables": [str(table) for table in (tables or [])], + "fullmap": str(fullmap), + "rig": { + "source_info": { + "infores_id": "infores:target-kg", + "terms_of_use_info": {"license_name": "CC0"}, + "data_access_locations": ["PMC - https://pmc.ncbi.nlm.nih.gov/"], + "source_status": "unknown", + }, + "ingest_info": {"utility": "Target test.", "scope": "Target test."}, + "provenance_info": {"contributions": ["Target test"]}, + "artifact_base_url": "https://example.org/target-kg", + "artifact_base_path": str(tmp_path / "published"), + }, + } + path.write_text(yaml.safe_dump(data, sort_keys=False)) + return path -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.""" +def test_supervisor_appends_to_supplied_graph_and_uses_metadata(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A successful result updates the exact target and audits with its graph identity/RIG.""" + from tablassert.graph_target import prepare_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) + target_path: Path = _target_graph(tmp_path, fullmap_db) + prepared = prepare_graph(target_path) state_dir: Path = tmp_path / "state" result = run_supervisor( ["PMC1"], - fullmap=fullmap_db, + graph=prepared.graph, + graph_path=prepared.path, 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)) + data: dict[str, Any] = yaml.safe_load(target_path.read_text()) + assert data["name"] == "TARGET_KG" + assert data["version"] == "9.0.0" + assert data["rig"]["source_info"]["infores_id"] == "infores:target-kg" + assert data["tables"] == [str(Path(str(rec.best_config_path)).resolve())] + assert Path(data["tables"][0]).is_absolute() + build_dir = tmp_path / "w" / "builds" / "PMC1" + assert (build_dir / "artifacts" / "TARGET_KG_9.0.0.nodes.ndjson").is_file() + assert not (build_dir / "artifacts" / "agent_0.0.1.nodes.ndjson").exists() + + +def test_supervisor_skipped_does_not_append_or_overwrite_existing_entry(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A failed rerun leaves a prior successful target entry and config untouched.""" + from tablassert.graph_target import prepare_graph + + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\n") + good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + target_path: Path = _target_graph(tmp_path, fullmap_db) + prepared = prepare_graph(target_path) state_dir: Path = tmp_path / "state" + _patch_fetch(monkeypatch, table) - 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( + first = run_supervisor( ["PMC1"], - fullmap=fullmap_db, + graph=prepared.graph, + graph_path=prepared.path, 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") + rec: ConfigRecord = first["records"]["PMC1"] # pyright: ignore[reportIndexIssue] + assert rec.status == "MAPPED" + old_config = Path(str(rec.best_config_path)) + before_graph = target_path.read_bytes() + before_config = old_config.read_bytes() - monkeypatch.setattr(agent_mod, "register_build", boom) + def fail_fetch(*args: object, **kwargs: object) -> list[Path]: + raise FileNotFoundError("rerun failed") - result = run_supervisor( + monkeypatch.setattr("tablassert.agent.fetch_pmc_article", fail_fetch) + second = run_supervisor( ["PMC1"], - fullmap=fullmap_db, + graph=prepared.graph, + graph_path=prepared.path, 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 + assert second["records"]["PMC1"].status == "SKIPPED" # pyright: ignore[reportIndexIssue] + assert target_path.read_bytes() == before_graph + assert old_config.read_bytes() == before_config -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. +def test_supervisor_target_rerun_replaces_same_pmc_entry(tmp_path: Path, fullmap_db: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A successful rerun is processed again and still leaves one entry for its PMC.""" + from tablassert.graph_target import prepare_graph - 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) + table: Path = _write_table(tmp_path, "good.tsv", "brca1\tmapk1\n") + calls: list[str] = _patch_fetch(monkeypatch, table) good_yaml: str = yaml.safe_dump(_column_cfg(table), sort_keys=False) + target_path: Path = _target_graph(tmp_path, fullmap_db) + prepared = prepare_graph(target_path) 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") + kwargs = { + "graph": prepared.graph, + "graph_path": prepared.path, + "build_model_factory": lambda: make_fake_model(final_yaml=good_yaml), + "map_threshold": 0.8, + "state_dir": state_dir, + "workdir": tmp_path / "w", + } + first = run_supervisor(["PMC1"], **kwargs) # type: ignore[arg-type] + second = run_supervisor(["PMC1"], **kwargs) # type: ignore[arg-type] 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" + assert calls.count("PMC1") == 2 + tables: list[str] = yaml.safe_load(target_path.read_text())["tables"] + assert len(tables) == 1 + assert tables[0] == str((state_dir / "configs" / "PMC1.yaml").resolve()) diff --git a/tests/test_docs_cli_coverage.py b/tests/test_docs_cli_coverage.py index ba9613b..b92e5fe 100644 --- a/tests/test_docs_cli_coverage.py +++ b/tests/test_docs_cli_coverage.py @@ -25,7 +25,6 @@ "agent": ("agent.md",), "build-fullmap": ("cli.md", "fullmap.md"), "build-kg": ("cli.md",), - "rebuild-agent-graph": ("cli.md",), "validate": ("cli.md",), "validate-kgx": ("cli.md",), } @@ -33,7 +32,6 @@ "agent": ("agent.md", "cli.md"), "build-fullmap": ("cli.md", "fullmap.md"), "build-kg": ("cli.md",), - "rebuild-agent-graph": ("cli.md",), "validate": ("cli.md",), "validate-kgx": ("cli.md",), } diff --git a/tests/test_graph_registry.py b/tests/test_graph_registry.py deleted file mode 100644 index 8f6f848..0000000 --- a/tests/test_graph_registry.py +++ /dev/null @@ -1,313 +0,0 @@ -"""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_NAME, GRAPH_VERSION, _registry_rig, 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["rig"]["source_info"]["infores_id"] == "infores:tablassert-agent" - 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, - "tables": [str(foreign.resolve())], - "fullmap": str(fullmap.resolve()), - "rig": _registry_rig(state_dir), - } - (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, - "tables": [str(stale.resolve())], - "fullmap": str(fullmap.resolve()), - "rig": _registry_rig(state_dir), - } - (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" diff --git a/tests/test_graph_target.py b/tests/test_graph_target.py new file mode 100644 index 0000000..fe483cc --- /dev/null +++ b/tests/test_graph_target.py @@ -0,0 +1,112 @@ +"""Tests for caller-owned target graph preparation and atomic agent appends.""" + +from __future__ import annotations + +import multiprocessing +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tablassert.errors import GraphValidationError +from tablassert.graph_target import append_successful_config, prepare_graph +from tablassert.models import Graph + + +def _rig() -> dict[str, Any]: + return { + "source_info": { + "infores_id": "infores:target-test", + "terms_of_use_info": {"license_name": "CC0"}, + "data_access_locations": ["Test source - https://example.org/data"], + "source_status": "unknown", + }, + "ingest_info": {"utility": "Target test.", "scope": "Target test."}, + "provenance_info": {"contributions": ["Test author"]}, + "artifact_base_url": "https://example.org/target-test", + "artifact_base_path": "./output", + } + + +def _write_graph(path: Path, tables: list[str] | None = None) -> Path: + path.write_text( + yaml.safe_dump( + {"name": "TARGET", "version": "2.0.0", "tables": [] if tables is None else tables, "fullmap": "./fullmap", "rig": _rig()}, sort_keys=False + ) + ) + return path + + +def _write_config(path: Path) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("template: {}\n") + return path + + +def test_prepare_graph_resolves_execution_paths_without_rewriting_yaml(tmp_path: Path) -> None: + target = _write_graph(tmp_path / "graph.yaml") + before = target.read_text() + + prepared = prepare_graph(target) + + assert prepared.path == target.resolve() + assert prepared.graph.fullmap == (tmp_path / "fullmap").resolve() + assert prepared.graph.rig.artifact_base_path == (tmp_path / "output").resolve() + assert target.read_text() == before + + +def test_append_preserves_metadata_and_replaces_matching_pmc(tmp_path: Path) -> None: + old = _write_config(tmp_path / "old" / "PMC1.yaml") + other = _write_config(tmp_path / "other" / "PMC2.yaml") + target = _write_graph(tmp_path / "graph.yaml", [str(old), str(other)]) + new = _write_config(tmp_path / "new" / "PMC1.yaml") + + append_successful_config(target, "PMC1", new) + + data: dict[str, Any] = yaml.safe_load(target.read_text()) + assert data["name"] == "TARGET" + assert data["version"] == "2.0.0" + assert data["fullmap"] == "./fullmap" + assert data["rig"] == _rig() + assert data["tables"] == [str(other), str(new.resolve())] + Graph.model_validate(data) + assert (tmp_path / "graph.yaml.lock").is_file() + assert not (tmp_path / ".graph.yaml.tmp").exists() + + +def test_append_rejects_invalid_target_without_changing_it(tmp_path: Path) -> None: + target = tmp_path / "graph.yaml" + target.write_text("name: not a graph\n") + before = target.read_bytes() + config = _write_config(tmp_path / "PMC1.yaml") + + with pytest.raises(GraphValidationError): + append_successful_config(target, "PMC1", config) + + assert target.read_bytes() == before + assert not list(tmp_path.glob("*.corrupt-*")) + + +def _append_worker(args: tuple[str, str, str]) -> str: + target, pmc_id, config = args + append_successful_config(Path(target), pmc_id, Path(config)) + return pmc_id + + +def test_concurrent_appends_do_not_lose_entries(tmp_path: Path) -> None: + target = _write_graph(tmp_path / "graph.yaml") + tasks: list[tuple[str, str, str]] = [] + for i in range(8): + config = _write_config(tmp_path / "configs" / f"PMC{i}.yaml") + tasks.append((str(target), f"PMC{i}", str(config))) + + with multiprocessing.Pool(processes=8) as pool: + assert sorted(pool.map(_append_worker, tasks)) == [f"PMC{i}" for i in range(8)] + + data: dict[str, Any] = yaml.safe_load(target.read_text()) + tables: list[str] = data["tables"] + assert len(tables) == 8 + assert {Path(table).stem for table in tables} == {f"PMC{i}" for i in range(8)} + assert all(Path(table).is_absolute() for table in tables) + Graph.model_validate(data)