def port_spines(
project: str,
*,
discover_fn: Callable[[str], dict[str, Any]],
get_graph: Any,
graphs_dir: "Path | None" = None,
) -> int:
"""Port a project's legacy materialized spines into its directory (idempotent).
For each ``discover_fn(project)`` result NOT yet ported (per the
``_ported_spines.json`` marker), create a ``spine``-state org carrying the
reverse-engineered embedded schema + ``direction=incoming``, DE-REGISTER its
synthesis graph as a project source, and record it in the marker — all inside
ONE owner-locked critical section so concurrent first-listings can neither
double-port nor lose a manifest write.
``discover_fn`` is :func:`server.discover_spines` (injected to keep this module
free of a hard server dependency and to make the routine testable); it surfaces
the live apex + dimension structure. Non-destructive + reversible: no note is
modified and no edge is rewritten. Returns the number of orgs created.
"""
base = Path(graphs_dir) if graphs_dir is not None else _org.GRAPHS_DIR
ot, on = _org._validate_owner("project", project)
owner_dir = _org._owner_dir(base, ot, on)
# Discovery (the heavy graph scan) + the cheap marker pre-check run OUTSIDE the
# lock; the lock re-validates against the marker so the pre-check is only an
# optimization, never a correctness dependency.
try:
disc = discover_fn(project) or {}
except Exception: # noqa: BLE001 — discovery failure must never fail a listing
_org.logger.warning("discover_spines failed during port for project '%s'", project, exc_info=True)
return 0
if not isinstance(disc, dict) or disc.get("error"):
return 0
spines = [
s for s in (disc.get("spines") or [])
if isinstance(s, dict) and str(s.get("graph") or "").strip()
]
if not spines:
return 0
already = _org._read_ported_marker(owner_dir)
pending = [s for s in spines if str(s.get("graph") or "").strip() not in already]
if not pending:
return 0
ported = 0
wrote = False
with _org.review_write_lock(_org._owner_lock_name(ot, on), graphs_dir=base):
# Re-read the marker under the lock to close the double-port race, and
# collect the synthesis graphs already represented by an org so a structure
# already overlaid collapses by identity rather than minting a second
# overlay (design §9 de-dup). Widened beyond ``spine_ref``: a MIGRATED
# review-table org is a ``lens`` (``spine_ref=''``) that still DESCRIBES a
# synthesis graph via its row-axis/column ``graph`` scope, so it must
# collapse too — :func:`_org_synthesis_graphs` matches the table-org's
# underlying graph against the discovered spine.
marker = _org._read_ported_marker(owner_dir)
existing_refs: set[str] = set()
for o in _org._scan_owner_orgs(owner_dir, ot, on):
existing_refs |= _org_synthesis_graphs(o)
# Collision-DETECTION universe handed to ``_resolve_blank_scoping``. This is
# a SCAN-ONLY set — widening it can only cause MORE collisions to be
# detected (more degrade-skips), never fewer; it re-registers NO source and
# de-registers nothing. The invariant it must establish: NO graph that hosts
# a dimension node colliding with a pending spine's blank-attached id can be
# invisible to the collision scan, regardless of de-registration / cross-
# call / cross-session timing. A bare manifest read is INSUFFICIENT because
# porting DE-REGISTERS each ported spine's synthesis graph as a source, so a
# foreign spine B ported in an EARLIER ``port_spines`` call has its graph in
# NEITHER the live sources NOR this loop's ``pending`` — its colliding dim
# node would go undetected and B's still-registered base claims would leak
# into a later spine's matrix (the 8th P0 leak, cross-call/de-registered gap).
#
# Completeness by construction: the universe is the UNION of every place a
# spine's synthesis graph can be recorded across time —
# {live project sources} — currently-registered graphs
# ∪ {spine graphs pending in THIS loop} — same-call siblings (re-review BUG 1)
# ∪ {de-registered graphs in the ported marker} — graphs ported+de-registered
# in EARLIER calls (the 8th leak)
# ∪ {synthesis graphs referenced by existing orgs} — any already-ported/
# overlaid graph, even if
# its marker entry was lost
# The marker term and the org-ref term are independently sufficient for the
# cross-call case (either one names B's de-registered graph); taking BOTH
# makes detection robust to a missing/rebuilt marker or a hand-authored
# overlay. ``marker`` and ``existing_refs`` are read above under the lock and
# frozen here via ``|`` (set union returns a NEW set), so the in-loop
# mutation of those sets does not retroactively widen this snapshot. A graph
# is only flagged as colliding when it actually HOSTS one of the spine's dim
# ids (``ids & set(zg.notes)``), so adding non-colliding graphs to the scan
# never invents a false collision — a no-collision / fully-recorded spine
# still ports.
try:
from zettelkasten.graph import load_project
_pdata = load_project(project, graphs_dir=base) or {}
pre_port_sources = {g for g in (_pdata.get("sources") or []) if g}
except Exception: # noqa: BLE001 — best-effort; fall back to pending graphs only
pre_port_sources = set()
source_universe = (
pre_port_sources
| {
str(s.get("graph") or "").strip()
for s in pending
if str(s.get("graph") or "").strip()
}
| {g for g in marker if g}
| {g for g in existing_refs if g}
)
# The COLLISION-HOST universe widens the project-local ``source_universe``
# to every synthesis graph on disk CROSS-PROJECT — so a colliding spine in
# a DIFFERENT project that shares only the base corpus is detected and the
# porting spine degrade-skips instead of leaking the foreign spine's rows
# (residual 1, the cross-project shared-corpus leak). Computed ONCE under
# the owner lock and threaded down to ``_resolve_blank_scoping``; the
# shared ``host_index`` caches each host graph's note-id set so the whole
# ``pending`` loop reads each graph at most once. Still a PURE SCAN: it can
# only ever flag MORE collisions (more degrade-skips), never re-register a
# source nor relax a skip — so a no-collision / fully-recorded spine still
# ports exactly as before.
host_universe = source_universe | _cross_graph_host_universe(base)
host_index: dict[str, set[str]] = {}
for s in pending:
graph = str(s.get("graph") or "").strip()
if graph in marker:
continue
if graph in existing_refs:
# De-dup by identity: an org already overlays OR describes this
# synthesis graph (a promoted/ported spine via ``spine_ref``, or a
# migrated table-org via its axis/column ``graph`` scope). Collapse
# to that one entry — still de-register the source so the overlay
# leaves the base composite — and mark it so the next listing skips
# the (heavy) re-check. Mark ONLY on a confirmed de-register so a
# swallowed manifest-write failure leaves the spine re-portable
# rather than permanently double-counted (re-critic P2-5).
if _org._deregister_project_source_locked(project, graph, base):
marker.add(graph)
wrote = True
continue
try:
org = _ported_spine_org(
s,
project,
base,
get_graph=get_graph,
source_universe=source_universe,
host_universe=host_universe,
host_index=host_index,
)
except ValueError as exc:
# DEGRADE SAFELY: a spine whose live structure can't be cleanly
# reverse-engineered is SKIPPED (never ported as a broken schema)
# and deliberately NOT marked, so a later-completed spine can still
# port. The signal is logged, never swallowed.
_org.logger.warning(
"skipping un-portable spine '%s' for project '%s': %s", graph, project, exc
)
continue
_org._save_organization_locked(base, org)
existing_refs.add(graph)
ported += 1
wrote = True
# Mark the spine ONLY after a CONFIRMED successful de-register. A
# swallowed ``_projects/<P>.yaml`` write failure must leave the spine
# UN-marked so the next listing retries (the saved org collapses by
# identity on the retry and re-attempts the de-register) — otherwise
# the spine would be permanently BOTH a base source AND an overlay, a
# double-count the marker would forever skip re-trying (re-critic P2-5).
if _org._deregister_project_source_locked(project, graph, base):
marker.add(graph)
else:
_org.logger.warning(
"ported spine org for synthesis graph '%s' (project '%s') was "
"saved but its source de-register FAILED; leaving it UN-marked "
"so the next listing retries the de-register",
graph,
project,
)
if wrote:
_org._write_ported_marker(owner_dir, marker)
return ported