coordinator.extraction

coordinator.extraction

Grounded-extraction pipeline helper.

This module backs the create_extraction_graph coordinator tool (registered only when the zettelkasten capability is enabled). It performs the synchronous, tool-side prep a grounded-extraction run needs -- ingesting each source into the zettelkasten and seeding a per-source hub note -- and then emits the fixed extractor -> scribe -> auditor task graph (plus a trailing memory task).

Prep writes notes directly by importing the zettelkasten library functions. This runs inside the coordinator process: the zettelkasten code is always importable (it is a core feature) and GRAPHS_DIR resolves from ANGELO_WORKSPACE, so the coordinator writes into the same .zettelkasten/ tree the zettelkasten MCP server uses. Doing prep here -- not as a graph task -- is what lets the extractor (a planner) run first: a planner cannot depend on an implementer, so the ingest + hub-seeding cannot be a graph task.

The trio agents/personas and the schema registry live in the zettelkasten capability; this module only orchestrates them.

ExtractionError

Bases: Exception

Raised when source prep fails irrecoverably (bad ingest / create).

Source code in coordinator/extraction.py

class ExtractionError(Exception):
    """Raised when source prep fails irrecoverably (bad ingest / create)."""

StructurePersistError

Bases: ExtractionError

Raised when an ENFORCING keyed structure/rubric write failed for EVERY source -- a per-source PERSIST failure (the quarantine case), NOT a genuine config/scaffold error (an empty resolved synthesis-graph name, a wrapped SpineError, etc).

A subclass of :class:ExtractionError so existing except ExtractionError handlers (e.g. reconcile_structures) keep catching it. Callers that must distinguish "could not persist the rubric for any source" (DEMOTE the run to flat -- the sources still extract context-less into the flat graph) from "the run is mis-configured" (PROPAGATE as a visible error) narrow on this subtype: only a persist failure demotes; a plain ExtractionError raises.

Source code in coordinator/extraction.py

class StructurePersistError(ExtractionError):
    """Raised when an ENFORCING keyed structure/rubric write failed for EVERY
    source -- a per-source PERSIST failure (the quarantine case), NOT a genuine
    config/scaffold error (an empty resolved synthesis-graph name, a wrapped
    ``SpineError``, etc).

    A subclass of :class:`ExtractionError` so existing ``except ExtractionError``
    handlers (e.g. ``reconcile_structures``) keep catching it. Callers that must
    distinguish "could not persist the rubric for any source" (DEMOTE the run to
    flat -- the sources still extract context-less into the flat graph) from
    "the run is mis-configured" (PROPAGATE as a visible error) narrow on this
    subtype: only a persist failure demotes; a plain ``ExtractionError`` raises.
    """

describe_sources

describe_sources(prepped: list[dict], limit: int = 3, title_max: int = 60) -> str

Human-readable summary of the sources in an extraction run.

Lists up to limit source titles (each truncated to title_max chars); any beyond the limit collapse into a +N more suffix. Used to give a coordinator run's goal/title a source-specific label instead of the generic Grounded extraction: <project> (schema: <schema>), so a dashboard full of extraction runs is distinguishable at a glance. Returns "" when no source carries a usable title/name (callers omit the source clause in that case).

Source code in coordinator/extraction.py

def describe_sources(prepped: list[dict], limit: int = 3, title_max: int = 60) -> str:
    """Human-readable summary of the sources in an extraction run.

    Lists up to ``limit`` source titles (each truncated to ``title_max`` chars);
    any beyond the limit collapse into a ``+N more`` suffix. Used to give a
    coordinator run's goal/title a source-specific label instead of the generic
    ``Grounded extraction: <project> (schema: <schema>)``, so a dashboard full of
    extraction runs is distinguishable at a glance. Returns ``""`` when no source
    carries a usable title/name (callers omit the source clause in that case).
    """
    titles: list[str] = []
    for p in prepped:
        t = ((p.get("title") or p.get("name") or "") if isinstance(p, dict) else "").strip()
        if not t:
            continue
        if len(t) > title_max:
            t = t[: title_max - 1].rstrip() + "…"
        titles.append(t)
    if not titles:
        return ""
    shown = titles[:limit]
    body = ", ".join(f'"{t}"' for t in shown)
    remaining = len(titles) - len(shown)
    if remaining > 0:
        body += f" +{remaining} more"
    return body

prep_sources

prep_sources(sources: list[dict], schema_obj: dict, project: str, hub_per_source: bool = True, projects: list[str] | None = None) -> list[dict]

Ingest each source and seed its hub note; return per-source prep records.

Each returned record: {name, title, hub_id, content_hash, source_path, existing} where name is the source's zettelkasten graph (folder) name.

A source dict accepts: name (required), optional title, authors, year, venue, doi, doc_type, and either path (the tool ingests it) or a precomputed content_hash (+ optional source_path).

EXTRACT-ONCE, ATTACH-TO-MANY-PROJECTS: projects (when given) is the FULL set of projects the source is registered into; the source is still ingested/created ONCE (hash-deduped) and the hub seeded once -- only project MEMBERSHIP fans out. The positional project stays the back-compat single-target shim and the primary used for the hub body text; absent projects it is the sole target, so single-project callers are byte-identical.

Only the LEGACY FLAT extraction block is persisted here (schema-level rubric), once per source and project-agnostic. The per-context (project, synthesis_graph) keyed entry is written later by :func:prep_spine / :func:reconcile_meta_tags, once structure resolution knows the REAL synthesis graph -- prep cannot know it yet (see the flat-write note below), so pre-seeding a keyed entry here only orphans it.

Source code in coordinator/extraction.py

def prep_sources(
    sources: list[dict],
    schema_obj: dict,
    project: str,
    hub_per_source: bool = True,
    projects: list[str] | None = None,
) -> list[dict]:
    """Ingest each source and seed its hub note; return per-source prep records.

    Each returned record: ``{name, title, hub_id, content_hash, source_path,
    existing}`` where ``name`` is the source's zettelkasten graph (folder) name.

    A source dict accepts: ``name`` (required), optional ``title``, ``authors``,
    ``year``, ``venue``, ``doi``, ``doc_type``, and either ``path`` (the tool
    ingests it) or a precomputed ``content_hash`` (+ optional ``source_path``).

    EXTRACT-ONCE, ATTACH-TO-MANY-PROJECTS: ``projects`` (when given) is the FULL
    set of projects the source is registered into; the source is still
    ingested/created ONCE (hash-deduped) and the hub seeded once -- only project
    MEMBERSHIP fans out. The positional ``project`` stays the back-compat
    single-target shim and the primary used for the hub body text; absent
    ``projects`` it is the sole target, so single-project callers are byte-identical.

    Only the LEGACY FLAT extraction block is persisted here (schema-level rubric),
    once per source and project-agnostic. The per-context ``(project,
    synthesis_graph)`` keyed entry is written later by :func:`prep_spine` /
    :func:`reconcile_meta_tags`, once structure resolution knows the REAL synthesis
    graph -- prep cannot know it yet (see the flat-write note below), so pre-seeding
    a keyed entry here only orphans it.
    """
    # Imported lazily: keeps coordinator startup free of the zettelkasten import
    # cost, and the capability is only enabled when zettelkasten is in play.
    from zettelkasten.graph import write_extraction_meta
    from zettelkasten.server import (
        add_note,
        add_to_project,
        create_source,
        find_by_title,
        ingest_source,
    )

    if not sources:
        raise ExtractionError("create_extraction_graph requires at least one source.")

    # Schema-level fields persisted onto every source so the zettelkasten server
    # can enforce the run's contract at write time (strict tags) and report
    # coverage deterministically. ``prep_spine`` augments this with the
    # materialized ``structure`` once the spine exists.
    schema_name = str(schema_obj.get("name") or "") if isinstance(schema_obj, dict) else ""
    schema_strict = bool(schema_obj.get("strict", True)) if isinstance(schema_obj, dict) else True
    schema_grounded = bool(schema_obj.get("grounded", True)) if isinstance(schema_obj, dict) else True
    schema_tags = list(schema_obj.get("tags") or []) if isinstance(schema_obj, dict) else []

    # A grounded-extraction run is "extract into graph X for project P", so each
    # source graph must be a MEMBER of P — otherwise the per-source graphs are
    # orphaned from the project (and the synthesis spine has nothing to roll up
    # in the composite). For a multi-project run the SAME source is registered
    # into EVERY target project (membership fans out; ingestion does not). Ensure
    # each project's manifest exists, then register each source graph below.
    # ``add_to_project`` / ``_ensure_project`` are idempotent and only append.
    target_projects = [
        p for p in (projects if projects is not None else [project])
        if (p or "").strip()
    ]
    for tp in target_projects:
        _ensure_project(tp)

    hub_spec = schema_obj.get("hub") if isinstance(schema_obj, dict) else None
    hub_type = "concept"
    hub_template = "{source_title}: Overview"
    if isinstance(hub_spec, dict):
        hub_type = str(hub_spec.get("type") or hub_type)
        hub_template = str(hub_spec.get("title_template") or hub_template)

    prepped: list[dict] = []
    for s in sources:
        # A ``kind: data`` source references an EXISTING dataset box (its ``graph``),
        # not a document to ingest/slice. Skip prose ingest + create_source entirely
        # and seed only what the data trio + shared spine need: project membership,
        # a per-source hub, and the flat rubric meta -- so a data source attaches to
        # the SAME spine dimension nodes as a prose source under one synthesis_label.
        if (s.get("kind") or "").strip().lower() == "data":
            prepped.append(_prep_data_source(
                s, project, target_projects, hub_spec, hub_type, hub_template,
                schema_name=schema_name, schema_strict=schema_strict,
                schema_grounded=schema_grounded, schema_tags=schema_tags,
                hub_per_source=hub_per_source,
                add_to_project=add_to_project, find_by_title=find_by_title,
                add_note=add_note, write_extraction_meta=write_extraction_meta,
            ))
            continue

        name = (s.get("name") or "").strip()
        if not name:
            raise ExtractionError(f"Source is missing a 'name': {s!r}")
        title = (s.get("title") or name).strip()

        content_hash = (s.get("content_hash") or "").strip()
        source_path = (s.get("source_path") or s.get("path") or "").strip()
        existing = ""
        # Slice metadata for the fan-out planner. Populated from the ingest result
        # (path sources, always cache_full) or the cached page-index sidecar
        # (content_hash-only sources). Defaults keep small/legacy sources single-trio.
        num_pages = 0
        total_chars = 0
        outline: list[dict] = []

        if s.get("path"):
            # cache_full: cache the WHOLE document (+ page/outline sidecar) so a
            # later chapter is addressable -- re-ingesting an existing source here
            # is the "force-refresh full-text cache" the book workaround did by hand.
            ing = _as_dict(ingest_source(path=s["path"], cache_full=True))
            if "error" in ing:
                raise ExtractionError(
                    f"Ingest failed for source '{name}': {ing.get('error')}"
                )
            content_hash = ing.get("content_hash", content_hash)
            source_path = ing.get("source_path", source_path)
            existing = ing.get("existing_source", "") or ""
            num_pages = int(ing.get("num_pages") or 0)
            total_chars = int(ing.get("total_chars") or 0)
            outline = ing.get("outline") or []
        elif content_hash:
            # No path to re-ingest: read the disposable page-index sidecar if one
            # was cached by a prior full ingest, so a pre-ingested book still fans
            # out. Absent sidecar -> zero metadata -> single full slice (safe).
            try:
                from zettelkasten.ingest import read_page_index

                idx = read_page_index(content_hash) or {}
                num_pages = int(idx.get("num_pages") or 0)
                total_chars = int(idx.get("total_chars") or 0)
                outline = idx.get("outline") or []
            except Exception:  # noqa: BLE001 — sidecar read is best-effort
                pass

        # GROWABLE SOURCE: a live/streamed source (``refresh: True``) whose
        # document grows over time re-targets its ONE box each pass -- an existing
        # box is refreshed (content pointer updated, grounding corpus busted) rather
        # than erroring, so incremental extraction accumulates claims into one box.
        cs = _as_dict(create_source(
            name=name,
            doc_type=s.get("doc_type", ""),
            title=title,
            authors=s.get("authors"),
            year=s.get("year"),
            venue=s.get("venue", ""),
            doi=s.get("doi", ""),
            abstract=s.get("abstract", ""),
            content_hash=content_hash,
            source_path=source_path,
            update=bool(s.get("refresh")),
            date=str(s.get("date") or ""),
        ))
        # An existing folder (AlreadyExists) or content-hash dedup are both fine:
        # reuse the resolved graph name and keep going.
        graph_name = cs.get("name", name)
        if "error" in cs and cs.get("type") != "AlreadyExists":
            raise ExtractionError(
                f"create_source failed for '{name}': {cs.get('error')}"
            )

        # Tie "extract into graph X" to "X is a member of EACH target project".
        # Idempotent and best-effort: a registration hiccup must not abort prep.
        if graph_name:
            for tp in target_projects:
                try:
                    add_to_project(project=tp, source=graph_name)
                except Exception:  # noqa: BLE001 — membership is best-effort
                    logger.warning(
                        "could not register source '%s' into project '%s'",
                        graph_name, tp,
                    )

        hub_id = ""
        if hub_per_source and isinstance(hub_spec, dict):
            hub_title = _safe_format(hub_template, {
                "source_title": title,
                "title": title,
                "persona": title,
                "source": graph_name,
                "name": graph_name,
                "year": str(s.get("year") or ""),
            })
            # Dedup hubs so re-running prep on the same corpus is idempotent.
            try:
                found = _as_dict(find_by_title(graph_name, hub_title))
                hub_id = found.get("id") or found.get("note_id") or ""
            except Exception:
                hub_id = ""
            if not hub_id:
                created = _as_dict(add_note(
                    graph=graph_name,
                    title=hub_title,
                    type=hub_type if hub_type else "concept",
                    body=(
                        f"Per-source hub for grounded extraction of '{title}' "
                        f"under the schema for project '{project}'. Dimension "
                        "claim notes link here."
                    ),
                    tags=["hub", "extraction-hub"],
                ))
                hub_id = created.get("id", "") or created.get("note_id", "")

        # Persist the run's rubric onto the source (the LEGACY FLAT block) so
        # write-time enforcement and the coverage matrix are model-independent for
        # no-context (flat-checklist) runs and as the back-compat fallback. The
        # PER-CONTEXT ``(project, synthesis_graph)`` rubric is NOT pre-seeded here:
        # prep cannot yet know the REAL synthesis graph (a chosen promoted spine's
        # ``spine_ref`` differs from what this schema would mint, and a forwarded
        # ``synthesis_label`` changes the minted name), so a pre-seed lands on an
        # ORPHAN key and accumulates. The keyed entry is instead written under the
        # REAL key by ``prep_spine`` / ``reconcile_meta_tags`` once structure
        # resolution picks the actual spine. Best-effort: a meta-write hiccup must
        # not abort prep (enforcement simply stays off for it).
        try:
            write_extraction_meta(
                graph_name,
                schema=schema_name,
                strict=schema_strict,
                grounded=schema_grounded,
                tags=schema_tags,
            )
        except Exception:  # noqa: BLE001 — rubric persistence is best-effort
            logger.warning(
                "could not persist extraction rubric onto source '%s'", graph_name
            )

        prepped.append({
            "name": graph_name,
            "title": title,
            "hub_id": hub_id,
            "content_hash": content_hash,
            "source_path": source_path,
            "existing": existing,
            # Book-scale slice metadata (0/[] for a paper) consumed by plan_slices.
            "num_pages": num_pages,
            "total_chars": total_chars,
            "outline": outline,
            # Caller override: explicit per-source slice list, threaded verbatim.
            "slices": s.get("slices") or None,
            "doc_type": (s.get("doc_type") or "").strip(),
        })

    return prepped

prep_spine

prep_spine(schema_obj: dict, project: str, prepped: list[dict], synthesis_label: str = '', parent_label: str = '') -> dict | None

Pre-create the schema's materialized structure (the spine), if declared.

When the schema carries a synthesis block this creates, idempotently: a dedicated synthesis graph (registered as a project source), the apex synthesis node, an optional spec stub, and one node per materialized dimension. It wires dimension nodes -> apex, spec -> apex, and apex -> each per-source hub, so per-source claims (which the scribe later attaches to their dimension node) reach the hub through the spine rather than flat.

parent_label (build-time sub-spine wiring, design §3) declares THIS spine a sub-spine of another: after the child apex exists, the parent spine's synthesis graph + apex are resolved and a primary child-apex --component-of--> parent-apex cross-graph edge is written as the child materializes. The parent graph is resolved via the parent's ORG spine_ref (schema-independent, so a HETEROGENEOUS tier stack whose parent uses a different synthesis.graph template still resolves correctly), falling back to this schema's template only for the homogeneous same-schema case. A parent resolved through a real spine org that has no apex is a HARD ERROR (a broken parent, never silently skipped); a template-fallback parent that is not built yet is logged and skipped so the child still materializes. STABILITY RULE: a hand-set / hand-moved parent edge WINS — if the child apex already carries a primary parent (e.g. a dashboard connect), a re-seed does NOT overwrite it.

Returns the structure record threaded into task context, or None when the schema declares no synthesis block (the run stays a flat checklist).

Source code in coordinator/extraction.py

def prep_spine(
    schema_obj: dict,
    project: str,
    prepped: list[dict],
    synthesis_label: str = "",
    parent_label: str = "",
) -> dict | None:
    """Pre-create the schema's materialized structure (the spine), if declared.

    When the schema carries a ``synthesis`` block this creates, idempotently:
    a dedicated synthesis graph (registered as a project source), the apex
    synthesis node, an optional spec stub, and one node per *materialized*
    dimension. It wires dimension nodes -> apex, spec -> apex, and apex -> each
    per-source hub, so per-source claims (which the scribe later attaches to
    their dimension node) reach the hub through the spine rather than flat.

    ``parent_label`` (build-time sub-spine wiring, design §3) declares THIS spine a
    sub-spine of another: after the child apex exists, the parent spine's synthesis
    graph + apex are resolved and a primary ``child-apex --component-of-->
    parent-apex`` cross-graph edge is written as the child materializes. The parent
    graph is resolved via the parent's ORG ``spine_ref`` (schema-independent, so a
    HETEROGENEOUS tier stack whose parent uses a different ``synthesis.graph``
    template still resolves correctly), falling back to this schema's template only
    for the homogeneous same-schema case. A parent resolved through a real spine org
    that has no apex is a HARD ERROR (a broken parent, never silently skipped); a
    template-fallback parent that is not built yet is logged and skipped so the
    child still materializes. STABILITY RULE: a hand-set / hand-moved parent edge
    WINS — if the child apex already carries a primary parent (e.g. a dashboard
    ``connect``), a re-seed does NOT overwrite it.

    Returns the structure record threaded into task context, or ``None`` when the
    schema declares no ``synthesis`` block (the run stays a flat checklist).
    """
    syn = schema_obj.get("synthesis") if isinstance(schema_obj, dict) else None
    if not syn:
        return None

    from zettelkasten.server import _get_graph, add_to_project, link_notes
    from zettelkasten.spine import (
        ServerSpineGraphOps,
        SpineError,
        SubSpineCycleError,
        build_spine_skeleton,
        connect_sub_spine_edge,
        find_apex_id,
    )

    label = (
        synthesis_label or project or str(schema_obj.get("name") or "") or "synthesis"
    ).strip()
    fields = {
        "label": label,
        "persona": label,
        "project": project,
        "schema": str(schema_obj.get("name") or ""),
    }
    graph_name = _synthesis_graph_name(schema_obj, project, synthesis_label)
    if not graph_name:
        raise ExtractionError(
            f"Synthesis graph name resolved empty for schema "
            f"'{schema_obj.get('name')}' (label='{label}')."
        )

    apex = syn["node"]
    dim_node = syn["dimension_node"]

    # The materialized dimensions, keyed by their schema tag so the returned
    # ``dimension_nodes`` map (tag -> node id) matches the STRUCTURE block the
    # scribe reads. The empty-scaffold path attaches no members; the scribe wires
    # each claim to its dimension node during extraction.
    dims: list[dict[str, Any]] = []
    for d in schema_obj.get("dimensions", []):
        if not d.get("materialize", False):
            continue
        tag = d["tag"]
        dims.append({
            "key": tag,
            "tag": tag,
            "title": _safe_format(dim_node["title_template"], dict(fields, dimension=tag)),
            "type": dim_node["type"],
            "body": (
                f"Dimension '{tag}' for '{label}': {d.get('desc', '')}. "
                f"Source claims for this dimension attach here."
            ),
        })

    spec_spec: "dict[str, Any] | None" = None
    if syn.get("spec"):
        spec = syn["spec"]
        spec_spec = {
            "title": _safe_format(spec["title_template"], fields),
            "type": spec["type"],
            "body": f"Spec stub for '{label}'.",
            "relation": spec["relation"],
        }

    ops = ServerSpineGraphOps()
    try:
        spine_skeleton = build_spine_skeleton(
            ops,
            graph_name=graph_name,
            graph_description=f"Synthesis spine for '{label}' (project '{project}').",
            apex_title=_safe_format(apex["title_template"], fields),
            apex_type=apex["type"],
            apex_body=(
                f"Synthesis apex for '{label}'. Materialized dimension nodes roll up "
                f"here; this node links to each per-source hub."
            ),
            apex_relation=dim_node["relation"],
            attach_relation=dim_node["attach_relation"],
            dimensions=dims,
            spec=spec_spec,
        )
    except SpineError as exc:
        raise ExtractionError(str(exc)) from exc

    # Register the synthesis graph as a normal project source (idempotent). The
    # promotion path deliberately skips this (spines are NOT project sources); a
    # grounded-extraction spine stays a registered source so the scribe's
    # cross-graph attach resolves it.
    add_to_project(project=project, source=graph_name)

    apex_id = spine_skeleton["apex_id"]
    # Apex -> each per-source hub (cross-graph), so hub linkage rolls up.
    if apex_id:
        for p in prepped:
            hub_id = p.get("hub_id")
            src_graph = p.get("name")
            if hub_id and src_graph:
                link_notes(
                    graph=graph_name, source_id=apex_id, target_id=hub_id,
                    relation=apex["link_relation"], target_graph=src_graph,
                )

    # BUILD-TIME SUB-SPINE SEED (design §3): when a ``parent_label`` is declared,
    # hang this child spine beneath the parent by writing a primary apex→apex
    # ``component-of`` edge. The parent graph name is minted by the SAME resolution
    # a ``synthesis_label`` uses, so a run labeled ``E/P`` with parent ``Value``
    # resolves the ``Value`` spine's synthesis graph + apex. The parent must
    # already exist; a missing/empty parent (no apex) is logged and skipped rather
    # than aborting the child's materialization. STABILITY RULE: ``respect_existing``
    # makes a pre-existing hand-authored primary parent edge WIN — a re-seed never
    # silently reverts a dashboard ``connect`` or a hand-moved parent.
    parent_label = (parent_label or "").strip()
    if parent_label and apex_id:
        # Resolve the parent's ACTUAL synthesis graph via its org ``spine_ref``
        # (schema-independent — correct for a heterogeneous tier stack), falling
        # back to the child-schema template ONLY for the homogeneous same-schema
        # case. The fallback path is best-effort (parent may not be built yet); a
        # parent resolved through a real spine ORG must resolve fully or it is a
        # HARD ERROR, never a silent skip.
        resolved_via_org = _resolve_parent_spine_graph(project, parent_label)
        parent_graph = resolved_via_org or _synthesis_graph_name(
            schema_obj, project, parent_label
        )
        if not parent_graph:
            raise ExtractionError(
                f"prep_spine: declared parent_label '{parent_label}' for child "
                f"spine '{graph_name}' could not be resolved to any parent spine "
                "graph."
            )
        if parent_graph == graph_name:
            raise ExtractionError(
                f"prep_spine: declared parent_label '{parent_label}' resolves to "
                f"the child's OWN graph '{graph_name}' — a spine cannot be its own "
                "parent."
            )
        parent_apex = find_apex_id(ops, parent_graph, _get_graph)
        if not parent_apex:
            if resolved_via_org:
                # The parent is a KNOWN promoted/ported spine org, but its
                # ``spine_ref`` graph carries no apex — a structurally BROKEN
                # parent (matching ``connect_sub_spine``'s ``_resolve_spine_apex``),
                # not a not-yet-built one. Fail loud rather than silently skip.
                raise ExtractionError(
                    f"prep_spine: declared parent spine '{parent_label}' (org "
                    f"spine_ref '{parent_graph}') has no apex node; the parent "
                    "spine is broken."
                )
            # Template-fallback path only: the parent tier is simply not built yet
            # (the parent "must exist or be declared earlier in the same setup").
            # Log + skip rather than aborting the child's materialization.
            logger.warning(
                "prep_spine: parent spine graph '%s' (label '%s') has no apex "
                "node; child '%s' NOT nested (declare/build the parent first)",
                parent_graph, parent_label, graph_name,
            )
        else:
            try:
                seed = connect_sub_spine_edge(
                    ops,
                    child_graph=graph_name,
                    child_apex_id=apex_id,
                    parent_graph=parent_graph,
                    parent_apex_id=parent_apex,
                    respect_existing=True,
                )
            except SubSpineCycleError:
                # A build-time seed must never abort the child over a cycle; the
                # explicit connect front door rejects it loudly instead.
                logger.warning(
                    "prep_spine: nesting child '%s' beneath parent '%s' would "
                    "create a component-of cycle; sub-spine seed skipped",
                    graph_name, parent_graph,
                )
            else:
                if not seed.get("connected") and seed.get("kept_existing"):
                    logger.info(
                        "prep_spine: child '%s' already has a primary parent %r; "
                        "stability rule kept it over parent_label '%s'",
                        graph_name, seed["kept_existing"], parent_label,
                    )

    structure = {
        "synthesis_graph": graph_name,
        "apex_id": apex_id,
        "spec_id": spine_skeleton["spec_id"],
        "dimension_nodes": spine_skeleton["dimension_nodes"],
        "attach_relation": dim_node["attach_relation"],
        "apex_relation": dim_node["relation"],
    }
    # Persist per-context (keyed by project + this spine's synthesis_graph) so a
    # source extracted under several schemas/projects keeps an independent
    # structure rather than last-write-wins clobbering. (The stream/templates.py
    # multi-schema path drives this writer once per schema; keying the write here
    # is what makes those per-schema writes non-clobbering.) This is the
    # NAMED-SCHEMA path's ONLY keyed write (no ``reconcile_meta_tags`` runs here),
    # so it MUST carry THIS schema's OWN rubric (tags/strict/grounded/schema) into
    # the keyed entry -- the keyed entry is no longer seeded from the flat block,
    # which in a multi-schema run holds the FIRST schema's tags.
    schema_tags = list(schema_obj.get("tags") or []) if isinstance(schema_obj, dict) else []
    schema_strict = bool(schema_obj.get("strict", True)) if isinstance(schema_obj, dict) else True
    schema_grounded = bool(schema_obj.get("grounded", True)) if isinstance(schema_obj, dict) else True
    schema_name = str(schema_obj.get("name") or "") if isinstance(schema_obj, dict) else ""
    quarantined = _persist_structure_meta(
        structure, prepped, project=project,
        schema=schema_name, strict=schema_strict, grounded=schema_grounded,
        tags=schema_tags,
    )
    # Surface per-source QUARANTINE (P2) to the caller WITHOUT changing the return
    # type (``stream/templates.py`` and other callers consume the structure dict
    # directly). The gx handler pops this and excludes the named sources; consumers
    # that ignore extra keys (build_extraction_tasks, _structure_context) are
    # unaffected. ``_persist_structure_meta`` already raised if NO source survived.
    if quarantined:
        structure["quarantined"] = quarantined
    return structure

prep_spine_from_org

prep_spine_from_org(org: dict, project: str, prepped: list[dict], *, get_graph: Any, materialize: bool = True) -> dict | None

Source the extraction STRUCTURE from a PROMOTED spine org (design §6-7).

Unlike :func:prep_spine -- which mints a FRESH scaffold from a named schema's synthesis block -- this reads an already materialized spine: a state=='spine' org whose spine_ref names the synthesis graph and whose schema carries the EMBEDDED expanded rubric (both landed in phase 2). The spine graph is read DIRECTLY (promoted spines are intentionally NOT project sources, so discover_spines never sees them): the apex + dimension nodes are found by a tag walk over spine.APEX_TAGS / spine.DIMENSION_TAG and mapped back to the embedded schema's dimension tags.

Returns the SAME dict shape :func:prep_spine returns (and that :func:build_extraction_tasks / :func:_structure_context consume): {synthesis_graph, apex_id, spec_id, dimension_nodes, attach_relation, apex_relation}. CRITICAL: attach_relation is the org's EMBEDDED relation (spine-member for a promoted v2 spine), NOT input-to. The v2 readback (_spine_membership_index / spine_readback_matrix) only sees spine-member edges, so attaching extracted claims via input-to would make them invisible in matrix/outline/graph -- this is correctness, not preference.

Degrades to None (caller skips this spine) when the org is not a promoted spine, its spine_ref graph is missing/empty, its embedded schema has no synthesis block, or no materialized dimension node matches a schema tag.

materialize (default True) gates the SOLE write this function performs: the apex -> per-source-hub rollup edges (:func:link_spine_hubs). The tag-walk / dimension detection above is ALWAYS read-only. Scope resolution must evaluate every candidate without side effects, so it calls with materialize=False (detection only) and then invokes :func:link_spine_hubs ONCE for the chosen spine. The default True preserves any direct/legacy caller.

Source code in coordinator/extraction.py

def prep_spine_from_org(
    org: dict,
    project: str,
    prepped: list[dict],
    *,
    get_graph: Any,
    materialize: bool = True,
) -> dict | None:
    """Source the extraction STRUCTURE from a PROMOTED spine org (design §6-7).

    Unlike :func:`prep_spine` -- which mints a FRESH scaffold from a named
    schema's ``synthesis`` block -- this reads an *already materialized* spine: a
    ``state=='spine'`` org whose ``spine_ref`` names the synthesis graph and whose
    ``schema`` carries the EMBEDDED expanded rubric (both landed in phase 2). The
    spine graph is read DIRECTLY (promoted spines are intentionally NOT project
    sources, so ``discover_spines`` never sees them): the apex + dimension nodes
    are found by a tag walk over ``spine.APEX_TAGS`` / ``spine.DIMENSION_TAG`` and
    mapped back to the embedded schema's dimension tags.

    Returns the SAME dict shape :func:`prep_spine` returns (and that
    :func:`build_extraction_tasks` / :func:`_structure_context` consume):
    ``{synthesis_graph, apex_id, spec_id, dimension_nodes, attach_relation,
    apex_relation}``. CRITICAL: ``attach_relation`` is the org's EMBEDDED relation
    (``spine-member`` for a promoted v2 spine), NOT ``input-to``. The v2 readback
    (``_spine_membership_index`` / ``spine_readback_matrix``) only sees ``spine-member``
    edges, so attaching extracted claims via ``input-to`` would make them invisible
    in matrix/outline/graph -- this is correctness, not preference.

    Degrades to ``None`` (caller skips this spine) when the org is not a promoted
    spine, its ``spine_ref`` graph is missing/empty, its embedded ``schema`` has no
    ``synthesis`` block, or no materialized dimension node matches a schema tag.

    ``materialize`` (default ``True``) gates the SOLE write this function performs:
    the apex -> per-source-hub rollup edges (:func:`link_spine_hubs`). The tag-walk
    / dimension detection above is ALWAYS read-only. Scope resolution must evaluate
    every candidate without side effects, so it calls with ``materialize=False``
    (detection only) and then invokes :func:`link_spine_hubs` ONCE for the chosen
    spine. The default ``True`` preserves any direct/legacy caller.
    """
    if not isinstance(org, dict) or str(org.get("state") or "") != "spine":
        return None
    spine_graph = str(org.get("spine_ref") or "").strip()
    if not spine_graph:
        return None
    schema = org.get("schema") if isinstance(org.get("schema"), dict) else {}
    syn = schema.get("synthesis") if isinstance(schema, dict) else None
    if not isinstance(syn, dict):
        return None

    from zettelkasten.spine import APEX_TAGS, DIMENSION_TAG

    # Read the materialized spine graph directly. A missing/unloadable or empty
    # graph degrades gracefully -- the caller falls back rather than crashing.
    try:
        zg = get_graph(spine_graph)
    except Exception:  # noqa: BLE001 — a missing spine graph just means "skip it"
        return None
    notes = list(getattr(zg, "notes", {}).values())
    if not notes:
        return None

    # attach/apex relations come from the org's EMBEDDED schema: for a promoted v2
    # spine ``attach_relation`` is ``spine-member`` (the spine-side membership
    # edge), NOT extraction prep's ``input-to``.
    dim_node = syn.get("dimension_node") if isinstance(syn.get("dimension_node"), dict) else {}
    attach_relation = str(dim_node.get("attach_relation") or "").strip()
    apex_relation = str(dim_node.get("relation") or "").strip()

    # Tag walk: the apex (APEX_TAGS) and the materialized dimension nodes
    # (DIMENSION_TAG + the schema dimension tag). build_spine_skeleton / derive_dimension_tags
    # stamp the SAME tag on the node and embed it in the schema, so matching by tag
    # keeps the embedded-schema tags and the node tags in agreement.
    apex_set = set(APEX_TAGS)
    apex_id = ""
    nodes_by_tag: dict[str, str] = {}
    for note in notes:
        tags = set(note.tags or [])
        if not apex_id and apex_set <= tags:
            apex_id = note.id
            continue
        if DIMENSION_TAG in tags:
            for t in tags:
                if t and t != DIMENSION_TAG:
                    nodes_by_tag.setdefault(t, note.id)

    # Restrict + order the map to the embedded schema's MATERIALIZED dimensions,
    # keyed by the schema tag the scribe routes claims on.
    dimension_nodes: dict[str, str] = {}
    missing_tags: list[str] = []
    for d in schema.get("dimensions", []):
        if not isinstance(d, dict):
            continue
        # Default ``False`` MATCHES prep_spine's materializable filter: only a
        # dimension explicitly marked ``materialize`` is EXPECTED to have a node in
        # the spine graph. (A real embedded schema is ``expand_spec``-expanded, so
        # every synthesis dimension already carries an explicit ``materialize``
        # flag.) A dim with no flag is neither routed nor counted missing, so it
        # never produces a FALSE ``partial`` for a spine materialized per its intent.
        if not d.get("materialize", False):
            continue
        tag = str(d.get("tag") or "").strip()
        if not tag:
            continue
        nid = nodes_by_tag.get(tag)
        if nid:
            dimension_nodes[tag] = nid
        else:
            missing_tags.append(tag)

    if not dimension_nodes:
        return None

    # Partial coverage is allowed (we route the matched dims), but a materialized
    # dimension whose embedded-schema tag has NO node in the spine graph would be
    # SILENTLY dropped -- make that drop OBSERVABLE so the divergence is noticed.
    if missing_tags:
        logger.warning(
            "prep_spine_from_org: spine '%s' has no materialized node for "
            "embedded-schema dimension tag(s) %s; extracting PARTIAL coverage over "
            "the %d matched dimension(s) %s",
            spine_graph, missing_tags, len(dimension_nodes),
            sorted(dimension_nodes),
        )

    # The apex -> hub link relation comes from the embedded schema's synthesis
    # ``node``. Computed unconditionally (read-only) and threaded onto the returned
    # structure so the caller can drive the deferred link write without re-reading.
    node = syn.get("node") if isinstance(syn.get("node"), dict) else {}
    link_relation = str(node.get("link_relation") or "").strip() or "related"

    structure = {
        "synthesis_graph": spine_graph,
        "apex_id": apex_id,
        "spec_id": "",
        "dimension_nodes": dimension_nodes,
        "attach_relation": attach_relation or "spine-member",
        "apex_relation": apex_relation or "component-of",
        # Carried for the deferred apex->hub link step (link_spine_hubs); other
        # consumers (_structure_context, build_extraction_tasks) ignore extra keys.
        "link_relation": link_relation,
        # The CHOSEN spine's EMBEDDED rubric, carried so the keyed extraction-meta
        # write (``reconcile_meta_tags``) persists THIS spine's strict/grounded/
        # schema -- NOT the named registry schema's, which can differ from a
        # scoped spine's embedded values and would write a WRONG keyed rubric. A
        # field absent from the embedded schema is ``None`` here so the caller can
        # fall back to the named schema's value. (``tags`` already come from the
        # spine via ``dimension_nodes``.)
        "embedded_strict": (
            schema.get("strict") if isinstance(schema, dict) else None
        ),
        "embedded_grounded": (
            schema.get("grounded") if isinstance(schema, dict) else None
        ),
        "embedded_schema_name": (
            str(schema.get("name") or "") if isinstance(schema, dict) else ""
        ),
        # PARTIAL coverage: embedded-schema dimension tags with NO node in the spine
        # graph (logged above). Surfaced so a half-materialized chosen spine is
        # distinguishable from a fully-covered one in the resolution / tool JSON.
        "missing_dims": list(missing_tags),
    }
    # Detection above is a PURE READ. BOTH side effects -- the apex -> per-source
    # hub rollup edges AND the per-source extraction-meta -- are deferred behind
    # ``materialize`` so scope resolution can evaluate every candidate without
    # writing; the caller materializes ONLY the chosen spine. (Note: persisting the
    # source meta is last-write-wins, so leaving it ungated would let an IGNORED
    # candidate clobber the chosen spine's source meta -- gating it is correctness.)
    if materialize:
        materialize_structure(structure, prepped, project=project)
    return structure

link_spine_hubs

link_spine_hubs(structure: dict, prepped: list[dict]) -> None

Write the apex -> per-source-hub rollup edges for a resolved spine.

Factored out of :func:prep_spine_from_org so SCOPE RESOLUTION can stay a PURE READ: detection runs with materialize=False (zero graph writes) for EVERY candidate, the caller picks exactly ONE, then materializes it ONCE. The spine graph is NOT (re)registered as a project source -- promoted spines are deliberately not project sources. Best-effort: never abort.

No-op without an apex id / synthesis graph / prepared sources, so it is safe on any structure dict (including the named-schema fallback, which has no apex of this form).

Source code in coordinator/extraction.py

def link_spine_hubs(structure: dict, prepped: list[dict]) -> None:
    """Write the apex -> per-source-hub rollup edges for a resolved spine.

    Factored out of :func:`prep_spine_from_org` so SCOPE RESOLUTION can stay a
    PURE READ: detection runs with ``materialize=False`` (zero graph writes) for
    EVERY candidate, the caller picks exactly ONE, then materializes it ONCE. The
    spine graph is NOT (re)registered as a project source -- promoted spines are
    deliberately not project sources. Best-effort: never abort.

    No-op without an apex id / synthesis graph / prepared sources, so it is safe on
    any structure dict (including the named-schema fallback, which has no apex of
    this form).
    """
    apex_id = str(structure.get("apex_id") or "").strip()
    spine_graph = str(structure.get("synthesis_graph") or "").strip()
    if not apex_id or not spine_graph or not prepped:
        return
    from zettelkasten.server import link_notes

    link_relation = str(structure.get("link_relation") or "").strip() or "related"
    for p in prepped:
        hub_id = p.get("hub_id")
        src_graph = p.get("name")
        if hub_id and src_graph:
            try:
                link_notes(
                    graph=spine_graph, source_id=apex_id, target_id=hub_id,
                    relation=link_relation, target_graph=src_graph,
                )
            except Exception:  # noqa: BLE001 — hub rollup link is best-effort
                logger.warning(
                    "could not link spine apex '%s' -> hub '%s' (graph '%s')",
                    apex_id, hub_id, src_graph,
                )

materialize_structure

materialize_structure(structure: dict, prepped: list[dict], project: str = '') -> list[dict]

Perform the deferred WRITES for the CHOSEN spine structure.

:func:prep_spine_from_org defers BOTH of its side effects -- the apex -> per-source-hub rollup edges AND the per-source extraction-meta -- behind its materialize flag so scope resolution can detect every candidate as a pure read. Once the caller picks ONE structure it calls this ONCE, so exactly the chosen spine is materialized; ignored/broken candidates leave ZERO writes.

project (when known) keys the persisted structure meta by (project, synthesis_graph) so the chosen spine's structure does not clobber another context's; absent it, the legacy flat block is written. A per-structure project (set by the multi-project resolver) takes precedence over the passed project argument, so each spine is keyed under ITS OWN project even when materialized in a multi-project run.

Returns the quarantine list from :func:_persist_structure_meta (empty here: this structure-only write passes no strict, so it is never enforcing -- the spine path's enforcing rubric write is :func:reconcile_meta_tags).

Source code in coordinator/extraction.py

def materialize_structure(
    structure: dict, prepped: list[dict], project: str = ""
) -> list[dict]:
    """Perform the deferred WRITES for the CHOSEN spine structure.

    :func:`prep_spine_from_org` defers BOTH of its side effects -- the apex ->
    per-source-hub rollup edges AND the per-source extraction-meta -- behind its
    ``materialize`` flag so scope resolution can detect every candidate as a pure
    read. Once the caller picks ONE structure it calls this ONCE, so exactly the
    chosen spine is materialized; ignored/broken candidates leave ZERO writes.

    ``project`` (when known) keys the persisted structure meta by
    ``(project, synthesis_graph)`` so the chosen spine's structure does not
    clobber another context's; absent it, the legacy flat block is written. A
    per-structure ``project`` (set by the multi-project resolver) takes precedence
    over the passed ``project`` argument, so each spine is keyed under ITS OWN
    project even when materialized in a multi-project run.

    Returns the quarantine list from :func:`_persist_structure_meta` (empty here:
    this structure-only write passes no ``strict``, so it is never enforcing -- the
    spine path's enforcing rubric write is :func:`reconcile_meta_tags`).
    """
    link_spine_hubs(structure, prepped)
    eff_project = (
        structure.get("project") if isinstance(structure, dict) else ""
    ) or project
    return _persist_structure_meta(structure, prepped, project=eff_project)

resolve_extraction_structure

resolve_extraction_structure(project: str, prepped: list[dict], spines: list[str], *, get_graph: Any, present_fn: 'Callable[[str, str], bool] | None' = None, multi: bool = False) -> dict

Resolve the spine scope to a structure resolution dict.

Shared by the coordinator create_extraction_graph tool AND the stream / daemon path (phase 3b): both turn a spines scope into the STRUCTURE(s) the trio tasks fill. get_graph is injected (the zettelkasten _get_graph) so this module never imports the zettelkasten MCP server, and present_fn (defaulting to :func:_explicit_spine_present) is injectable purely so the coordinator wrapper can keep its existing monkeypatch seam.

Builds the ordered, de-duped scope and returns {"structure", "structures", "ignored_spines", "broken_spines", "unknown_spines"}:

• structure -- the chosen STRUCTURE dict (None to fall back to the named-schema path). Back-compat: equals structures[0] (or None).
• structures -- the list of materialized STRUCTURE dicts. With multi=False (default) this is [structure] (or []) -- the single chosen spine, byte-identical to the historical behavior. With multi=True it holds EVERY in-scope spine that yielded a valid structure (the opt-in full multi-fill).
• ignored_spines -- other in-scope org ids that ALSO yielded a valid structure but were NOT used. In single mode (multi=False) this run fills exactly ONE structure, so every OTHER valid in-scope spine is deferred here (to a later resync). In multi mode every valid spine is filled, so this shrinks to only entries beyond any optional cap (none today -- there is no cap, so it stays empty under multi=True).
• broken_spines -- scoped org ids that CLAIM to be a promoted spine (state=='spine' with a spine_ref) but yielded no structure (their spine graph is missing/empty / no materialized dimension matches).
• unknown_spines -- EXPLICIT opt-ins (caller-named in spines) that resolved to no structure and are NOT broken-promoted: a nonexistent org id, a saved-but-UN-promoted lens, or a state=='spine' org with a BLANK spine_ref. Surfaced so an explicit opt-in never silently vanishes into the default. The auto-included DEFAULT falling through is NOT flagged (legacy no-scope behavior).

No-silent-loss for an EXPLICIT demand: if a caller-named org is PRESENT on disk (present_fn) but cannot be loaded/normalized -- a corrupt org file, or load_organization raising / returning None via a caught ValueError -- this function RAISES ExtractionError rather than bucketing it unknown and silently degrading to the named-schema fallback. The caller explicitly demanded THAT spine; extracting into a DIFFERENT structure would be a silent wrong-structure run. Only a genuinely ABSENT explicit id (a typo) stays graceful unknown. Default/auto spine resolution failures CONTINUE to degrade gracefully (broken/silent) -- only EXPLICIT opt-ins fail loud.

Scope resolution is a PURE READ: each candidate is detected via :func:prep_spine_from_org with materialize=False (zero graph writes), and only the CHOSEN structure is materialized (apex->hub edges + source meta).

Precedence: EXPLICIT opt-ins take precedence over the project default. The scope is the de-duped opt-in org ids in caller order FIRST, then the project default appended last (still auto-included). The FIRST scoped org that yields a valid structure is CHOSEN; every OTHER scoped org that ALSO yields a valid structure is recorded in ignored_spines. So spines=['B'] with project default A uses B and reports A as ignored. With no opt-ins the default is the sole candidate (design §7). De-duping by org id means an org that is BOTH the default and an explicit opt-in is processed once (no double-extraction).

MULTI-SPINE (multi=True, opt-in): fill EVERY in-scope spine that yields a valid structure rather than collapsing to the first. The scope is already LIST-shaped and detection already evaluates EVERY candidate as a pure read, so this just collects ALL valid structures and materializes each (materialize_structure per structure -- each writes its own keyed (project, synthesis_graph) extraction-meta, so the contexts stay independent). Downstream (build_extraction_tasks / _structure_context / the scribe persona) then carries MULTIPLE numbered STRUCTURE blocks. The semantics: a claim attaches to the matching dimension node of EVERY in-scope spine whose dimension it fits; a spine whose dimensions it does NOT fit gets no membership edge and the claim stays connected to those spines only through its organic inter-note edges (membership is per-dimension and opt-in, so non-match == no edge, automatically -- heterogeneous schemas are handled for free). Default multi=False keeps "first wins" and returns exactly ONE structure (and structures=[structure]), byte-identical to today.

Source code in coordinator/extraction.py

def resolve_extraction_structure(
    project: str,
    prepped: list[dict],
    spines: list[str],
    *,
    get_graph: Any,
    present_fn: "Callable[[str, str], bool] | None" = None,
    multi: bool = False,
) -> dict:
    """Resolve the spine scope to a structure resolution dict.

    Shared by the coordinator ``create_extraction_graph`` tool AND the stream /
    daemon path (phase 3b): both turn a ``spines`` scope into the
    STRUCTURE(s) the trio tasks fill. ``get_graph`` is injected (the zettelkasten
    ``_get_graph``) so this module never imports the zettelkasten MCP server, and
    ``present_fn`` (defaulting to :func:`_explicit_spine_present`) is injectable
    purely so the coordinator wrapper can keep its existing monkeypatch seam.

    Builds the ordered, de-duped scope and returns
    ``{"structure", "structures", "ignored_spines", "broken_spines",
    "unknown_spines"}``:

    * ``structure`` -- the chosen STRUCTURE dict (``None`` to fall back to the
      named-schema path). Back-compat: equals ``structures[0]`` (or ``None``).
    * ``structures`` -- the list of materialized STRUCTURE dicts. With
      ``multi=False`` (default) this is ``[structure]`` (or ``[]``) -- the
      single chosen spine, byte-identical to the historical behavior. With
      ``multi=True`` it holds EVERY in-scope spine that yielded a valid
      structure (the opt-in full multi-fill).
    * ``ignored_spines`` -- other in-scope org ids that ALSO yielded a valid
      structure but were NOT used. In single mode (``multi=False``) this run
      fills exactly ONE structure, so every OTHER valid in-scope spine is
      deferred here (to a later ``resync``). In multi mode every valid spine is
      filled, so this shrinks to only entries beyond any optional cap (none
      today -- there is no cap, so it stays empty under ``multi=True``).
    * ``broken_spines`` -- scoped org ids that CLAIM to be a promoted spine
      (``state=='spine'`` with a ``spine_ref``) but yielded no structure
      (their spine graph is missing/empty / no materialized dimension matches).
    * ``unknown_spines`` -- EXPLICIT opt-ins (caller-named in ``spines``) that
      resolved to no structure and are NOT broken-promoted: a nonexistent org
      id, a saved-but-UN-promoted lens, or a ``state=='spine'`` org with a BLANK
      ``spine_ref``. Surfaced so an explicit opt-in never silently vanishes into
      the default. The auto-included DEFAULT falling through is NOT flagged
      (legacy no-scope behavior).

    No-silent-loss for an EXPLICIT demand: if a caller-named org is PRESENT on
    disk (``present_fn``) but cannot be loaded/normalized -- a corrupt org file,
    or ``load_organization`` raising / returning ``None`` via a caught
    ``ValueError`` -- this function RAISES ``ExtractionError`` rather than
    bucketing it ``unknown`` and silently degrading to the named-schema
    fallback. The caller explicitly demanded THAT spine; extracting into a
    DIFFERENT structure would be a silent wrong-structure run. Only a genuinely
    ABSENT explicit id (a typo) stays graceful ``unknown``. Default/auto spine
    resolution failures CONTINUE to degrade gracefully (``broken``/silent) --
    only EXPLICIT opt-ins fail loud.

    Scope resolution is a PURE READ: each candidate is detected via
    :func:`prep_spine_from_org` with ``materialize=False`` (zero graph
    writes), and only the CHOSEN structure is materialized (apex->hub edges +
    source meta).

    Precedence: EXPLICIT opt-ins take precedence over the project default. The
    scope is the de-duped opt-in org ids in caller order FIRST, then the project
    default appended last (still auto-included). The FIRST scoped org that yields
    a valid structure is CHOSEN; every OTHER scoped org that ALSO yields a valid
    structure is recorded in ``ignored_spines``. So ``spines=['B']`` with project
    default ``A`` uses B and reports A as ignored. With no opt-ins the default is
    the sole candidate (design §7). De-duping by org id means an org that is BOTH
    the default and an explicit opt-in is processed once (no double-extraction).

    MULTI-SPINE (``multi=True``, opt-in): fill EVERY in-scope spine that yields a
    valid structure rather than collapsing to the first. The scope is already
    LIST-shaped and detection already evaluates EVERY candidate as a pure read,
    so this just collects ALL valid structures and materializes each
    (``materialize_structure`` per structure -- each writes its own keyed
    ``(project, synthesis_graph)`` extraction-meta, so the contexts stay
    independent). Downstream (``build_extraction_tasks`` / ``_structure_context``
    / the scribe persona) then carries MULTIPLE numbered STRUCTURE blocks. The
    semantics: a claim attaches to the matching dimension node of EVERY in-scope
    spine whose dimension it fits; a spine whose dimensions it does NOT fit gets
    no membership edge and the claim stays connected to those spines only through
    its organic inter-note edges (membership is per-dimension and opt-in, so
    non-match == no edge, automatically -- heterogeneous schemas are handled for
    free). Default ``multi=False`` keeps "first wins" and returns exactly ONE
    ``structure`` (and ``structures=[structure]``), byte-identical to today.
    """
    from zettelkasten import graph as zk_graph
    from zettelkasten import organizations as zk_orgs

    present = present_fn or _explicit_spine_present

    # Explicit opt-ins FIRST (caller order, de-duped) so an explicit choice wins
    # over the project default; the default is appended last, still auto-included.
    # ``explicit_ids`` records which scoped ids the CALLER named so an explicit
    # opt-in that fails to resolve can be surfaced (never silently dropped),
    # while the auto-included default may legitimately fall through.
    scoped_ids: list[str] = []
    explicit_ids: set[str] = set()
    for sid in spines or []:
        sid = str(sid or "").strip()
        if sid and sid not in scoped_ids:
            scoped_ids.append(sid)
            explicit_ids.add(sid)

    default_spine = None
    try:
        default_spine = zk_graph.project_default_spine(
            zk_graph.load_project(project)
        )
    except ValueError as e:
        # A corrupt manifest is real data loss -- do NOT swallow it silently.
        logger.warning(
            "extraction scope: could not load project '%s' for its default "
            "spine: %s", project, e,
        )
    except Exception as e:  # noqa: BLE001 — a malformed value just means "no default"
        logger.warning(
            "extraction scope: default-spine lookup for project '%s' failed: %s",
            project, e,
        )
    if default_spine and default_spine not in scoped_ids:
        scoped_ids.append(default_spine)

    chosen_structures: list[dict] = []
    ignored_spines: list[str] = []
    broken_spines: list[str] = []
    unknown_spines: list[str] = []

    def _explicit_failed_to_load(org_id: str, reason: str) -> None:
        """Bucket an EXPLICIT opt-in that did not load.

        A caller-named org that is really PRESENT on disk but cannot be
        loaded/normalized is a corrupt / migration-or-normalize failure: the
        caller demanded THIS spine, so silently routing its claims into the
        named-schema fallback would be a silent wrong-structure run. FAIL LOUD.
        A genuinely ABSENT id is a typo / nonexistent org -> graceful
        ``unknown`` (the id is simply wrong).
        """
        if present(project, org_id):
            raise ExtractionError(
                f"explicit spine opt-in '{org_id}' exists in project "
                f"'{project}' but could not be loaded ({reason}); refusing to "
                f"silently extract against the named-schema fallback -- the "
                f"caller explicitly demanded this spine. Repair / re-promote "
                f"the corrupt spine org, or drop it from spines=."
            )
        unknown_spines.append(org_id)

    for org_id in scoped_ids:
        try:
            org = zk_orgs.load_organization("project", project, org_id)
        except Exception as e:  # noqa: BLE001 — an unreadable scoped org is broken
            logger.warning(
                "extraction scope: scoped spine org '%s' could not be loaded: %s",
                org_id, e,
            )
            # An EXPLICIT opt-in: a PRESENT-but-unreadable org fails loud (its
            # file exists on disk -- a real corrupt/migration error, not a
            # typo); a genuinely absent id is a typo -> ``unknown``. A
            # non-explicit / auto-included default load failure stays ``broken``
            # (degrades gracefully).
            if org_id in explicit_ids:
                _explicit_failed_to_load(org_id, str(e))
            else:
                broken_spines.append(org_id)
            continue
        if not org:
            # ``load_organization`` returns None for BOTH a missing file (a
            # typo) AND a present-but-corrupt file (unparseable JSON / a caught
            # normalize ValueError). For an EXPLICIT opt-in, disambiguate by
            # disk presence: a present-but-unloadable org fails loud (never
            # silently routes into the default), an absent id is ``unknown``.
            # The auto-included default silently falling through stays legitimate
            # (it may simply not exist yet).
            if org_id in explicit_ids:
                _explicit_failed_to_load(
                    org_id,
                    "corrupt org file or migration/normalize error",
                )
            continue
        # PURE READ: detect the structure with NO graph writes. Only the chosen
        # spine is materialized (below), so an ignored/broken candidate leaves
        # zero side effects -- the "fills exactly ONE structure" guarantee.
        spine = prep_spine_from_org(
            org, project, prepped, get_graph=get_graph, materialize=False
        )
        if spine:
            if multi or not chosen_structures:
                # multi=True: fill EVERY valid in-scope spine. Default
                # (multi=False): the FIRST valid spine wins (byte-identical to
                # the historical "first wins").
                chosen_structures.append(spine)
            else:
                # Single mode: a second in-scope spine that ALSO resolved is
                # deferred to a later resync (this run fills exactly one).
                ignored_spines.append(org_id)
        elif (
            isinstance(org, dict)
            and str(org.get("state") or "") == "spine"
            and str(org.get("spine_ref") or "").strip()
        ):
            # A SCOPED promoted spine that yielded nothing is BROKEN (its
            # spine_ref graph is missing/empty / no materialized dim matched) --
            # distinct from a non-spine org, which is just the legacy fallback.
            broken_spines.append(org_id)
        elif org_id in explicit_ids:
            # An EXPLICIT opt-in that resolved to no structure and is NOT a
            # broken promoted spine (a saved-but-UN-promoted lens, a
            # ``state=='spine'`` org with a BLANK spine_ref, or a non-spine org):
            # surface it so the explicit intent never vanishes into the default.
            unknown_spines.append(org_id)
    # Materialize ONLY the chosen spine(s) (the deferred-write step). Ignored
    # / broken / unknown candidates were detected as pure reads and stay so.
    # Thread ``project`` so each structure is persisted under its OWN per-context
    # ``(project, synthesis_graph)`` key -- WITHOUT it the keyed write path is
    # dead in production and cross-project runs clobber each other's structure
    # last-write-wins (P0). The key is each structure's own ``synthesis_graph``
    # (the spine_ref), set inside ``_persist_structure_meta``. In single mode
    # this loop runs at most ONCE (byte-identical to materializing the lone
    # chosen spine); in multi mode each in-scope spine is materialized
    # independently under its own keyed context.
    for s in chosen_structures:
        materialize_structure(s, prepped, project=project)
    return {
        "structure": chosen_structures[0] if chosen_structures else None,
        "structures": list(chosen_structures),
        "ignored_spines": ignored_spines,
        "broken_spines": broken_spines,
        "unknown_spines": unknown_spines,
    }

resolve_extraction_structures_multi

resolve_extraction_structures_multi(projects: list[str], prepped: list[dict], spines: list[str], *, get_graph: Any, present_fn: 'Callable[[str, str], bool] | None' = None, multi: bool = False) -> dict

Resolve the spine scope across SEVERAL projects and union the structures.

The "extract-once, attach-to-many-projects" entrypoint. A source is prepped / ingested ONCE (the caller already ran :func:prep_sources for the full project set); this then calls :func:resolve_extraction_structure ONCE PER PROJECT -- each resolves and materializes that project's own spine scope under its OWN (project, synthesis_graph) key, so the contexts stay independent (no clobber) -- stamps structure["project"] = <that project> on every returned structure so the downstream render / reconcile keys each spine under ITS OWN project, and concatenates the per-project structure lists. The per-project diagnostics (ignored_spines / broken_spines / unknown_spines) are UNIONed. An explicit spine id that does not resolve in a given project's own org tree simply lands in that project's unknown_spines (best-effort).

With a single project this is exactly :func:resolve_extraction_structure plus a project stamp -- the multi-project codepaths stay inert.

Source code in coordinator/extraction.py

def resolve_extraction_structures_multi(
    projects: list[str],
    prepped: list[dict],
    spines: list[str],
    *,
    get_graph: Any,
    present_fn: "Callable[[str, str], bool] | None" = None,
    multi: bool = False,
) -> dict:
    """Resolve the spine scope across SEVERAL projects and union the structures.

    The "extract-once, attach-to-many-projects" entrypoint. A source is prepped /
    ingested ONCE (the caller already ran :func:`prep_sources` for the full project
    set); this then calls :func:`resolve_extraction_structure` ONCE PER PROJECT --
    each resolves and materializes that project's own spine scope under its OWN
    ``(project, synthesis_graph)`` key, so the contexts stay independent (no
    clobber) -- stamps ``structure["project"] = <that project>`` on every returned
    structure so the downstream render / reconcile keys each spine under ITS OWN
    project, and concatenates the per-project structure lists. The per-project
    diagnostics (``ignored_spines`` / ``broken_spines`` / ``unknown_spines``) are
    UNIONed. An explicit spine id that does not resolve in a given project's own
    org tree simply lands in that project's ``unknown_spines`` (best-effort).

    With a single project this is exactly :func:`resolve_extraction_structure`
    plus a ``project`` stamp -- the multi-project codepaths stay inert.
    """
    all_structures: list[dict] = []
    ignored: list[str] = []
    broken: list[str] = []
    unknown: list[str] = []
    for p in projects:
        res = resolve_extraction_structure(
            p, prepped, spines, get_graph=get_graph,
            present_fn=present_fn, multi=multi,
        )
        structs = res.get("structures")
        if structs is None:
            single = res.get("structure")
            structs = [single] if single is not None else []
        for s in structs:
            if isinstance(s, dict):
                # Key each spine under ITS OWN project for the keyed meta /
                # rendered CONTEXT_KEY downstream.
                s["project"] = p
            all_structures.append(s)
        ignored.extend(res.get("ignored_spines") or [])
        broken.extend(res.get("broken_spines") or [])
        unknown.extend(res.get("unknown_spines") or [])
    return {
        "structure": all_structures[0] if all_structures else None,
        "structures": all_structures,
        "ignored_spines": ignored,
        "broken_spines": broken,
        "unknown_spines": unknown,
    }

reconcile_meta_tags

reconcile_meta_tags(prepped: list[dict], structure: dict, project: str = '', *, strict: bool | None = None, grounded: bool | None = None, schema: str = '') -> list[dict]

Reconcile each source's persisted strict-tag VOCABULARY to a chosen spine.

prep_sources persists the NAMED registry schema's tags (and strict intent) onto every source BEFORE structure resolution. When a spines=-sourced structure wins, its dimension_nodes are keyed by the SPINE's EMBEDDED dimension tags -- an independent vocabulary. add_note strict enforcement checks a note's tags against the persisted tags, so a scribe tagging claims with the spine's tag (e.g. claim) would be REJECTED whenever the spine tags are not a subset of the named schema's (e.g. {finding, method}) -- silently breaking the spines= feature.

Re-write each prepared source's persisted tags to the CHOSEN structure's dimension tags so the scribe's spine-tagged claims pass enforcement. No-op for a structure with no dimension tags (so the named-schema fallback, which never calls this, is unaffected).

Returns the list of QUARANTINED sources -- [{"source", "reason"}] for each source whose ENFORCING keyed write failed -- so the caller can EXCLUDE them from the run and report them. Empty for a clean / non-enforcing run.

The per-source write is best-effort EXCEPT for an ENFORCING keyed run (a keyed write whose effective strict is True). There a failed write is NOT silently swallowed -- otherwise the scribe's full (project, synthesis_graph) key would miss against a populated contexts list and reject every claim later (the no-silent-loss P1). But to avoid stranding the other healthy sources (the P2 availability regression), a single enforcing source's write failure QUARANTINES just that source (recorded in the returned list, loudly logged) and the rest continue; only when NO source survives does this raise :class:ExtractionError. Non-enforcing / flat / non-strict writes stay best-effort, as does a missing-_meta.yaml (FileNotFoundError) -- a structural "not a real prepped source" condition, not the transient write failure the escalation targets.

When project is known the reconcile targets the per-context (project, synthesis_graph) entry (the chosen spine's synthesis_graph); without a project the legacy FLAT block is reconciled (back-compat) and ONLY tags is written there (preserving whatever strict/grounded prep wrote flat).

In the KEYED branch strict/grounded/schema (the run's rubric) are written INTO the keyed entry alongside the spine tags. This is the spine path's keyed rubric write: the keyed entry is no longer seeded from the flat block, so the run's strict intent must be written here explicitly or write-time enforcement would silently stay off for the spine context.

The keyed strict/grounded/schema PREFER the chosen spine's EMBEDDED rubric carried on structure (embedded_strict / embedded_grounded / embedded_schema_name, set by :func:prep_spine_from_org) over the strict/grounded/schema arguments. A scoped spine's embedded strict/grounded can differ from the named registry schema the caller passes; persisting the named value would write the WRONG keyed rubric. The argument is the FALLBACK, used only when the embedded field is absent (None / empty).

Source code in coordinator/extraction.py

def reconcile_meta_tags(
    prepped: list[dict],
    structure: dict,
    project: str = "",
    *,
    strict: bool | None = None,
    grounded: bool | None = None,
    schema: str = "",
) -> list[dict]:
    """Reconcile each source's persisted strict-tag VOCABULARY to a chosen spine.

    ``prep_sources`` persists the NAMED registry schema's ``tags`` (and ``strict``
    intent) onto every source BEFORE structure resolution. When a ``spines=``-sourced
    structure wins, its ``dimension_nodes`` are keyed by the SPINE's EMBEDDED
    dimension tags -- an independent vocabulary. ``add_note`` strict enforcement
    checks a note's tags against the persisted ``tags``, so a scribe tagging claims
    with the spine's tag (e.g. ``claim``) would be REJECTED whenever the spine tags
    are not a subset of the named schema's (e.g. ``{finding, method}``) -- silently
    breaking the ``spines=`` feature.

    Re-write each prepared source's persisted ``tags`` to the CHOSEN structure's
    dimension tags so the scribe's spine-tagged claims pass enforcement. No-op for
    a structure with no dimension tags (so the named-schema fallback, which never
    calls this, is unaffected).

    Returns the list of QUARANTINED sources -- ``[{"source", "reason"}]`` for each
    source whose ENFORCING keyed write failed -- so the caller can EXCLUDE them
    from the run and report them. Empty for a clean / non-enforcing run.

    The per-source write is best-effort EXCEPT for an ENFORCING keyed run (a keyed
    write whose effective ``strict`` is True). There a failed write is NOT silently
    swallowed -- otherwise the scribe's full ``(project, synthesis_graph)`` key
    would miss against a populated ``contexts`` list and reject every claim later
    (the no-silent-loss P1). But to avoid stranding the other healthy sources (the
    P2 availability regression), a single enforcing source's write failure
    QUARANTINES just that source (recorded in the returned list, loudly logged) and
    the rest continue; only when NO source survives does this raise
    :class:`ExtractionError`. Non-enforcing / flat / non-strict writes stay
    best-effort, as does a missing-``_meta.yaml`` (``FileNotFoundError``) -- a
    structural "not a real prepped source" condition, not the transient write
    failure the escalation targets.

    When ``project`` is known the reconcile targets the per-context
    ``(project, synthesis_graph)`` entry (the chosen spine's ``synthesis_graph``);
    without a project the legacy FLAT block is reconciled (back-compat) and ONLY
    ``tags`` is written there (preserving whatever ``strict``/``grounded`` prep
    wrote flat).

    In the KEYED branch ``strict``/``grounded``/``schema`` (the run's rubric) are
    written INTO the keyed entry alongside the spine tags. This is the spine
    path's keyed rubric write: the keyed entry is no longer seeded from the flat
    block, so the run's ``strict`` intent must be written here explicitly or
    write-time enforcement would silently stay off for the spine context.

    The keyed strict/grounded/schema PREFER the chosen spine's EMBEDDED rubric
    carried on ``structure`` (``embedded_strict`` / ``embedded_grounded`` /
    ``embedded_schema_name``, set by :func:`prep_spine_from_org`) over the
    ``strict``/``grounded``/``schema`` arguments. A scoped spine's embedded
    strict/grounded can differ from the named registry schema the caller passes;
    persisting the named value would write the WRONG keyed rubric. The argument is
    the FALLBACK, used only when the embedded field is absent (``None`` / empty).
    """
    if not structure:
        return []
    tags = list((structure.get("dimension_nodes") or {}).keys())
    if not tags:
        return []
    from zettelkasten.graph import write_extraction_meta

    # Prefer the chosen spine's EMBEDDED rubric over the named-schema fallback the
    # caller passes (see docstring): a spine's own strict/grounded/schema is the
    # authority for its keyed entry.
    eff_strict = structure.get("embedded_strict")
    if eff_strict is None:
        eff_strict = strict
    eff_grounded = structure.get("embedded_grounded")
    if eff_grounded is None:
        eff_grounded = grounded
    eff_schema = str(structure.get("embedded_schema_name") or "") or schema

    # Per-structure project (set by the multi-project resolver) keys this spine's
    # reconcile under ITS OWN project; the passed ``project`` is the default/
    # fallback so single-project callers are byte-identical.
    eff_project = (structure.get("project") or project)

    synthesis_graph = str(structure.get("synthesis_graph", "") or "")
    keyed = bool(eff_project) and bool(synthesis_graph)
    # The spine path's keyed RUBRIC write (strict/grounded/schema/tags). When the
    # effective rubric is STRICT this is the enforcing run's context: if it never
    # lands, the scribe's full (project, synthesis_graph) key misses against a
    # populated ``contexts`` list -> ExtractionContextNotFound -> every claim of an
    # otherwise-correct run rejected. So a write failure for an enforcing keyed run
    # surfaces LOUD at prep time (operator retries) rather than silently mass-
    # rejecting later. Non-enforcing / flat / non-strict writes stay best-effort.
    # Coerce to bool to match what ``write_extraction_meta`` persists (``bool(strict)``).
    # A spine's embedded ``strict`` may be a truthy non-bool (e.g. ``1`` from YAML); the
    # identity check ``is True`` would skip enforcement here while the read path still
    # enforces, swallowing a keyed-write failure and reopening the per-source silent loss.
    enforcing = keyed and bool(eff_strict)
    quarantined: list[dict] = []
    succeeded = 0
    for p in prepped:
        src = p.get("name")
        if not src:
            continue
        try:
            if keyed:
                # Carry the run's rubric (strict/grounded/schema) into the keyed
                # entry too, since it is not seeded from the flat block. None
                # leaves a key untouched, so a caller passing only the structure
                # still just reconciles tags.
                write_extraction_meta(
                    src, project=eff_project, synthesis_graph=synthesis_graph,
                    schema=eff_schema, strict=eff_strict, grounded=eff_grounded,
                    tags=tags,
                )
            else:
                # Flat back-compat: only tags change; strict/grounded stay as prep
                # wrote them (omitted ⇒ untouched).
                write_extraction_meta(src, tags=tags)
            succeeded += 1
        except FileNotFoundError:
            # Source has no ``_meta.yaml`` (not a real prepped source). See the
            # matching note in ``_persist_structure_meta``: a structural condition
            # prep_sources rules out in production, not the transient write failure
            # (repro_a) the escalation targets -- stay best-effort.
            logger.warning(
                "could not reconcile extraction tags onto source '%s' "
                "(no _meta.yaml -- not a prepared source)", src
            )
        except Exception as exc:  # noqa: BLE001 — see enforcing/best-effort split
            if enforcing:
                # Per-source QUARANTINE (P2): record + loudly log this source and
                # KEEP GOING so the other healthy sources still extract. The caller
                # excludes the quarantined sources from the run. (Aborting loud only
                # if none survive -- see below.)
                logger.warning(
                    "quarantining source '%s': could not reconcile the keyed "
                    "extraction rubric for the ENFORCING context (project='%s', "
                    "synthesis_graph='%s'): %s. Excluding it from this run and "
                    "continuing with the remaining sources.",
                    src, eff_project, synthesis_graph, exc,
                )
                quarantined.append({"source": src, "reason": str(exc)})
                continue
            logger.warning(
                "could not reconcile extraction tags onto source '%s'", src
            )
    if enforcing and quarantined and succeeded == 0:
        # Every enforcing source failed -> nothing left to extract: abort LOUD
        # rather than emit an empty run. Raise the ``StructurePersistError``
        # SUBTYPE (a per-source persist failure) so callers can demote to flat
        # while genuine config/scaffold ``ExtractionError``s still propagate.
        raise StructurePersistError(
            f"could not reconcile the keyed extraction rubric for the ENFORCING "
            f"context (project='{eff_project}', synthesis_graph='{synthesis_graph}') "
            f"onto ANY source -- every source's write failed: {quarantined}. "
            f"Refusing to proceed with no sources; fix the sources' _meta.yaml and "
            f"re-create the extraction graph."
        )
    return quarantined

reconcile_structures

reconcile_structures(prepped: list[dict], structures: 'list[dict | None]', *, project: str, schema: str, strict: bool, grounded: bool) -> 'tuple[dict[str, list[dict]], list[dict], list[dict]]'

Reconcile each structure's keyed extraction-meta into a PER-SOURCE map.

The flat semantic graph is the always-present substrate; a spine is a SEPARATE additive overlay. So this NEVER aborts a run and NEVER drops a source: a source whose keyed rubric write FAILED for a structure simply does not have that structure available (it is DEMOTED to flat for that spine, and still extracts into the flat graph). A spine that no source could join is merely unused (still materialized -- cleanup deferred).

Runs :func:reconcile_meta_tags once per non-empty structure so every (project, synthesis_graph) context gets its OWN keyed strict-tag meta. For each structure, a source is AVAILABLE when its enforcing keyed write SUCCEEDED. reconcile_meta_tags still surfaces failures the two historical ways -- a per-source quarantined list (partial failure) and an :class:ExtractionError (EVERY source failed) -- but here BOTH are absorbed as demotions instead of exclusion/abort: a partial failure demotes just the failing source(s) for that structure, and a total failure demotes ALL sources for that structure (leaving it unused). Nothing propagates.

Returns (source_structures, demoted, unused):

• source_structures -- {source_name -> [available structure, ...]} for EVERY prepped source (preserving the input structure order; a source that joined no spine maps to [] and renders flat).
• demoted -- lightweight diagnostics: one {"source", "synthesis_graph", "reason"} per (source, structure) pair that fell back to flat because the keyed write failed.
• unused -- the structure dicts that NO source could join (zero members), so the caller can drop them from the rendered/surviving set and surface them as unused spines.

Source code in coordinator/extraction.py

def reconcile_structures(
    prepped: list[dict],
    structures: "list[dict | None]",
    *,
    project: str,
    schema: str,
    strict: bool,
    grounded: bool,
) -> "tuple[dict[str, list[dict]], list[dict], list[dict]]":
    """Reconcile each structure's keyed extraction-meta into a PER-SOURCE map.

    The flat semantic graph is the always-present substrate; a spine is a
    SEPARATE additive overlay. So this NEVER aborts a run and NEVER drops a
    source: a source whose keyed rubric write FAILED for a structure simply does
    not have that structure available (it is DEMOTED to flat for that spine, and
    still extracts into the flat graph). A spine that no source could join is
    merely unused (still materialized -- cleanup deferred).

    Runs :func:`reconcile_meta_tags` once per non-empty structure so every
    ``(project, synthesis_graph)`` context gets its OWN keyed strict-tag meta.
    For each structure, a source is AVAILABLE when its enforcing keyed write
    SUCCEEDED. ``reconcile_meta_tags`` still surfaces failures the two historical
    ways -- a per-source ``quarantined`` list (partial failure) and an
    :class:`ExtractionError` (EVERY source failed) -- but here BOTH are absorbed
    as demotions instead of exclusion/abort: a partial failure demotes just the
    failing source(s) for that structure, and a total failure demotes ALL sources
    for that structure (leaving it unused). Nothing propagates.

    Returns ``(source_structures, demoted, unused)``:

    * ``source_structures`` -- ``{source_name -> [available structure, ...]}`` for
      EVERY prepped source (preserving the input structure order; a source that
      joined no spine maps to ``[]`` and renders flat).
    * ``demoted`` -- lightweight diagnostics: one
      ``{"source", "synthesis_graph", "reason"}`` per (source, structure) pair
      that fell back to flat because the keyed write failed.
    * ``unused`` -- the structure dicts that NO source could join (zero members),
      so the caller can drop them from the rendered/surviving set and surface
      them as unused spines.
    """
    # Only real (dimension-bearing) spines are tracked: a structure with no
    # dimension_nodes renders flat regardless, so it never grants/denies
    # membership and must not skew the per-source availability map.
    present = [s for s in structures if s and s.get("dimension_nodes")]
    names = [p.get("name") for p in prepped if p.get("name")]
    source_structures: dict[str, list[dict]] = {n: [] for n in names}
    # ``materialize_structure`` (called by the resolver BEFORE this reconcile)
    # persists a per-source NON-ENFORCING (structure-only) keyed block onto each
    # source optimistically. So a spine that ends up ``unused`` below (no source
    # joined its ENFORCING rubric), or a single source DEMOTED from a surviving
    # spine, leaves an orphaned structure-only keyed entry behind. We prune those
    # at the end of this function (see ``_prune_orphaned_meta`` call) via the
    # ``demoted`` list -- the guarded ``prune_extraction_context`` only removes a
    # structure-only orphan and never an enforcing enrollment, so it is safe
    # across runs. NOT cleaned: the apex->hub rollup EDGES ``link_spine_hubs``
    # wrote -- edges carry no enforcing marker, so a failed-this-run edge is
    # indistinguishable from one a PRIOR run legitimately created over the same
    # source; removing them would risk stranding a real rollup, so they are left.
    demoted: list[dict] = []
    unused: list[dict] = []
    # Each structure may carry its OWN project (multi-project run). The orphan-meta
    # prune below must target the right ``(project, synthesis_graph)`` key, so we
    # record one ``(source, project, synthesis_graph)`` target per demotion using
    # THIS structure's project -- NOT a ``synthesis_graph -> project`` map, which
    # would last-write-wins-collapse two projects that legitimately share a graph
    # name and then prune one project's meta under the other. Falls back to the
    # passed ``project`` for single-project structures (byte-identical).
    prune_targets: list[tuple[str, str, str]] = []
    for s in present:
        sg = str(s.get("synthesis_graph") or "")
        s_proj = (s.get("project") or project)
        try:
            q = reconcile_meta_tags(
                prepped, s, project=project, schema=schema,
                strict=strict, grounded=grounded,
            ) or []
        except StructurePersistError as exc:
            # NARROW (symmetric with the named path): only a per-source PERSIST/
            # rubric WRITE failure (every enforcing source's keyed write failed)
            # demotes a spine to flat. ``reconcile_meta_tags`` raises this subtype
            # for exactly that case. A genuine config/scaffold ``ExtractionError``
            # (a mis-configured run, NOT a transient persist failure) is NOT caught
            # here -- it PROPAGATES, just like the named-schema path, instead of
            # being silently masked as a per-source demotion.
            # INVARIANT: ``reconcile_meta_tags`` is the SOLE raiser inside this try,
            # and it raises ONLY the ``StructurePersistError`` subtype for the
            # transient/per-source persist failure this clause demotes. A bare
            # ``ExtractionError`` raised inside the try (a non-persist config error)
            # would fall through this narrow ``except`` and PROPAGATE -- aborting
            # the run rather than being mis-demoted to flat. Keep that split intact:
            # do not broaden this to ``except ExtractionError``.
            # This structure's enforcing keyed write failed for EVERY source ->
            # demote ALL sources to flat for THIS spine; the spine is unused. The
            # run continues (flat for this spine), nothing raises.
            logger.warning(
                "extraction: STRUCTURE (synthesis_graph='%s') could not persist "
                "its enforcing keyed extraction-meta for ANY source (%s); demoting "
                "every source to FLAT for this spine (it joins no member).",
                sg, exc,
            )
            for n in names:
                demoted.append(
                    {"source": n, "synthesis_graph": sg, "reason": str(exc)}
                )
                prune_targets.append((n, s_proj, sg))
            unused.append(s)
            continue
        failed = {entry.get("source"): entry.get("reason", "") for entry in q}
        members = 0
        for n in names:
            if n in failed:
                # Per-source demote: this source's keyed write failed for this
                # spine. It still extracts flat; it just does not join this spine.
                demoted.append(
                    {"source": n, "synthesis_graph": sg, "reason": failed[n]}
                )
                prune_targets.append((n, s_proj, sg))
            else:
                source_structures[n].append(s)
                members += 1
        if members == 0:
            # No source joined (every source was in the partial-failure list, yet
            # reconcile did not raise) -> the spine is unused.
            unused.append(s)
    # Best-effort cleanup of the orphaned structure-only keyed meta that
    # materialize_structure wrote for every (source, spine) that fell back to
    # flat -- ``demoted`` covers BOTH a fully-unused spine (all sources demoted)
    # and a single source demoted from a surviving spine. ``prune_extraction_context``
    # self-guards (it only removes a non-enforcing orphan), so it never strands a
    # real enrollment and a failure here never aborts the run.
    if project and prune_targets:
        from zettelkasten.graph import prune_extraction_context
        for src, prune_project, sg in prune_targets:
            if not sg or not src:
                continue
            # Prune under the spine's OWN project (multi-project), defaulting to
            # the run-global ``project`` for a single-project structure.
            try:
                prune_extraction_context(
                    src, project=prune_project, synthesis_graph=sg,
                )
            except Exception as exc:  # noqa: BLE001 — cleanup is best-effort
                logger.warning(
                    "extraction: could not prune orphaned keyed meta for source "
                    "'%s' (synthesis_graph='%s'): %s", src, sg, exc,
                )
    return source_structures, demoted, unused

finalize_source_structures

finalize_source_structures(structures: 'list[dict | None]', source_structures: 'dict[str, list[dict]]') -> 'tuple[list[dict], dict[str, list[dict]] | None]'

Collapse a per-source availability map to (surviving, per_source).

surviving is the structures with >=1 member source, in the input order (used for the run JSON / prep record + the back-compat singular structure + the shared structures= arg). per_source is the map to thread into :func:build_extraction_tasks ONLY when sources have HETEROGENEOUS available sets (some source was demoted from a SURVIVING spine); when every source sees the SAME full surviving set -- the common no-demotion case, and also the case where the only demotions were from spines that NO source joined -- it returns None so the caller renders the shared (byte-identical) path.

Source code in coordinator/extraction.py

def finalize_source_structures(
    structures: "list[dict | None]",
    source_structures: "dict[str, list[dict]]",
) -> "tuple[list[dict], dict[str, list[dict]] | None]":
    """Collapse a per-source availability map to ``(surviving, per_source)``.

    ``surviving`` is the structures with >=1 member source, in the input order
    (used for the run JSON / prep record + the back-compat singular ``structure``
    + the shared ``structures=`` arg). ``per_source`` is the map to thread into
    :func:`build_extraction_tasks` ONLY when sources have HETEROGENEOUS available
    sets (some source was demoted from a SURVIVING spine); when every source sees
    the SAME full surviving set -- the common no-demotion case, and also the case
    where the only demotions were from spines that NO source joined -- it returns
    ``None`` so the caller renders the shared (byte-identical) path.
    """
    member_ids: set[int] = set()
    for lst in source_structures.values():
        for s in lst:
            member_ids.add(id(s))
    surviving = [
        s for s in structures
        if s and s.get("dimension_nodes") and id(s) in member_ids
    ]
    homogeneous = all(lst == surviving for lst in source_structures.values())
    return surviving, (None if homogeneous else source_structures)

projects_touching

projects_touching(prepped: list[dict], launch_project: str) -> list[str]

Every project any prepped source belongs to (launch project first).

The synthesizer's connectivity work is per-project: a source registered into projects B and C (not just the launch project A) should have its claims placed on B's and C's spines and synthesized into their _cross hubs too. This reverse-lookups source -> projects by scanning the project manifests (there is no dedicated index): a project is "touched" when its sources list contains any of this run's source graph names. The launch_project is always included and placed first (it is the run's primary even if its manifest write lags).

Best-effort: a manifest that cannot be listed just means fewer touched projects, never an error -- returns at least [launch_project].

Source code in coordinator/extraction.py

def projects_touching(prepped: list[dict], launch_project: str) -> list[str]:
    """Every project any prepped source belongs to (launch project first).

    The synthesizer's connectivity work is per-project: a source registered into
    projects B and C (not just the launch project A) should have its claims placed
    on B's and C's spines and synthesized into their ``_cross`` hubs too. This
    reverse-lookups source -> projects by scanning the project manifests (there is
    no dedicated index): a project is "touched" when its ``sources`` list contains
    any of this run's source graph names. The ``launch_project`` is always included
    and placed first (it is the run's primary even if its manifest write lags).

    Best-effort: a manifest that cannot be listed just means fewer touched
    projects, never an error -- returns at least ``[launch_project]``.
    """
    names = {p.get("name") for p in prepped if p.get("name")}
    ordered: list[str] = [launch_project] if launch_project else []
    seen = set(ordered)
    try:
        from zettelkasten.graph import list_projects

        for pdata in list_projects():
            pname = pdata.get("name")
            if not pname or pname in seen:
                continue
            srcs = set(pdata.get("sources") or [])
            if srcs & names:
                ordered.append(pname)
                seen.add(pname)
    except Exception as exc:  # noqa: BLE001 — reverse lookup is best-effort
        logger.warning(
            "synthesis: could not enumerate projects touching the run's sources "
            "(%s); scoping synthesis to the launch project only", exc,
        )
    return ordered

build_synthesis_context

build_synthesis_context(prepped: list[dict], launch_project: str, *, get_graph: Any, present_fn: 'Callable[[str, str], bool] | None' = None, return_structures: bool = False) -> 'str | tuple[str, list[dict]]'

Materialize every spine layout of every touched project + render the block.

Two jobs, both feeding the trailing synthesizer task:

1. MATERIALIZE. Reuse the multi-project resolver with multi=True so EVERY promoted spine of EVERY project the sources touch gets its structure meta persisted under its own (project, synthesis_graph) key (and its apex -> hub rollup edges written). This is the prerequisite that lets the synthesizer's note(action="attach") resolve a dimension for a spine the per-source scribe never chose. It is idempotent with the main run's own materialization (identical edges/meta dedup), so re-doing the launch project's chosen spine here is harmless.
2. RENDER. Emit one SPINES block per (project, spine) giving the project, the synthesis graph, and the dimension MAP -- each dimension's tag, node id, and a human description read from the dimension node itself -- so the synthesizer can judge, semantically, which dimension each claim falls under.

Returns the rendered SPINES: ... context string (empty "" when no project in scope has a promoted spine, which tells the caller to skip the synthesizer's placement step). Fully best-effort: any failure returns "" rather than aborting graph creation -- synthesis is additive, never load-bearing.

return_structures (default False) is an additive opt-in for the STANDALONE synthesis-prep path (prep_synthesis_contexts): when True the function returns (context_str, structures) where structures is the list of resolved+materialized spine structure dicts (each carrying project and synthesis_graph), so a caller can report exactly which (project, synthesis_graph) contexts it just registered. The default keeps the historical bare-string return byte-identical for create_extraction_graph.

Source code in coordinator/extraction.py

def build_synthesis_context(
    prepped: list[dict],
    launch_project: str,
    *,
    get_graph: Any,
    present_fn: "Callable[[str, str], bool] | None" = None,
    return_structures: bool = False,
) -> "str | tuple[str, list[dict]]":
    """Materialize every spine layout of every touched project + render the block.

    Two jobs, both feeding the trailing ``synthesizer`` task:

    1. MATERIALIZE. Reuse the multi-project resolver with ``multi=True`` so EVERY
       promoted spine of EVERY project the sources touch gets its structure meta
       persisted under its own ``(project, synthesis_graph)`` key (and its apex ->
       hub rollup edges written). This is the prerequisite that lets the
       synthesizer's ``note(action="attach")`` resolve a dimension for a spine the
       per-source scribe never chose. It is idempotent with the main run's own
       materialization (identical edges/meta dedup), so re-doing the launch
       project's chosen spine here is harmless.
    2. RENDER. Emit one ``SPINES`` block per (project, spine) giving the project,
       the synthesis graph, and the dimension MAP -- each dimension's ``tag``, node
       id, and a human description read from the dimension node itself -- so the
       synthesizer can judge, semantically, which dimension each claim falls under.

    Returns the rendered ``SPINES: ...`` context string (empty ``""`` when no
    project in scope has a promoted spine, which tells the caller to skip the
    synthesizer's placement step). Fully best-effort: any failure returns ``""``
    rather than aborting graph creation -- synthesis is additive, never load-bearing.

    ``return_structures`` (default ``False``) is an additive opt-in for the
    STANDALONE synthesis-prep path (``prep_synthesis_contexts``): when ``True`` the
    function returns ``(context_str, structures)`` where ``structures`` is the list
    of resolved+materialized spine structure dicts (each carrying ``project`` and
    ``synthesis_graph``), so a caller can report exactly which
    ``(project, synthesis_graph)`` contexts it just registered. The default keeps
    the historical bare-string return byte-identical for ``create_extraction_graph``.
    """
    # Defined before the try so the early-exit returns can reference it under
    # ``return_structures`` (empty list == nothing materialized).
    structures: list[dict] = []

    def _ret(context_str: str) -> "str | tuple[str, list[dict]]":
        return (context_str, structures) if return_structures else context_str

    try:
        from zettelkasten import organizations as zk_orgs

        projects = projects_touching(prepped, launch_project)
        if not projects:
            return _ret("")
        # A project's spine LAYOUTS are its promoted orgs (state=="spine" with a
        # spine_ref) -- NOT just its ``default_spine`` (which is often unset: the
        # intrinsic lens). ``resolve_extraction_structure`` only scopes to the
        # spine ids handed to it plus the default, so to cover EVERY layout we must
        # enumerate each project's promoted orgs and pass their ids in explicitly.
        # We resolve per project (each id is valid only in its own org tree) with
        # ``multi=True`` so every enumerated spine is filled + materialized, and
        # stamp ``structure["project"]`` so downstream keying/render is per project.
        # ``structures`` is declared in the enclosing scope (above the try) so the
        # early-exit returns can surface it under ``return_structures``.
        # Per-project nested-spine (tier) composition, so each rendered SPINES
        # block can be annotated with its parent/child spines. READ-ONLY: derived
        # from existing component-of edges, never written (see _nested_tier_map).
        tier_by_project: dict[str, dict] = {}
        for proj in projects:
            try:
                orgs = zk_orgs.list_organizations("project", proj)
            except Exception:  # noqa: BLE001 — a project with no orgs is fine
                orgs = []
            spine_ids = [
                str(o.get("id") or "")
                for o in orgs
                if isinstance(o, dict)
                and str(o.get("state") or "") == "spine"
                and str(o.get("spine_ref") or "").strip()
                and o.get("id")
            ]
            if not spine_ids:
                continue
            tier = _nested_tier_map(orgs, get_graph=get_graph)
            if tier:
                tier_by_project[proj] = tier
            try:
                res = resolve_extraction_structure(
                    proj, prepped, spine_ids, get_graph=get_graph,
                    present_fn=present_fn, multi=True,
                )
            except Exception as exc:  # noqa: BLE001 — isolate a bad project
                logger.warning(
                    "synthesis: could not resolve spines for project '%s' (%s); "
                    "skipping its layouts", proj, exc,
                )
                continue
            for s in (res.get("structures") or []):
                if isinstance(s, dict) and s.get("dimension_nodes"):
                    s["project"] = proj
                    structures.append(s)
    except Exception as exc:  # noqa: BLE001 — synthesis prep is never load-bearing
        logger.warning(
            "synthesis: could not resolve/materialize touched-project spines "
            "(%s); the synthesizer will run synthesis-only (no cross-spine "
            "placement)", exc,
        )
        return _ret("")

    if not structures:
        return _ret("")

    blocks: list[str] = []
    for s in structures:
        proj = str(s.get("project") or launch_project)
        sg = str(s.get("synthesis_graph") or "")
        dim_nodes = s.get("dimension_nodes") or {}
        if not sg or not dim_nodes:
            continue
        # Read each dimension node for a human description (title + first body
        # line) so the synthesizer places by MEANING, not tag-string match.
        try:
            zg = get_graph(sg)
            notes = getattr(zg, "notes", {}) or {}
        except Exception:  # noqa: BLE001 — a missing spine graph just omits descs
            notes = {}
        dim_lines: list[str] = []
        for tag, nid in dim_nodes.items():
            note = notes.get(nid)
            desc = ""
            if note is not None:
                title = (getattr(note, "title", "") or "").strip()
                body = (getattr(note, "body", "") or "").strip().splitlines()
                first = body[0].strip() if body else ""
                desc = f"{title} — {first}" if first else title
            dim_lines.append(
                f"    - {tag}  (node {nid}): {desc}" if desc
                else f"    - {tag}  (node {nid})"
            )
        # NESTED annotation: when this spine sits in a tier hierarchy, tell the
        # synthesizer its parent/child spines so it places at the deepest fitting
        # tier (deepest-owner) rather than redundantly on an ancestor.
        nested_lines = _render_nested_annotation(
            (tier_by_project.get(proj) or {}).get(sg)
        )
        block_body = "\n".join(dim_lines + nested_lines)
        blocks.append(f"  PROJECT {proj} / SPINE {sg}:\n" + block_body)

    if not blocks:
        return _ret("")
    return _ret(
        "SPINES (each block = one project x one spine layout the run's sources "
        "touch; attach claims that fit, passing the block's project AND synthesis "
        "graph):\n" + "\n".join(blocks)
    )

plan_slices

plan_slices(prep: dict, *, window: int = SLICE_WINDOW_DEFAULT, overlap: int = SLICE_OVERLAP_DEFAULT, threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT, page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT, explicit: 'list[dict] | None' = None) -> list[dict]

Plan the SLICE fan-out for one prepared source.

Returns a list of slice dicts, each {index, total, kind, label, char_start, char_end, page_start, page_end} where kind is full | chars | pages | chapter and a None char_end means "to the end".

A source at or below the size threshold (chars <= threshold and pages <= page_threshold) yields a SINGLE full slice -- byte-identical to the historical single-trio run. Above it: caller explicit slices verbatim, else chapter-grouped windows when the source carries a cached outline, else plain char windows.

Source code in coordinator/extraction.py

def plan_slices(
    prep: dict,
    *,
    window: int = SLICE_WINDOW_DEFAULT,
    overlap: int = SLICE_OVERLAP_DEFAULT,
    threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT,
    page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT,
    explicit: "list[dict] | None" = None,
) -> list[dict]:
    """Plan the SLICE fan-out for one prepared source.

    Returns a list of slice dicts, each ``{index, total, kind, label, char_start,
    char_end, page_start, page_end}`` where ``kind`` is ``full`` | ``chars`` |
    ``pages`` | ``chapter`` and a ``None`` ``char_end`` means "to the end".

    A source at or below the size threshold (chars ``<= threshold`` and pages
    ``<= page_threshold``) yields a SINGLE ``full`` slice -- byte-identical to the
    historical single-trio run. Above it: caller ``explicit`` slices verbatim,
    else chapter-grouped windows when the source carries a cached outline, else
    plain char windows.
    """
    total = int(prep.get("total_chars") or 0)
    num_pages = int(prep.get("num_pages") or 0)
    outline = prep.get("outline") or []

    if explicit:
        spans = [_normalize_explicit_slice(s, outline, total) for s in explicit]
        finalized = _finalize_slices([s for s in spans if s], kind_default="chars")
        if finalized:
            return finalized
        # An unusable explicit list degrades to a single full slice rather than
        # emitting zero trios (which would silently drop the source).

    small = total <= 0 or (
        total <= threshold and (num_pages <= 0 or num_pages <= page_threshold)
    )
    if small:
        return [{
            "index": 0, "total": 1, "kind": "full", "label": "full",
            "char_start": 0, "char_end": None,
            "page_start": None, "page_end": None,
        }]

    spans = None
    if outline:
        spans = _chapter_windows(outline, total, window, overlap)
    if not spans:
        spans = [
            {"char_start": s, "char_end": e}
            for s, e in _char_windows(0, total, window, overlap)
        ]
        return _finalize_slices(spans, kind_default="chars")
    return _finalize_slices(spans, kind_default="chapter")

slice_read_call

slice_read_call(sl: dict) -> str

Render the source(action="fulltext", ...) call a slice's extractor uses.

Source code in coordinator/extraction.py

def slice_read_call(sl: dict) -> str:
    """Render the ``source(action="fulltext", ...)`` call a slice's extractor uses."""
    kind = sl.get("kind")
    if kind == "full":
        return 'source(action="fulltext")'
    if kind == "pages" and (sl.get("page_start") is not None or sl.get("page_end") is not None):
        ps = sl.get("page_start")
        pe = sl.get("page_end")
        parts = []
        if ps is not None:
            parts.append(f"page_start={ps}")
        if pe is not None:
            parts.append(f"page_end={pe}")
        return f'source(action="fulltext", {", ".join(parts)})'
    start = int(sl.get("char_start") or 0)
    end = sl.get("char_end")
    if end is not None:
        span = max(1, int(end) - start)
        return f'source(action="fulltext", offset={start}, max_chars={span})'
    return f'source(action="fulltext", offset={start})'

slice_label

slice_label(sl: dict) -> str

Human-readable SLICE line for a slice's task context.

Source code in coordinator/extraction.py

def slice_label(sl: dict) -> str:
    """Human-readable SLICE line for a slice's task context."""
    if sl.get("kind") == "full":
        return "full"
    idx = int(sl.get("index", 0)) + 1
    total = int(sl.get("total", 1))
    label = sl.get("label") or ""
    loc = ""
    if sl.get("kind") == "pages":
        loc = f"pages {sl.get('page_start')}-{sl.get('page_end')}"
    elif sl.get("char_end") is not None:
        loc = f"chars [{sl.get('char_start')}:{sl.get('char_end')}]"
    else:
        loc = f"chars [{sl.get('char_start')}:end]"
    head = f"{label} ({loc})" if label else loc
    return f"{head} — part {idx} of {total}"

build_extraction_tasks

build_extraction_tasks(prepped: list[dict], schema_name: str, project: str, batch_size: int = 0, include_memory: bool = True, structure: dict | None = None, structures: 'list[dict | None] | None' = None, source_structures: 'dict[str, list[dict | None]] | None' = None, synthesis_context: str = '', include_synthesis: bool = False, connection_context: str = '', slice_window: int = SLICE_WINDOW_DEFAULT, slice_overlap: int = SLICE_OVERLAP_DEFAULT, slice_threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT, slice_page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT, reextract: bool = False, slice_report: 'dict[str, dict] | None' = None) -> list[dict]

Emit the per-source extractor -> scribe -> auditor DAG (+ memory task).

BOOK-SCALE FAN-OUT: a source is sliced by :func:plan_slices (char windows, chapter-snapped when the source carries a cached PDF outline, or a caller's explicit slices list). A source at/below the size threshold yields ONE full slice and its trio is BYTE-IDENTICAL to the historical single-trio run. A large source emits one extractor -> scribe pair PER slice (all sharing the per-source hub + structure), then ONE auditor depending on ALL that source's scribes -- so the coverage matrix stays per-source and every slice is committed before the audit.

INCREMENTAL SKIP: planned slices already covered by the source's completed-slice ledger (stamped by a prior run's auditor) are dropped, so a re-run only mines the not-yet-extracted regions. A source whose EVERY slice is already done emits NO trio and is reported as skipped in slice_report. reextract=True bypasses the filter (re-mines everything). When slice_report is a dict it is populated with {source_name: {"done": [labels], "todo": [labels]}} so the caller can surface what was skipped vs planned; it never affects the emitted tasks.

Each scribe/data-scribe declares writes: [".zettelkasten/<src>"] — a write-lease scoped to its SOURCE BOX. Scribes on DIFFERENT sources get disjoint lease keys and run in parallel; scribes on the SAME box (e.g. two per-schema scribes over one source in a union run) serialize at claim time, since their search-then-add is not atomic across calls and would otherwise duplicate notes. The trailing synthesizer keeps writes: [] (it legitimately touches many boxes). The extractor (planner) and auditor (checker) take no write lease.

When include_synthesis is set (the create_extraction_graph tool opts in via its synthesize flag; other callers such as the stream daemon keep the default False so their graph shape is unchanged), a single trailing synthesizer task is appended after ALL auditors: it places the run's claims onto every touched project's other spine layouts (using synthesis_context) and wires cross-source themes into _cross hubs. The memory task then trails the synthesizer so the run is recorded last.

Each task carries schema=schema_name so the coordinator flows the expanded rubric into its context. Extractors are independent (one wave) unless batch_size > 0, in which case each batch's extractors chain to the previous batch's extractors (planner -> planner, which the validator allows) so the orchestrator fans out one batch at a time.

The flat semantic graph is the always-present substrate; a spine is an additive overlay. Rendering is therefore PER SOURCE, keyed on the structures AVAILABLE to each source:

• source_structures (preferred) maps source_name -> [structure, ...] -- the structures whose keyed extraction-meta write SUCCEEDED for that source. A source whose write failed for a given spine simply does NOT have it in the list (demoted to flat for that spine), and is NEVER dropped from the run. Sources absent from the map (or mapped to []) render FLAT.
• When source_structures is None (callers not in spine mode, or a clean run with NO per-source demotions) the run falls back to the shared structures / structure list applied uniformly to every source -- byte-identical to the historical behavior.

structures carries the MULTI-SPINE list (one entry per in-scope spine); structure (singular) is the back-compat scalar (defaults to [structure] when structures is not given). With ZERO or ONE valid structure (per source) the rendered STRUCTURE block + scribe/auditor instructions are BYTE-IDENTICAL to the historical single-/no-structure behavior; with MORE than one the task context carries numbered STRUCTURE blocks and the scribe attaches each claim to EVERY spine whose dimension_nodes contains the claim's tag.

Source code in coordinator/extraction.py

def build_extraction_tasks(
    prepped: list[dict],
    schema_name: str,
    project: str,
    batch_size: int = 0,
    include_memory: bool = True,
    structure: dict | None = None,
    structures: "list[dict | None] | None" = None,
    source_structures: "dict[str, list[dict | None]] | None" = None,
    synthesis_context: str = "",
    include_synthesis: bool = False,
    connection_context: str = "",
    slice_window: int = SLICE_WINDOW_DEFAULT,
    slice_overlap: int = SLICE_OVERLAP_DEFAULT,
    slice_threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT,
    slice_page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT,
    reextract: bool = False,
    slice_report: "dict[str, dict] | None" = None,
) -> list[dict]:
    """Emit the per-source extractor -> scribe -> auditor DAG (+ memory task).

    BOOK-SCALE FAN-OUT: a source is sliced by :func:`plan_slices` (char windows,
    chapter-snapped when the source carries a cached PDF outline, or a caller's
    explicit ``slices`` list). A source at/below the size threshold yields ONE
    ``full`` slice and its trio is BYTE-IDENTICAL to the historical single-trio
    run. A large source emits one ``extractor -> scribe`` pair PER slice (all
    sharing the per-source hub + structure), then ONE ``auditor`` depending on
    ALL that source's scribes -- so the coverage matrix stays per-source and every
    slice is committed before the audit.

    INCREMENTAL SKIP: planned slices already covered by the source's
    completed-slice ledger (stamped by a prior run's auditor) are dropped, so a
    re-run only mines the not-yet-extracted regions. A source whose EVERY slice is
    already done emits NO trio and is reported as skipped in ``slice_report``.
    ``reextract=True`` bypasses the filter (re-mines everything). When
    ``slice_report`` is a dict it is populated with ``{source_name: {"done":
    [labels], "todo": [labels]}}`` so the caller can surface what was skipped vs
    planned; it never affects the emitted tasks.

    Each scribe/data-scribe declares ``writes: [".zettelkasten/<src>"]`` — a
    write-lease scoped to its SOURCE BOX. Scribes on DIFFERENT sources get
    disjoint lease keys and run in parallel; scribes on the SAME box (e.g. two
    per-schema scribes over one source in a union run) serialize at claim time,
    since their search-then-add is not atomic across calls and would otherwise
    duplicate notes. The trailing synthesizer keeps ``writes: []`` (it
    legitimately touches many boxes). The extractor (planner) and auditor
    (checker) take no write lease.

    When ``include_synthesis`` is set (the ``create_extraction_graph`` tool opts
    in via its ``synthesize`` flag; other callers such as the stream daemon keep
    the default ``False`` so their graph shape is unchanged), a single trailing
    ``synthesizer`` task is appended after ALL auditors: it places the run's
    claims onto every touched project's other spine layouts (using
    ``synthesis_context``) and wires cross-source themes into ``_cross`` hubs. The
    ``memory`` task then trails the synthesizer so the run is recorded last.

    Each task carries ``schema=schema_name`` so the coordinator flows the
    expanded rubric into its context. Extractors are independent (one wave)
    unless ``batch_size`` > 0, in which case each batch's extractors chain to the
    previous batch's extractors (planner -> planner, which the validator allows)
    so the orchestrator fans out one batch at a time.

    The flat semantic graph is the always-present substrate; a spine is an
    additive overlay. Rendering is therefore PER SOURCE, keyed on the structures
    AVAILABLE to each source:

    * ``source_structures`` (preferred) maps ``source_name -> [structure, ...]``
      -- the structures whose keyed extraction-meta write SUCCEEDED for that
      source. A source whose write failed for a given spine simply does NOT have
      it in the list (demoted to flat for that spine), and is NEVER dropped from
      the run. Sources absent from the map (or mapped to ``[]``) render FLAT.
    * When ``source_structures`` is ``None`` (callers not in spine mode, or a
      clean run with NO per-source demotions) the run falls back to the shared
      ``structures`` / ``structure`` list applied uniformly to every source --
      byte-identical to the historical behavior.

    ``structures`` carries the MULTI-SPINE list (one entry per in-scope spine);
    ``structure`` (singular) is the back-compat scalar (defaults to
    ``[structure]`` when ``structures`` is not given). With ZERO or ONE valid
    structure (per source) the rendered STRUCTURE block + scribe/auditor
    instructions are BYTE-IDENTICAL to the historical single-/no-structure
    behavior; with MORE than one the task context carries numbered STRUCTURE
    blocks and the scribe attaches each claim to EVERY spine whose
    dimension_nodes contains the claim's tag.
    """
    if not prepped:
        raise ExtractionError("No prepared sources to build a graph from.")

    if structures is None:
        structures = [structure] if structure is not None else []
    # The shared (uniform) structure list used when no per-source map is given.
    shared_structures = list(structures)

    # Resolve the NAMED schema's FULL dimension tag set ONCE, via the existing
    # capability resolver -- the exact tags the EXTRACTOR tags claims with (the
    # schema's strict flat vocabulary = expand_schema's ``"tags"`` = ALL declared
    # dimensions, materialized or not). This is the token space the single-spine
    # render gates FULL vs PARTIAL on (a spine is FULL only when its
    # dimension_nodes is a SUPERSET of these tags). It must span ALL named tags,
    # NOT just the materialized subset: a ``materialize: false`` dimension is
    # still a valid extracted claim tag but has NO spine node, so a spine
    # covering only the materialized dims must read PARTIAL (conditional) -- else
    # the ``materialize: false`` claim is keyed-routed and then rejected by the
    # spine's materialized-only keyed rubric and silently LOST. ``None`` when the
    # name is unresolvable -> the render defaults to the SAFE conditional fallback.
    named_dims: "frozenset[str] | None" = None
    try:
        from coordinator.contrib import expand_schema

        expanded = expand_schema(schema_name)
        if isinstance(expanded, dict):
            named_dims = frozenset(
                t.strip()
                for t in (expanded.get("tags") or [])
                if isinstance(t, str) and t.strip()
            )
    except Exception:  # noqa: BLE001 — an unresolvable schema name just means
        # "no named vocab" -> the single-spine render uses the safe conditional
        # fallback rather than crashing the whole graph build.
        named_dims = None

    tasks: list[dict] = []
    auditor_aliases: list[str] = []
    # First-extractor alias per source, so batch chaining can gate a source's
    # extractors on the previous batch's anchor regardless of how each was sliced.
    first_ext_by_source: list[str] = []

    for i, p in enumerate(prepped):
        src = p["name"]
        hub = p.get("hub_id", "")

        # Per-source available structures: the source-centric map when supplied,
        # else the shared list (byte-identical to the old uniform render).
        if source_structures is None:
            src_structures = shared_structures
        else:
            src_structures = source_structures.get(src, [])
        structure_block, scribe_instr, coverage_call = _render_source_blocks(
            src, project, src_structures, named_dims=named_dims
        )

        # DATA source: build the data trio (data-extractor -> data-scribe ->
        # auditor) instead of the prose trio, and SKIP slicing entirely (a data
        # source references an existing dataset box, not a document). The rendered
        # ``structure_block`` / ``coverage_call`` are agent-agnostic, so a data
        # source sharing a synthesis_label threads the SAME spine dimension nodes as
        # the prose sources -- the synthesizer/spine wiring treats its claims like
        # prose claims. Prose sources (no ``kind``) never enter this branch, so the
        # prose task list/aliases/structure stay byte-identical.
        if p.get("kind") == "data":
            dext, dscr = f"dext_{i}", f"dscr_{i}"
            first_ext_by_source.append(dext)
            # Gate on the anchor extractor of the source ``batch_size`` back; a
            # fully-skipped predecessor has an EMPTY anchor, so skip it (a broken
            # "" dependency would otherwise dangle -- see the prose branch below).
            batch_dep = []
            if batch_size and batch_size > 0 and i >= batch_size:
                prev_anchor = first_ext_by_source[i - batch_size]
                if prev_anchor:
                    batch_dep = [prev_anchor]

            derivations = p.get("derivations") or []
            deriv_line = (
                f"DERIVATIONS: {', '.join(str(d) for d in derivations)}\n"
                if derivations else ""
            )
            source_dir = (
                f"SOURCE: {src}  (zettelkasten dataset box; read its type=dataset "
                "notes via dataset(action=\"values\"))\n"
            )
            ctx = (
                f"PROJECT: {project}\n"
                f"{source_dir}"
                f"HUB: {hub or '(none)'}  (graph: {src})\n"
                f"{deriv_line}"
                f"SCHEMA: {schema_name}  (the full dimension rubric is provided in "
                "this task's context object; attend to the evidence:data "
                "dimensions)\n"
                f"{structure_block}"
            )
            tasks.append({
                "alias": dext,
                "agent": "data-extractor",
                "schema": schema_name,
                "depends_on": list(batch_dep),
                "description": (
                    f"Read the dataset source '{src}' and emit grounded "
                    f"data-citation candidates for the '{schema_name}' schema's "
                    f"evidence:data dimensions.\n{ctx}"
                ),
            })
            tasks.append({
                "alias": dscr,
                "agent": "data-scribe",
                "schema": schema_name,
                # Write-lease scoped to THIS source box (``.zettelkasten/<src>``):
                # data-scribes on DIFFERENT sources get disjoint keys and run in
                # parallel, while two scribes on the SAME box serialize (their
                # search-then-add is not atomic across calls -> would otherwise
                # duplicate notes). Mirrors the prose scribe.
                "writes": [f".zettelkasten/{src}"],
                "depends_on": [dext],
                "description": (
                    f"Commit the data-extractor's candidates for '{src}' as claim "
                    f"notes with method:data grounding, attaching each to its "
                    f"schema dimension node exactly as the prose scribe does.\n{ctx}"
                ),
            })
            aud = f"aud_{i}"
            tasks.append({
                "alias": aud,
                "agent": "auditor",
                "schema": schema_name,
                "depends_on": [dscr],
                "description": (
                    f"Audit '{src}': grounding integrity (every data claim's "
                    f"citation re-computes to its asserted value within tolerance), "
                    f"hub/spine linkage, and coverage -- read the deterministic "
                    f"matrix via {coverage_call} and interpret it.\n"
                    f"PROJECT: {project}\n"
                    f"{source_dir}"
                    f"HUB: {hub or '(none)'}  (graph: {src})\n"
                    f"SCHEMA: {schema_name}  (the full dimension rubric is provided "
                    "in this task's context object)\n"
                    f"{structure_block}"
                ),
            })
            auditor_aliases.append(aud)
            continue

        slices = plan_slices(
            p,
            window=slice_window,
            overlap=slice_overlap,
            threshold=slice_threshold,
            page_threshold=slice_page_threshold,
            explicit=p.get("slices"),
        )
        # INCREMENTAL SKIP: drop planned slices already covered by this source's
        # completed-slice ledger (keyed on the same context the auditor stamps).
        # ``reextract`` bypasses the filter. ``done``/``todo`` are surfaced via
        # ``slice_report`` so the caller can report skipped vs planned.
        todo_slices, done_slices = _partition_slices(
            slices, p, project, src_structures, schema=schema_name,
            reextract=reextract,
        )
        if slice_report is not None:
            slice_report[src] = {
                "done": [slice_label(s) for s in done_slices],
                "todo": [slice_label(s) for s in todo_slices],
            }
        if not todo_slices:
            # Every slice already extracted -> emit NO trio (reported as skipped).
            # Keep ``first_ext_by_source`` index-aligned with ``prepped`` so batch
            # chaining still resolves the right anchor for later sources.
            first_ext_by_source.append("")
            continue

        single_full = len(todo_slices) == 1 and todo_slices[0].get("kind") == "full"
        # This source's anchor (first) extractor alias, recorded for batch chaining.
        anchor = f"ext_{i}" if single_full else f"ext_{i}_{int(todo_slices[0]['index'])}"
        first_ext_by_source.append(anchor)

        # Batch chaining is per-source: gate this source's extractors on the
        # anchor extractor of the source `batch_size` back (planner -> planner).
        # A skipped predecessor has an empty anchor; fall back to no gate.
        batch_dep: list[str] = []
        if batch_size and batch_size > 0 and i >= batch_size:
            prev_anchor = first_ext_by_source[i - batch_size]
            if prev_anchor:
                batch_dep = [prev_anchor]

        source_dir = (
            f"SOURCE: {src}  (zettelkasten graph with ingested fulltext; "
            "read via source.fulltext)\n"
        )
        # Auditor runs at the SOURCE level (audits the whole source, reads the
        # per-source coverage matrix), so its ctx is always the full-source view.
        # It carries the SLICES this run extracted (+ content_hash) so the persona
        # stamps EXACTLY them via mark_extracted on a PASS.
        aud_ctx = (
            f"PROJECT: {project}\n"
            f"{source_dir}"
            f"HUB: {hub or '(none)'}  (graph: {src})\n"
            "SLICE: full\n"
            f"SCHEMA: {schema_name}  (the full dimension rubric is provided in "
            "this task's context object)\n"
            f"{_render_slices_block(todo_slices, str(p.get('content_hash') or ''))}"
            f"{structure_block}"
        )

        source_scribes: list[str] = []
        for sl in todo_slices:
            if single_full:
                ext, scr = f"ext_{i}", f"scr_{i}"
                slice_line = "SLICE: full\n"
                read_line = ""
                ext_desc = (
                    f"Read source '{src}' and emit grounded claim+quote candidates "
                    f"mapped to the '{schema_name}' schema dimensions."
                )
            else:
                k = int(sl["index"])
                ext, scr = f"ext_{i}_{k}", f"scr_{i}_{k}"
                slice_line = f"SLICE: {slice_label(sl)}\n"
                read_line = (
                    f"SLICE_READ: {slice_read_call(sl)}  (read ONLY this window; "
                    "quotes must be verbatim substrings of it)\n"
                )
                ext_desc = (
                    f"Read your assigned SLICE of source '{src}' and emit grounded "
                    f"claim+quote candidates mapped to the '{schema_name}' schema "
                    "dimensions. Cover the dimensions THIS slice supports; other "
                    "slices of the same source cover the rest."
                )

            ctx = (
                f"PROJECT: {project}\n"
                f"{source_dir}"
                f"HUB: {hub or '(none)'}  (graph: {src})\n"
                f"{slice_line}"
                f"{read_line}"
                f"SCHEMA: {schema_name}  (the full dimension rubric is provided in "
                "this task's context object)\n"
                f"{structure_block}"
            )

            tasks.append({
                "alias": ext,
                "agent": "extractor",
                "schema": schema_name,
                "depends_on": list(batch_dep),
                "description": f"{ext_desc}\n{ctx}",
            })
            tasks.append({
                "alias": scr,
                "agent": "scribe",
                "schema": schema_name,
                # Write-lease scoped to THIS source box (``.zettelkasten/<src>``):
                # scribes on DIFFERENT source boxes get disjoint lease keys and run
                # in parallel; scribes on the SAME box (e.g. two per-schema scribes
                # in a union run) serialize, since their search-then-add is not
                # atomic across calls and would otherwise duplicate notes.
                "writes": [f".zettelkasten/{src}"],
                "depends_on": [ext],
                "description": f"{scribe_instr}\n{ctx}",
            })
            source_scribes.append(scr)

        aud = f"aud_{i}"
        tasks.append({
            "alias": aud,
            "agent": "auditor",
            "schema": schema_name,
            "depends_on": source_scribes,
            "description": (
                f"Audit '{src}': grounding integrity (every claim has a verbatim "
                f"supporting quote), hub/spine linkage, and coverage -- read the "
                f"deterministic matrix via {coverage_call} and interpret it.\n{aud_ctx}"
            ),
        })
        auditor_aliases.append(aud)

    # Trailing cross-cutting wave: the synthesizer runs ONCE after every source's
    # trio, placing the run's claims onto every touched project's OTHER spine
    # layouts and wiring cross-source themes into `_cross` hubs. It depends on ALL
    # auditors (every source must be committed + verified before cross-linking) and
    # is an implementer (it writes edges/hubs), so it declares ``writes: []``
    # (dynamic write-lease mode -- the zettelkasten store enforces per-box locking,
    # so it takes no claim-time whole-workspace lease). Emitted only when there
    # is cross-cutting work to describe (a rendered SPINES block and/or a project
    # to synthesize); the memory task then trails IT so the run is recorded last.
    # DISCOVERY OFFLOAD: the deterministic SPINES (cross-spine placement) and
    # CONNECTIONS (flat cross-graph candidate) blocks are threaded in below so the
    # synthesizer adjudicates a prepared worklist rather than re-discovering it.
    final_deps = list(auditor_aliases)
    if include_synthesis and auditor_aliases:
        synth_blocks = synthesis_context or (
            "SPINES: (none — no additional spine layouts resolved for the touched "
            "projects; do the flat-graph CONNECT job, plus cross-source SYNTHESIS "
            "only if a second source exists)"
        )
        if connection_context:
            synth_blocks += "\n" + connection_context
        tasks.append({
            "alias": "synth",
            "agent": "synthesizer",
            "writes": [],
            "depends_on": list(auditor_aliases),
            "description": (
                "Place this run's claims onto every spine layout of every project "
                f"the sources touch (launch project: '{project}'), wire the "
                "flat-graph world relations between notes (within and across "
                "sources), and synthesize cross-source themes into the shared "
                "`_cross` hubs. If the corpus is a SINGLE source (this run's one "
                "source and no other source in the touched project), the "
                "cross-source and hub jobs collapse to no-ops — do cross-spine "
                "PLACEMENT and INTRA-source CONNECT only; detect the mode yourself "
                "as your persona describes.\n"
                "SOURCES:\n"
                + "\n".join(
                    f"  - {p['name']}  (hub: {p.get('hub_id') or '(none)'})"
                    for p in prepped
                )
                + "\n"
                + synth_blocks
            ),
        })
        final_deps.append("synth")

    if include_memory:
        tasks.append({
            "alias": "memory",
            "agent": "memory",
            "depends_on": final_deps,
            "description": (
                f"Record the grounded-extraction run for project '{project}' "
                f"(schema '{schema_name}'): sources processed, coverage, any "
                "grounding/coverage issues the auditors flagged, and the "
                "cross-spine/cross-source connectivity the synthesizer added."
            ),
        })

    return tasks

build_union_extraction_tasks

build_union_extraction_tasks(prepped: list[dict], schema_specs: list[dict], project: str, *, batch_size: int = 0, include_memory: bool = False, synthesis_context: str = '', include_synthesis: bool = False, connection_context: str = '', slice_window: int = SLICE_WINDOW_DEFAULT, slice_overlap: int = SLICE_OVERLAP_DEFAULT, slice_threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT, slice_page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT, reextract: bool = False, slice_report: 'dict[str, dict] | None' = None) -> list[dict]

Emit a UNION extractor -> per-schema scribe -> per-schema auditor DAG.

The extract-once fan-out for a MULTI-SCHEMA run: instead of one extractor per schema re-reading the same source, each source/slice gets ONE shared extractor carrying ALL the schemas' rubrics (a SCHEMAS list). It tags each candidate with the schema: it belongs to; then ONE scribe PER schema (each depending on the shared extractor, each carrying its own schema so the persona filters candidates to that schema) commits its claims, and ONE auditor PER schema audits + stamps that schema's slices.

schema_specs is a list of per-schema dicts, each {name, structure, structures, source_structures, prefix} where prefix is the alias namespace the caller derived (its existing per-schema uniqueness logic). The shared extractor takes NO prefix (it is one node across schemas); the scribes and auditors take their schema's prefix.

A SINGLE-element schema_specs is delegated verbatim to :func:build_extraction_tasks (with the spec's prefix applied), so a one-schema run stays BYTE-IDENTICAL to today: one extractor, one scribe, one auditor per slice.

Incremental skip, reextract, slice_report, and the trailing synthesizer/memory tasks behave exactly as in :func:build_extraction_tasks. The per-schema skip is honored independently: a slice already done for schema A but not schema B is still read (for B), and only B's scribe processes it.

Source code in coordinator/extraction.py

def build_union_extraction_tasks(
    prepped: list[dict],
    schema_specs: list[dict],
    project: str,
    *,
    batch_size: int = 0,
    include_memory: bool = False,
    synthesis_context: str = "",
    include_synthesis: bool = False,
    connection_context: str = "",
    slice_window: int = SLICE_WINDOW_DEFAULT,
    slice_overlap: int = SLICE_OVERLAP_DEFAULT,
    slice_threshold: int = SLICE_CHAR_THRESHOLD_DEFAULT,
    slice_page_threshold: int = SLICE_PAGE_THRESHOLD_DEFAULT,
    reextract: bool = False,
    slice_report: "dict[str, dict] | None" = None,
) -> list[dict]:
    """Emit a UNION extractor -> per-schema scribe -> per-schema auditor DAG.

    The extract-once fan-out for a MULTI-SCHEMA run: instead of one extractor per
    schema re-reading the same source, each source/slice gets ONE shared extractor
    carrying ALL the schemas' rubrics (a ``SCHEMAS`` list). It tags each candidate
    with the ``schema:`` it belongs to; then ONE scribe PER schema (each depending
    on the shared extractor, each carrying its own ``schema`` so the persona
    filters candidates to that schema) commits its claims, and ONE auditor PER
    schema audits + stamps that schema's slices.

    ``schema_specs`` is a list of per-schema dicts, each ``{name, structure,
    structures, source_structures, prefix}`` where ``prefix`` is the alias
    namespace the caller derived (its existing per-schema uniqueness logic). The
    shared extractor takes NO prefix (it is one node across schemas); the scribes
    and auditors take their schema's prefix.

    A SINGLE-element ``schema_specs`` is delegated verbatim to
    :func:`build_extraction_tasks` (with the spec's prefix applied), so a
    one-schema run stays BYTE-IDENTICAL to today: one extractor, one scribe, one
    auditor per slice.

    Incremental skip, ``reextract``, ``slice_report``, and the trailing
    synthesizer/memory tasks behave exactly as in :func:`build_extraction_tasks`.
    The per-schema skip is honored independently: a slice already done for schema
    A but not schema B is still read (for B), and only B's scribe processes it.
    """
    if not prepped:
        raise ExtractionError("No prepared sources to build a graph from.")
    if not schema_specs:
        raise ExtractionError(
            "build_union_extraction_tasks requires at least one schema spec."
        )

    # SINGLE SCHEMA: delegate to the canonical builder so the trio shape is
    # byte-identical, then apply the spec's alias prefix (empty for a single run).
    # Only NON-DEFAULT extra kwargs are forwarded, so the common single-schema
    # call is byte-identical to the historical direct ``build_extraction_tasks``
    # invocation (the same minimal arg set callers/tests already rely on).
    if len(schema_specs) == 1:
        spec = schema_specs[0]
        call_kwargs: dict = {
            "batch_size": batch_size,
            "include_memory": include_memory,
            "structure": spec.get("structure"),
            "structures": spec.get("structures"),
        }
        if spec.get("source_structures") is not None:
            call_kwargs["source_structures"] = spec["source_structures"]
        if synthesis_context:
            call_kwargs["synthesis_context"] = synthesis_context
        if include_synthesis:
            call_kwargs["include_synthesis"] = include_synthesis
        if connection_context:
            call_kwargs["connection_context"] = connection_context
        if slice_window != SLICE_WINDOW_DEFAULT:
            call_kwargs["slice_window"] = slice_window
        if slice_overlap != SLICE_OVERLAP_DEFAULT:
            call_kwargs["slice_overlap"] = slice_overlap
        if slice_threshold != SLICE_CHAR_THRESHOLD_DEFAULT:
            call_kwargs["slice_threshold"] = slice_threshold
        if slice_page_threshold != SLICE_PAGE_THRESHOLD_DEFAULT:
            call_kwargs["slice_page_threshold"] = slice_page_threshold
        if reextract:
            call_kwargs["reextract"] = reextract
        if slice_report is not None:
            call_kwargs["slice_report"] = slice_report
        tasks = build_extraction_tasks(prepped, spec["name"], project, **call_kwargs)
        prefix = spec.get("prefix") or ""
        if prefix:
            for t in tasks:
                t["alias"] = prefix + t["alias"]
                t["depends_on"] = [prefix + d for d in t.get("depends_on", [])]
        return tasks

    # MULTI-SCHEMA union. Resolve each schema's named dimension vocab once (for
    # the FULL/PARTIAL render gate), mirroring build_extraction_tasks.
    from coordinator.contrib import expand_schema

    named_dims_by_spec: "list[frozenset[str] | None]" = []
    for spec in schema_specs:
        nd: "frozenset[str] | None" = None
        try:
            exp = expand_schema(spec["name"])
            if isinstance(exp, dict):
                nd = frozenset(
                    t.strip()
                    for t in (exp.get("tags") or [])
                    if isinstance(t, str) and t.strip()
                )
        except Exception:  # noqa: BLE001 — unresolvable name -> safe conditional
            nd = None
        named_dims_by_spec.append(nd)

    schemas_all = _render_schemas_block([s["name"] for s in schema_specs])

    tasks: list[dict] = []
    auditor_aliases: list[str] = []
    first_ext_by_source: list[str] = []

    for i, p in enumerate(prepped):
        src = p["name"]
        hub = p.get("hub_id", "")

        # Per (source, spec) available structures + rendered blocks (once).
        spec_src_structs: list[list] = []
        rendered: list[tuple] = []
        for si, spec in enumerate(schema_specs):
            ss = spec.get("source_structures")
            src_structs = (
                ss.get(src, []) if ss is not None else (spec.get("structures") or [])
            )
            spec_src_structs.append(src_structs)
            rendered.append(
                _render_source_blocks(
                    src, project, src_structs, named_dims=named_dims_by_spec[si]
                )
            )

        # DATA source: one shared data-extractor over all schemas -> per-schema
        # data-scribe -> per-schema auditor. Data sources are never sliced.
        if p.get("kind") == "data":
            dext = f"dext_{i}"
            first_ext_by_source.append(dext)
            # Batch chaining: gate on the anchor extractor of the source
            # ``batch_size`` back (empty when that source was skipped).
            batch_dep = []
            if batch_size and batch_size > 0 and i >= batch_size:
                prev = first_ext_by_source[i - batch_size]
                if prev:
                    batch_dep = [prev]
            derivations = p.get("derivations") or []
            deriv_line = (
                f"DERIVATIONS: {', '.join(str(d) for d in derivations)}\n"
                if derivations else ""
            )
            source_dir = (
                f"SOURCE: {src}  (zettelkasten dataset box; read its type=dataset "
                "notes via dataset(action=\"values\"))\n"
            )
            ext_ctx = (
                f"PROJECT: {project}\n"
                f"{source_dir}"
                f"HUB: {hub or '(none)'}  (graph: {src})\n"
                f"{deriv_line}{schemas_all}"
            )
            tasks.append({
                "alias": dext,
                "agent": "data-extractor",
                "schema": schema_specs[0]["name"],
                "depends_on": list(batch_dep),
                "description": (
                    f"Read the dataset source '{src}' and emit grounded "
                    "data-citation candidates for the evidence:data dimensions of "
                    "EVERY schema in SCHEMAS, tagging each candidate with the "
                    "`schema:` it belongs to.\n" + ext_ctx
                ),
            })
            # A data source is never sliced and emits EVERY schema, so this is
            # always a genuine multi-schema fan-out here (len(schema_specs) > 1):
            # stamp the union skip signal on all data-scribes and designate the
            # first (si == 0) as primary so exactly one adopts untagged candidates.
            for si, spec in enumerate(schema_specs):
                prefix = spec.get("prefix") or ""
                structure_block, _scribe_instr, coverage_call = rendered[si]
                union_line = _UNION_SCRIBE_SIGNAL.format(schema=spec["name"])
                if si == 0:
                    union_line += _UNION_PRIMARY_SCRIBE_SIGNAL
                sctx = (
                    f"PROJECT: {project}\n"
                    f"{source_dir}"
                    f"HUB: {hub or '(none)'}  (graph: {src})\n"
                    f"{deriv_line}"
                    f"{union_line}"
                    f"SCHEMA: {spec['name']}  (the full dimension rubric is "
                    "provided in this task's context object; attend to the "
                    "evidence:data dimensions)\n"
                    f"{structure_block}"
                )
                dscr = f"{prefix}dscr_{i}"
                tasks.append({
                    "alias": dscr,
                    "agent": "data-scribe",
                    "schema": spec["name"],
                    # Scoped to this source box so same-source per-schema scribes
                    # serialize while different sources parallelize (see
                    # build_extraction_tasks).
                    "writes": [f".zettelkasten/{src}"],
                    "depends_on": [dext],
                    "description": (
                        f"Commit the data-extractor's candidates for '{src}' whose "
                        f"schema: tag is '{spec['name']}' as claim notes with "
                        "method:data grounding, attaching each to its schema "
                        f"dimension node exactly as the prose scribe does.\n{sctx}"
                    ),
                })
                aud = f"{prefix}aud_{i}"
                tasks.append({
                    "alias": aud,
                    "agent": "auditor",
                    "schema": spec["name"],
                    "depends_on": [dscr],
                    "description": (
                        f"Audit '{src}' for schema '{spec['name']}': grounding "
                        "integrity (every data claim's citation re-computes to its "
                        "asserted value within tolerance), hub/spine linkage, and "
                        f"coverage -- read the deterministic matrix via "
                        f"{coverage_call} and interpret it.\n"
                        f"PROJECT: {project}\n"
                        f"{source_dir}"
                        f"HUB: {hub or '(none)'}  (graph: {src})\n"
                        f"SCHEMA: {spec['name']}  (the full dimension rubric is "
                        "provided in this task's context object)\n"
                        f"{structure_block}"
                    ),
                })
                auditor_aliases.append(aud)
            continue

        # PROSE source: plan slices once (source-level), then split per schema.
        planned = plan_slices(
            p,
            window=slice_window,
            overlap=slice_overlap,
            threshold=slice_threshold,
            page_threshold=slice_page_threshold,
            explicit=p.get("slices"),
        )
        spec_todo_idx: list[set] = []
        spec_todo_slices: list[list] = []
        for si in range(len(schema_specs)):
            todo, _done = _partition_slices(
                planned, p, project, spec_src_structs[si],
                schema=schema_specs[si]["name"], reextract=reextract,
            )
            spec_todo_slices.append(todo)
            spec_todo_idx.append({int(s["index"]) for s in todo})
        needed: set = set().union(*spec_todo_idx) if spec_todo_idx else set()

        if slice_report is not None:
            slice_report[src] = {
                "done": [slice_label(s) for s in planned if int(s["index"]) not in needed],
                "todo": [slice_label(s) for s in planned if int(s["index"]) in needed],
            }

        if not needed:
            # Every slice done for every schema -> emit no trio (skipped). Keep the
            # anchor list index-aligned with ``prepped`` for batch chaining.
            first_ext_by_source.append("")
            continue

        single_full = len(planned) == 1 and planned[0].get("kind") == "full"
        anchor = f"ext_{i}" if single_full else f"ext_{i}_{min(needed)}"
        first_ext_by_source.append(anchor)
        batch_dep = []
        if batch_size and batch_size > 0 and i >= batch_size:
            prev = first_ext_by_source[i - batch_size]
            if prev:
                batch_dep = [prev]

        source_dir = (
            f"SOURCE: {src}  (zettelkasten graph with ingested fulltext; "
            "read via source.fulltext)\n"
        )

        source_scribes_by_spec: list[list[str]] = [[] for _ in schema_specs]
        for sl in planned:
            k = int(sl["index"])
            if k not in needed:
                continue
            specs_here = [si for si in range(len(schema_specs)) if k in spec_todo_idx[si]]
            if single_full:
                ext = f"ext_{i}"
                slice_line = "SLICE: full\n"
                read_line = ""
                ext_desc = (
                    f"Read source '{src}' and emit grounded claim+quote candidates "
                    "for EVERY schema in SCHEMAS, tagging each candidate with the "
                    "`schema:` it belongs to."
                )
            else:
                ext = f"ext_{i}_{k}"
                slice_line = f"SLICE: {slice_label(sl)}\n"
                read_line = (
                    f"SLICE_READ: {slice_read_call(sl)}  (read ONLY this window; "
                    "quotes must be verbatim substrings of it)\n"
                )
                ext_desc = (
                    f"Read your assigned SLICE of source '{src}' and emit grounded "
                    "claim+quote candidates for EVERY schema in SCHEMAS, tagging "
                    "each candidate with the `schema:` it belongs to. Cover the "
                    "dimensions THIS slice supports; other slices of the same "
                    "source cover the rest."
                )
            # SCHEMAS carries only the schemas that still need THIS slice, so a
            # slice already done for some schema isn't re-extracted for it.
            schemas_here = _render_schemas_block(
                [schema_specs[si]["name"] for si in specs_here]
            )
            ext_ctx = (
                f"PROJECT: {project}\n"
                f"{source_dir}"
                f"HUB: {hub or '(none)'}  (graph: {src})\n"
                f"{slice_line}{read_line}{schemas_here}"
            )
            tasks.append({
                "alias": ext,
                "agent": "extractor",
                "schema": schema_specs[specs_here[0]]["name"],
                "depends_on": list(batch_dep),
                "description": f"{ext_desc}\n{ext_ctx}",
            })
            # PER-SLICE schema set drives the scribe's mode. A slice that emits
            # only ONE schema (the others were incremental-skipped for it) is
            # SINGLE-SCHEMA for that slice: it carries NO union signal so its
            # scribe keeps the byte-identical "missing -> mine" behavior and
            # adopts untagged candidates. A slice emitting 2+ schemas stamps the
            # union skip signal on all of them AND designates the FIRST EMITTED
            # schema (specs_here[0], not the global first) as the slice primary so
            # exactly one scribe adopts untagged candidates.
            slice_multi = len(specs_here) > 1
            slice_primary_si = specs_here[0] if slice_multi else None
            for si in specs_here:
                spec = schema_specs[si]
                prefix = spec.get("prefix") or ""
                structure_block, scribe_instr, _coverage = rendered[si]
                if slice_multi:
                    union_line = _UNION_SCRIBE_SIGNAL.format(schema=spec["name"])
                    if si == slice_primary_si:
                        union_line += _UNION_PRIMARY_SCRIBE_SIGNAL
                else:
                    union_line = ""
                sctx = (
                    f"PROJECT: {project}\n"
                    f"{source_dir}"
                    f"HUB: {hub or '(none)'}  (graph: {src})\n"
                    f"{slice_line}{read_line}"
                    f"{union_line}"
                    f"SCHEMA: {spec['name']}  (the full dimension rubric is "
                    "provided in this task's context object)\n"
                    f"{structure_block}"
                )
                scr = f"{prefix}scr_{i}" if single_full else f"{prefix}scr_{i}_{k}"
                tasks.append({
                    "alias": scr,
                    "agent": "scribe",
                    "schema": spec["name"],
                    # Scoped to this source box so same-source per-schema scribes
                    # serialize while different sources parallelize (see
                    # build_extraction_tasks).
                    "writes": [f".zettelkasten/{src}"],
                    "depends_on": [ext],
                    "description": f"{scribe_instr}\n{sctx}",
                })
                source_scribes_by_spec[si].append(scr)

        # Per-schema auditor over that schema's scribes for this source, stamping
        # exactly that schema's todo slices.
        content_hash = str(p.get("content_hash") or "")
        for si, spec in enumerate(schema_specs):
            if not source_scribes_by_spec[si]:
                continue
            prefix = spec.get("prefix") or ""
            structure_block, _scribe_instr, coverage_call = rendered[si]
            aud_ctx = (
                f"PROJECT: {project}\n"
                f"{source_dir}"
                f"HUB: {hub or '(none)'}  (graph: {src})\n"
                "SLICE: full\n"
                f"SCHEMA: {spec['name']}  (the full dimension rubric is provided "
                "in this task's context object)\n"
                f"{_render_slices_block(spec_todo_slices[si], content_hash)}"
                f"{structure_block}"
            )
            aud = f"{prefix}aud_{i}"
            tasks.append({
                "alias": aud,
                "agent": "auditor",
                "schema": spec["name"],
                "depends_on": list(source_scribes_by_spec[si]),
                "description": (
                    f"Audit '{src}' for schema '{spec['name']}': grounding "
                    "integrity (every claim has a verbatim supporting quote), "
                    "hub/spine linkage, and coverage -- read the deterministic "
                    f"matrix via {coverage_call} and interpret it.\n{aud_ctx}"
                ),
            })
            auditor_aliases.append(aud)

    # Trailing synthesizer + memory, mirroring build_extraction_tasks.
    final_deps = list(auditor_aliases)
    if include_synthesis and auditor_aliases:
        synth_blocks = synthesis_context or (
            "SPINES: (none — no additional spine layouts resolved for the touched "
            "projects; do the flat-graph CONNECT job, plus cross-source SYNTHESIS "
            "only if a second source exists)"
        )
        if connection_context:
            synth_blocks += "\n" + connection_context
        tasks.append({
            "alias": "synth",
            "agent": "synthesizer",
            "writes": [],
            "depends_on": list(auditor_aliases),
            "description": (
                "Place this run's claims onto every spine layout of every project "
                f"the sources touch (launch project: '{project}'), wire the "
                "flat-graph world relations between notes (within and across "
                "sources), and synthesize cross-source themes into the shared "
                "`_cross` hubs.\n"
                "SOURCES:\n"
                + "\n".join(
                    f"  - {p['name']}  (hub: {p.get('hub_id') or '(none)'})"
                    for p in prepped
                )
                + "\n"
                + synth_blocks
            ),
        })
        final_deps.append("synth")

    if include_memory:
        schema_names = ", ".join(s["name"] for s in schema_specs)
        tasks.append({
            "alias": "memory",
            "agent": "memory",
            "depends_on": final_deps,
            "description": (
                f"Record the grounded-extraction run for project '{project}' "
                f"(schemas: {schema_names}): sources processed, per-schema "
                "coverage, and any grounding/coverage issues the auditors flagged."
            ),
        })

    return tasks