Repository navigation
feat: schema drift core analyzer, prune plan and apply (1/5) - #83
Merged
Merged
Conversation
Adds schema_drift.py: deterministic, LLM-free detection of declared-but-unused concepts/properties (used/locked/structural/prunable), prune plan write/load/validate, and apply_prune with backup, schema_history delta (schema_prune trigger) and schema.ttl regeneration. Nodes/edges untouched. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Reviewer's GuideThis PR adds the core schema-drift and pruning engine: it streams graph exports to classify unused schema entries, produces fingerprinted and validated prune plans, and safely applies schema-only removals with TTL regeneration, backups, history deltas, and extensive tests; CLI, configuration, and graph mutation remain out of scope. Sequence diagram for validated schema pruningsequenceDiagram
participant Caller
participant Analyzer as analyze_drift
participant Planner as plan_prune
participant Validator as validate_plan
participant Applier as apply_prune
participant Files as Schema files
Caller->>Analyzer: analyze_drift(schema, nodes_iter, edges_iter, locked)
Analyzer-->>Caller: DriftReport with fingerprint and classifications
Caller->>Planner: plan_prune(report)
Planner-->>Caller: PrunePlan
Caller->>Validator: validate_plan(plan, report)
Validator-->>Caller: validated plan
Caller->>Applier: apply_prune(plan, intermediate_dir, report)
Applier->>Files: read schema.json
Applier->>Files: export_ttl(pruned, [], {})
Applier->>Files: validate_schema_ttl(ttl)
Applier->>Files: write_schema(..., TRIGGER_SCHEMA_PRUNE)
Applier-->>Caller: pruned schema
Flow diagram for schema drift analysis and pruningflowchart LR
Nodes["nodes.jsonl"] --> Analyze[analyze_drift]
Edges["edges.jsonl"] --> Analyze
Schema["schema.json"] --> Analyze
Locked["LockedSchema"] --> Analyze
Analyze --> Report["DriftReport"]
Report --> Record[record_drift]
Report --> Plan[plan_prune]
Plan --> PlanFile["schema_prune_plan.json"]
PlanFile --> Load[load_plan]
Load --> Validate[validate_plan]
Report --> Validate
Validate --> Apply[apply_prune]
Apply --> NewSchema["schema.json + schema.ttl"]
Apply --> Backup["schema.pre_prune.json"]
Apply --> History["schema_prune history delta"]
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 11 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="src/mykg/schema_drift.py" line_range="436-441" />
<code_context>
+ from mykg.exporter import export_ttl
+ from mykg.schema_validator import validate_schema_ttl
+
+ if report is not None:
+ validate_plan(plan, report)
+ intermediate_dir = Path(intermediate_dir)
+ schema_path = intermediate_dir / "schema.json"
+ try:
+ schema = json.loads(schema_path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as exc:
+ raise SchemaDriftError(f"Cannot read {schema_path}: {exc}") from exc
</code_context>
<issue_to_address>
**New graph entries remain undeclared**
When the graph changes after the report is computed but before pruning applies, `validate_plan` checks only the supplied report, and `apply_prune` does not verify that it still represents the current graph. The prune removes newly used concepts or properties, leaving graph entries undeclared in the saved schema.
Recompute or verify the report against the current graph immediately before applying the prune.
</issue_to_address>
### Comment 2
<location path="src/mykg/schema_drift.py" line_range="436" />
<code_context>
+ from mykg.exporter import export_ttl
+ from mykg.schema_validator import validate_schema_ttl
+
+ if report is not None:
+ validate_plan(plan, report)
+ intermediate_dir = Path(intermediate_dir)
</code_context>
<issue_to_address>
**Used schema entries are removed**
When a caller applies a plan without passing a report, `apply_prune` skips `validate_plan` and checks only schema membership and dangling references, so it removes used, locked, or structural entries named by a hand-edited or stale plan. The caller gets a pruned schema that validation would reject, while graph data remains untouched.
Require a fresh report and validate the plan before applying any removals.
</issue_to_address>
### Comment 3
<location path="src/mykg/schema_drift.py" line_range="465" />
<code_context>
+ raise SchemaDriftError(f"Pruned schema failed validation: {result.errors}")
+
+ shutil.copy2(schema_path, intermediate_dir / BACKUP_FILE)
+ write_schema(
+ pruned,
+ intermediate_dir,
</code_context>
<issue_to_address>
**Prune outputs disagree after failure**
When a filesystem operation fails or the process stops while prune outputs are being written, `write_schema` updates `schema.json` before appending history, and `apply_prune` writes `schema.ttl` afterward. A failure between these writes leaves the schema, TTL, and history describing different versions, so later pipeline steps read inconsistent state.
Stage the prune outputs and commit them as one recoverable transaction, rolling back if any write fails.
</issue_to_address>
### Comment 4
<location path="src/mykg/schema_drift.py" line_range="465-473" />
<code_context>
+ raise SchemaDriftError(f"Pruned schema failed validation: {result.errors}")
+
+ shutil.copy2(schema_path, intermediate_dir / BACKUP_FILE)
+ write_schema(
+ pruned,
+ intermediate_dir,
+ TRIGGER_SCHEMA_PRUNE,
+ extra={
+ "pruned": [{"kind": e.kind, "name": e.name, "reason": e.reason} for e in plan.remove]
+ },
+ )
+ (intermediate_dir / "schema.ttl").write_text(ttl, encoding="utf-8")
+ log.info("schema_drift — pruned %d concept(s), %d property(ies)", len(rm_c), len(rm_p))
+ return pruned
</code_context>
<issue_to_address>
**Concurrent schema updates are lost**
When two prune or pipeline writers update the same intermediate directory concurrently, `apply_prune` writes whole-schema replacements without serialization, and concurrent `write_schema` calls can choose the same history sequence and overwrite a delta. Callers see lost schema updates and incomplete history.
Serialize schema and history updates under a shared lock, including history-sequence allocation.
</issue_to_address>
### Comment 5
<location path="src/mykg/schema_drift.py" line_range="465-473" />
<code_context>
+ raise SchemaDriftError(f"Pruned schema failed validation: {result.errors}")
+
+ shutil.copy2(schema_path, intermediate_dir / BACKUP_FILE)
+ write_schema(
+ pruned,
+ intermediate_dir,
+ TRIGGER_SCHEMA_PRUNE,
+ extra={
+ "pruned": [{"kind": e.kind, "name": e.name, "reason": e.reason} for e in plan.remove]
+ },
+ )
+ (intermediate_dir / "schema.ttl").write_text(ttl, encoding="utf-8")
+ log.info("schema_drift — pruned %d concept(s), %d property(ies)", len(rm_c), len(rm_p))
+ return pruned
</code_context>
<issue_to_address>
**Resumed extraction uses stale vocabulary**
When `flattened_schema.json` exists when a session resumes after pruning, `apply_prune` leaves `flattened_schema.json` unchanged, and the orchestrator can skip `schema_flatten` because that file exists. `run_pass2_step` then uses the stale vocabulary, so removed concepts and attributes remain available to extraction and can reappear in the graph.
Invalidate or regenerate `flattened_schema.json` when pruning so resumed sessions flatten the pruned schema.
Also at `src/mykg/schema_drift.py:431`.
</issue_to_address>
### Comment 6
<location path="src/mykg/schema_drift.py" line_range="154" />
<code_context>
+ if not line:
+ continue
+ try:
+ yield json.loads(line)
+ except json.JSONDecodeError as exc:
+ raise SchemaDriftError(f"{path}:{lineno} is not valid JSON: {exc}") from exc
</code_context>
<issue_to_address>
**Malformed JSON values escape the documented error**
When a JSONL line parses successfully but its row or attributes value is not an object, `stream_jsonl` yields any value returned by `json.loads`, and `analyze_drift` calls `.get()` on rows and their attributes. Invalid shapes therefore raise an uncaught `AttributeError` instead of `SchemaDriftError`, so callers cannot handle corrupt input through the documented error type.
Validate row and attributes shapes and raise `SchemaDriftError` for values that are not objects.
Also at `src/mykg/schema_drift.py:201-207`, `src/mykg/schema_drift.py:210`.
</issue_to_address>
### Comment 7
<location path="src/mykg/schema_drift.py" line_range="403" />
<code_context>
+ }
+ rm_c: set[str] = set()
+ rm_p: set[str] = set()
+ for e in plan.remove:
+ row = rows[e.kind].get(e.name)
+ if row is None:
</code_context>
<issue_to_address>
**Plan totals disagree with removals**
When a caller edits a plan’s `remove` list without updating its totals or keep entries, `validate_plan` checks removals but does not reconcile `keep` or `totals` with the edited list. `apply_prune` follows `remove`, so the plan can report counts that differ from the schema it actually produces.
Recompute `keep` and `totals` from the validated removals, or reject plans whose values do not match.
Also at `src/mykg/schema_drift.py:411`.
</issue_to_address>
### Comment 8
<location path="src/mykg/schema_drift.py" line_range="465" />
<code_context>
+ raise SchemaDriftError(f"Pruned schema failed validation: {result.errors}")
+
+ shutil.copy2(schema_path, intermediate_dir / BACKUP_FILE)
+ write_schema(
+ pruned,
+ intermediate_dir,
</code_context>
<issue_to_address>
**No-op prune creates misleading history**
When a caller applies a plan with no entries to remove, `apply_prune` still copies the schema to the pre-prune backup and calls `write_schema`, creating an empty prune history entry. After a real prune, a no-op apply also replaces the useful backup with the already-pruned schema.
Return without writing a backup or history entry when `plan.remove` is empty.
</issue_to_address>
### Comment 9
<location path="tests/test_schema_drift.py" line_range="97" />
<code_context>
+ "Project",
+ "Service",
+ }
+ assert len([p for p in report.properties if p.classification == "prunable"]) == 8
+ plan = plan_prune(report)
+ assert plan.totals.before.concepts == 5 and plan.totals.before.properties == 11
</code_context>
<issue_to_address>
**Wrong properties pass pruning checks**
When the analyzer misclassifies an edge-bearing property as prunable and an unused property as retained while the totals stay unchanged, `test_article_example_counts_and_prune` checks only the number of prunable properties, and the apply/history checks likewise compare counts, not names. Swapping `works_at` into the prunable set and `depends_on` out of it can preserve those counts and pass while pruning a property with edges.
Assert the exact prunable-property names and exact property names remaining after apply.
Also at `tests/test_schema_drift.py:318`.
</issue_to_address>
### Comment 10
<location path="tests/test_schema_drift.py" line_range="319" />
<code_context>
+ assert delta["timestamp"]
+ assert delta["concepts_removed"] == ["Project", "Service"]
+ assert len(delta["properties_removed"]) == 8
+ assert {e["name"] for e in delta["pruned"]} >= {"Project", "Service", "owns_project"}
+
+ ttl = (inter / "schema.ttl").read_text(encoding="utf-8")
</code_context>
<issue_to_address>
**Extra prune records go unnoticed**
When the history delta includes extra names that were not in the prune plan, `test_apply_side_effects_and_jsonl_untouched` uses `>=` for the `delta["pruned"]` names, so an apply implementation that adds unrelated entries to the history record still passes the audit check.
Compare the recorded `pruned` entries against the exact expected plan entries.
</issue_to_address>
### Comment 11
<location path="src/mykg/schema_drift.py" line_range="473" />
<code_context>
+ "pruned": [{"kind": e.kind, "name": e.name, "reason": e.reason} for e in plan.remove]
+ },
+ )
+ (intermediate_dir / "schema.ttl").write_text(ttl, encoding="utf-8")
+ log.info("schema_drift — pruned %d concept(s), %d property(ies)", len(rm_c), len(rm_p))
+ return pruned
</code_context>
<issue_to_address>
**Exported graph retains pruned schema**
When the session already has an exported `knowledge_graph.ttl` when pruning is applied, `apply_prune` updates only the intermediate schema files, leaving an existing `output/knowledge_graph.ttl` unchanged. That export still declares removed concepts and properties, so users see a graph schema that disagrees with the pruned `schema.json` and `schema.ttl`.
Regenerate or invalidate `knowledge_graph.ttl` when applying a prune so consumers do not keep using the pre-prune schema export.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 11 findings to address first, and a mistaken prune removes concept or property definitions from the persisted schema and TTL, which can affect later pipeline runs even though graph nodes and edges remain untouched. The pre-prune backup and history delta make the bounded change repairable, but reverting the code alone does not restore the schema.
Blocking findings: src/mykg/schema_drift.py:441, src/mykg/schema_drift.py:436, src/mykg/schema_drift.py:465, src/mykg/schema_drift.py:473, src/mykg/schema_drift.py:473, and 6 more
Comment on lines
+436
to
+441
| if report is not None: | ||
| validate_plan(plan, report) | ||
| intermediate_dir = Path(intermediate_dir) | ||
| schema_path = intermediate_dir / "schema.json" | ||
| try: | ||
| schema = json.loads(schema_path.read_text(encoding="utf-8")) |
There was a problem hiding this comment.
🟠 High · New graph entries remain undeclared
When the graph changes after the report is computed but before pruning applies, validate_plan checks only the supplied report, and apply_prune does not verify that it still represents the current graph. The prune removes newly used concepts or properties, leaving graph entries undeclared in the saved schema.
Recompute or verify the report against the current graph immediately before applying the prune.
Prompt for AI agents
In `src/mykg/schema_drift.py` at lines 436-441:
**New graph entries remain undeclared**
When the graph changes after the report is computed but before pruning applies, `validate_plan` checks only the supplied report, and `apply_prune` does not verify that it still represents the current graph. The prune removes newly used concepts or properties, leaving graph entries undeclared in the saved schema.
Recompute or verify the report against the current graph immediately before applying the prune.| raise SchemaDriftError(f"Pruned schema failed validation: {result.errors}") | ||
|
|
||
| shutil.copy2(schema_path, intermediate_dir / BACKUP_FILE) | ||
| write_schema( |
There was a problem hiding this comment.
🟠 High · Prune outputs disagree after failure
When a filesystem operation fails or the process stops while prune outputs are being written, write_schema updates schema.json before appending history, and apply_prune writes schema.ttl afterward. A failure between these writes leaves the schema, TTL, and history describing different versions, so later pipeline steps read inconsistent state.
Stage the prune outputs and commit them as one recoverable transaction, rolling back if any write fails.
Prompt for AI agents
In `src/mykg/schema_drift.py` at line 465:
**Prune outputs disagree after failure**
When a filesystem operation fails or the process stops while prune outputs are being written, `write_schema` updates `schema.json` before appending history, and `apply_prune` writes `schema.ttl` afterward. A failure between these writes leaves the schema, TTL, and history describing different versions, so later pipeline steps read inconsistent state.
Stage the prune outputs and commit them as one recoverable transaction, rolling back if any write fails.
Comment on lines
+465
to
+473
| write_schema( | ||
| pruned, | ||
| intermediate_dir, | ||
| TRIGGER_SCHEMA_PRUNE, | ||
| extra={ | ||
| "pruned": [{"kind": e.kind, "name": e.name, "reason": e.reason} for e in plan.remove] | ||
| }, | ||
| ) | ||
| (intermediate_dir / "schema.ttl").write_text(ttl, encoding="utf-8") |
There was a problem hiding this comment.
🟡 Medium · Concurrent schema updates are lost
When two prune or pipeline writers update the same intermediate directory concurrently, apply_prune writes whole-schema replacements without serialization, and concurrent write_schema calls can choose the same history sequence and overwrite a delta. Callers see lost schema updates and incomplete history.
Serialize schema and history updates under a shared lock, including history-sequence allocation.
Prompt for AI agents
In `src/mykg/schema_drift.py` at lines 465-473:
**Concurrent schema updates are lost**
When two prune or pipeline writers update the same intermediate directory concurrently, `apply_prune` writes whole-schema replacements without serialization, and concurrent `write_schema` calls can choose the same history sequence and overwrite a delta. Callers see lost schema updates and incomplete history.
Serialize schema and history updates under a shared lock, including history-sequence allocation.| "pruned": [{"kind": e.kind, "name": e.name, "reason": e.reason} for e in plan.remove] | ||
| }, | ||
| ) | ||
| (intermediate_dir / "schema.ttl").write_text(ttl, encoding="utf-8") |
There was a problem hiding this comment.
🟡 Medium · Exported graph retains pruned schema
When the session already has an exported knowledge_graph.ttl when pruning is applied, apply_prune updates only the intermediate schema files, leaving an existing output/knowledge_graph.ttl unchanged. That export still declares removed concepts and properties, so users see a graph schema that disagrees with the pruned schema.json and schema.ttl.
Regenerate or invalidate knowledge_graph.ttl when applying a prune so consumers do not keep using the pre-prune schema export.
Prompt for AI agents
In `src/mykg/schema_drift.py` at line 473:
**Exported graph retains pruned schema**
When the session already has an exported `knowledge_graph.ttl` when pruning is applied, `apply_prune` updates only the intermediate schema files, leaving an existing `output/knowledge_graph.ttl` unchanged. That export still declares removed concepts and properties, so users see a graph schema that disagrees with the pruned `schema.json` and `schema.ttl`.
Regenerate or invalidate `knowledge_graph.ttl` when applying a prune so consumers do not keep using the pre-prune schema export.…egen, shape validation, no-op prune) - apply_prune(plan, dir, report): report required, always validate_plan first - recompute keep/totals/reasons from validated removals (recompute_plan) - empty plan returns early: no backup, history or writes - regenerate flattened_schema.json (if present) and write schema.ttl atomically - reject schema.json changed since the report was computed - SchemaDriftError for malformed JSONL rows, attributes, types, schema and locked shapes - fingerprint includes locked class/property names Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Owner
Author
|
Review fixes pushed. Accepted: required report + validate_plan in apply_prune (83.1/83.2), atomic schema.ttl and idempotent write order (83.3, partly), flattened_schema.json regenerated when it exists (83.5), SchemaDriftError for malformed JSONL/schema/locked shapes (83.6), keep/totals recomputed from validated removals (83.7), empty plan is a no-op (83.8), exact-name tests (83.9/83.10), locked names in the fingerprint. Declined: cross-process locking (83.4) and core regenerating knowledge_graph.ttl (83.11, handled in CLI unit). |
This was referenced Oct 23, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Unit 1 of 5. Adds src/mykg/schema_drift.py (analyze_drift, record_drift, plan_prune, write_plan, load_plan, validate_plan, apply_prune), the schema_prune history trigger, and tests. No CLI yet; no config keys; nodes/edges never modified.
Tests: tests/test_schema_drift.py (article 5/11 -> 3/3 example etc.). Full suite passes except a live-Anthropic network test.
🤖 Generated with Claude Code
Summary by Sourcery
Add the core schema drift analyzer and safe pruning workflow for removing unused schema concepts and properties.
New Features:
Enhancements:
Tests: