Extraction-schema registry for the grounded-extraction bundle.
A schema is the reusable rubric that parameterizes a grounded-extraction run:
a named set of dimensions, each with a tag that becomes the note tag the
scribe writes (so the schema is simultaneously the extraction rubric and the
cross-source comparison index). See extraction-schemas.yaml for the format.
Lifecycle mirrors agents.yaml: built-in defaults ship inside this package
(extraction-schemas.yaml next to this module) and are always available; a
project may extend or override them with its own extraction-schemas.yaml
placed next to agents.yaml (.cursor/ or .claude/). Both files are
loaded once and cached; call :func:reset_cache in tests after editing them.
This module is imported by the coordinator (via the capability contribution) to
expand a schema name into the fully-expanded object it flows into task context.
Drop the loaded-registry cache (used by tests after editing the files).
Source code in zettelkasten/extraction_schemas.py
| def reset_cache() -> None:
"""Drop the loaded-registry cache (used by tests after editing the files)."""
global _raw_cache
_raw_cache = None
|
expand_schema
expand_schema(name: str) -> dict | None
Expand a registry schema name into its fully-resolved object, or None.
Applies per-schema defaults (grounded/strict/link_relation),
normalizes the dimension list, validates the optional structural relations
backbone, synthesis block, and per-dimension connect_to hints, and
enforces tag uniqueness. Returns None when name is not in the registry;
raises ValueError when the schema exists but is malformed (duplicate tags,
dangling tag refs, invalid relations) so the error is actionable.
Source code in zettelkasten/extraction_schemas.py
| def expand_schema(name: str) -> dict | None:
"""Expand a registry schema ``name`` into its fully-resolved object, or ``None``.
Applies per-schema defaults (``grounded``/``strict``/``link_relation``),
normalizes the dimension list, validates the optional structural ``relations``
backbone, ``synthesis`` block, and per-dimension ``connect_to`` hints, and
enforces tag uniqueness. Returns ``None`` when ``name`` is not in the registry;
raises ``ValueError`` when the schema exists but is malformed (duplicate tags,
dangling tag refs, invalid relations) so the error is actionable.
"""
if not name:
return None
raw = _load_raw().get(name)
if raw is None:
return None
return expand_spec(name, raw)
|
expand_spec
expand_spec(name: str, raw: dict) -> dict
Expand a raw schema raw dict (not from the registry) for name.
Same validation + expansion as :func:expand_schema, but on a caller-supplied
spec -- used by the schema tool to validate a draft before saving. Raises
ValueError on a malformed spec; never returns None.
Source code in zettelkasten/extraction_schemas.py
| def expand_spec(name: str, raw: dict) -> dict:
"""Expand a raw schema ``raw`` dict (not from the registry) for ``name``.
Same validation + expansion as :func:`expand_schema`, but on a caller-supplied
spec -- used by the ``schema`` tool to validate a draft before saving. Raises
``ValueError`` on a malformed spec; never returns ``None``.
"""
if not isinstance(raw, dict):
raise ValueError(f"Schema '{name}' spec must be a mapping.")
dims = _normalize_dimensions(raw.get("dimensions"), name)
tags = {d["tag"] for d in dims}
from zettelkasten.graph import VALID_RELATIONS
link_relation = str(raw.get("link_relation") or _DEFAULT_LINK_RELATION)
if link_relation not in VALID_RELATIONS:
raise ValueError(
f"Schema '{name}' link_relation '{link_relation}' is not a valid relation. "
f"Must be one of: {sorted(VALID_RELATIONS)}"
)
# Validate per-dimension attach hints against the tag set + relation vocab.
for d in dims:
ct = d.get("connect_to")
if ct and ct not in tags:
raise ValueError(
f"Schema '{name}' dimension '{d['tag']}' connect_to references unknown "
f"tag '{ct}'."
)
if ct == d["tag"]:
raise ValueError(
f"Schema '{name}' dimension '{d['tag']}' cannot connect_to itself."
)
ar = d.get("attach_relation")
if ar and ar not in VALID_RELATIONS:
raise ValueError(
f"Schema '{name}' dimension '{d['tag']}' relation '{ar}' is not a valid "
f"relation. Must be one of: {sorted(VALID_RELATIONS)}"
)
# Evidence channel: validate the enum WHEN DECLARED, but surface the key
# only when the schema explicitly set it (matching the optional-field
# convention used by connect_to/note_type). Injecting a default ``text``
# onto every dimension perturbs the prose-extractor context, so a schema
# with no ``evidence: data`` must expand byte-identically to before; the
# data trio still reads ``evidence: data`` when it is declared.
ev = d.get("evidence")
if ev is not None and ev not in _VALID_EVIDENCE:
raise ValueError(
f"Schema '{name}' dimension '{d['tag']}' evidence '{ev}' is not a "
f"valid evidence channel. Must be one of: {sorted(_VALID_EVIDENCE)}"
)
relations = _build_relations(raw.get("relations"), tags, name)
synthesis = _build_synthesis(raw.get("synthesis"), link_relation, name)
# Resolve effective ``materialize`` per dimension. Only meaningful when the
# schema declares a ``synthesis`` block; then every dimension materializes a
# node unless it explicitly opted out with ``materialize: false``.
for d in dims:
if synthesis is not None:
d["materialize"] = bool(d.get("materialize", True))
else:
d.pop("materialize", None)
hub = raw.get("hub")
return {
"name": name,
"description": str(raw.get("description") or ""),
"grounded": bool(raw.get("grounded", _DEFAULT_GROUNDED)),
"strict": bool(raw.get("strict", _DEFAULT_STRICT)),
"link_relation": link_relation,
"hub": hub if isinstance(hub, dict) else None,
"dimensions": dims,
"tags": [d["tag"] for d in dims],
"relations": relations,
"synthesis": synthesis,
}
|
list_schemas() -> list[str]
Sorted names of all schemas in the merged registry.
Source code in zettelkasten/extraction_schemas.py
| def list_schemas() -> list[str]:
"""Sorted names of all schemas in the merged registry."""
return sorted(_load_raw().keys())
|
save_schema(name: str, spec: dict) -> dict
Validate a draft schema spec and write it into the project registry.
Validates by expanding the spec (raising ValueError on a malformed draft),
then merges the raw block under name into the project's
extraction-schemas.yaml (creating the file + parent dir on first save),
overriding any built-in of the same name. Drops the registry cache so the new
schema is visible to subsequent reads in this process. Returns the write path
and the expanded schema. NOTE: a running MCP server still holds the old
in-memory registry until restarted -- the caller surfaces that to the user.
Source code in zettelkasten/extraction_schemas.py
| def save_schema(name: str, spec: dict) -> dict:
"""Validate a draft schema ``spec`` and write it into the project registry.
Validates by expanding the spec (raising ``ValueError`` on a malformed draft),
then merges the raw block under ``name`` into the project's
``extraction-schemas.yaml`` (creating the file + parent dir on first save),
overriding any built-in of the same name. Drops the registry cache so the new
schema is visible to subsequent reads in this process. Returns the write path
and the expanded schema. NOTE: a running MCP server still holds the old
in-memory registry until restarted -- the caller surfaces that to the user.
"""
name = (name or "").strip()
if not name:
raise ValueError("save_schema requires a non-empty schema name.")
# Validate first; never write a malformed block.
expanded = expand_spec(name, spec)
path = _project_schema_write_path()
path.parent.mkdir(parents=True, exist_ok=True)
existing: dict = {}
if path.exists():
loaded = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
if isinstance(loaded, dict):
existing = loaded
existing[name] = spec
path.write_text(
yaml.safe_dump(existing, sort_keys=False, allow_unicode=True),
encoding="utf-8",
)
reset_cache()
return {"name": name, "path": str(path), "schema": expanded}
|