feat: add shared agent graph registry

docs/agent.md

@@ -194,6 +194,8 @@ the fetched downloads, and the build outputs **all co-locate** under it:
 ```text
 .tablassert/agent/                       # = state_dir (the workspace root)
   state.json                             # supervisor checkpoint (atomic; unchanged location)
+  graph.yaml                             # SHARED aggregate graph registry (flock-serialized, atomic)
+  graph.yaml.lock                        # sidecar lock file for graph.yaml (exclusive flock)
   configs/<pmc_id>.yaml                  # best / accepted config (ALL configs in ONE folder)
   configs/<pmc_id>.derived.yaml          # initial agent-derived config
   downloads/<pmc_id>/<prefix>/...        # fetched PMC payload (main text + metadata + tables) — stable, persists
@@ -203,6 +205,8 @@ the fetched downloads, and the build outputs **all co-locate** under it:
 | Path | Contents | Lifecycle |
 | --- | --- | --- |
 | `state.json` | supervisor checkpoint: `{pmc_id, status, config_path, coverage_history[], qc_pass_rate, attempts, last_edits, best_coverage, best_config_path}` per record | written **atomically** (tmp write + `os.replace`) after each config and each improve iteration; git-ignored |
+| `graph.yaml` | SHARED aggregate graph registry: one `tables` entry per successful (`MAPPED` / `BUILT_UNMEASURED`) build | maintained under an exclusive `graph.yaml.lock` flock; atomic writes; see [Parallel agents and the shared graph registry](#parallel-agents-and-the-shared-graph-registry) |
+| `graph.yaml.lock` | sidecar lock file serializing registry read-modify-write | created on first registration; never deleted |
 | `configs/<pmc_id>.yaml` | the best / accepted config for the article | the reuse entry point (below) |
 | `configs/<pmc_id>.derived.yaml` | the agent's initial derived config | kept for provenance |
 | `downloads/<pmc_id>/<prefix>/` | fetched PMC payload (main text + metadata + tables) | **stable** — persists across runs |
@@ -219,14 +223,55 @@ ready-to-build `graph.yaml` (wrapping `table.yaml` with the resolved fullmap) in
 
 ```bash
 cd .tablassert/agent/builds/PMC11708054
-tablassert build-kg graph.yaml
+tablassert build-kg -f graph.yaml
 ```
 
 !!! warning "Not relocatable"
     `source.local` in the best config is an **absolute** path into `downloads/<pmc_id>/`. 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:
+
+```bash
+# fan out over DISJOINT pmc sets, all pointed at one shared state dir
+tablassert agent PMC1 PMC2 --fullmap ./fullmap --state-dir ./shared &
+tablassert agent PMC3 PMC4 --fullmap ./fullmap --state-dir ./shared &
+wait
+
+# one build of the whole registered graph
+tablassert build-kg -f ./shared/graph.yaml
+```
+
+Use **disjoint pmc sets**: each process owns its own ids. The shared registry itself is fully
+cross-process safe, but the per-process checkpoint (`state.json`) read-modify cycle is not
+cross-process locked, so two processes must not own the same pmc id.
+
+**How the registry works.** Every build that ends `MAPPED` or `BUILT_UNMEASURED` (both are
+successful builds) UPSERTS its best config into `<state-dir>/graph.yaml`:
+
+- **Concurrency-safe** — each registration takes an EXCLUSIVE `flock` on the
+  `<state-dir>/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-<UTC timestamp>` and rebuilt fresh with a warning, so unattended
+  parallel runs never wedge on a damaged file.
+- **Registered statuses** — only `MAPPED` and `BUILT_UNMEASURED` register; `SKIPPED` never does. A
+  re-run that SKIPS an already-MAPPED pmc keeps the existing entry (resume skips terminal records
+  entirely, so nothing rewrites them). `tablassert rebuild-agent-graph --state-dir ./shared
+  --fullmap ./fullmap` reconstructs the registry from `state.json` and prunes stale entries
+  (deleted configs, non-registered statuses).
+
 ## The tools
 
 | Tool | Kind | Purpose |

docs/cli.md

@@ -1,8 +1,9 @@
 # CLI Reference
 
 Tablassert extracts knowledge assertions from tabular data into KGX NDJSON. The `tablassert` app
-exposes **four subcommands** — `agent`, `build-fullmap`, `build-kg`, `validate` — plus an app-level
-`--version` flag. Run `tablassert --help` (or `<command> --help`) for the live surface.
+exposes **five subcommands** — `agent`, `build-fullmap`, `build-kg`, `rebuild-agent-graph`,
+`validate` — plus an app-level `--version` flag. Run `tablassert --help` (or `<command> --help`)
+for the live surface.
 
 ## Command index
 
@@ -11,6 +12,7 @@ exposes **four subcommands** — `agent`, `build-fullmap`, `build-kg`, `validate
 | [`agent`](#agent) | Autonomously derive, build, audit, and improve KG configs from PMC articles |
 | [`build-fullmap`](#build-fullmap) | Build the embedded fullmap redb used for entity resolution |
 | [`build-kg`](#build-kg) | Build a KGX NDJSON knowledge graph from a YAML configuration |
+| [`rebuild-agent-graph`](#rebuild-agent-graph) | Rebuild the shared agent graph registry from the supervisor checkpoint |
 | [`validate`](#validate) | Validate a graph or table configuration without executing it |
 
 ## App flags
@@ -137,6 +139,37 @@ Output is written to the current directory as `{name}_{version}.nodes.ndjson`,
 
 ---
 
+## rebuild-agent-graph
+
+Use this to rebuild the SHARED agent graph registry (`<state-dir>/graph.yaml`) from the supervisor
+checkpoint (`<state-dir>/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

src/tablassert/agent.py

@@ -33,6 +33,7 @@
 from tablassert.enums import EncodingMethods
 from tablassert.errors import GraphValidationError, QcRuntimeMissingError, SectionValidationError, TablassertValidationError
 from tablassert.fullmap import distinct, fullmap_db_path, is_lock_contention, lookup_rows
+from tablassert.graph_registry import REGISTERED_STATUSES, register_build
 from tablassert.lib import Tcode
 from tablassert.log import cat
 from tablassert.models import NodeEncoding, Section
@@ -2768,7 +2769,9 @@ def run_supervisor(
 
             best_path: Path = best_config_path(state_dir, pmc_id)
             best_path.write_text(current_config)
-            rec.best_config_path = str(best_path)
+            # Persist the ABSOLUTE path: a relative --state-dir would otherwise store a CWD-relative
+            # entry that rebuild_graph (which may run from a different CWD) could not locate.
+            rec.best_config_path = str(best_path.resolve())
             rec.config_path = str(best_path)
             if current_cov >= map_threshold:
                 # Optional semantic gate (W1): when a real judge model is configured, MAPPED additionally
@@ -2804,6 +2807,18 @@ def run_supervisor(
                     f"SKIPPED: could not reach map_threshold={map_threshold} after {max_improve_iters} "
                     f"improve iters (best coverage {current_cov:.3f})"
                 )
+            # Shared graph registry: successful builds upsert into <state_dir>/graph.yaml so concurrent
+            # agents over one --state-dir converge on a single aggregate config. SKIPPED never registers,
+            # and registration must NEVER flip a successful status: on any error log + note and keep the
+            # status. A re-run that SKIPS an already-MAPPED pmc keeps its registry entry: resume skips
+            # terminal records entirely, and rebuild-agent-graph prunes stale entries from state.json.
+            if rec.status in REGISTERED_STATUSES:
+                try:
+                    register_build(state_dir, pmc_id, best_path, fullmap)
+                except Exception as reg_exc:  # a registry failure is a note, never a status change
+                    logger.error("graph registry update failed for {pmc}: {error}", pmc=pmc_id, error=reg_exc)
+                    note: str = f"graph registry update failed (status kept {rec.status}): {reg_exc}"
+                    rec.notes = f"{rec.notes}; {note}" if rec.notes else note
             save_state(state_dir, state)
         except Exception as exc:  # one bad pmc never aborts the batch
             rec.status = "SKIPPED"

src/tablassert/cli.py

@@ -717,6 +717,37 @@ def metric(key: str, default: float) -> float:
     )
 
 
+@APP.command(name="rebuild-agent-graph")
+def rebuild_agent_graph(
+    state_dir: Annotated[Path, cyclopts.Parameter(name=["--state-dir", "-sd"])] = Path(".tablassert") / "agent",
+    *,
+    fullmap: Annotated[Path, cyclopts.Parameter(name=["--fullmap", "-f"])],
+) -> None:
+    """Rebuild the shared agent graph registry from the supervisor checkpoint.
+
+    Reconstructs ``<state-dir>/graph.yaml`` from ``<state-dir>/state.json``: every MAPPED /
+    BUILT_UNMEASURED record whose best config still exists on disk becomes a ``tables`` entry
+    (sorted by pmc id); stale entries (deleted configs, non-registered statuses) are pruned.
+    Parallel ``tablassert agent`` runs maintain the registry incrementally; this command
+    reconstructs it deterministically. Concurrency-safe: the same exclusive ``graph.yaml.lock``
+    flock + atomic write the agent registration uses.
+
+    Args:
+        state_dir: Agent state directory holding ``state.json`` + ``configs/``.
+        fullmap: Fullmap redb file or base directory recorded in the registry (first-wins: an
+            existing registry fullmap that differs is kept with a warning).
+    """
+    import yaml
+
+    from tablassert.graph_registry import rebuild_graph
+
+    graph_path: Path = rebuild_graph(state_dir, fullmap)
+    data: object = yaml.safe_load(graph_path.read_text(encoding="utf-8"))
+    tables: object = data.get("tables", []) if isinstance(data, dict) else []
+    count: int = len(tables) if isinstance(tables, list) else 0
+    print(f"tablassert rebuild-agent-graph: wrote {graph_path} with {count} table config(s).")
+
+
 def build_fullmap_pipeline(
     output: Path, progress: PipelineProgress, cache: Path = Path("./fullmap/downloads"), version: str = BABEL_VERSION, threads: int | None = None
 ) -> None: