class ExtractionTemplate:
"""Grounded extraction over a corpus, one spine per schema."""
name = "extraction"
def build(
self,
*,
documents: list[Document],
schemas: list[str],
project: str = "",
projects: list[str] | None = None,
options: dict[str, Any] | None = None,
) -> GraphSpec:
from coordinator import contrib as coord_contrib
from coordinator import extraction
opts = options or {}
if not schemas:
raise ValueError(
"The extraction template requires at least one schema. For "
"schema-free ingestion use a semantic template."
)
# EXTRACT-ONCE, ATTACH-TO-MANY-PROJECTS: ``projects`` (when given) is the
# full target set; the singular ``project`` is a back-compat shim and the
# primary used for naming/goal. A SINGLE project keeps every codepath
# byte-identical (no multi-project branch, no new kwargs threaded).
# Dedup while preserving order so ``projects=["A", "A"]`` collapses to a
# single-project run (no spurious multi-project codepath, no ``"A, A"``
# goal/memory text).
normalized = list(dict.fromkeys(
s for p in (projects if projects is not None else [project])
if (s := (p or "").strip())
))
if not normalized:
raise ValueError(
"The extraction template requires at least one project "
"(pass project= or projects=)."
)
primary = normalized[0]
multi_project = len(normalized) > 1
sources = [d.as_source() for d in documents]
hub_per_source = bool(opts.get("hub_per_source", True))
batch_size = int(opts.get("batch_size", 0) or 0)
synthesis_label = str(opts.get("synthesis_label", "") or "")
# Spine-sourced extraction (phase 3b). ``off`` (the default for direct
# programmatic callers) keeps the legacy named-schema scaffold path
# byte-identical; the daemon passes ``auto`` (from ``stream.yaml``
# ``defaults.spine``) to opt in. ``spines`` are EXPLICIT extra org ids;
# the routed project's default spine is auto-included by the resolver.
spine_mode = str(opts.get("spine", "off") or "off").lower()
explicit_spines = [
str(x).strip() for x in (opts.get("spines", []) or []) if str(x).strip()
]
# Opt-in FULL multi-spine fill (default off, byte-identical to today): when
# enabled the resolver fills EVERY in-scope spine and the trio attaches each
# claim to each fitting spine. Only meaningful when ``spine_mode`` sources.
multi_spine = bool(opts.get("multi_spine", False))
# Expand + validate all schemas up front (first seeds the per-source hubs).
schema_objs: dict[str, dict] = {}
for schema_name in schemas:
schema_obj = coord_contrib.expand_schema(schema_name)
if schema_obj is None:
raise ValueError(
f"Unknown schema {schema_name!r}. Available: "
f"{coord_contrib.list_schemas()}"
)
schema_objs[schema_name] = schema_obj
# prep_sources runs once (shared source graphs) for ALL target projects;
# the first schema seeds the per-source hubs (matches the legacy per-schema
# loop's first pass). The ``projects`` kwarg is threaded ONLY for a genuine
# multi-project run so the single-project call stays byte-identical.
prep_kwargs = {"projects": normalized} if multi_project else {}
prepped = extraction.prep_sources(
sources, schema_objs[schemas[0]], primary,
hub_per_source=hub_per_source, **prep_kwargs,
)
# When enabled, resolve each target project's spine scope. The resolver
# auto-includes each project's default spine and detects every candidate
# as a PURE READ, materializing only the chosen one(s). A multi-project run
# resolves per project and unions the (project-stamped) structures; a
# single project takes the byte-identical single-resolver path.
structure: dict | None = None
resolved_structures: list[dict] = []
spine_info: dict[str, Any] = {}
if spine_mode != "off":
from zettelkasten.server import _get_graph
if multi_project:
resolution = extraction.resolve_extraction_structures_multi(
normalized, prepped, explicit_spines, get_graph=_get_graph,
multi=multi_spine,
)
else:
resolution = extraction.resolve_extraction_structure(
primary, prepped, explicit_spines, get_graph=_get_graph,
multi=multi_spine,
)
structure = resolution.get("structure")
# The full materialized list (back-compat: ``[structure]`` when the
# resolver predates the multi-spine return). ``multi_spine`` fills
# every in-scope spine; the default keeps exactly one.
resolved_structures = resolution.get("structures")
if resolved_structures is None:
resolved_structures = [structure] if structure is not None else []
spine_info = {
"ignored_spines": resolution.get("ignored_spines", []),
"broken_spines": resolution.get("broken_spines", []),
"unknown_spines": resolution.get("unknown_spines", []),
}
if resolution.get("unknown_spines"):
logger.warning(
"stream: explicit spine opt-in(s) %s did not resolve to a "
"usable promoted spine; they were NOT used",
resolution["unknown_spines"],
)
if resolution.get("ignored_spines"):
logger.warning(
"stream: chose one in-scope spine; %d other(s) %s also "
"resolved and are deferred to resync",
len(resolution["ignored_spines"]), resolution["ignored_spines"],
)
if resolution.get("broken_spines"):
logger.warning(
"stream: scoped spine(s) %s are BROKEN (spine_ref graph "
"missing/empty or no materialized dimension matched)",
resolution["broken_spines"],
)
tasks: list[dict] = []
auditor_aliases: list[str] = []
structures: dict[str, dict | None] = {}
if structure is not None:
# SPINE-SOURCED: the promoted spine IS the organizing structure, so
# the run sources from it (the named schema informs only the rubric
# fallback + the task's schema attribute, mirroring the coordinator's
# single-schema spine path). If the route selected MULTIPLE named
# schemas they are SUPERSEDED for this run -- the spine's embedded
# rubric governs -- so name them in the log rather than silently fan
# out parallel scaffolds.
named = schemas[0]
if len(schemas) > 1:
logger.info(
"stream: project '%s' has a promoted spine; sourcing "
"structure from it and superseding the other routed named "
"schema(s) %s for this run.",
primary, schemas[1:],
)
sobj = schema_objs[named]
# Reconcile each source's persisted strict-tag vocabulary to the
# chosen spine's tags so spine-tagged claims pass add_note strict
# enforcement. reconcile PREFERS the spine's EMBEDDED rubric (carried
# on each structure); the named values are only a fallback. MULTI-SPINE:
# reconcile EACH resolved structure so every spine gets its OWN keyed
# (project, synthesis_graph) rubric. DEMOTE-TO-FLAT: the flat semantic
# graph is the always-present substrate and a spine is an additive
# overlay, so a source whose enforcing write fails is NEVER dropped --
# it just does not join that spine and extracts flat. The run is never
# aborted for a spine-rubric failure; a spine no source could join is
# simply unused (its orphaned per-source meta is pruned by reconcile;
# the spine graph itself stays materialized).
source_structures, demoted, unused = extraction.reconcile_structures(
prepped, resolved_structures, project=primary,
schema=str(sobj.get("name") or named),
strict=bool(sobj.get("strict", True)),
grounded=bool(sobj.get("grounded", True)),
)
# SURVIVING spines (>=1 member) drive the prep record + the back-compat
# singular ``structure``; ``per_source`` is set only when sources have
# heterogeneous available sets (some demotion from a survivor), so the
# clean-run render stays byte-identical to today.
resolved_structures, per_source = extraction.finalize_source_structures(
resolved_structures, source_structures
)
structure = resolved_structures[0] if resolved_structures else None
spine_info["demoted_sources"] = demoted
spine_info["unused_structures"] = [
{"synthesis_graph": s.get("synthesis_graph", "")}
for s in unused if isinstance(s, dict)
]
# Back-compat: nothing is EXCLUDED anymore, so ``quarantined_sources``
# is always empty.
spine_info["quarantined_sources"] = []
# Thread the per-source availability map ONLY when there are
# heterogeneous demotions; omit the kwarg otherwise so the shared
# render stays byte-identical (and back-compat callers are unaffected).
extra: dict = {} if per_source is None else {"source_structures": per_source}
spine_tasks = extraction.build_extraction_tasks(
prepped,
named,
primary,
batch_size=batch_size,
include_memory=False,
structure=structure,
structures=resolved_structures,
**extra,
)
for t in spine_tasks:
tasks.append(t)
if t["agent"] == "auditor":
auditor_aliases.append(t["alias"])
structures[named] = structure
# MULTI-SPINE: also surface the additional filled spines in the prep
# record so the run reflects every spine the claims attached to.
# Single-spine runs add nothing here, so the prep record stays
# byte-identical to today. A MULTI-project run keys each extra spine by
# its OWN (project, synthesis_graph): >=3 projects whose default/promoted
# spines share a ``spine_ref`` graph name would otherwise collide on the
# bare synthesis_graph and drop a spine. Single-project keeps the bare
# synthesis_graph key (byte-identical).
for s in resolved_structures:
if s is structure or not (s and s.get("dimension_nodes")):
continue
base = s.get("synthesis_graph") or f"spine-{len(structures)}"
if multi_project:
# Key per (project, synthesis_graph): two projects may
# legitimately share a spine_ref graph NAME, so the bare
# ``base`` would collide and drop a spine. Use the RAW project
# (NOT ``_slug``, which is many-to-one and would re-collapse
# slug-aliasing project names); a dedup suffix is the defensive
# backstop for the rare ``__``-separator coincidence.
proj = s.get("project") or primary
key = f"{proj}__{base}"
n = 1
while key in structures:
key = f"{proj}__{base}__{n}"
n += 1
else:
key = base
structures[key] = s
else:
# NAMED-SCHEMA scaffold path: one fresh spine per schema over the
# shared sources, namespaced trio per schema. Mirrors the coordinator's
# named-schema fallback DEMOTE-TO-FLAT semantics so the daemon never
# mass-rejects a multi-context source: a source whose ENFORCING
# keyed-meta write fails is NEVER dropped -- it is demoted to FLAT for
# that schema's spine (extracts context-less / hub-linked) and the run
# is never aborted for a per-source persist failure. Only a genuine
# scaffold/config ``ExtractionError`` propagates (a mis-configured run
# must surface, not silently degrade to flat).
demoted: list[dict] = []
# MINT one fresh spine per (project, schema) pair over the shared
# sources. The cross-product PLAN is the structure identity, and each
# entry is addressed by its INDEX into the plan -- NOT by a ``_slug``
# derived label, which is many-to-one and would collide for distinct
# project or schema names that slug-alias (e.g. ``A-B``/``A_B`` or
# ``s-1``/``s.1``). For a SINGLE project this is exactly the per-schema
# loop (project == primary), byte-identical to today; for several it is
# the (project, schema) cross product with each structure stamped with
# its OWN project. The trio is NOT emitted per (project, schema) -- that
# would re-read each source ONCE PER PROJECT (re-extraction). Instead the
# per-project structures are minted here, then GROUPED by schema below
# so a SINGLE trio per schema fans each claim across every project's
# spine (extract-once).
if multi_project:
plan = [(p, sch) for p in normalized for sch in schemas]
else:
plan = [(primary, sch) for sch in schemas]
# Index-addressed parallel arrays: ``built[i]`` / ``per_source_by_idx[i]``
# belong to ``plan[i]``. The index IS the injective structure identity.
built: list[dict | None] = [None] * len(plan)
per_source_by_idx: list[dict | None] = [None] * len(plan)
for idx, (proj, schema_name) in enumerate(plan):
# PER-PROJECT SPINE NAMING: an explicit ``synthesis_label`` is
# project-INDEPENDENT, so projA and projB would mint the SAME
# synthesis graph name and SHARE apex/dimension nodes. For a
# multi-project run fold the project into the label so each project
# gets its OWN spine. Single-project (and the empty-label case,
# where the label already IS the project) keep the byte-identical
# minted name.
eff_label = (
f"{synthesis_label}-{proj}"
if (multi_project and synthesis_label)
else synthesis_label
)
try:
s = extraction.prep_spine(
schema_objs[schema_name], proj, prepped,
synthesis_label=eff_label,
)
except extraction.StructurePersistError as exc:
# Total per-source persist failure for THIS (project, schema)
# spine -> demote it to FLAT (sources still extract
# context-less). A non-persist (config/scaffold) ExtractionError
# is NOT caught here and propagates.
logger.warning(
"stream: named-schema '%s' (project '%s') structure could "
"not be persisted for ANY source (%s); demoting it to FLAT "
"(sources still extract context-less into the flat graph)",
schema_name, proj, exc,
)
demoted.extend(
{"source": p["name"], "synthesis_graph": "",
"reason": str(exc)}
for p in prepped
)
built[idx] = None
per_source_by_idx[idx] = None
continue
# Stamp the per-structure project so its keyed (project,
# synthesis_graph) meta + rendered CONTEXT_KEY key under THIS
# project (no-op for a single project: it equals ``primary``).
if isinstance(s, dict) and multi_project:
s["project"] = proj
# Pop the per-source quarantine ``prep_spine`` stashed and DEMOTE
# those sources to FLAT for this spine (never exclude): the failing
# sources keep extracting context-less, the healthy ones keep the
# keyed context. ``finalize_source_structures`` yields a per-source
# map only when the available sets are heterogeneous, so a clean run
# (no quarantine) threads NO ``source_structures`` and stays
# byte-identical to the legacy named-schema render.
q = (s.pop("quarantined", []) or []) if isinstance(s, dict) else []
per_source: "dict[str, list] | None" = None
if q and s is not None:
sg = s.get("synthesis_graph", "")
q_names = {x.get("source") for x in q}
demoted.extend(
{"source": x.get("source"), "synthesis_graph": sg,
"reason": x.get("reason", "")}
for x in q
)
source_structures = {
p["name"]: ([] if p["name"] in q_names else [s])
for p in prepped
}
surviving, per_source = extraction.finalize_source_structures(
[s], source_structures
)
s = surviving[0] if surviving else None
built[idx] = s
per_source_by_idx[idx] = per_source
# SAFETY NET (spine collision across distinct (project, schema)): even
# with the injective index identity, a schema whose ``synthesis.graph``
# template ignores the label/project could mint the SAME graph name for
# two distinct (project, schema) pairs. Refuse to proceed rather than
# silently merge their spines (shared apex / dimension nodes corrupting
# synthesis). The check reads the REAL index-addressed structures, so
# the message names the actual colliding pairs (never a graph a pair
# never mints, the bug of computing it off an already-clobbered dict).
if multi_project:
sg_owner: dict[str, tuple[str, str]] = {}
for idx, (proj, sch) in enumerate(plan):
# Use the INTENDED synthesis-graph name (what ``prep_spine``
# mints via the same ``_synthesis_graph_name`` call), NOT the
# post-demotion ``built[idx]``: a (project, schema) that demoted
# via StructurePersistError -- AFTER ``build_spine_skeleton`` already
# materialized a colliding spine -- would be ``None`` here and
# hide the collision. The label fold mirrors the prep loop above.
eff_label = (
f"{synthesis_label}-{proj}" if synthesis_label else synthesis_label
)
sg = extraction._synthesis_graph_name(
schema_objs[sch], proj, eff_label
)
if not sg:
continue
prev = sg_owner.get(sg)
if prev is not None and prev != (proj, sch):
pp, psch = prev
raise extraction.ExtractionError(
f"Multi-project synthesis graph collision: (project "
f"'{pp}', schema '{psch}') and (project '{proj}', "
f"schema '{sch}') both mint synthesis graph '{sg}'. "
f"Use distinct synthesis labels or a project/schema-"
f"scoped synthesis.graph template so each keeps its "
f"own spine."
)
sg_owner[sg] = (proj, sch)
# GROUP the minted per-(project, schema) structures by SCHEMA and emit
# a UNION fan-out (extract-once): one shared extractor per source/slice
# carrying ALL schemas' rubrics -> one scribe PER schema -> a per-schema
# auditor. ``structures=`` is the same multi-structure mechanism the
# spine-sourced path uses; the scribe fans each claim across the
# projects' spines. A single-project run passes a one-element list,
# byte-identical to the historical single-structure call. The alias
# axis is per-SCHEMA, not per (project, schema). ``idxs`` are positions
# into ``plan``/``built`` (the injective identity).
schema_groups: "dict[str, list[int]]" = {}
for idx, (proj, schema_name) in enumerate(plan):
schema_groups.setdefault(schema_name, []).append(idx)
single = len(schema_groups) == 1
# ALIAS UNIQUENESS: the per-schema prefix is byte-identical to today
# (``_slug(schema_name)_``) WHENEVER the schema slugs are unique -- the
# common case, incl. single-project multi-schema. Only if two schemas
# slug-collide (e.g. ``s-1`` vs ``s.1``) disambiguate with a stable
# index so ``create_graph`` never sees duplicate aliases.
schema_slugs = [_slug(s) for s in schema_groups]
slugs_unique = len(set(schema_slugs)) == len(schema_slugs)
# Build ONE spec per schema (structures + per-source availability map +
# alias prefix), then hand them to the union builder in a SINGLE call so
# each source is read ONCE across all schemas instead of once per schema.
schema_specs: list[dict] = []
for s_idx, (schema_name, idxs) in enumerate(schema_groups.items()):
# The structures for this schema (one per project), in project
# order; drop fully-demoted (None) spines -- their sources still
# extract flat via the trio's flat render.
schema_structures = [
built[i] for i in idxs if built[i] is not None
]
# Combine the per-source availability maps across this schema's
# structures. When NO entry demoted a source (the common clean
# case) leave ``source_structures`` unset so the shared,
# byte-identical render runs; otherwise build the union map.
maps_present = any(
per_source_by_idx[i] is not None for i in idxs
)
if maps_present:
combined: dict[str, list] = {p["name"]: [] for p in prepped}
for i in idxs:
st = built[i]
if st is None:
continue
m = per_source_by_idx[i]
for n in combined:
if m is None or any(x is st for x in m.get(n, [])):
combined[n].append(st)
source_structures = combined
else:
source_structures = None
if single:
prefix = ""
elif slugs_unique:
prefix = f"{_slug(schema_name)}_"
else:
prefix = f"{_slug(schema_name)}_{s_idx}_"
schema_specs.append({
"name": schema_name,
"structure": schema_structures[0] if schema_structures else None,
"structures": schema_structures,
"source_structures": source_structures,
"prefix": prefix,
})
# PRIMARY-SCRIBE untagged adoption is now decided PER SLICE inside
# ``build_union_extraction_tasks`` (a slice that emits only one schema
# is single-schema and adopts untagged; a slice emitting 2+ marks its
# first EMITTED schema primary). This is correct under incremental skip,
# where the global-first schema can be skipped for a given slice — a
# global primary loop here would then leave that slice with no primary
# and silently drop untagged candidates.
union_tasks = extraction.build_union_extraction_tasks(
prepped,
schema_specs,
primary,
batch_size=batch_size,
include_memory=False,
)
for t in union_tasks:
tasks.append(t)
if t["agent"] == "auditor":
auditor_aliases.append(t["alias"])
# EXTERNALLY-VISIBLE prep-record keys: one guaranteed-UNIQUE key per
# (project, schema). A SINGLE project keeps the byte-identical
# schema-name key (single-schema -> ``"s1"``, multi-schema -> per-schema
# names). A MULTI-project run keys each entry on its structure's ACTUAL
# per-(project, schema) synthesis_graph NAME -- guaranteed DISTINCT (the
# per-project label fold mints a distinct spine per project, and the
# collision guard above refuses any duplicate), so the mapping is
# injective WITHOUT a lossy ``_slug`` composite (the structure identity
# is already the injective ``plan`` index; this only NAMES it for the
# prep record). A fully-demoted (None) entry has no synthesis_graph, so
# it falls back to the RAW ``project__schema`` pair (raw, not slugged, so
# slug-aliasing names stay distinct); a defensive dedup suffix guarantees
# uniqueness even where a fallback string coincides with a real graph
# name.
for idx, (proj, sch) in enumerate(plan):
st = built[idx]
if multi_project:
sg = (st.get("synthesis_graph") if isinstance(st, dict) else "") or ""
key = sg or f"{proj}__{sch}"
base_key = key
n = 1
while key in structures:
key = f"{base_key}__{n}"
n += 1
else:
key = sch
structures[key] = st
# Surface the DEMOTE-TO-FLAT diagnostics (mirroring the coordinator and
# the spine-sourced branch): nothing is EXCLUDED, so quarantined_sources
# is always empty. Empty lists keep ``any(spine_info.values())`` falsy,
# so a clean named-schema run still emits NO ``spine`` key (byte-identical).
spine_info["demoted_sources"] = demoted
spine_info["quarantined_sources"] = []
# EMPTY-GRAPH NO-OP GUARD: when incremental skip left NO extraction trios
# (every source already fully covered -> no auditors), do NOT append the
# trailing memory task. Appending it would yield a lone-memory graph that
# `create_graph` turns into a run + a spurious memory session for work that
# did not happen. Gating on ``auditor_aliases`` (empty iff no trios) instead
# returns an EMPTY ``tasks`` list -- the stream caller's clean no-op signal.
# Mirrors the coordinator's create_extraction_graph no-op guard.
if auditor_aliases and opts.get("include_memory", True):
tasks.append(
{
"alias": "memory",
"agent": "memory",
"depends_on": auditor_aliases,
"description": (
f"Record the grounded-extraction run for project "
f"'{', '.join(normalized)}' (schemas: {', '.join(schemas)}): "
"sources processed, per-schema coverage, and any grounding "
"issues."
),
}
)
prep_record = {
"sources": [
{"name": p["name"], "title": p["title"], "hub_id": p["hub_id"]}
for p in prepped
],
"structures": {
name: (
{
"synthesis_graph": s.get("synthesis_graph", ""),
"apex_id": s.get("apex_id", ""),
"dimension_nodes": s.get("dimension_nodes", {}),
}
if s
else None
)
for name, s in structures.items()
},
}
# Surface the spine scope decision only when spine-sourcing actually
# happened (a structure resolved) or the resolver flagged something
# (ignored/broken/unknown/quarantined). An ``auto`` run with no default
# spine resolves nothing and flags nothing, so this stays absent —
# keeping the prep byte-identical to the ``off`` / legacy named-schema
# path. (``spine_info`` is a dict of empty lists in that case, which is
# still truthy, so test for real content rather than the dict itself.)
if structure is not None or any(spine_info.values()):
prep_record["spine"] = spine_info
src_desc = extraction.describe_sources(prepped)
goal = (
f"Grounded extraction: {src_desc} → {', '.join(normalized)} "
f"(schemas: {', '.join(schemas)})"
if src_desc
else (
f"Grounded extraction: {', '.join(normalized)} "
f"(schemas: {', '.join(schemas)})"
)
)
return GraphSpec(tasks=tasks, goal=goal, prep=prep_record)