A moved file or directory keeps its shape judgments and snippet locations #213

Merged
bvandeusen merged 2 commits from dev into main 2026-10-22 15:29:37 -04:00
10 changed files with 338 additions and 10 deletions
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "grimoire",
"description": "Grimoire for Claude Code: connects the grimoire MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Grimoire skills (using-grimoire, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting, family-canon), and syncs your saved Grimoire Processes as skills (/grimoire:sync).",
"version": "2026.10.21.2344",
"version": "2026.10.22.1907",
"author": {
"name": "Bryan Van Deusen"
},
+10 -2
View File
@@ -41,6 +41,14 @@ row carries a status:
nothing. Rows, never prose — a consumer list in a note or verification
detail cannot be sorted, queried, or diffed.
Rows are keyed by repo and path, so a move in the tree is a move in the
ledger. When a file or directory moves, run `move_path(project_id,
from_path, to_path)` before the next refresh; it carries the snippets
recorded there too. For a repo renamed on its
forge, run `bind_repo(new_url, project_id, moved_from=old_url)`. Otherwise
the refresh marks every judged row vanished and lists the same shapes again,
unjudged.
## The writer judges; audits check that it happened
The ledger is filled from your answers, given while you still know them. When
@@ -157,8 +165,8 @@ Three questions the ledger answers mechanically:
it), `variant` with the why if the departure is deliberate. The write-path
hint asks the same question in-band the moment such a shape is written.
- **History** — `shape_history(project_id, path, symbol?)`: the current rows
plus every `classified` / `vanished` / `reappeared` / `drifted` event with
its commit — "instance of #N from <date>, re-judged variant of #M because
plus every `classified` / `vanished` / `reappeared` / `drifted` / `moved`
event with its commit — "instance of #N from <date>, re-judged variant of #M because
R, vanished at C". Rows, not recollection.
- **Recheck** — `list_shapes(project_id, flag="recheck")`: judged
instances/variants whose body changed since judged. The judgment stands;
+1 -1
View File
@@ -200,7 +200,7 @@ _WRITE_TOOLS = frozenset({
"merge_snippets", "unmerge_snippet",
"create_process", "update_process", "delete_process",
"classify_shapes", "classify_shapes_by_rule", "confirm_shape_proposals",
"refresh_pattern_coverage",
"refresh_pattern_coverage", "move_path",
# design systems
"create_design_system", "update_design_system", "delete_design_system",
"create_design_token", "update_design_token", "delete_design_token",
+44 -2
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
from grimoire.mcp._context import current_user_id
from grimoire.services import coverage as coverage_svc
from grimoire.services import shape_ledger as shape_ledger_svc
from grimoire.services import snippets as snippets_svc
async def classify_shapes(
@@ -263,7 +264,8 @@ async def shape_history(
snippet_id, who, why — one per judgment, so a shape that was an instance
of #N and later a variant of #M shows both), `vanished`, `reappeared`,
`drifted` (the body moved under a judgment; see list_shapes flag=
"recheck"). Each event carries the commit the tree was read at.
"recheck"), `moved` (move_path carried the row; the event's `path`
is the one it left). Each event carries the commit the tree was read at.
Read it as a timeline: "instance of #N from <first classified at>,
re-judged variant of #M at <at> because <reason>, vanished at <commit>".
@@ -398,10 +400,50 @@ async def refresh_pattern_coverage(project_id: int) -> dict:
}
async def move_path(project_id: int, from_path: str, to_path: str) -> dict:
"""Carry what the record knows about a file or directory that moved in
the tree: its shape judgments and its snippets' recorded locations.
Both are keyed by path. After `git mv src/old src/new`, the next coverage
refresh would mark every row under src/old vanished and list the same
shapes again under src/new, unclassified, with every judgment left
behind. Each snippet recorded under src/old would also read `missing`.
Run this once the move is committed, before that refresh.
- Every ledger row under `from_path` is re-keyed to the same place under
`to_path`, in every repo bound to the project. It keeps its judgment,
uses edges and history, and gains a `moved` event (shape_history).
- Every snippet of the project recorded under `from_path` has its
location re-pathed.
A ledger row already at the new path is replaced when nobody judged it
(the refresh ran first, or the sync stamped it `scoped`). A judged one
stands; the old row stays where it is and is listed in `kept`. The
refresh that follows compares bodies as always, so a judged shape whose
body changed in the move comes back flagged `recheck`.
Args:
project_id: The project whose record moves. Requires write access.
from_path: The file or directory as it was recorded.
to_path: Where it lives now. Neither path may contain the other.
Returns {"moved": N, "replaced": N, "kept": [{path, symbol, kind}],
"snippets_moved": [ids], "snippets_read_only": [ids]}. A snippet shared
with you read-only stays where it is, and its owner moves it.
"""
uid = current_user_id()
out = await shape_ledger_svc.move_shapes(uid, project_id, from_path, to_path)
snippets = await snippets_svc.move_locations(uid, project_id, from_path, to_path)
out["snippets_moved"] = snippets["moved"]
out["snippets_read_only"] = snippets["read_only"]
if not (out["moved"] or out["kept"] or snippets["moved"] or snippets["read_only"]):
raise ValueError(f"nothing in project {project_id} is recorded under `{from_path}`")
return out
def register(mcp) -> None:
for fn in (
classify_shapes, classify_shapes_by_rule, list_shapes,
refresh_pattern_coverage, confirm_shape_proposals, shape_history,
stamps_to_review,
stamps_to_review, move_path,
):
mcp.tool(name=fn.__name__)(fn)
+4 -3
View File
@@ -311,8 +311,9 @@ class CodeShapeConsumer(Base):
# What a shape's history records. Not "appeared" — first_seen and
# created_at already say that on the row; history is for what CHANGED:
SHAPE_EVENTS = ("classified", "vanished", "reappeared", "drifted")
# created_at already say that on the row; history is for what CHANGED.
# A `moved` event carries the path the shape left; the row says where it went.
SHAPE_EVENTS = ("classified", "vanished", "reappeared", "drifted", "moved")
class CodeShapeEvent(Base):
@@ -321,7 +322,7 @@ class CodeShapeEvent(Base):
"We used #N here from <date>, #M replaced it at commit C, reason R" is a
question the ledger row alone cannot answer once it has moved on; this
table keeps each judgment (status, snippet, who, why, at which commit)
and each presence change (vanished / reappeared / drifted) as it
and each presence change (vanished / reappeared / drifted / moved) as it
happened. Denormalised path/symbol/kind so a directory's history reads
without joining; `snippet_id` is deliberately FK-free — history outlives
the snippet it names, which is the point.
+74
View File
@@ -170,6 +170,80 @@ async def sync_repo_shapes(
await session.commit()
async def move_shapes(
user_id: int, project_id: int, from_path: str, to_path: str,
) -> dict:
"""Carry the ledger rows under ``from_path`` (a file or a directory) to
the same place under ``to_path``, in every repo of the project.
Rows are keyed by path, so a directory moved in the tree would otherwise
vanish every judged row at the next sync and start the same shapes again,
unclassified, under the new path. A moved row is the same row — its id,
judgment, uses edges and history ride along — and a `moved` event
records the path it left. A row already at the target (the sync ran
first, or a provisional judgment landed there) is replaced when nobody
judged it; a judged one stands, and the old row is left where it is and
reported in ``kept``. Bodies are not compared here: the next sync does
that as always, and asks for a recheck where a judged body changed.
"""
from sqlalchemy import or_
from grimoire.services import access
from grimoire.services.snippets import _path_touches
old = (from_path or "").strip().strip("/")
new = (to_path or "").strip().strip("/")
if not old or not new:
raise ValueError("from_path and to_path are both required")
if _path_touches(old, new) or _path_touches(new, old):
raise ValueError(f"`{old}` and `{new}` overlap; move to a path outside the source")
if not await access.can_write_project(user_id, project_id):
raise ValueError(f"project {project_id} not found or no write access")
now = datetime.now(timezone.utc)
async with async_session() as session:
def beneath(prefix: str):
return select(CodeShape).where(
CodeShape.project_id == project_id,
or_(CodeShape.path == prefix, CodeShape.path.like(prefix + "/%")),
)
# LIKE reads `_` as a wildcard; the prefix test is the real filter.
sources = [
r for r in (await session.execute(beneath(old))).scalars().all()
if _path_touches(old, r.path)
]
targets = {
(r.repo_key, r.path, r.symbol, r.kind): r
for r in (await session.execute(beneath(new))).scalars().all()
}
moved, replaced, kept = 0, 0, []
for row in sources:
dest = new + row.path[len(old):]
there = targets.get((row.repo_key, dest, row.symbol, row.kind))
if there is not None:
if there.status not in _MECHANICAL_TODO:
kept.append({"path": row.path, "symbol": row.symbol, "kind": row.kind})
continue
# The target is the live sighting when the sync got there
# first; the moved row takes over its presence.
row.vanished_at = there.vanished_at
row.last_seen_commit = there.last_seen_commit or row.last_seen_commit
await session.delete(there)
await session.flush()
replaced += 1
session.add(CodeShapeEvent(
shape_id=row.id, project_id=project_id,
path=row.path, symbol=row.symbol, kind=row.kind,
event="moved", status=row.status, snippet_id=row.snippet_id,
classified_by=row.classified_by, reason=row.reason,
commit=row.last_seen_commit or "", at=now,
))
row.path = dest
moved += 1
await session.commit()
return {"moved": moved, "replaced": replaced, "kept": kept}
def _event(row: CodeShape, event: str, at: datetime, *, commit: str = "") -> CodeShapeEvent:
"""A history row for a state change on ``row`` — status/snippet/by/reason
are the row's CURRENT values, which for `classified` is the judgment
+46
View File
@@ -1280,6 +1280,52 @@ def _path_touches(recorded: str, changed: str) -> bool:
return changed == recorded or changed.startswith(recorded + "/")
async def move_locations(
user_id: int, project_id: int, from_path: str, to_path: str,
) -> dict:
"""Re-path the recorded locations under ``from_path`` (a file or a
directory) to the same place under ``to_path``, on every live snippet of
the project the caller can edit.
The twin of the shape ledger's move: a snippet recorded at a path that
moved reads `missing` at its next check, and its canonical stamp no longer
covers the shape that moved. Each snippet is rewritten through
update_snippet, so its body and indexed fields stay one record. A
snippet the caller can only read is left as it is and listed in
``read_only``.
"""
old = (from_path or "").strip().strip("/")
new = (to_path or "").strip().strip("/")
async with async_session() as session:
notes = (
await session.execute(
select(Note).where(
Note.project_id == project_id,
Note.note_type == SNIPPET_NOTE_TYPE,
Note.deleted_at.is_(None),
)
)
).scalars().all()
moved: list[int] = []
read_only: list[int] = []
for note in notes:
locations = snippet_fields(note).get("locations") or []
if not any(_path_touches(old, loc.get("path") or "") for loc in locations):
continue
repathed = [
{**loc, "path": new + loc["path"].strip("/")[len(old):]}
if _path_touches(old, loc.get("path") or "") else loc
for loc in locations
]
try:
await update_snippet(user_id, int(note.id), locations=repathed)
except PermissionError:
read_only.append(int(note.id))
continue
moved.append(int(note.id))
return {"moved": moved, "read_only": read_only}
async def invalidate_for_push(
repo_key: str,
changed: list[str],
+149
View File
@@ -0,0 +1,149 @@
"""Real-Postgres integration tests for moving a directory's shape judgments.
Ledger rows are keyed by path. A directory moved in the tree has to carry its
rows to the new path, or the next sync vanishes every judgment and lists the
same shapes again, unjudged.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select
from grimoire.models import async_session
from grimoire.models.code_shape import CodeShape, CodeShapeEvent
from grimoire.models.project import Project
from grimoire.services.shape_ledger import classify_shapes, move_shapes, sync_repo_shapes
from tests.helpers import ensure_user
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
REPO = "git.example.com/alice/widget"
OLD_TREE = [
("src/old/app.py", "sym", "make_app"),
("src/old/sub/util.py", "sym", "helper"),
("src/older/keep.py", "sym", "untouched"),
]
NEW_TREE = [
("src/new/app.py", "sym", "make_app"),
("src/new/sub/util.py", "sym", "helper"),
("src/older/keep.py", "sym", "untouched"),
]
@pytest_asyncio.fixture
async def seeded():
"""A project whose ledger was synced before the move, one shape judged."""
async with async_session() as s:
owner = await ensure_user(s, "shape_move_owner")
other = await ensure_user(s, "shape_move_other")
project = Project(user_id=owner.id, title="Moved directory")
s.add(project)
await s.flush()
ids = {"owner": owner.id, "other": other.id, "pid": project.id}
await s.commit()
await sync_repo_shapes(ids["pid"], REPO, OLD_TREE, seen_marker="c1")
await classify_shapes(ids["owner"], ids["pid"], [
{"path": "src/old/sub/util.py", "symbol": "helper", "status": "exempt",
"reason": "one caller"},
])
return ids
async def _ledger(pid: int) -> dict[tuple[str, str], tuple[str, bool]]:
async with async_session() as s:
rows = await s.execute(select(CodeShape).where(CodeShape.project_id == pid))
return {
(r.path, r.symbol): (r.status, r.vanished_at is None)
for r in rows.scalars()
}
@pytest.mark.integration
async def test_the_judgments_move_and_the_next_sync_finds_them(seeded):
pid = seeded["pid"]
result = await move_shapes(seeded["owner"], pid, "src/old/", "src/new")
assert result == {"moved": 2, "replaced": 0, "kept": []}
await sync_repo_shapes(pid, REPO, NEW_TREE, seen_marker="c2")
assert await _ledger(pid) == {
("src/new/app.py", "make_app"): ("unclassified", True),
("src/new/sub/util.py", "helper"): ("exempt", True),
("src/older/keep.py", "untouched"): ("unclassified", True),
}
async with async_session() as s:
moved = (await s.execute(
select(CodeShapeEvent.path).where(
CodeShapeEvent.project_id == pid, CodeShapeEvent.event == "moved",
)
)).scalars().all()
assert sorted(moved) == ["src/old/app.py", "src/old/sub/util.py"]
@pytest.mark.integration
async def test_after_the_sync_got_there_first_the_judgment_replaces_the_fresh_row(seeded):
pid = seeded["pid"]
await sync_repo_shapes(pid, REPO, NEW_TREE, seen_marker="c2")
result = await move_shapes(seeded["owner"], pid, "src/old", "src/new")
assert result == {"moved": 2, "replaced": 2, "kept": []}
assert await _ledger(pid) == {
("src/new/app.py", "make_app"): ("unclassified", True),
("src/new/sub/util.py", "helper"): ("exempt", True),
("src/older/keep.py", "untouched"): ("unclassified", True),
}
@pytest.mark.integration
async def test_a_judgment_already_at_the_new_path_stands(seeded):
pid = seeded["pid"]
await sync_repo_shapes(pid, REPO, OLD_TREE + NEW_TREE, seen_marker="c2")
await classify_shapes(seeded["owner"], pid, [
{"path": "src/new/sub/util.py", "symbol": "helper", "status": "exempt",
"reason": "judged again after the move"},
])
result = await move_shapes(seeded["owner"], pid, "src/old", "src/new")
assert result["moved"] == 1 and result["replaced"] == 1
assert result["kept"] == [
{"path": "src/old/sub/util.py", "symbol": "helper", "kind": "sym"},
]
ledger = await _ledger(pid)
assert ledger[("src/new/sub/util.py", "helper")] == ("exempt", True)
assert ("src/old/sub/util.py", "helper") in ledger
@pytest.mark.integration
async def test_overlapping_and_unwritable_moves_are_refused(seeded):
owner, other, pid = seeded["owner"], seeded["other"], seeded["pid"]
with pytest.raises(ValueError, match="overlap"):
await move_shapes(owner, pid, "src/old", "src/old/inner")
assert await move_shapes(owner, pid, "src/gone", "src/new") == {
"moved": 0, "replaced": 0, "kept": [],
}
with pytest.raises(ValueError, match="no write access"):
await move_shapes(other, pid, "src/old", "src/new")
assert ("src/old/app.py", "make_app") in await _ledger(pid)
@pytest.mark.integration
async def test_the_snippets_recorded_there_move_with_it(seeded):
from grimoire.services import snippets as snippets_svc
owner, pid = seeded["owner"], seeded["pid"]
note = await snippets_svc.create_snippet(
owner, name="move_helper", code="def helper():\n pass\n",
language="python", project_id=pid, locations=[
{"repo": "Widget", "path": "src/old/sub/util.py", "symbol": "helper"},
{"repo": "Widget", "path": "src/older/keep.py", "symbol": "untouched"},
],
)
result = await snippets_svc.move_locations(owner, pid, "src/old", "src/new")
assert result == {"moved": [int(note.id)], "read_only": []}
moved = await snippets_svc.get_snippet(owner, int(note.id))
assert [loc["path"] for loc in snippets_svc.snippet_fields(moved)["locations"]] == [
"src/new/sub/util.py", "src/older/keep.py",
]
+8
View File
@@ -383,6 +383,14 @@ def test_bind_repo_tool_can_move_a_renamed_repo():
assert "moved_from" in tool.parameters.get("properties", {})
def test_move_path_is_mounted_as_a_write_tool():
"""A directory moved in the tree carries its judgments to the new path."""
from grimoire.mcp.server import _WRITE_TOOLS, build_mcp_server
tool = build_mcp_server()._tool_manager.get_tool("move_path")
assert {"from_path", "to_path"} <= set(tool.parameters.get("properties", {}))
assert "move_path" in _WRITE_TOOLS
def test_coverage_routes_are_registered():
from grimoire.app import create_app
+1 -1
View File
@@ -642,5 +642,5 @@ def test_history_and_divergence_columns_are_pinned():
fk = next(iter(ev.c.shape_id.foreign_keys))
assert fk.ondelete == "CASCADE" and fk.column.table.name == "code_shapes"
assert not ev.c.snippet_id.foreign_keys # history outlives the snippet
assert SHAPE_EVENTS == ("classified", "vanished", "reappeared", "drifted")
assert SHAPE_EVENTS == ("classified", "vanished", "reappeared", "drifted", "moved")
assert "code_shape_events" in Base.metadata.tables