feat: resolve accountability ownership reviews

This commit is contained in:
2026-05-24 09:53:44 +02:00
parent c27d71a511
commit a55f1a45d6
7 changed files with 592 additions and 3 deletions

View File

@@ -39,6 +39,12 @@ Identity projection schema:
schemas/accountability-identity-projection.schema.yaml schemas/accountability-identity-projection.schema.yaml
``` ```
Ownership review schema:
```text
schemas/accountability-ownership-review.schema.yaml
```
## Required Sections ## Required Sections
- `netkingdom`: root id, name, and king actor. - `netkingdom`: root id, name, and king actor.
@@ -104,3 +110,35 @@ railiance-fabric discover-roots \
The store is intentionally separate from accepted registry graph snapshots. It The store is intentionally separate from accepted registry graph snapshots. It
keeps raw evidence runs, evidence items, and identity candidates available for keeps raw evidence runs, evidence items, and identity candidates available for
inspection before any candidate is promoted. inspection before any candidate is promoted.
## Ownership Review
To resolve ownership and containment from the normalized identities:
```bash
railiance-fabric discover-roots \
--ownership-review \
--store-db .railiance-fabric/accountability-evidence.sqlite3
```
The ownership review inherits owners from fabric/subfabric containment when
possible, applies explicit owner evidence from discovery roots, and marks
unresolved or ambiguous candidates as `needs_review`. Accepted candidates must
have a resolved owner and containment unless they are actors or the netkingdom
root.
To persist a reviewer decision for a stable identity candidate:
```bash
railiance-fabric review-identity identity:repository:example-repo \
--store-db .railiance-fabric/accountability-evidence.sqlite3 \
--decision accept \
--owner-actor-id actor.railiance.primary-lord \
--fabric-id fabric.railiance.primary \
--reviewer operator \
--note "accepted from reviewed checkout evidence"
```
Reviewer decisions are keyed by stable identity key. Later rescans apply the
latest decision for that key, so ordinary evidence refreshes do not lose
reviewed ownership choices.

View File

@@ -63,6 +63,19 @@ railiance-fabric discover-roots --identity-projection
railiance-fabric discover-roots --store-db .railiance-fabric/accountability-evidence.sqlite3 railiance-fabric discover-roots --store-db .railiance-fabric/accountability-evidence.sqlite3
``` ```
To inspect ownership blockers and apply review decisions:
```bash
railiance-fabric discover-roots --ownership-review \
--store-db .railiance-fabric/accountability-evidence.sqlite3
railiance-fabric review-identity <stable-key> \
--store-db .railiance-fabric/accountability-evidence.sqlite3 \
--decision accept \
--owner-actor-id actor.railiance.primary-lord \
--fabric-id fabric.railiance.primary
```
The financial export must satisfy these invariants: The financial export must satisfy these invariants:
- every accepted node has resolvable ownership; - every accepted node has resolvable ownership;

View File

@@ -194,6 +194,58 @@ def build_identity_projection(
return projection return projection
def build_ownership_review(
identity_projection: dict[str, Any],
manifest: dict[str, Any],
*,
review_decisions: dict[str, dict[str, Any]] | None = None,
) -> dict[str, Any]:
review_decisions = review_decisions or {}
actor_roles = {
actor.get("id"): actor.get("role", "")
for actor in manifest.get("actors", [])
if isinstance(actor, dict)
}
fabric_owners: dict[str, str] = {}
fabric_kinds: dict[str, str] = {}
for fabric in manifest.get("fabrics", []):
if not isinstance(fabric, dict):
continue
fabric_id = str(fabric.get("id") or "")
fabric_kinds[fabric_id] = str(fabric.get("kind") or "")
fabric_owners[fabric_id] = str(fabric.get("tenant_actor_id") or fabric.get("lord_actor_id") or "")
items = [
_ownership_item(candidate, actor_roles, fabric_owners, fabric_kinds, review_decisions.get(candidate["stable_key"]))
for candidate in identity_projection.get("identity_candidates", [])
if isinstance(candidate, dict)
]
review = {
"apiVersion": "railiance.fabric/v1alpha2",
"kind": "AccountabilityOwnershipReview",
"generated_at": _utc_now(),
"evidence_run": identity_projection.get("evidence_run", {}),
"items": sorted(items, key=lambda item: item["stable_key"]),
"summary": {
"total": len(items),
"accepted": sum(1 for item in items if item["review_state"] == "accepted"),
"needs_review": sum(1 for item in items if item["review_state"] == "needs_review"),
"unresolved_ownership": sum(
1 for item in items if item["ownership"]["resolution"] == "unresolved"
),
"ambiguous_containment": sum(
1 for item in items if item["containment"]["status"] == "ambiguous"
),
},
}
validator = draft202012_validator(repo_root() / "schemas" / "accountability-ownership-review.schema.yaml")
errors = sorted(validator.iter_errors(review), key=lambda error: list(error.path))
if errors:
location = ".".join(str(part) for part in errors[0].path) or "<root>"
raise ValueError(f"invalid accountability ownership review at {location}: {errors[0].message}")
return review
@dataclass(frozen=True) @dataclass(frozen=True)
class AccountabilityEvidenceStore: class AccountabilityEvidenceStore:
path: Path path: Path
@@ -252,6 +304,21 @@ class AccountabilityEvidenceStore:
create index if not exists idx_accountability_identity_candidates_run create index if not exists idx_accountability_identity_candidates_run
on accountability_identity_candidates(run_id); on accountability_identity_candidates(run_id);
create table if not exists accountability_review_decisions (
id integer primary key autoincrement,
stable_key text not null,
decision text not null,
reviewer text not null,
owner_actor_id text,
fabric_id text,
subfabric_id text,
note text,
created_at text not null
);
create index if not exists idx_accountability_review_decisions_stable_key
on accountability_review_decisions(stable_key, id desc);
""" """
) )
@@ -350,6 +417,59 @@ class AccountabilityEvidenceStore:
).fetchone() ).fetchone()
return dict(row) if row else None return dict(row) if row else None
def add_review_decision(
self,
*,
stable_key: str,
decision: str,
reviewer: str,
owner_actor_id: str = "",
fabric_id: str = "",
subfabric_id: str = "",
note: str = "",
) -> dict[str, Any]:
self.init_schema()
created_at = _utc_now()
with self._connect() as db:
cursor = db.execute(
"""
insert into accountability_review_decisions (
stable_key, decision, reviewer, owner_actor_id, fabric_id,
subfabric_id, note, created_at
) values (?, ?, ?, ?, ?, ?, ?, ?)
""",
(stable_key, decision, reviewer, owner_actor_id, fabric_id, subfabric_id, note, created_at),
)
decision_id = int(cursor.lastrowid)
return {
"id": decision_id,
"stable_key": stable_key,
"decision": decision,
"reviewer": reviewer,
"owner_actor_id": owner_actor_id,
"fabric_id": fabric_id,
"subfabric_id": subfabric_id,
"note": note,
"created_at": created_at,
}
def latest_review_decisions(self) -> dict[str, dict[str, Any]]:
self.init_schema()
with self._connect() as db:
rows = db.execute(
"""
select *
from accountability_review_decisions
where id in (
select max(id)
from accountability_review_decisions
group by stable_key
)
order by stable_key
"""
).fetchall()
return {row["stable_key"]: dict(row) for row in rows}
def list_evidence(self, run_id: int) -> list[dict[str, Any]]: def list_evidence(self, run_id: int) -> list[dict[str, Any]]:
with self._connect() as db: with self._connect() as db:
rows = db.execute( rows = db.execute(
@@ -491,6 +611,136 @@ def _identity_from_evidence(root: dict[str, Any], item: dict[str, Any]) -> dict[
return None return None
def _ownership_item(
candidate: dict[str, Any],
actor_roles: dict[str, str],
fabric_owners: dict[str, str],
fabric_kinds: dict[str, str],
decision: dict[str, Any] | None,
) -> dict[str, Any]:
attributes = candidate.get("attributes") if isinstance(candidate.get("attributes"), dict) else {}
blockers: list[str] = []
fabric_id = str(candidate.get("fabric_id") or "")
subfabric_id = str(candidate.get("subfabric_id") or "")
owner_actor_id = str(candidate.get("owner_actor_id") or "")
resolution = "explicit" if owner_actor_id else "unresolved"
inherited_from = ""
if attributes.get("ambiguous_aliases"):
blockers.append("ambiguous_identity")
if attributes.get("ambiguous_owner_actor_ids"):
blockers.append("ambiguous_ownership")
if candidate.get("identity_type") not in {"Actor", "Netkingdom"}:
if not fabric_id:
blockers.append("containment_unresolved")
elif fabric_id not in fabric_kinds:
blockers.append("unknown_fabric")
if subfabric_id and subfabric_id not in fabric_kinds:
blockers.append("unknown_subfabric")
if not owner_actor_id and subfabric_id and fabric_owners.get(subfabric_id):
owner_actor_id = fabric_owners[subfabric_id]
resolution = "inherited"
inherited_from = subfabric_id
if not owner_actor_id and fabric_id and fabric_owners.get(fabric_id):
owner_actor_id = fabric_owners[fabric_id]
resolution = "inherited"
inherited_from = fabric_id
if candidate.get("identity_type") == "Actor" and not owner_actor_id:
owner_actor_id = str(candidate.get("graph_id") or "")
resolution = "explicit" if owner_actor_id else "unresolved"
decision_payload: dict[str, Any] | None = None
if decision:
decision_payload = {
"decision": decision.get("decision", ""),
"reviewer": decision.get("reviewer", ""),
"note": decision.get("note", ""),
"created_at": decision.get("created_at", ""),
}
if decision.get("fabric_id"):
fabric_id = str(decision["fabric_id"])
if decision.get("subfabric_id"):
subfabric_id = str(decision["subfabric_id"])
if decision.get("owner_actor_id"):
owner_actor_id = str(decision["owner_actor_id"])
resolution = "review_decision"
inherited_from = ""
if decision.get("decision") == "accept":
blockers = [
blocker
for blocker in blockers
if blocker
not in {
"ambiguous_identity",
"ambiguous_ownership",
"containment_unresolved",
"unknown_fabric",
"unknown_subfabric",
"ownership_unresolved",
"unknown_owner_actor",
}
]
if candidate.get("identity_type") not in {"Actor", "Netkingdom"}:
if not fabric_id:
blockers.append("containment_unresolved")
elif fabric_id not in fabric_kinds:
blockers.append("unknown_fabric")
if subfabric_id and subfabric_id not in fabric_kinds:
blockers.append("unknown_subfabric")
if not owner_actor_id:
blockers.append("ownership_unresolved")
resolution = "unresolved"
if owner_actor_id and owner_actor_id not in actor_roles:
blockers.append("unknown_owner_actor")
review_state = "candidate"
if blockers:
review_state = "needs_review"
if decision:
if decision.get("decision") == "reject":
review_state = "rejected"
elif decision.get("decision") == "needs_review":
review_state = "needs_review"
elif decision.get("decision") == "accept" and owner_actor_id and not blockers:
review_state = "accepted"
elif decision.get("decision") == "accept":
review_state = "needs_review"
blockers.append("accepted_without_resolved_owner_or_containment")
containment_status = "resolved"
if "containment_unresolved" in blockers:
containment_status = "unresolved"
elif "unknown_fabric" in blockers or "unknown_subfabric" in blockers:
containment_status = "ambiguous"
item = {
"stable_key": candidate["stable_key"],
"identity_type": candidate["identity_type"],
"label": candidate["label"],
"review_state": review_state,
"ownership": {
"owner_actor_id": owner_actor_id,
"owner_role": actor_roles.get(owner_actor_id, ""),
"resolution": resolution,
},
"containment": {
"fabric_id": fabric_id,
"subfabric_id": subfabric_id,
"status": containment_status,
},
"blockers": _unique_strings(blockers),
"evidence_ids": candidate.get("evidence_ids", []),
}
if inherited_from:
item["ownership"]["inherited_from"] = inherited_from
if decision_payload:
item["decision"] = decision_payload
return item
def _add_identity_candidate( def _add_identity_candidate(
candidates: dict[str, dict[str, Any]], candidates: dict[str, dict[str, Any]],
*, *,
@@ -535,6 +785,11 @@ def _add_identity_candidate(
existing["aliases"] = _unique_strings([*existing.get("aliases", []), *incoming["aliases"]]) existing["aliases"] = _unique_strings([*existing.get("aliases", []), *incoming["aliases"]])
existing["evidence_ids"] = _unique_strings([*existing.get("evidence_ids", []), *incoming["evidence_ids"]]) existing["evidence_ids"] = _unique_strings([*existing.get("evidence_ids", []), *incoming["evidence_ids"]])
existing["attributes"] = {**existing.get("attributes", {}), **incoming["attributes"]} existing["attributes"] = {**existing.get("attributes", {}), **incoming["attributes"]}
if incoming.get("owner_actor_id") and existing.get("owner_actor_id") and incoming["owner_actor_id"] != existing["owner_actor_id"]:
existing["attributes"]["ambiguous_owner_actor_ids"] = _unique_strings(
[existing["owner_actor_id"], incoming["owner_actor_id"], *existing["attributes"].get("ambiguous_owner_actor_ids", [])]
)
existing["review_state"] = "needs_review"
for key in ("fabric_id", "subfabric_id", "owner_actor_id", "graph_id"): for key in ("fabric_id", "subfabric_id", "owner_actor_id", "graph_id"):
if incoming.get(key) and not existing.get(key): if incoming.get(key) and not existing.get(key):
existing[key] = incoming[key] existing[key] = incoming[key]

View File

@@ -18,6 +18,7 @@ from .accountability_roots import (
DEFAULT_ROOT_MANIFEST_PATH, DEFAULT_ROOT_MANIFEST_PATH,
AccountabilityEvidenceStore, AccountabilityEvidenceStore,
build_identity_projection, build_identity_projection,
build_ownership_review,
collect_accountability_root_evidence, collect_accountability_root_evidence,
load_accountability_root_manifest, load_accountability_root_manifest,
) )
@@ -123,8 +124,22 @@ def build_parser() -> argparse.ArgumentParser:
discover_roots.add_argument("--include-remote", action="store_true", help="Allow HTTP reads from configured remote roots.") discover_roots.add_argument("--include-remote", action="store_true", help="Allow HTTP reads from configured remote roots.")
discover_roots.add_argument("--max-items-per-root", type=int, default=200) discover_roots.add_argument("--max-items-per-root", type=int, default=200)
discover_roots.add_argument("--identity-projection", action="store_true", help="Print normalized identity candidates instead of raw evidence.") discover_roots.add_argument("--identity-projection", action="store_true", help="Print normalized identity candidates instead of raw evidence.")
discover_roots.add_argument("--ownership-review", action="store_true", help="Print ownership resolution and review blockers.")
discover_roots.add_argument("--store-db", type=Path, default=None, help="Persist evidence and identity candidates in a SQLite store.") discover_roots.add_argument("--store-db", type=Path, default=None, help="Persist evidence and identity candidates in a SQLite store.")
review_identity = sub.add_parser(
"review-identity",
help="Persist a review decision for a stable accountability identity candidate.",
)
review_identity.add_argument("stable_key")
review_identity.add_argument("--store-db", type=Path, required=True)
review_identity.add_argument("--decision", choices=["accept", "needs_review", "reject"], required=True)
review_identity.add_argument("--owner-actor-id", default="")
review_identity.add_argument("--fabric-id", default="")
review_identity.add_argument("--subfabric-id", default="")
review_identity.add_argument("--reviewer", default="operator")
review_identity.add_argument("--note", default="")
registry = sub.add_parser("registry", help="Feed a running Railiance Fabric registry service.") registry = sub.add_parser("registry", help="Feed a running Railiance Fabric registry service.")
registry_sub = registry.add_subparsers(dest="registry_command", required=True) registry_sub = registry.add_subparsers(dest="registry_command", required=True)
@@ -345,10 +360,26 @@ def main(argv: list[str] | None = None) -> int:
max_items_per_root=args.max_items_per_root, max_items_per_root=args.max_items_per_root,
) )
projection = build_identity_projection(payload, manifest) projection = build_identity_projection(payload, manifest)
store = AccountabilityEvidenceStore(args.store_db) if args.store_db else None
decisions = store.latest_review_decisions() if store else {}
ownership_review = build_ownership_review(projection, manifest, review_decisions=decisions)
if args.store_db: if args.store_db:
store = AccountabilityEvidenceStore(args.store_db)
store.add_evidence_run(payload, projection) store.add_evidence_run(payload, projection)
print(json.dumps(projection if args.identity_projection else payload, indent=2, sort_keys=True)) output = ownership_review if args.ownership_review else projection if args.identity_projection else payload
print(json.dumps(output, indent=2, sort_keys=True))
return 0
if args.command == "review-identity":
decision = AccountabilityEvidenceStore(args.store_db).add_review_decision(
stable_key=args.stable_key,
decision=args.decision,
reviewer=args.reviewer,
owner_actor_id=args.owner_actor_id,
fabric_id=args.fabric_id,
subfabric_id=args.subfabric_id,
note=args.note,
)
print(json.dumps(decision, indent=2, sort_keys=True))
return 0 return 0
if args.command == "registry": if args.command == "registry":

View File

@@ -0,0 +1,149 @@
$schema: "https://json-schema.org/draft/2020-12/schema"
$id: "https://railiance.local/fabric/schemas/accountability-ownership-review.schema.yaml"
title: "AccountabilityOwnershipReview"
type: object
additionalProperties: false
required:
- apiVersion
- kind
- generated_at
- evidence_run
- items
- summary
properties:
apiVersion:
type: string
const: "railiance.fabric/v1alpha2"
kind:
type: string
const: AccountabilityOwnershipReview
generated_at:
type: string
format: date-time
evidence_run:
type: object
additionalProperties: true
items:
type: array
items:
$ref: "#/$defs/reviewItem"
summary:
type: object
additionalProperties: false
required:
- total
- accepted
- needs_review
- unresolved_ownership
- ambiguous_containment
properties:
total:
type: integer
minimum: 0
accepted:
type: integer
minimum: 0
needs_review:
type: integer
minimum: 0
unresolved_ownership:
type: integer
minimum: 0
ambiguous_containment:
type: integer
minimum: 0
$defs:
reviewItem:
type: object
additionalProperties: false
required:
- stable_key
- identity_type
- label
- review_state
- ownership
- containment
- blockers
- evidence_ids
properties:
stable_key:
type: string
minLength: 3
identity_type:
type: string
minLength: 1
label:
type: string
minLength: 1
review_state:
type: string
enum:
- candidate
- accepted
- needs_review
- rejected
ownership:
type: object
additionalProperties: false
required:
- owner_actor_id
- owner_role
- resolution
properties:
owner_actor_id:
type: string
owner_role:
type: string
resolution:
type: string
enum:
- explicit
- inherited
- review_decision
- unresolved
inherited_from:
type: string
containment:
type: object
additionalProperties: false
required:
- fabric_id
- subfabric_id
- status
properties:
fabric_id:
type: string
subfabric_id:
type: string
status:
type: string
enum:
- resolved
- ambiguous
- unresolved
blockers:
type: array
items:
type: string
evidence_ids:
type: array
items:
type: string
decision:
type: object
additionalProperties: false
required:
- decision
- reviewer
- note
- created_at
properties:
decision:
type: string
reviewer:
type: string
note:
type: string
created_at:
type: string

View File

@@ -4,6 +4,7 @@ from pathlib import Path
from railiance_fabric.accountability_roots import ( from railiance_fabric.accountability_roots import (
AccountabilityEvidenceStore, AccountabilityEvidenceStore,
build_identity_projection, build_identity_projection,
build_ownership_review,
collect_accountability_root_evidence, collect_accountability_root_evidence,
load_accountability_root_manifest, load_accountability_root_manifest,
) )
@@ -77,6 +78,40 @@ def test_evidence_store_persists_runs_items_and_identities(tmp_path: Path) -> No
assert stored["identity_candidate_count"] == len(store.list_identity_candidates(stored["run_id"])) assert stored["identity_candidate_count"] == len(store.list_identity_candidates(stored["run_id"]))
def test_ownership_review_flags_ambiguity_and_applies_review_decisions(tmp_path: Path) -> None:
manifest_path = _fixture_manifest(tmp_path)
manifest = load_accountability_root_manifest(manifest_path)
projection = build_identity_projection(collect_accountability_root_evidence(manifest_path), manifest)
review = build_ownership_review(projection, manifest)
validator = draft202012_validator(Path("schemas/accountability-ownership-review.schema.yaml"))
assert list(validator.iter_errors(review)) == []
repo_key = "identity:repository:fixture-repo"
repo_item = next(item for item in review["items"] if item["stable_key"] == repo_key)
assert repo_item["review_state"] == "needs_review"
assert "ambiguous_ownership" in repo_item["blockers"]
store = AccountabilityEvidenceStore(tmp_path / "evidence.sqlite3")
store.add_review_decision(
stable_key=repo_key,
decision="accept",
reviewer="tester",
owner_actor_id="actor.fixture.lord",
fabric_id="fabric.fixture.primary",
note="fixture checkout owner wins over registry root",
)
accepted_review = build_ownership_review(
projection,
manifest,
review_decisions=store.latest_review_decisions(),
)
accepted_item = next(item for item in accepted_review["items"] if item["stable_key"] == repo_key)
assert accepted_item["review_state"] == "accepted"
assert accepted_item["ownership"]["resolution"] == "review_decision"
assert accepted_item["ownership"]["owner_actor_id"] == "actor.fixture.lord"
def test_discover_roots_cli_prints_evidence_json(tmp_path: Path, capsys) -> None: def test_discover_roots_cli_prints_evidence_json(tmp_path: Path, capsys) -> None:
manifest = _fixture_manifest(tmp_path) manifest = _fixture_manifest(tmp_path)
@@ -110,6 +145,52 @@ def test_discover_roots_cli_can_print_identities_and_store(tmp_path: Path, capsy
assert AccountabilityEvidenceStore(store_path).latest_run() is not None assert AccountabilityEvidenceStore(store_path).latest_run() is not None
def test_review_identity_cli_persists_decision_for_ownership_review(tmp_path: Path, capsys) -> None:
manifest = _fixture_manifest(tmp_path)
store_path = tmp_path / "evidence.sqlite3"
repo_key = "identity:repository:fixture-repo"
assert (
cli_main(
[
"review-identity",
repo_key,
"--store-db",
str(store_path),
"--decision",
"accept",
"--owner-actor-id",
"actor.fixture.lord",
"--fabric-id",
"fabric.fixture.primary",
"--reviewer",
"tester",
]
)
== 0
)
decision_payload = json.loads(capsys.readouterr().out)
assert decision_payload["stable_key"] == repo_key
assert (
cli_main(
[
"discover-roots",
"--manifest",
str(manifest),
"--ownership-review",
"--store-db",
str(store_path),
]
)
== 0
)
review_payload = json.loads(capsys.readouterr().out)
repo_item = next(item for item in review_payload["items"] if item["stable_key"] == repo_key)
assert repo_item["review_state"] == "accepted"
assert repo_item["decision"]["reviewer"] == "tester"
def _fixture_manifest(tmp_path: Path) -> Path: def _fixture_manifest(tmp_path: Path) -> Path:
workspace = tmp_path / "workspace" workspace = tmp_path / "workspace"
repo = workspace / "fixture-repo" repo = workspace / "fixture-repo"

View File

@@ -183,7 +183,7 @@ Result:
```task ```task
id: RAIL-FAB-WP-0018-T04 id: RAIL-FAB-WP-0018-T04
status: todo status: done
priority: high priority: high
state_hub_task_id: "670be2c2-6bec-4534-ae6a-ab0186ce0a8d" state_hub_task_id: "670be2c2-6bec-4534-ae6a-ab0186ce0a8d"
``` ```
@@ -205,6 +205,28 @@ Done when:
- unresolved or ambiguous nodes are visible before promotion; - unresolved or ambiguous nodes are visible before promotion;
- review decisions survive ordinary rescans. - review decisions survive ordinary rescans.
Result:
- Added `schemas/accountability-ownership-review.schema.yaml` for ownership
resolution and review-blocker payloads.
- Added `build_ownership_review()` to resolve explicit owners, inherit owners
from fabric/subfabric containment, flag unresolved ownership, flag unknown or
ambiguous containment, and surface ambiguous owner evidence.
- Extended `AccountabilityEvidenceStore` with durable review decisions keyed by
stable identity candidate key.
- Added `railiance-fabric discover-roots --ownership-review` and
`railiance-fabric review-identity` so operators can inspect blockers and
persist accept/needs-review/reject decisions across rescans.
- Added tests proving ambiguous ownership is visible, review decisions can
accept a stable identity, decisions survive a later ownership-review run, and
accepted items cannot silently lack a resolved owner.
- Documented ownership review and reviewer decisions in the manifest and
operator docs.
- Verified with
`python3 -m pytest tests/test_accountability_roots.py tests/test_accountability_root_adapters.py -q`,
`python3 -m railiance_fabric.cli discover-roots --max-items-per-root 5 --ownership-review --store-db /tmp/railiance-root-ownership.sqlite3`,
and full `python3 -m pytest`.
## T05 - Implement Snapshot Deltas And Freshness Triggers ## T05 - Implement Snapshot Deltas And Freshness Triggers
```task ```task