appdb/config_versions.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Versioned ``project_config/`` bundle -- history, diff, rollback (phase 3).
``docs/admin-panel-architecture.md`` §6 and the phase 3 spec: the nine
``project_config/*.yaml`` files (see :data:`CONFIG_FILENAMES`) become
editable, versioned, and reversible through the application database
(:mod:`appdb.engine`), the same durable store phase 2 already put the key
store and role grants in.
Shape (spec §1, §6.1)
----------------------
**Bundle-versioned, with per-file restore.** One row in
``config_bundle_versions`` (:data:`appdb.models.config_bundle_versions`) is
a full snapshot of all nine files -- never a diff, so restoring any depth
is a read, not a replay. The *active* version is always the highest
``version_id`` whose ``status`` is ``"applied"``; there is deliberately no
separate flag to keep in sync with that.
**Revert, never reset.** :func:`restore` never mutates or deletes an
earlier row -- it creates a brand-new version whose content equals the
target's, going through the exact same validate-diff-dry_run-apply path as
any other edit (:func:`propose_or_apply`). Restoring version 3 when
versions 4 and 5 already exist leaves both of them exactly where they were
and creates version 6; the restore itself is just another version, and is
therefore itself reversible.
**Propose and approve (spec §3.1).** ``schema.yaml`` is the guard's
allowlist (security admin's, per the architecture's §2). An operations
principal may still *propose* a change that touches it: this module then
creates the version as an unapplied ``"draft"`` -- carrying its diff and
dry-run results, computed once, at proposal time, so what a reviewer sees
is exactly what they are approving -- and only a principal holding the
security capability can promote it (:func:`approve_draft`) or reject it
(:func:`reject_draft`). Approval re-checks optimistic locking (below)
against the *current* active version, not the one open when the draft was
created, so a change nobody has seen yet cannot be approved out from under
a more recent edit.
Validated, every time (spec §4)
---------------------------------
Before anything is written, :func:`propose_or_apply` (1) parses and
validates every file through the real loaders in
:mod:`knowledge.config_loader` and :mod:`schema_data.registry` -- a
malformed file is rejected with the loader's own error, and the active
configuration is left completely untouched; (2) when ``schema.yaml``
changed, runs the same schema-agnostic structural-invariant check
``tests/test_schema_registry_snapshot.py``'s ``TestAllowlistStructuralInvariants``
pins (:func:`schema_data.registry.check_allowlist_structural_invariants`
-- called, not reimplemented: both this module and that test call the one
function); (3) computes a structural diff -- for ``schema.yaml``,
specifically which tables and columns enter and leave the allowlist, not a
text diff of YAML; and (4) dry-runs the golden set through the real
:mod:`eval` harness in offline mode (no LLM, no database -- see
:mod:`eval.runner`'s own docstring for what that mode does and does not
prove) and, for a ``schema.yaml`` change specifically, cross-checks every
table the golden set's own reference SQL touches
(:func:`security.sql_guard.extract_touched_tables`) against the
*candidate* allowlist -- this is what catches "a schema.yaml from five
versions ago names tables the warehouse no longer has" before it is
applied, not after.
Optimistic locking, not merging (spec §8)
--------------------------------------------
Every write here (:func:`propose_or_apply`, :func:`approve_draft`) takes
the version its caller had open (``based_on_version``) and refuses with
:class:`StaleVersionError` -- naming who changed it and when -- if the
active version has moved on since. No three-way merge is attempted
anywhere in this module.
What takes effect immediately, and what needs a restart
------------------------------------------------------------
Applying a version always writes the bundle to
``cfg.settings.project_config_dir`` (so the very next process restart
picks it up unconditionally) and, for the eight files that do not feed
``security.sql_guard``'s allowlist, also clears their in-process caches
(:mod:`knowledge`'s lazy ``knowledge.*`` loader modules, and
:mod:`prompt_engine.static_prefix`'s ``lru_cache``s when a prefix-visible
file changed) so a long-running server reflects the change on its very
next request, the same "immediate, not eventually" bar
:mod:`appdb.key_store` already holds revocation to.
``schema.yaml`` is deliberately excluded from that live-reload:
``security.sql_guard`` resolves its table/column allowlist from a
module-level binding computed once, at import time
(:data:`schema_data.columns.TABLE_COLUMNS` frozen into
``security.sql_guard``'s own namespace), and mutating that binding for one
in-flight admin request would race every concurrently-executing analyst
request validating SQL against it -- a security-critical global has no
business being swapped out from under a live server. A ``schema.yaml``
change is therefore validated and dry-run exactly as thoroughly as any
other file, written to disk immediately, and takes effect for the guard
(and the prompt's schema block, which is fed by the same frozen import) on
the next restart -- exactly the "an innocuous-looking edit ... halves
throughput with no error anywhere" risk §6 of the architecture describes
being made *visible* via the prefix version, not eliminated by an
unsafe live patch of security-critical state.
Export (spec §7)
------------------
Every applied version is also written to ``cfg.settings.config_export_dir``
when configured -- an operator pointing that at their own git repository
gets a commit per change, and an off-box backup, without this module's
correctness depending on git being installed at all (see
:func:`import_bundle_from_directory` for the other half of that round
trip).
"""
from __future__ import annotations
import hashlib
import importlib
import json
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Sequence
from sqlalchemy import func, select
import config as cfg
from appdb.engine import get_app_engine
from appdb.models import config_bundle_versions
from core.yaml_loading import safe_load_strict
from security.auth import OPERATIONS_CAPABILITY, SECURITY_CAPABILITY
#: The nine ``project_config/*.yaml`` files this module versions as one
#: bundle (spec §1; architecture §6). ``relationships.yaml`` (a separate,
#: optional file consumed by ``database/relationship_map.py``'s schema
#: auto-discovery, not by :mod:`knowledge.config_loader`) is deliberately
#: NOT one of these nine.
CONFIG_FILENAMES: tuple[str, ...] = (
"aliases.yaml",
"business_rules.yaml",
"entities.yaml",
"examples.yaml",
"metrics.yaml",
"retrieval_hints.yaml",
"session_policy.yaml",
"memory_policy.yaml",
"schema.yaml",
)
#: The one file of the nine that is security-admin's, not operations'
#: (spec §3; architecture §2) -- it is the guard's allowlist, not merely
#: descriptive domain data.
SECURITY_ONLY_FILES: frozenset[str] = frozenset({"schema.yaml"})
#: Which of the nine files feed the byte-identical static prompt prefix
#: (:mod:`prompt_engine.static_prefix`) -- only these three warrant
#: clearing that module's ``lru_cache``s on apply. ``schema.yaml`` also
#: feeds the prefix but is excluded here on purpose -- see the module
#: docstring's "What takes effect immediately" section.
_PREFIX_RELEVANT_FILES: frozenset[str] = frozenset(
{"business_rules.yaml", "metrics.yaml", "examples.yaml"}
)
#: filename -> the dotted module name of its ``knowledge.*`` lazy-loader
#: shim, for cache invalidation on apply. ``schema.yaml`` has no entry
#: here -- see the module docstring.
_KNOWLEDGE_CACHE_MODULES: dict[str, str] = {
"aliases.yaml": "knowledge.aliases",
"business_rules.yaml": "knowledge.business_rules",
"entities.yaml": "knowledge.entities",
"examples.yaml": "knowledge.examples",
"metrics.yaml": "knowledge.metrics",
"retrieval_hints.yaml": "knowledge.retrieval_hints",
"session_policy.yaml": "knowledge.session_policy",
"memory_policy.yaml": "knowledge.memory_policy",
}
# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------
class ConfigVersionValidationError(ValueError):
"""A candidate bundle failed to parse/validate through a real loader.
``str(self)`` is the *first* failing file's own error message --
matching :func:`knowledge.config_loader._load_validated`'s own "surface
the first error" convention -- and :attr:`errors` carries every
failing file's message, in file order, for a caller that wants the
complete picture (e.g. a panel showing every problem at once).
"""
def __init__(self, errors: list[str]) -> None:
super().__init__(errors[0] if errors else "invalid configuration")
self.errors = errors
class VersionNotFoundError(LookupError):
"""No ``config_bundle_versions`` row matches the given ``version_id``."""
class StaleVersionError(RuntimeError):
"""A save/approval was based on a version that is no longer active
(spec §8) -- never merged, always refused. The message names which
version is now active, who created it, and when."""
class NotADraftError(RuntimeError):
""":func:`approve_draft` / :func:`reject_draft` called on a version
whose ``status`` is not ``"draft"``."""
# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------
def _now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _hash_content(files: dict[str, str]) -> str:
canonical = json.dumps(files, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def _content_dict(row: dict[str, Any]) -> dict[str, str]:
return json.loads(row["content_json"])
def _text_validators() -> dict[str, Any]:
"""``{filename: callable(text) -> parsed model}`` -- the real,
in-memory text validators for each of :data:`CONFIG_FILENAMES` (see
module docstring's "Validated, every time" section).
Every entry is a genuine loader function
(:func:`knowledge.config_loader.validate_yaml_text` /
:func:`schema_data.registry.validate_schema_yaml_text`), not a
reimplementation -- both raise the exact same
``"[filename] validation error at ...'"`` :class:`ValueError`
:mod:`knowledge.config_loader`'s file-based loaders raise for the same
content on disk. Deliberately the *text*-based entry points, not the
file-based ``load_*`` functions: those read
``cfg.settings.project_config_dir`` and this module must validate a
candidate bundle that may never be written to disk, without the
thread-unsafe ``cfg.override_settings`` detour (see
:func:`~knowledge.config_loader.validate_yaml_text`'s docstring).
Deferred import: this module must not force-import the whole
knowledge/schema layer merely by being imported.
"""
from knowledge.config_loader import (
AliasesConfig,
BusinessRulesConfig,
EntitiesConfig,
ExamplesConfig,
MemoryPolicyConfig,
MetricsConfig,
RetrievalHintsConfig,
SessionPolicyConfig,
validate_yaml_text,
)
return {
"aliases.yaml": lambda text: validate_yaml_text("aliases.yaml", text, AliasesConfig),
"business_rules.yaml": lambda text: validate_yaml_text(
"business_rules.yaml", text, BusinessRulesConfig
),
"entities.yaml": lambda text: validate_yaml_text("entities.yaml", text, EntitiesConfig),
"examples.yaml": lambda text: validate_yaml_text("examples.yaml", text, ExamplesConfig),
"metrics.yaml": lambda text: validate_yaml_text("metrics.yaml", text, MetricsConfig),
"retrieval_hints.yaml": lambda text: validate_yaml_text(
"retrieval_hints.yaml", text, RetrievalHintsConfig
),
"session_policy.yaml": lambda text: validate_yaml_text(
"session_policy.yaml", text, SessionPolicyConfig
),
"memory_policy.yaml": lambda text: validate_yaml_text(
"memory_policy.yaml", text, MemoryPolicyConfig
),
"schema.yaml": _validate_candidate_schema,
}
def _validate_candidate_schema(text: str) -> Any:
"""Validate a candidate ``schema.yaml``, including its data sources.
Beyond :func:`schema_data.registry.validate_schema_yaml_text`, every
table's ``datasource`` must name a source in this deployment's
``datasources.yaml`` (or be empty); otherwise the draft would be
approved and then fail at start-up.
"""
from database.datasources import check_table_datasources
from schema_data.registry import validate_schema_yaml_text
parsed = validate_schema_yaml_text(text)
check_table_datasources(
{name: table.datasource for name, table in parsed.tables.items()}
)
return parsed
def _public(row: dict[str, Any], *, include_content: bool = False) -> dict[str, Any]:
out = {
"version_id": row["version_id"],
"status": row["status"],
"content_hash": row["content_hash"],
"based_on_version": row["based_on_version"],
"restored_from_version": row["restored_from_version"],
"restored_file": row["restored_file"],
"created_at": row["created_at"],
"created_by": row["created_by"],
"created_by_capability": row["created_by_capability"],
"reviewed_by": row["reviewed_by"],
"reviewed_at": row["reviewed_at"],
"review_note": row["review_note"],
"diff": json.loads(row["diff_json"]) if row["diff_json"] else None,
"dry_run": json.loads(row["dry_run_json"]) if row["dry_run_json"] else None,
}
if include_content:
out["files"] = _content_dict(row)
return out
# ---------------------------------------------------------------------------
# Bootstrap -- lazy, on first access, mirroring knowledge/*.py's own
# "loaded on first access, never at import time" discipline.
# ---------------------------------------------------------------------------
def _ensure_bootstrapped(conn) -> None:
count = conn.execute(select(func.count()).select_from(config_bundle_versions)).scalar()
if count:
return
base_dir = Path(cfg.settings.project_config_dir)
files = {
filename: (base_dir / filename).read_text(encoding="utf-8")
if (base_dir / filename).exists() else ""
for filename in CONFIG_FILENAMES
}
conn.execute(
config_bundle_versions.insert().values(
status="applied",
content_json=json.dumps(files, sort_keys=True),
content_hash=_hash_content(files),
based_on_version=None,
restored_from_version=None,
restored_file=None,
created_at=_now_iso(),
created_by="bootstrap",
created_by_capability=OPERATIONS_CAPABILITY,
reviewed_by=None,
reviewed_at=None,
review_note=None,
diff_json=None,
dry_run_json=None,
)
)
def _active_version_row(conn) -> dict[str, Any] | None:
row = (
conn.execute(
select(config_bundle_versions)
.where(config_bundle_versions.c.status == "applied")
.order_by(config_bundle_versions.c.version_id.desc())
.limit(1)
)
.mappings()
.first()
)
return dict(row) if row else None
def _version_row(conn, version_id: int) -> dict[str, Any]:
row = (
conn.execute(
select(config_bundle_versions).where(
config_bundle_versions.c.version_id == version_id
)
)
.mappings()
.first()
)
if row is None:
raise VersionNotFoundError(f"no config version {version_id!r}")
return dict(row)
# ---------------------------------------------------------------------------
# Public read API
# ---------------------------------------------------------------------------
def get_active_version() -> dict[str, Any]:
"""The current active bundle version, with its content."""
engine = get_app_engine()
with engine.begin() as conn:
_ensure_bootstrapped(conn)
row = _active_version_row(conn)
assert row is not None # _ensure_bootstrapped guarantees at least one applied row
return _public(row, include_content=True)
_version_id_lock = threading.Lock()
_cached_version_id: int | None = None
_cached_version_id_at: float = 0.0
#: Which application database :data:`_cached_version_id` was read from.
#: Without this the cache is a process-global keyed on nothing, and a
#: version id read from one database is served for another the moment
#: ``APP_DB_URL`` is re-pointed -- which is exactly what a test fixture
#: does between cases, so the failure shows up as a suite that passes
#: alone and fails in order.
_cached_version_id_source: str | None = None
def invalidate_active_version_id_cache() -> None:
"""Throw away the cached active version id.
Called by every function in this module that applies a version, so a
change is live in this process on the very next request rather than
after ``config_version_cache_ttl_seconds`` elapses.
"""
global _cached_version_id, _cached_version_id_at, _cached_version_id_source
with _version_id_lock:
_cached_version_id = None
_cached_version_id_at = 0.0
_cached_version_id_source = None
def get_active_version_id() -> int:
"""The current active bundle version's ``version_id`` alone -- the
"configuration version identifier" :mod:`prompt_engine.static_prefix`'s
``prefix_version_for_config`` derives the prefix version from (spec
§6).
This one is on the per-request path: ``api/runner.py`` folds it into
every query-cache key, so it reads a single integer column rather than
:func:`get_active_version`'s full nine-file snapshot, and caches it
for ``cfg.settings.config_version_cache_ttl_seconds`` -- the same
short-TTL-plus-explicit-invalidation shape :mod:`appdb.key_store` uses,
for the same reason.
"""
global _cached_version_id, _cached_version_id_at, _cached_version_id_source
ttl = cfg.settings.config_version_cache_ttl_seconds
source = cfg.settings.app_db_url
now = time.monotonic()
with _version_id_lock:
if (
_cached_version_id is not None
and _cached_version_id_source == source
and ttl > 0
and now - _cached_version_id_at < ttl
):
return _cached_version_id
engine = get_app_engine()
with engine.begin() as conn:
_ensure_bootstrapped(conn)
version_id = conn.execute(
select(config_bundle_versions.c.version_id)
.where(config_bundle_versions.c.status == "applied")
.order_by(config_bundle_versions.c.version_id.desc())
.limit(1)
).scalar()
assert version_id is not None # _ensure_bootstrapped guarantees one applied row
with _version_id_lock:
_cached_version_id = int(version_id)
_cached_version_id_at = now
_cached_version_id_source = source
return int(version_id)
def get_version(version_id: int) -> dict[str, Any]:
"""One version's full metadata and content.
Raises
------
VersionNotFoundError
"""
engine = get_app_engine()
with engine.begin() as conn:
row = _version_row(conn, version_id)
return _public(row, include_content=True)
def list_versions() -> list[dict[str, Any]]:
"""Every version's metadata (never content), newest first."""
engine = get_app_engine()
with engine.begin() as conn:
_ensure_bootstrapped(conn)
rows = (
conn.execute(
select(config_bundle_versions).order_by(
config_bundle_versions.c.version_id.desc()
)
)
.mappings()
.all()
)
return [_public(dict(row)) for row in rows]
# ---------------------------------------------------------------------------
# Validation, diff, dry-run (spec §4, §5)
# ---------------------------------------------------------------------------
def _validate_bundle(files: dict[str, str]) -> None:
"""Parse and validate every one of :data:`CONFIG_FILENAMES` through the
real loaders' in-memory text-validation entry points (see
:func:`_text_validators`) -- never against the live
``project_config/`` on disk, and never via
:func:`config.override_settings` (thread-unsafe: see
:func:`knowledge.config_loader.validate_yaml_text`'s docstring), so a
bad candidate never disturbs the active configuration and this
function is safe to call from a concurrently-served admin request.
Raises
------
ConfigVersionValidationError
Carrying every failing file's own loader error.
"""
errors: list[str] = []
validators = _text_validators()
for filename in CONFIG_FILENAMES:
try:
validators[filename](files.get(filename, ""))
except ValueError as exc:
errors.append(str(exc))
if errors:
raise ConfigVersionValidationError(errors)
def _schema_table_columns(schema_yaml_text: str) -> dict[str, set[str]]:
"""``{table: {column, ...}}`` for the tables schema.yaml text describes
with a ``columns`` key -- i.e. the guard allowlist a candidate
``schema.yaml`` would produce, computed without touching
:mod:`schema_data.registry`'s process-lifetime cache at all."""
from schema_data.registry import SchemaConfig
if not schema_yaml_text.strip():
return {}
parsed = SchemaConfig.model_validate(safe_load_strict(schema_yaml_text) or {})
return {
name: set(table.columns or {})
for name, table in parsed.tables.items()
if table.columns
}
def _diff_schema(old_text: str, new_text: str) -> dict[str, Any]:
"""Which tables and columns enter and leave the allowlist (spec §4.3) --
never a text diff of the YAML."""
old_tables = _schema_table_columns(old_text)
new_tables = _schema_table_columns(new_text)
tables_changed_columns: dict[str, dict[str, list[str]]] = {}
for table in sorted(set(old_tables) & set(new_tables)):
added = sorted(new_tables[table] - old_tables[table])
removed = sorted(old_tables[table] - new_tables[table])
if added or removed:
tables_changed_columns[table] = {"columns_added": added, "columns_removed": removed}
return {
"tables_added": sorted(set(new_tables) - set(old_tables)),
"tables_removed": sorted(set(old_tables) - set(new_tables)),
"tables_changed_columns": tables_changed_columns,
}
def _diff_bundles(old: dict[str, str], new: dict[str, str]) -> dict[str, Any]:
files_changed = sorted(f for f in CONFIG_FILENAMES if old.get(f, "") != new.get(f, ""))
diff: dict[str, Any] = {"files_changed": files_changed}
if "schema.yaml" in files_changed:
diff["schema"] = _diff_schema(old.get("schema.yaml", ""), new.get("schema.yaml", ""))
return diff
def _dry_run_bundle(files: dict[str, str]) -> dict[str, Any]:
"""Run the golden set through the real, offline :mod:`eval` harness
(spec §5) -- never reimplemented, only called -- plus, when
``schema.yaml`` is present, cross-check every table the golden set's
reference SQL touches against the *candidate* allowlist.
Offline mode replays the golden set's own ``expected_sql`` (see
:mod:`eval.runner`'s docstring), so its accuracy figure is not a
measure of generation quality against this candidate's domain
knowledge -- what it *does* prove, every time, is that the harness
still runs end to end and that the golden set's own reference queries
still parse and pass the guard. The schema table cross-check below is
what actually catches "this candidate schema.yaml would make a
currently-working golden question fail."
"""
from eval.report import build_report
from eval.runner import load_golden_cases, make_offline_executor, make_offline_generator, run_golden_set
from security.sql_guard import extract_touched_tables
golden_path = Path(cfg.settings.eval_golden_path)
if not golden_path.exists():
return {"ran": False, "reason": f"golden set not found: {golden_path}"}
cases = load_golden_cases(golden_path)
generate_fn = make_offline_generator(cases)
execute_fn = make_offline_executor(cases)
report = build_report(run_golden_set(cases, generate_fn, execute_fn), mode="offline")
missing_tables: list[str] = []
schema_text = files.get("schema.yaml", "")
if schema_text.strip():
candidate_tables = set(_schema_table_columns(schema_text))
touched: set[str] = set()
for case in cases:
if case.expected_sql:
touched.update(extract_touched_tables(case.expected_sql))
missing_tables = sorted(touched - candidate_tables)
return {
"ran": True,
"mode": report.mode,
"total": report.total,
"passed": report.passed,
"accuracy_pct": report.accuracy_pct,
"guard_rejections": report.guard_rejections,
"status_counts": dict(report.status_counts),
"golden_set_tables_missing_from_candidate_schema": missing_tables,
}
# ---------------------------------------------------------------------------
# Cache invalidation on apply -- see module docstring's "What takes effect
# immediately, and what needs a restart".
# ---------------------------------------------------------------------------
def _invalidate_runtime_caches(files_changed: Sequence[str]) -> None:
invalidate_active_version_id_cache()
touched_prefix_relevant = False
for filename in files_changed:
module_name = _KNOWLEDGE_CACHE_MODULES.get(filename)
if module_name is None:
continue # schema.yaml (or an unrecognised name): left alone
module = importlib.import_module(module_name)
module._cache.clear()
if filename in _PREFIX_RELEVANT_FILES:
touched_prefix_relevant = True
if touched_prefix_relevant:
from prompt_engine.static_prefix import build_static_prefix, prefix_version
# static_prefix_token_estimate/should_use_static_prefix are plain
# functions that call build_static_prefix internally -- they carry
# no lru_cache of their own, so clearing these two is sufficient.
build_static_prefix.cache_clear()
prefix_version.cache_clear()
def _write_bundle_to_disk(files: dict[str, str]) -> None:
base_dir = Path(cfg.settings.project_config_dir)
base_dir.mkdir(parents=True, exist_ok=True)
for filename in CONFIG_FILENAMES:
(base_dir / filename).write_text(files.get(filename, ""), encoding="utf-8")
def _write_export(files: dict[str, str], version_id: int) -> None:
"""Write *files* to ``cfg.settings.config_export_dir``, when configured
(spec §7). A no-op when that setting is empty (the default)."""
export_dir = cfg.settings.config_export_dir
if not export_dir:
return
target = Path(export_dir)
target.mkdir(parents=True, exist_ok=True)
for filename in CONFIG_FILENAMES:
(target / filename).write_text(files.get(filename, ""), encoding="utf-8")
manifest = {"version_id": version_id, "exported_at": _now_iso()}
(target / ".config_export_manifest.json").write_text(
json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8"
)
# ---------------------------------------------------------------------------
# Write API
# ---------------------------------------------------------------------------
def propose_or_apply(
files: dict[str, str],
*,
based_on_version: int,
actor_principal_id: str,
actor_capabilities: frozenset[str] | set[str],
) -> dict[str, Any]:
"""Validate, diff and dry-run *files* against the active version, then
either apply it as a new version or -- if it touches ``schema.yaml``
and *actor_capabilities* lacks the security capability -- save it as an
unapplied draft for a security admin to review (spec §3.1).
*files* need not name every one of :data:`CONFIG_FILENAMES`: any
filename not given is carried forward unchanged from the version named
by *based_on_version*.
Parameters
----------
files:
``{filename: yaml_text}`` for the file(s) being changed.
based_on_version:
The ``version_id`` the caller had open (optimistic locking, spec
§8).
actor_principal_id, actor_capabilities:
Who is saving, and which capabilities they hold.
Returns
-------
dict
The newly created version's public representation (see
:func:`get_version`) -- its ``status`` tells the caller whether it
was applied immediately or saved as a draft.
Raises
------
StaleVersionError
*based_on_version* is not the current active version.
ConfigVersionValidationError
A file failed to parse/validate through its real loader.
ValueError
*files* names a filename outside :data:`CONFIG_FILENAMES`, or
``schema.yaml``'s structural invariants fail (spec §4 item 2).
"""
unknown = set(files) - set(CONFIG_FILENAMES)
if unknown:
raise ValueError(f"unknown config file(s): {sorted(unknown)}")
engine = get_app_engine()
with engine.begin() as conn:
_ensure_bootstrapped(conn)
active = _active_version_row(conn)
assert active is not None
if active["version_id"] != based_on_version:
raise StaleVersionError(
f"based_on_version={based_on_version} is no longer the "
f"active configuration -- version {active['version_id']} "
f"was created at {active['created_at']} by "
f"{active['created_by']!r}. Reload the current "
"configuration and re-apply your change on top of it."
)
base_content = _content_dict(active)
new_content = dict(base_content)
new_content.update(files)
_validate_bundle(new_content)
if "schema.yaml" in files and new_content["schema.yaml"].strip():
from schema_data.registry import SchemaConfig, check_allowlist_structural_invariants
parsed = SchemaConfig.model_validate(safe_load_strict(new_content["schema.yaml"]) or {})
table_columns = {
name: t.columns for name, t in parsed.tables.items() if t.columns
}
table_descriptions = {name: t.description for name, t in parsed.tables.items()}
relationships = {
f"{r.from_table} -> {r.to_table}": r.join_sql for r in parsed.relationships
}
violations = check_allowlist_structural_invariants(
table_columns, table_descriptions, relationships
)
if violations:
raise ValueError(
"schema.yaml failed the guard allowlist's structural "
f"invariants: {'; '.join(violations)}"
)
diff = _diff_bundles(base_content, new_content)
dry_run = _dry_run_bundle(new_content)
# The role split (spec §3) turns on whether any SECURITY_ONLY_FILES
# entry actually changed -- not merely whether one was named in
# *files* (a caller may pass a filename whose content happens to be
# byte-identical to the active version's).
schema_changed = bool(SECURITY_ONLY_FILES & set(diff["files_changed"]))
if schema_changed and SECURITY_CAPABILITY not in actor_capabilities:
status = "draft"
created_by_capability = OPERATIONS_CAPABILITY
else:
status = "applied"
created_by_capability = (
SECURITY_CAPABILITY if schema_changed else OPERATIONS_CAPABILITY
)
now = _now_iso()
result = conn.execute(
config_bundle_versions.insert().values(
status=status,
content_json=json.dumps(new_content, sort_keys=True),
content_hash=_hash_content(new_content),
based_on_version=based_on_version,
restored_from_version=None,
restored_file=None,
created_at=now,
created_by=actor_principal_id,
created_by_capability=created_by_capability,
reviewed_by=None,
reviewed_at=None,
review_note=None,
diff_json=json.dumps(diff, sort_keys=True),
dry_run_json=json.dumps(dry_run, sort_keys=True),
)
)
version_id = result.inserted_primary_key[0]
if status == "applied":
_write_bundle_to_disk(new_content)
_write_export(new_content, version_id)
_invalidate_runtime_caches(diff["files_changed"])
return get_version(version_id)
def _stamp_restore_provenance(
version_id: int, from_version_id: int, filename: str | None
) -> None:
engine = get_app_engine()
with engine.begin() as conn:
conn.execute(
config_bundle_versions.update()
.where(config_bundle_versions.c.version_id == version_id)
.values(restored_from_version=from_version_id, restored_file=filename)
)
def restore(
*,
from_version_id: int,
filename: str | None,
actor_principal_id: str,
actor_capabilities: frozenset[str] | set[str],
) -> dict[str, Any]:
"""Restore *filename* (or, when ``None``, the whole bundle) from
``from_version_id`` -- spec §2's "revert, never reset": this creates a
brand-new version and never touches ``from_version_id`` or any version
in between it and the current active one.
Goes through the exact same :func:`propose_or_apply` path as any other
edit (optimistic locking against the *current* active version,
validation, diff, dry-run, and the same operations/security split for
a ``schema.yaml`` restore) -- a rollback is an edit like any other and
gets the same treatment (spec §4).
Raises
------
VersionNotFoundError
*from_version_id* does not exist.
ValueError
*filename* is not one of :data:`CONFIG_FILENAMES`.
"""
if filename is not None and filename not in CONFIG_FILENAMES:
raise ValueError(f"unknown config file: {filename!r}")
engine = get_app_engine()
with engine.begin() as conn:
_ensure_bootstrapped(conn)
active = _active_version_row(conn)
assert active is not None
source = _version_row(conn, from_version_id)
source_content = _content_dict(source)
files = dict(source_content) if filename is None else {filename: source_content.get(filename, "")}
result = propose_or_apply(
files,
based_on_version=active["version_id"],
actor_principal_id=actor_principal_id,
actor_capabilities=actor_capabilities,
)
_stamp_restore_provenance(result["version_id"], from_version_id, filename)
return get_version(result["version_id"])
def approve_draft(version_id: int, *, actor_principal_id: str) -> dict[str, Any]:
"""Promote a pending draft to a real, applied version (spec §3.1).
Raises
------
VersionNotFoundError
NotADraftError
*version_id*'s status is not ``"draft"``.
StaleVersionError
The configuration has moved on since this draft was proposed.
"""
engine = get_app_engine()
with engine.begin() as conn:
draft = _version_row(conn, version_id)
if draft["status"] != "draft":
raise NotADraftError(
f"version {version_id} is not a pending draft (status={draft['status']!r})"
)
active = _active_version_row(conn)
assert active is not None
if active["version_id"] != draft["based_on_version"]:
raise StaleVersionError(
f"this draft was based on version {draft['based_on_version']}, "
f"but version {active['version_id']} is now the active "
f"configuration (created {active['created_at']} by "
f"{active['created_by']!r}). Reject this draft and re-propose "
"the change on top of the current configuration."
)
now = _now_iso()
conn.execute(
config_bundle_versions.update()
.where(config_bundle_versions.c.version_id == version_id)
.values(status="applied", reviewed_by=actor_principal_id, reviewed_at=now)
)
content = _content_dict(draft)
files_changed = json.loads(draft["diff_json"])["files_changed"] if draft["diff_json"] else []
_write_bundle_to_disk(content)
_write_export(content, version_id)
_invalidate_runtime_caches(files_changed)
return get_version(version_id)
def reject_draft(version_id: int, *, actor_principal_id: str, reason: str) -> dict[str, Any]:
"""Reject a pending draft -- it never becomes active, and the row is
kept (never deleted) recording who rejected it, when, and why (spec
§3.1).
Raises
------
VersionNotFoundError
NotADraftError
"""
engine = get_app_engine()
with engine.begin() as conn:
draft = _version_row(conn, version_id)
if draft["status"] != "draft":
raise NotADraftError(
f"version {version_id} is not a pending draft (status={draft['status']!r})"
)
conn.execute(
config_bundle_versions.update()
.where(config_bundle_versions.c.version_id == version_id)
.values(
status="rejected",
reviewed_by=actor_principal_id,
reviewed_at=_now_iso(),
review_note=reason,
)
)
return get_version(version_id)
# ---------------------------------------------------------------------------
# Export round trip (spec §7, §10)
# ---------------------------------------------------------------------------
def export_active_version(target_dir: str | Path) -> dict[str, Any]:
"""Write the current active version's bundle to *target_dir*, on
demand -- the same shape :func:`_write_export` writes automatically to
``cfg.settings.config_export_dir`` on every apply, usable for an
explicit "export now" action or to hand to
:func:`import_bundle_from_directory` when setting up a second
deployment."""
active = get_active_version()
files = active["files"]
target = Path(target_dir)
target.mkdir(parents=True, exist_ok=True)
for filename in CONFIG_FILENAMES:
(target / filename).write_text(files.get(filename, ""), encoding="utf-8")
manifest = {"version_id": active["version_id"], "exported_at": _now_iso()}
(target / ".config_export_manifest.json").write_text(
json.dumps(manifest, indent=2, sort_keys=True), encoding="utf-8"
)
return manifest
def import_bundle_from_directory(
source_dir: str | Path,
*,
actor_principal_id: str,
actor_capabilities: frozenset[str] | set[str],
) -> dict[str, Any]:
"""Import a previously exported bundle (see :func:`export_active_version`)
from *source_dir*.
This is a configuration change like any other, and takes exactly the
same route as one: :func:`propose_or_apply`, so the bundle is parsed
and validated through the real loaders, diffed, dry-run against the
golden set, and -- when it changes ``schema.yaml`` and the caller holds
only the operations capability -- held as a draft rather than applied.
An earlier shape seeded a bundle *unvalidated* when the deployment had
no version history yet, on the reasoning that a first version has
nothing to be validated against. That reasoning does not hold: the
checks that matter here are not comparisons against a previous version
but properties of the bundle itself -- does it parse, does
``schema.yaml`` still describe a coherent allowlist, does the golden
set still resolve against it -- and none of them need a predecessor.
It also made import the one way into a new version that skipped the
role split, in exactly the window (a fresh deployment, before anyone
has opened the configuration page) where the first admin action is most
likely to be an import. So the history is bootstrapped from disk first,
and the import is then proposed on top of it.
"""
source = Path(source_dir)
files = {
filename: (source / filename).read_text(encoding="utf-8")
if (source / filename).exists() else ""
for filename in CONFIG_FILENAMES
}
active = get_active_version() # bootstraps from disk if nothing exists yet
return propose_or_apply(
files,
based_on_version=active["version_id"],
actor_principal_id=actor_principal_id,
actor_capabilities=actor_capabilities,
)