fix(pfs): guidance --write resolves+builds before opening the batch (refs #689)

Mirrors stack pfs reaction's Ruling A13 discipline: every --family key
is resolved and validated up front (an unknown key fails before any
write), then build_guidance runs per family against a plain _read()
connection, and the _batch() block only ever calls write_guidance and
_publish() — it no longer holds the DuckDB single-writer lock while
FR/CPT text is scanned (#508-#514).

src/cli/pfs.py

@@ -369,6 +369,23 @@ def _print_guidance(rows: list[Any]) -> None:
         )
 
 
+def _resolve_guidance_families(con: Any, keys: list[str]) -> list[tuple[str, Any]]:
+    """(key, Family) per requested key, refreshed from *con* — raises
+    ``typer.BadParameter`` on the first unknown key, before any family's
+    rows are built or any write happens (mirrors ``_reaction_targets``'s
+    resolve-everything-up-front discipline)."""
+    refresh_from(con)
+    out: list[tuple[str, Any]] = []
+    for key in keys:
+        fam = FAMILIES.get(key)
+        if fam is None:
+            raise typer.BadParameter(
+                f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
+            )
+        out.append((key, fam))
+    return out
+
+
 @app.command()
 def guidance(
     family: list[str] = typer.Option(
@@ -384,19 +401,27 @@ def guidance(
     keys = [f.upper() for f in family]
     store = _store()
     if write:
+        # Ruling A13 (mirrors `stack pfs reaction`): resolve every family
+        # key up front and build every family's rows against a plain
+        # DuckDB read (I4) — build_guidance only reads pfs.cpt_* — never
+        # the write lock a notebook may be holding (#508-#514). The batch
+        # below only ever calls write_guidance and publishes.
+        con = _read()
+        try:
+            targets = _resolve_guidance_families(con, keys)
+            by_family: dict[str, list[Any]] = {}
+            rows: list[Any] = []
+            for key, fam in targets:
+                fam_rows = build_guidance(con, store, fam.codes, family=key)
+                by_family[key] = fam_rows
+                rows.extend(fam_rows)
+        finally:
+            con.close()
+
         with _batch() as con:
             ensure_tables(con)
-            refresh_from(con)
-            rows: list[Any] = []
-            for key in keys:
-                fam = FAMILIES.get(key)
-                if fam is None:
-                    raise typer.BadParameter(
-                        f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
-                    )
-                fam_rows = build_guidance(con, store, fam.codes, family=key)
+            for key, fam_rows in by_family.items():
                 write_guidance(con, key, fam_rows)
-                rows.extend(fam_rows)
         _publish()
         _print_guidance(rows)
         return

tests/cli/test_pfs_cli.py

@@ -624,6 +624,48 @@ class TestGuidance:
         for key in pfs_cli.HAND_FAMILIES:
             assert key in res.output
 
+    def test_unknown_key_among_several_errors_before_any_batch(self, con, monkeypatch):
+        # Every --family key must be resolved and validated up front — an
+        # unknown key later in the list must fail before build_guidance
+        # runs for any family and before the write lock is ever opened.
+        built: list[str] = []
+        monkeypatch.setattr(
+            pfs_cli,
+            "build_guidance",
+            lambda con_, store_, codes, *, family: built.append(family) or [],
+        )
+
+        def fail_batch():
+            raise AssertionError("must not open a RW duckdb_batch connection")
+
+        monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
+        res = runner.invoke(
+            app, ["pfs", "guidance", "--family", "CCM", "--family", "NOPE", "--write"]
+        )
+        assert res.exit_code != 0
+        assert built == []
+
+    def test_write_builds_every_family_before_opening_the_batch(self, con, monkeypatch):
+        order: list[str] = []
+
+        def fake_build(con_, store_, codes, *, family):
+            order.append("build")
+            return [self._row(family=family)]
+
+        real_batch = pfs_cli._batch
+
+        def recording_batch():
+            order.append("batch")
+            return real_batch()
+
+        monkeypatch.setattr(pfs_cli, "build_guidance", fake_build)
+        monkeypatch.setattr(pfs_cli, "_batch", recording_batch)
+        res = runner.invoke(
+            app, ["pfs", "guidance", "--family", "CCM", "--family", "APCM", "--write"]
+        )
+        assert res.exit_code == 0, res.output
+        assert order == ["build", "build", "batch"]
+
     def test_requires_family(self, con):
         res = runner.invoke(app, ["pfs", "guidance"])
         assert res.exit_code != 0