bvandeusenandClaude Opus 5
91cde6c3e4
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / TypeScript typecheck (push) Successful in 52s
CI & Build / integration (push) Successful in 52s
CI & Build / Python tests (push) Successful in 1m33s
CI & Build / Build & push image (push) Successful in 28s
#4204 gave the divergence check a structural gate, which silenced one of the five false prompts it was filed for. The other four are `def` helpers in a directory whose canon is an `async def` service unit. No refinement of `shape_form` reaches them: they differ from #2793's acceptance case — a hand-rolled sync `confirmDanger` where an async confirm helper is canon — only by the JOB they do, and a signature does not carry a job. The proposer's semantic arm already reads bodies per symbol, which is the comparison option 2 asked for and was thought to be missing. What it did not do was record its MISSES: a hit became `proposal_basis = "semantic"`, a miss left the row indistinguishable from one nobody had looked at. So `flag_divergence` could ask the proposer "do you agree this is the canon?" but never "did you check, and is it not?". `_semantic_canon` now reports whether an empty answer is evidence, and `flag_divergence` withholds the prompt when it is. The whole risk is in the negative, so only a conclusive miss is stored. A body too thin to embed, a row the per-refresh cap never reached, an arm that threw, and a result set that came back full — and may therefore have hidden the canon behind the limit — all stay "cannot tell" and still ask the question. That is the discipline `FORM_UNKNOWN` already enforces here: not knowing must make a check quieter, never more confident. `_SEMANTIC_LIMIT` is named for that reason; the number is load-bearing, not a tuning knob. No migration: `proposal_basis` is nullable Text with no CHECK constraint (verified in the model and across alembic/versions), so rule 36 does not bite. Nothing can mistake the miss for a proposal either — every reader keys on `proposed_snippet_id` or `proposal_group`, and `confirm_shape_proposals` requires the id non-NULL before it will confirm anything. `_PROPOSER_VERSION` 3 -> 4, per its own contract: rows remember the ruleset they were examined under, and without the bump no already-examined row would ever acquire a miss. Option 1 (widening `kind`) stays closed, on the merits rather than on cost: bucketing density by exact form takes the async canon out of a sync candidate's denominator and silences #2793's acceptance case by the identical mechanism, one layer down. The reasoning is on #4208. Tests: tests/test_divergence_meaning_gate.py pins the report contract, with the truncation case tested hardest — reading a cut-off as a negative would weaken the guard in proportion to how many snippets the operator has. The end-to-end discrimination is in test_integration_shape_classify.py on deliberately the SAME fixture as #2793's acceptance case, so the two runs differ in exactly one thing: whether the arm claims to have looked. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01821k5B3Ysecp9fNYs92Kuy
1023 lines
50 KiB
Python
1023 lines
50 KiB
Python
"""Real-Postgres integration tests for shape classification (#2789).
|
|
|
|
What mocks can't prove: the all-or-nothing batch against real rows, the
|
|
write-ACL gate, the todo query's filters, and the consumer map riding
|
|
get_snippet. Ledger rows are seeded through the same sync the coverage walk
|
|
uses — no forge needed, the sync takes extracted shapes directly.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.code_shape import CodeShape
|
|
from scribe.models.project import Project
|
|
from scribe.models.user import User
|
|
from scribe.services.shape_ledger import (
|
|
classify_shapes,
|
|
classify_shapes_where,
|
|
list_project_shapes,
|
|
snippet_consumers,
|
|
sync_repo_shapes,
|
|
)
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
REPO = "git.example.com/alice/widget"
|
|
SHAPES = [
|
|
("src/app.py", "sym", "make_app"),
|
|
("src/app.py", "sym", "Config"),
|
|
("src/util.py", "sym", "helper"),
|
|
("web/button.css", "css", "btn"),
|
|
]
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def seeded():
|
|
"""Owner + outsider, a project with a synced 4-shape ledger, one snippet."""
|
|
from scribe.services import snippets as snippets_svc
|
|
|
|
async with async_session() as s:
|
|
owner = await ensure_user(s, "classify_owner")
|
|
other = await ensure_user(s, "classify_other")
|
|
project = Project(user_id=owner.id, title="Classify target")
|
|
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, SHAPES, seen_marker="main")
|
|
snippet = await snippets_svc.create_snippet(
|
|
ids["owner"], name="cls_make_app", code="def make_app():\n pass\n",
|
|
language="python", repo="Widget", path="src/factory.py",
|
|
symbol="factory", project_id=ids["pid"],
|
|
)
|
|
ids["snippet"] = int(snippet.id)
|
|
return ids
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_classify_applies_judgments_and_reports_unmatched(seeded):
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
out = await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid},
|
|
{"path": "src/util.py", "symbol": "helper", "status": "exempt",
|
|
"reason": "test scaffolding, deliberately local"},
|
|
{"path": "web/button.css", "symbol": "btn", "status": "variant",
|
|
"snippet_id": sid, "reason": "darker focus ring for the toolbar"},
|
|
{"path": "gone.py", "symbol": "nothing", "status": "exempt",
|
|
"reason": "x"},
|
|
], via="audit")
|
|
assert out["classified"] == 3
|
|
assert out["unmatched"] == [{"path": "gone.py", "symbol": "nothing"}]
|
|
|
|
rows, total = await list_project_shapes(owner, pid)
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert total == 4
|
|
assert by_symbol["make_app"].status == "instance"
|
|
assert by_symbol["make_app"].snippet_id == sid
|
|
assert by_symbol["make_app"].classified_by == "audit"
|
|
assert by_symbol["helper"].status == "exempt"
|
|
assert by_symbol["helper"].reason == "test scaffolding, deliberately local"
|
|
assert by_symbol["btn"].status == "variant"
|
|
assert by_symbol["Config"].status == "unclassified"
|
|
|
|
# Withdrawing a judgment returns the shape to the todo, fields cleared.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
|
|
])
|
|
rows, _ = await list_project_shapes(owner, pid, status="unclassified")
|
|
assert {r.symbol for r in rows} == {"Config", "make_app"}
|
|
make_app = next(r for r in rows if r.symbol == "make_app")
|
|
assert make_app.snippet_id is None and make_app.classified_by is None
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_a_bad_batch_applies_nothing(seeded):
|
|
"""All-or-nothing (#2709's lesson): a caller must never learn later that
|
|
half a batch silently happened."""
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
with pytest.raises(ValueError) as err:
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid},
|
|
{"path": "src/util.py", "symbol": "helper", "status": "variant",
|
|
"snippet_id": sid}, # variant with no reason: structural error
|
|
])
|
|
assert "needs a reason" in str(err.value)
|
|
rows, _ = await list_project_shapes(owner, pid, status="unclassified")
|
|
assert len(rows) == 4 # including make_app — the valid half did NOT apply
|
|
|
|
# A snippet target the caller can't read is the same: nothing applies.
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": 999999999},
|
|
])
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_classification_is_write_gated_and_listing_read_gated(seeded):
|
|
other, pid, sid = seeded["other"], seeded["pid"], seeded["snippet"]
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes(other, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "exempt",
|
|
"reason": "not their call to make"},
|
|
])
|
|
assert await list_project_shapes(other, pid) == ([], 0)
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_rule_form_sweeps_unclassified_rows_only_and_applies_whole(seeded):
|
|
"""#2868: one judgment over a directory + glob; judged rows are left
|
|
alone unless include_judged; the same gates as the row form."""
|
|
owner, other, pid, sid = seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "settings holder"},
|
|
])
|
|
out = await classify_shapes_where(
|
|
owner, pid, path="src", status="instance", snippet_id=sid, via="audit",
|
|
)
|
|
# make_app + helper swept; Config (already judged) untouched; css not under src/.
|
|
assert out["classified"] == 2
|
|
assert out["sample"] == ["src/app.py::make_app", "src/util.py::helper"]
|
|
rows, _ = await list_project_shapes(owner, pid)
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert by_symbol["make_app"].status == "instance" and by_symbol["make_app"].classified_by == "audit"
|
|
assert by_symbol["Config"].status == "exempt" and by_symbol["Config"].reason == "settings holder"
|
|
assert by_symbol["btn"].status == "unclassified"
|
|
# Glob + kind narrow; include_judged re-judges.
|
|
out = await classify_shapes_where(
|
|
owner, pid, path="web", status="exempt", pattern="btn*", kind="css",
|
|
reason="one toolbar button", include_judged=True,
|
|
)
|
|
assert out["classified"] == 1
|
|
out = await classify_shapes_where(
|
|
owner, pid, path="src", status="unclassified", include_judged=True,
|
|
)
|
|
assert out["classified"] == 3 # withdrawal sweeps judged rows when asked
|
|
# Gates: reason for exempt, snippet for instance, write access, a path.
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes_where(owner, pid, path="src", status="exempt")
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes_where(owner, pid, path="src", status="instance")
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes_where(owner, pid, path="", status="exempt", reason="x")
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes_where(other, pid, path="src", status="exempt", reason="x")
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_sync_stamps_scoped_rows_and_unstamps_when_they_become_reachable(seeded):
|
|
"""#2869: by-construction one-offs arrive `scoped` (mechanical), count as
|
|
accounted, are reached by the sweep's default, and go back to the todo
|
|
if a later tree makes them ordinary. A judgment overrides the stamp."""
|
|
from scribe.services.coverage import ArchiveShape
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
scoped_shapes = [
|
|
ArchiveShape("web/Card.vue", "css", "card", ".card {", "s1", ".card { x: 1 }", True),
|
|
ArchiveShape("web/Card.vue", "sym", "load", "function load() {", "s2", "function load() {}", True),
|
|
ArchiveShape("src/util.py", "sym", "helper", "def helper():", "s3", "def helper(): pass", False),
|
|
]
|
|
await sync_repo_shapes(pid, REPO, scoped_shapes, seen_marker="main")
|
|
rows, _ = await list_project_shapes(owner, pid, path="web/Card.vue")
|
|
assert {r.status for r in rows} == {"scoped"}
|
|
assert all(r.classified_by == "mechanical" and "by construction" in (r.reason or "") for r in rows)
|
|
# The human todo excludes them; the sweep's default still reaches them.
|
|
assert (await list_project_shapes(owner, pid, status="unclassified", path="web/Card.vue"))[1] == 0
|
|
out = await classify_shapes_where(
|
|
owner, pid, path="web/Card.vue", status="instance", snippet_id=sid, kind="css",
|
|
)
|
|
assert out["classified"] == 1
|
|
# Re-synced as ordinary: the stamped sym returns to the todo; the
|
|
# judged css keeps its judgment.
|
|
plain = [ArchiveShape(s.path, s.kind, s.name, s.signature, s.body_sha, s.body, False) for s in scoped_shapes]
|
|
await sync_repo_shapes(pid, REPO, plain, seen_marker="main")
|
|
rows, _ = await list_project_shapes(owner, pid, path="web/Card.vue")
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert by_symbol["load"].status == "unclassified" and by_symbol["load"].classified_by is None
|
|
assert by_symbol["card"].status == "instance" and by_symbol["card"].snippet_id == sid
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_uses_edges_are_the_consumer_map(seeded):
|
|
"""#2870: a shape keeps ONE snippet_id (what it is) and any number of
|
|
uses edges (what it calls); the snippet's consumer map lists them,
|
|
list_shapes(uses=N) finds them, and a sweep can write them."""
|
|
from scribe.services import snippets as snippets_svc
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
helper = await snippets_svc.create_snippet(
|
|
owner, name="cls_hash_helper", code="def hash_token(raw):\n return raw\n",
|
|
language="python", repo="Widget", path="src/hash.py", symbol="hash_token",
|
|
project_id=pid,
|
|
)
|
|
hid = int(helper.id)
|
|
out = await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid, "uses": [hid]},
|
|
], via="audit")
|
|
assert out["classified"] == 1
|
|
rows, total = await list_project_shapes(owner, pid, uses=hid)
|
|
assert total == 1 and rows[0].symbol == "make_app" and rows[0].snippet_id == sid
|
|
consumers = await snippet_consumers(owner, hid)
|
|
assert consumers["instances"] == [] and len(consumers["uses"]) == 1
|
|
assert consumers["uses"][0]["symbol"] == "make_app" and consumers["uses"][0]["basis"] == "audit"
|
|
# A sweep writes uses too; an unknown snippet in uses applies nothing.
|
|
out = await classify_shapes_where(
|
|
owner, pid, path="src/util.py", status="exempt", reason="local", uses=[hid],
|
|
)
|
|
assert out["classified"] == 1
|
|
assert (await list_project_shapes(owner, pid, uses=hid))[1] == 2
|
|
# The proposer writes uses edges for JUDGED rows too: Config (exempt)
|
|
# names hash_token in its body → an edge, no proposal.
|
|
from scribe.services.shape_ledger import propose_for_repo
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "settings"},
|
|
])
|
|
defs = _defs(("src/app.py", "sym", "Config", "class Config:", "class Config:\n token = hash_token(raw)\n"))
|
|
with _quiet_semantic():
|
|
await propose_for_repo(owner, pid, REPO, defs)
|
|
rows, total = await list_project_shapes(owner, pid, uses=hid)
|
|
assert total == 3 and {r.symbol for r in rows} >= {"Config"}
|
|
cfg = next(r for r in rows if r.symbol == "Config")
|
|
assert cfg.status == "exempt" and cfg.proposal is None
|
|
with pytest.raises(ValueError):
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "Config", "status": "exempt", "reason": "x", "uses": [999999]},
|
|
])
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_list_filters_compose(seeded):
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid},
|
|
])
|
|
rows, total = await list_project_shapes(owner, pid, path="src")
|
|
assert total == 3 and all(r.path.startswith("src/") for r in rows)
|
|
# Directory semantics, not string prefix: "sr" matches nothing.
|
|
assert (await list_project_shapes(owner, pid, path="sr"))[1] == 0
|
|
rows, total = await list_project_shapes(owner, pid, snippet_id=sid)
|
|
assert total == 1 and rows[0].symbol == "make_app"
|
|
rows, total = await list_project_shapes(
|
|
owner, pid, status="unclassified", limit=2
|
|
)
|
|
assert total == 3 and len(rows) == 2 # paged, with the true total
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_get_snippet_carries_the_structured_consumer_map(seeded):
|
|
from scribe.mcp._context import _user_id_ctx
|
|
from scribe.mcp.tools.snippets import get_snippet
|
|
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid},
|
|
{"path": "web/button.css", "symbol": "btn", "status": "variant",
|
|
"snippet_id": sid, "reason": "darker focus ring"},
|
|
])
|
|
token = _user_id_ctx.set(owner)
|
|
try:
|
|
data = await get_snippet(snippet_id=sid)
|
|
finally:
|
|
_user_id_ctx.reset(token)
|
|
assert [i["path"] for i in data["instances"]] == ["src/app.py"]
|
|
assert data["variants"][0]["reason"] == "darker focus ring"
|
|
|
|
# The map is caller-scoped: an outsider asking the service directly gets
|
|
# silence, not another project's file layout.
|
|
consumers = await snippet_consumers(seeded["other"], sid)
|
|
assert consumers == {"instances": [], "variants": [], "uses": []}
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
|
|
"""The SET NULL companion (#2787): a judgment whose target is hard-deleted
|
|
rejoins the todo on the next sync instead of dangling target-less."""
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance",
|
|
"snippet_id": sid},
|
|
])
|
|
from scribe.models.note import Note
|
|
|
|
async with async_session() as s:
|
|
note = await s.get(Note, sid)
|
|
await s.delete(note) # hard delete, as purge_trash would
|
|
await s.commit()
|
|
await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="main")
|
|
async with async_session() as s:
|
|
row = (await s.execute(select(CodeShape).where(
|
|
CodeShape.project_id == pid, CodeShape.symbol == "make_app",
|
|
))).scalar_one()
|
|
assert row.status == "unclassified"
|
|
assert row.snippet_id is None
|
|
|
|
|
|
# --- #2791: the write-path feed lands hook evidence as rows -------------------
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_write_path_stamp_is_evidence_that_yields_to_judgment(seeded):
|
|
"""Pulled + referenced → every named shape of the snippet's kind becomes
|
|
an instance row, classified_by=hook, carrying the evidence as reason. A
|
|
later agent judgment on one of them stands against a re-stamp; the hook
|
|
may only overwrite nobody's judgment or its own. The outsider stamps
|
|
nothing (write-gated like every other ledger write)."""
|
|
from datetime import datetime, timezone
|
|
|
|
from scribe.services.shape_ledger import stamp_write_path_instances
|
|
|
|
owner, other, pid, sid = (
|
|
seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
|
|
)
|
|
pulled = {sid: datetime.now(timezone.utc)}
|
|
code = "app = factory()\nreturn app\n" # references the snippet's symbol
|
|
|
|
assert await stamp_write_path_instances(
|
|
other, pid, path="src/app.py", shapes=[("sym", "make_app")],
|
|
code=code, pulled=pulled,
|
|
) == []
|
|
|
|
stamped = await stamp_write_path_instances(
|
|
owner, pid, path="src/app.py",
|
|
shapes=[("sym", "make_app"), ("sym", "Config"), ("css", "nope")],
|
|
code=code, pulled=pulled,
|
|
)
|
|
assert {s["symbol"] for s in stamped} == {"make_app", "Config"} # css skipped: no css canon
|
|
rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert by_symbol["make_app"].status == "instance"
|
|
assert by_symbol["make_app"].classified_by == "hook"
|
|
assert by_symbol["make_app"].reason == f"hook: pulled #{sid}; payload references `factory`"
|
|
|
|
# A judgment lands; the next stamp must leave it alone but may re-stamp
|
|
# its own earlier row.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "exempt",
|
|
"reason": "the app factory is its own thing"},
|
|
])
|
|
again = await stamp_write_path_instances(
|
|
owner, pid, path="src/app.py",
|
|
shapes=[("sym", "make_app"), ("sym", "Config")], code=code, pulled=pulled,
|
|
)
|
|
assert {s["symbol"] for s in again} == {"Config"}
|
|
rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert by_symbol["make_app"].status == "exempt"
|
|
assert by_symbol["Config"].status == "instance"
|
|
|
|
# Neither pulled nor in play → nothing, even with shapes named.
|
|
assert await stamp_write_path_instances(
|
|
owner, pid, path="src/util.py", shapes=[("sym", "helper")],
|
|
code="print('unrelated')", pulled=pulled,
|
|
) == []
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_a_brand_new_shape_gets_a_provisional_row_the_sync_settles(seeded):
|
|
"""The shape being written right now has no ledger row yet. With the
|
|
hook's repo key it gets a provisional one — seen markers empty — so the
|
|
stamp survives until the next sync, which confirms it (sets the marker)
|
|
or stamps it vanished. Without a repo key only existing rows are touched."""
|
|
from datetime import datetime, timezone
|
|
|
|
from scribe.services.shape_ledger import stamp_write_path_instances
|
|
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
pulled = {sid: datetime.now(timezone.utc)}
|
|
code = "def build():\n return factory()\n"
|
|
|
|
assert await stamp_write_path_instances(
|
|
owner, pid, path="src/new.py", shapes=[("sym", "build")],
|
|
code=code, pulled=pulled, # no repo_key
|
|
) == []
|
|
stamped = await stamp_write_path_instances(
|
|
owner, pid, path="src/new.py", shapes=[("sym", "build")],
|
|
code=code, pulled=pulled, repo_key=REPO,
|
|
)
|
|
assert [s["symbol"] for s in stamped] == ["build"]
|
|
async with async_session() as s:
|
|
row = (await s.execute(select(CodeShape).where(
|
|
CodeShape.project_id == pid, CodeShape.path == "src/new.py",
|
|
))).scalar_one()
|
|
assert row.status == "instance" and row.classified_by == "hook"
|
|
# "Unset" is the column's empty default — the markers are non-null
|
|
# Text, and the sync is what first fills them.
|
|
assert row.first_seen_commit == "" and row.last_seen_commit == ""
|
|
|
|
# The sync sees the shape in the tree → confirmed, stamp intact.
|
|
await sync_repo_shapes(
|
|
pid, REPO, SHAPES + [("src/new.py", "sym", "build")], seen_marker="abc123",
|
|
)
|
|
rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
|
|
assert rows[0].status == "instance" and rows[0].last_seen_commit == "abc123"
|
|
|
|
# The sync no longer sees it → vanished, out of the live accounting.
|
|
await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="def456")
|
|
rows, _ = await list_project_shapes(owner, pid, path="src/new.py")
|
|
assert rows == []
|
|
rows, _ = await list_project_shapes(owner, pid, path="src/new.py", include_vanished=True)
|
|
assert rows[0].vanished_at is not None
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_recent_pulls_reads_the_usage_stream(seeded):
|
|
"""The "actually pulled it" half is the PULLED usage event, inside the
|
|
window; a surfacing alone is not a pull."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
|
|
from scribe.services.shape_ledger import recent_pulls
|
|
|
|
owner, sid = seeded["owner"], seeded["snippet"]
|
|
now = datetime.now(timezone.utc)
|
|
async with async_session() as s:
|
|
s.add_all([
|
|
NoteUsageEvent(user_id=owner, note_id=sid, event=PULLED, source="mcp_get_snippet"),
|
|
NoteUsageEvent(user_id=owner, note_id=sid + 1000, event=SURFACED, source="auto_inject"),
|
|
NoteUsageEvent(user_id=owner, note_id=sid + 2000, event=PULLED,
|
|
source="mcp_get_snippet", created_at=now - timedelta(days=2)),
|
|
])
|
|
await s.commit()
|
|
pulls = await recent_pulls(owner)
|
|
assert sid in pulls
|
|
assert sid + 1000 not in pulls
|
|
assert sid + 2000 not in pulls
|
|
|
|
|
|
# --- #2792: the mechanical proposer against real rows -------------------------
|
|
|
|
|
|
def _quiet_semantic():
|
|
"""The semantic basis needs the embedder; these tests prove the other
|
|
bases and the bookkeeping, so it answers "nothing" here."""
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from scribe.services import shape_ledger
|
|
return patch.object(shape_ledger, "_semantic_canon", AsyncMock(return_value=None))
|
|
|
|
|
|
def _defs(*items):
|
|
"""ArchiveShape-like records: (path, kind, name, signature, body_sha, body)."""
|
|
import hashlib
|
|
out = []
|
|
for path, kind, name, signature, body in items:
|
|
sha = hashlib.sha1(" ".join(body.split()).encode()).hexdigest()[:16]
|
|
out.append((path, kind, name, signature, sha, body))
|
|
return out
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_proposer_proposes_and_confirm_classifies(seeded):
|
|
"""Bodies in hand, the proposer records proposals on unclassified rows —
|
|
symbol (a second `factory` elsewhere), reference (a call site), and
|
|
nothing for the unrelated — skips rows whose content it already judged,
|
|
and a scoped confirm turns proposals into agent instances while a
|
|
classify on another retires its proposal."""
|
|
from scribe.services.shape_ledger import (
|
|
confirm_proposals, propose_for_repo,
|
|
)
|
|
|
|
owner, other, pid, sid = (
|
|
seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
|
|
)
|
|
defs = _defs(
|
|
("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n app = factory()\n return app"),
|
|
("src/app.py", "sym", "Config", "class Config:", "class Config:\n debug = False"),
|
|
("src/util.py", "sym", "helper", "def helper(x):", "def helper(x):\n return x"),
|
|
("src/dup.py", "sym", "factory", "def factory():", "def factory():\n return 1"),
|
|
("web/button.css", "css", "btn", ".btn {", ".btn {\n color: red;\n}"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, defs, seen_marker="main")
|
|
|
|
with _quiet_semantic():
|
|
stats = await propose_for_repo(owner, pid, REPO, defs)
|
|
assert stats == {"examined": 5, "proposed": 2, "semantic_checked": 2}
|
|
rows, total = await list_project_shapes(owner, pid, proposal="canon")
|
|
by_symbol = {r.symbol: r for r in rows}
|
|
assert total == 2
|
|
assert by_symbol["factory"].proposal == {"basis": "symbol", "score": 1.0, "snippet_id": sid}
|
|
assert by_symbol["make_app"].proposal == {"basis": "reference", "score": 0.9, "snippet_id": sid}
|
|
rows, _ = await list_project_shapes(owner, pid, proposal="reference")
|
|
assert [r.symbol for r in rows] == ["make_app"]
|
|
|
|
# Same content again → nothing re-examined (the semantic cap would
|
|
# otherwise be spent on the same rows every refresh). A cap that leaves
|
|
# rows unreached leaves them UNexamined, so the next refresh gets them.
|
|
with _quiet_semantic():
|
|
assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 0
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/util.py", "symbol": "helper", "status": "unclassified"},
|
|
])
|
|
assert (await propose_for_repo(owner, pid, REPO, defs, semantic_cap=0))["semantic_checked"] == 0
|
|
assert (await propose_for_repo(owner, pid, REPO, defs))["examined"] == 1
|
|
|
|
# Outsider can't confirm; the owner confirms by snippet, scoped.
|
|
with pytest.raises(ValueError):
|
|
await confirm_proposals(other, pid, snippet_id=sid)
|
|
assert await confirm_proposals(owner, pid, basis="symbol") == {"confirmed": 1}
|
|
rows, _ = await list_project_shapes(owner, pid, snippet_id=sid)
|
|
factory = next(r for r in rows if r.symbol == "factory")
|
|
assert factory.status == "instance" and factory.classified_by == "agent"
|
|
assert factory.reason == "confirmed symbol proposal (1.00)"
|
|
assert factory.proposal is None
|
|
|
|
# A judgment on a proposed row retires the proposal; withdrawing a
|
|
# judgment forgets the examination so the next pass proposes afresh.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "exempt", "reason": "bootstrap"},
|
|
])
|
|
rows, _ = await list_project_shapes(owner, pid, path="src/app.py")
|
|
make_app = next(r for r in rows if r.symbol == "make_app")
|
|
assert make_app.status == "exempt" and make_app.proposal is None
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
|
|
])
|
|
with _quiet_semantic():
|
|
assert (await propose_for_repo(owner, pid, REPO, defs))["proposed"] == 1
|
|
rows, _ = await list_project_shapes(owner, pid, proposal="canon")
|
|
assert [r.symbol for r in rows] == ["make_app"]
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_derive_groups_land_on_rows_and_in_the_summary(seeded):
|
|
"""Shapes with no canon hit that repeat — identical bodies in two files,
|
|
the same name in three — carry a derive proposal, and the readout ranks
|
|
the families. A canon proposal keeps a row out of any derive group."""
|
|
from scribe.services.shape_ledger import (
|
|
apply_derive_groups, live_rows, propose_for_repo, proposal_summary,
|
|
)
|
|
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
defs = _defs(
|
|
("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n return t.lower()"),
|
|
("a/two.py", "sym", "slug", "def slug(t):", "def slug(t):\n return t.lower()"),
|
|
("b/x.css", "css", "card", ".card {", ".card { padding: 1px }"),
|
|
("b/y.css", "css", "card", ".card {", ".card { padding: 2px }"),
|
|
("b/z.css", "css", "card", ".card {", ".card { padding: 3px }"),
|
|
("c/only.py", "sym", "alone", "def alone():", "def alone():\n return 0"),
|
|
("c/use.py", "sym", "boot", "def boot():", "def boot():\n return factory()"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, defs, seen_marker="main")
|
|
with _quiet_semantic():
|
|
await propose_for_repo(owner, pid, REPO, defs)
|
|
assert await apply_derive_groups(pid) == 5
|
|
|
|
rows, total = await list_project_shapes(owner, pid, proposal="derive")
|
|
assert total == 5
|
|
groups = {(r.path, r.symbol): r.proposal for r in rows}
|
|
assert groups[("a/one.py", "slug")]["group"] == groups[("a/two.py", "slug")]["group"]
|
|
assert groups[("a/one.py", "slug")]["group"].startswith("dup:")
|
|
assert groups[("b/x.css", "card")] == {"basis": "derive", "score": 3.0, "group": "name:css:card"}
|
|
rows, _ = await list_project_shapes(owner, pid, proposal="any")
|
|
assert {r.symbol for r in rows} == {"slug", "card", "boot"} # boot: reference proposal
|
|
|
|
summary = proposal_summary(await live_rows(pid))
|
|
assert summary["proposed"] == 1
|
|
# #2872: the body-identical group (a real copy) outranks the bigger name
|
|
# group (usually convention), even at size 2 vs 3.
|
|
order = [g["group"] for g in summary["derive_groups"]]
|
|
assert order[0].startswith("dup:") and order[1] == "name:css:card"
|
|
assert summary["derive_groups"][0]["files"] == 2
|
|
assert summary["derive_groups"][0]["label"] == "slug (identical body)"
|
|
assert summary["derive_groups"][1]["label"] == ".card"
|
|
assert summary["derive_groups"][0]["size"] == 2 and summary["derive_groups"][1]["size"] == 3
|
|
|
|
# One of the css copies gets judged → the group shrinks on the next pass
|
|
# but stays a family: a class in two files is already a recipe living in
|
|
# two places (css name floor 2, note 2917). Judge the second and it's gone.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "b/z.css", "symbol": "card", "status": "exempt", "reason": "print sheet"},
|
|
])
|
|
await apply_derive_groups(pid)
|
|
rows, _ = await list_project_shapes(owner, pid, proposal="derive")
|
|
assert {r.symbol for r in rows} == {"slug", "card"}
|
|
assert {r.path for r in rows if r.symbol == "card"} == {"b/x.css", "b/y.css"}
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "b/y.css", "symbol": "card", "status": "exempt", "reason": "print sheet"},
|
|
])
|
|
await apply_derive_groups(pid)
|
|
rows, _ = await list_project_shapes(owner, pid, proposal="derive")
|
|
assert {r.symbol for r in rows} == {"slug"} # 1 file < the css name floor
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_write_time_derive_names_the_family_or_the_canon_for_a_name(seeded):
|
|
"""#2900: against real rows — a name in a family → the family (other
|
|
files, count; for CSS a NAME family, never a body one — note 2917); a
|
|
name whose canonical row lives elsewhere → that canon; a judged row at
|
|
the path, the canon's own file, or an unknown name → silence."""
|
|
from scribe.services.shape_ledger import apply_derive_groups, write_time_derive
|
|
|
|
owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
|
|
defs = _defs(
|
|
("v/A.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
|
|
("v/B.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
|
|
("v/C.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
|
|
("src/factory.py", "sym", "factory", "def factory():", "def factory():\n return 1"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/factory.py", "symbol": "factory", "status": "canonical", "snippet_id": sid},
|
|
])
|
|
assert await apply_derive_groups(pid) >= 3
|
|
|
|
# A 4th copy about to be written → the family, naming the other files.
|
|
out = await write_time_derive(pid, "v/D.vue", [("css", "log-empty"), ("css", "unknown")])
|
|
assert len(out) == 1 and out[0]["symbol"] == "log-empty" and out[0]["kind"] == "css"
|
|
fam = out[0]["family"]
|
|
# Three identical bodies, and still a NAME family: CSS never groups by body.
|
|
assert fam["identical"] is False and fam["label"] == ".log-empty"
|
|
assert fam["files"] == ["v/A.vue", "v/B.vue", "v/C.vue"] and fam["file_count"] == 3
|
|
assert out[0]["key"] == fam["group"] == "name:css:log-empty"
|
|
# Editing one existing member still names the OTHER members.
|
|
out = await write_time_derive(pid, "v/A.vue", [("css", "log-empty")])
|
|
assert out[0]["family"]["files"] == ["v/B.vue", "v/C.vue"] and out[0]["family"]["size"] == 3
|
|
# The canon's name elsewhere → the canon; in the canon's own file → silence.
|
|
out = await write_time_derive(pid, "src/other.py", [("sym", "factory")])
|
|
assert out == [{"symbol": "factory", "kind": "sym", "key": f"canon:{sid}",
|
|
"canon": {"snippet_id": sid, "path": "src/factory.py", "label": "factory"}}]
|
|
assert await write_time_derive(pid, "src/factory.py", [("sym", "factory")]) == []
|
|
# A judged row at the path is not re-litigated.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "v/B.vue", "symbol": "log-empty", "status": "exempt", "reason": "print sheet"},
|
|
])
|
|
assert await write_time_derive(pid, "v/B.vue", [("css", "log-empty")]) == []
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(seeded):
|
|
"""#2899: the first sync seeds one `slug`; a later sync adds an identical
|
|
copy. Against the stamp between them, derive_new counts ONLY the
|
|
newcomer — the drift since the last refresh, not the whole family."""
|
|
from datetime import datetime, timezone
|
|
|
|
from scribe.services.shape_ledger import (
|
|
apply_derive_groups, derive_new_summary, live_rows,
|
|
)
|
|
|
|
owner, pid = seeded["owner"], seeded["pid"]
|
|
first = _defs(
|
|
("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n return t.lower()"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, first, seen_marker="m1")
|
|
stamp = datetime.now(timezone.utc)
|
|
second = _defs(
|
|
("a/one.py", "sym", "slug", "def slug(t):", "def slug(t):\n return t.lower()"),
|
|
("a/two.py", "sym", "slug", "def slug(t):", "def slug(t):\n return t.lower()"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, second, seen_marker="m2")
|
|
assert await apply_derive_groups(pid) == 2
|
|
|
|
rows = await live_rows(pid)
|
|
out = derive_new_summary(rows, since=stamp)
|
|
assert out["count"] == 1
|
|
assert out["examples"][0]["path"] == "a/two.py"
|
|
assert out["examples"][0]["label"] == "slug"
|
|
assert out["examples"][0]["group"].startswith("dup:")
|
|
assert derive_new_summary(rows, since=None)["count"] == 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_transition_and_prefix_references_clear_the_unused_css_flag(seeded):
|
|
"""#2970 end to end: the two class forms a template never spells out —
|
|
a transition `name=` and a concatenated name — travel from the real
|
|
extractor through resolution into the flag, so rules they reach stop
|
|
being reported as unused. Only a rule nothing can reach stays listed."""
|
|
from scribe.services.coverage import class_references
|
|
from scribe.services.shape_ledger import sync_repo_consumers, used_by_map, live_rows
|
|
|
|
pid, owner = seeded["pid"], seeded["owner"]
|
|
defs = _defs(
|
|
("assets/anim.css", "css", "toast-enter-active",
|
|
".toast-enter-active {", ".toast-enter-active { opacity: 0 }"),
|
|
("assets/anim.css", "css", "toast-leave-to",
|
|
".toast-leave-to {", ".toast-leave-to { opacity: 0 }"),
|
|
("assets/state.css", "css", "status-done",
|
|
".status-done {", ".status-done { color: green }"),
|
|
("assets/state.css", "css", "status-todo",
|
|
".status-todo {", ".status-todo { color: grey }"),
|
|
("assets/state.css", "css", "really-dead",
|
|
".really-dead {", ".really-dead { color: red }"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, defs, seen_marker="t1")
|
|
|
|
refs = {
|
|
# Vue applies the toast-* classes itself; the markup names none of them.
|
|
"v/Toast.vue": class_references(
|
|
"v/Toast.vue", '<transition-group name="toast"><li /></transition-group>'
|
|
),
|
|
# The board builds its status class; which one is unknowable.
|
|
"v/Board.vue": class_references("v/Board.vue", '<b :class="`status-${s}`" />'),
|
|
}
|
|
assert "toast-enter-active" in refs["v/Toast.vue"]
|
|
assert refs["v/Board.vue"] == {"status-*": 1}
|
|
|
|
await sync_repo_consumers(pid, REPO, refs)
|
|
live = [r for r in await live_rows(pid) if r.kind == "css"]
|
|
used = await used_by_map(live)
|
|
by = {(r.path, r.symbol): used[r.id]["count"] for r in live}
|
|
assert by[("assets/anim.css", "toast-enter-active")] == 1
|
|
assert by[("assets/anim.css", "toast-leave-to")] == 1
|
|
# the prefix credits BOTH candidates — the template does not say which
|
|
assert by[("assets/state.css", "status-done")] == 1
|
|
assert by[("assets/state.css", "status-todo")] == 1
|
|
assert by[("assets/state.css", "really-dead")] == 0
|
|
|
|
unused, n = await list_project_shapes(owner, pid, flag="unused-css")
|
|
assert n == 1 and [(r.path, r.symbol) for r in unused] == [
|
|
("assets/state.css", "really-dead")
|
|
]
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_consumer_map_syncs_edges_from_template_references(seeded):
|
|
"""Milestone 302: the consumer edges follow the archive — own-file
|
|
resolution for a scoped class, fan-out to the shared sheet for a class a
|
|
template does not define, counts refreshed and stale edges removed on
|
|
the next sync, and a vanished row's edges gone with it."""
|
|
from scribe.services.shape_ledger import consumers_of, live_rows, sync_repo_consumers
|
|
|
|
pid = seeded["pid"]
|
|
defs = _defs(
|
|
("v/A.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: red }"),
|
|
("v/B.vue", "css", "error-msg", ".error-msg {", ".error-msg { color: blue }"),
|
|
("assets/components.css", "css", "btn-primary", ".btn-primary {", ".btn-primary { x: 1 }"),
|
|
("assets/orphan.css", "css", "orphan", ".orphan {", ".orphan { y: 2 }"), # no template names it
|
|
)
|
|
await sync_repo_shapes(pid, REPO, defs, seen_marker="m1")
|
|
refs = {
|
|
"v/A.vue": {"error-msg": 2, "btn-primary": 1},
|
|
"v/B.vue": {"error-msg": 1},
|
|
"v/C.vue": {"error-msg": 1, "btn-primary": 4},
|
|
}
|
|
# A and B consume their OWN error-msg; C defines none, so its use fans
|
|
# out to both rows; btn-primary resolves to the shared sheet from A and C.
|
|
assert await sync_repo_consumers(pid, REPO, refs) == 6
|
|
rows = {(r.path, r.symbol): r.id for r in await live_rows(pid) if r.kind == "css"}
|
|
edges = await consumers_of(rows.values())
|
|
view = {(p, s): [(e.path, e.count) for e in edges.get(i, [])] for (p, s), i in rows.items()}
|
|
assert view[("v/A.vue", "error-msg")] == [("v/A.vue", 2), ("v/C.vue", 1)]
|
|
assert view[("v/B.vue", "error-msg")] == [("v/B.vue", 1), ("v/C.vue", 1)]
|
|
assert view[("assets/components.css", "btn-primary")] == [("v/A.vue", 1), ("v/C.vue", 4)]
|
|
assert view[("assets/orphan.css", "orphan")] == []
|
|
|
|
# The next tree: C stops using error-msg, A uses btn-primary twice now.
|
|
refs2 = {"v/A.vue": {"error-msg": 2, "btn-primary": 2}, "v/B.vue": {"error-msg": 1}}
|
|
assert await sync_repo_consumers(pid, REPO, refs2) == 3
|
|
edges = await consumers_of(rows.values())
|
|
assert [(e.path, e.count) for e in edges[rows[("v/B.vue", "error-msg")]]] == [("v/B.vue", 1)]
|
|
assert [(e.path, e.count) for e in edges[rows[("assets/components.css", "btn-primary")]]] == [("v/A.vue", 2)]
|
|
|
|
# The readout side: used_by per css row, the unused-css flag, and the
|
|
# family's consumers on the write-path check.
|
|
from scribe.services.shape_ledger import (
|
|
apply_derive_groups, used_by_map, write_time_derive,
|
|
)
|
|
owner = seeded["owner"]
|
|
live = [r for r in await live_rows(pid) if r.kind == "css"]
|
|
used = await used_by_map(live)
|
|
assert used[rows[("v/B.vue", "error-msg")]] == {"count": 1, "paths": ["v/B.vue"]}
|
|
assert used[rows[("assets/orphan.css", "orphan")]] == {"count": 0, "paths": []}
|
|
unused, n = await list_project_shapes(owner, pid, flag="unused-css")
|
|
assert n == 1 and [(r.path, r.symbol) for r in unused] == [("assets/orphan.css", "orphan")]
|
|
await apply_derive_groups(pid)
|
|
out = await write_time_derive(pid, "v/New.vue", [("css", "error-msg")])
|
|
assert out and out[0]["family"]["consumers"] == {"count": 2, "paths": ["v/A.vue", "v/B.vue"]}
|
|
|
|
# B's rule vanishes from the tree → its edges go with the pass.
|
|
await sync_repo_shapes(pid, REPO, [d for d in defs if d[0] != "v/B.vue"], seen_marker="m2")
|
|
await sync_repo_consumers(pid, REPO, refs2)
|
|
edges = await consumers_of(rows.values())
|
|
assert rows[("v/B.vue", "error-msg")] not in edges
|
|
|
|
|
|
# --- #2793: the divergence readout against real rows -------------------------
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_a_second_confirm_dialog_is_detected_and_named(seeded):
|
|
"""The milestone's acceptance case. A directory where one canon dominates
|
|
the judged siblings (a confirm helper with four instance call sites);
|
|
after a previous refresh, a new shape lands there that the proposer does
|
|
not match to the canon — it is flagged `diverges_from` the canon, the
|
|
readout names it, and the in-band check names it at write time. A
|
|
judgment clears the flag; a shape proposed AS the canon is not flagged."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from scribe.services.shape_ledger import (
|
|
divergence_summary, flag_divergence, live_rows, propose_for_repo,
|
|
write_time_divergence,
|
|
)
|
|
|
|
from scribe.services import snippets as snippets_svc
|
|
|
|
owner, pid = seeded["owner"], seeded["pid"]
|
|
# The confirm helper is TS canon: since #2871 a sym basis only proposes
|
|
# within the shape's language family, so the fixture's Python snippet
|
|
# says nothing about these Vue bodies — the canon must be one of theirs.
|
|
canon = await snippets_svc.create_snippet(
|
|
owner, name="cls_confirm_factory",
|
|
code="export async function factory(): Promise<boolean> {\n return true;\n}\n",
|
|
language="typescript", repo="Widget",
|
|
path="frontend/src/composables/useConfirm.ts", symbol="factory",
|
|
project_id=pid,
|
|
)
|
|
sid = int(canon.id)
|
|
comp = "frontend/src/components"
|
|
base = _defs(
|
|
*[(f"{comp}/{n}.vue", "sym", f"on{n}", f"async function on{n}() {{",
|
|
f"async function on{n}() {{\n const ok = await factory();\n if (!ok) return;\n}}")
|
|
for n in ("Trash", "Delete", "Remove", "Restore")],
|
|
)
|
|
await sync_repo_shapes(pid, REPO, base, seen_marker="aaa111")
|
|
await classify_shapes(owner, pid, [
|
|
{"path": f"{comp}/{n}.vue", "symbol": f"on{n}", "status": "instance", "snippet_id": sid}
|
|
for n in ("Trash", "Delete", "Remove", "Restore")
|
|
], via="audit")
|
|
previous = datetime.now(timezone.utc)
|
|
|
|
# Button B: a hand-rolled confirm that never touches the canon, plus a
|
|
# proper new instance (references the canon → the proposer claims it).
|
|
later = base + _defs(
|
|
(f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
|
|
"function confirmDanger() {\n return window.confirm('Really?');\n}"),
|
|
(f"{comp}/Proper.vue", "sym", "onPurge", "async function onPurge() {",
|
|
"async function onPurge() {\n const ok = await factory();\n if (!ok) return;\n}"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
|
|
with _quiet_semantic():
|
|
await propose_for_repo(owner, pid, REPO, later)
|
|
assert await flag_divergence(pid, since=None) == 0 # a first seed flags nothing
|
|
assert await flag_divergence(pid, since=previous - timedelta(seconds=1)) == 1
|
|
|
|
rows, total = await list_project_shapes(owner, pid, flag="divergence")
|
|
assert total == 1
|
|
assert rows[0].symbol == "confirmDanger" and rows[0].diverges_from == sid
|
|
summary = divergence_summary(await live_rows(pid))
|
|
assert summary["divergent"] == 1
|
|
assert summary["divergence"][0]["symbol"] == "confirmDanger"
|
|
assert summary["divergence"][0]["canon_snippet_id"] == sid
|
|
|
|
# In-band: the hook names the shape at write time → the check names the canon.
|
|
named = await write_time_divergence(
|
|
pid, f"{comp}/Danger.vue", [("sym", "confirmDanger")], stamped=[]
|
|
)
|
|
assert named == [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": sid,
|
|
"instances": 4, "judged": 4}]
|
|
# ...but an already-judged shape, or one just stamped as the canon's
|
|
# instance, is not re-litigated.
|
|
assert await write_time_divergence(pid, f"{comp}/Trash.vue", [("sym", "onTrash")], stamped=[]) == []
|
|
assert await write_time_divergence(
|
|
pid, f"{comp}/New.vue", [("sym", "onNew")],
|
|
stamped=[{"symbol": "onNew", "kind": "sym", "snippet_id": sid}],
|
|
) == []
|
|
# A directory with no dominant canon is silent.
|
|
assert await write_time_divergence(pid, "src/other.py", [("sym", "thing")], stamped=[]) == []
|
|
|
|
# The judgment answers the question and clears the flag.
|
|
await classify_shapes(owner, pid, [
|
|
{"path": f"{comp}/Danger.vue", "symbol": "confirmDanger", "status": "variant",
|
|
"snippet_id": sid, "reason": "native confirm is fine in the dev-only panel"},
|
|
])
|
|
rows, total = await list_project_shapes(owner, pid, flag="divergence")
|
|
assert total == 0
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_a_conclusive_meaning_miss_silences_what_the_signature_cannot(seeded):
|
|
"""#4208: the four false prompts #4204's form gate provably cannot reach.
|
|
|
|
THE FIXTURE IS THE ACCEPTANCE CASE ABOVE, DELIBERATELY. That is the whole
|
|
difficulty of this issue: a hand-rolled `confirmDanger` beside an async
|
|
confirm canon is structurally IDENTICAL to a registry helper beside an
|
|
async service canon — same family, same form contradiction, same directory
|
|
density. The form gate has to keep asking about both, so nothing derived
|
|
from a signature can separate them. The only difference is whether the
|
|
shape does the canon's JOB, and the only reading of that the ledger holds
|
|
is the proposer's per-symbol body comparison.
|
|
|
|
So the two runs differ in exactly one thing. In the test above the semantic
|
|
arm is quiet — it answers "nothing" without claiming to have looked — and
|
|
the prompt is RAISED, which is what milestone #2793 exists to produce. Here
|
|
it answers "I compared this body against the canons in its family and it is
|
|
none of them", and the prompt is WITHHELD. Holding the fixture identical is
|
|
what makes this a test of the meaning gate rather than of the setup.
|
|
|
|
Asserted on the stored basis as well as the outcome, so that a future
|
|
change which silences the prompt for some other reason fails here instead
|
|
of reading as a pass.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from scribe.services import shape_ledger
|
|
from scribe.services import snippets as snippets_svc
|
|
from scribe.services.shape_ledger import (
|
|
BASIS_NO_SEMANTIC_MATCH, flag_divergence, live_rows, propose_for_repo,
|
|
)
|
|
|
|
owner, pid = seeded["owner"], seeded["pid"]
|
|
canon = await snippets_svc.create_snippet(
|
|
owner, name="cls_confirm_factory_meaning",
|
|
code="export async function factory(): Promise<boolean> {\n return true;\n}\n",
|
|
language="typescript", repo="Widget",
|
|
path="frontend/src/composables/useConfirm.ts", symbol="factory",
|
|
project_id=pid,
|
|
)
|
|
sid = int(canon.id)
|
|
comp = "frontend/src/components"
|
|
base = _defs(
|
|
*[(f"{comp}/{n}.vue", "sym", f"on{n}", f"async function on{n}() {{",
|
|
f"async function on{n}() {{\n const ok = await factory();\n if (!ok) return;\n}}")
|
|
for n in ("Trash", "Delete", "Remove", "Restore")],
|
|
)
|
|
await sync_repo_shapes(pid, REPO, base, seen_marker="aaa111")
|
|
await classify_shapes(owner, pid, [
|
|
{"path": f"{comp}/{n}.vue", "symbol": f"on{n}", "status": "instance", "snippet_id": sid}
|
|
for n in ("Trash", "Delete", "Remove", "Restore")
|
|
], via="audit")
|
|
previous = datetime.now(timezone.utc)
|
|
|
|
later = base + _defs(
|
|
(f"{comp}/Danger.vue", "sym", "confirmDanger", "function confirmDanger() {",
|
|
"function confirmDanger() {\n return window.confirm('Really?');\n}"),
|
|
)
|
|
await sync_repo_shapes(pid, REPO, later, seen_marker="bbb222")
|
|
|
|
def _conclusive_miss(*_args, report=None, **_kw):
|
|
"""The arm ran, compared, and found no canon — the one empty answer
|
|
that is evidence. `_semantic_canon` itself decides when it may say
|
|
this (a result set shorter than the limit); the unit tests for that
|
|
judgment are in tests/test_divergence_meaning_gate.py."""
|
|
if report is not None:
|
|
report["conclusive"] = True
|
|
return None
|
|
|
|
with patch.object(shape_ledger, "_semantic_canon",
|
|
AsyncMock(side_effect=_conclusive_miss)):
|
|
await propose_for_repo(owner, pid, REPO, later)
|
|
|
|
rows = await live_rows(pid)
|
|
danger = next(r for r in rows if r.symbol == "confirmDanger")
|
|
assert danger.proposal_basis == BASIS_NO_SEMANTIC_MATCH
|
|
# The miss is not a proposal: nothing may read it as one.
|
|
assert danger.proposed_snippet_id is None
|
|
|
|
assert await flag_divergence(pid, since=previous - timedelta(seconds=1)) == 0
|
|
_, total = await list_project_shapes(owner, pid, flag="divergence")
|
|
assert total == 0, "a shape the proposer measured as unrelated must not be urged"
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_history_records_what_was_used_when_and_drift_asks_for_a_recheck(seeded):
|
|
from scribe.services.shape_ledger import shape_history
|
|
|
|
owner, other, pid, sid = (
|
|
seeded["owner"], seeded["other"], seeded["pid"], seeded["snippet"]
|
|
)
|
|
v1 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n return factory()"))
|
|
await sync_repo_shapes(pid, REPO, v1, seen_marker="c1")
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "instance", "snippet_id": sid},
|
|
])
|
|
# The body moves under the judgment → drifted + recheck; re-judging clears it.
|
|
v2 = _defs(("src/app.py", "sym", "make_app", "def make_app():", "def make_app():\n return factory(debug=True)"))
|
|
await sync_repo_shapes(pid, REPO, v2, seen_marker="c2")
|
|
rows, total = await list_project_shapes(owner, pid, flag="recheck")
|
|
assert total == 1 and rows[0].symbol == "make_app" and rows[0].status == "instance"
|
|
await classify_shapes(owner, pid, [
|
|
{"path": "src/app.py", "symbol": "make_app", "status": "variant", "snippet_id": sid,
|
|
"reason": "debug flag is deliberate here"},
|
|
])
|
|
rows, total = await list_project_shapes(owner, pid, flag="recheck")
|
|
assert total == 0
|
|
# Then it vanishes from the tree.
|
|
await sync_repo_shapes(pid, REPO, [], seen_marker="c3")
|
|
|
|
history = await shape_history(owner, pid, "src/app.py", symbol="make_app")
|
|
shape = history["shapes"][0]
|
|
assert shape["status"] == "variant" and shape["vanished_at"] is not None
|
|
# The seeded fixture synced this row first (marker "main"); v1/v2 are
|
|
# later sightings — first_seen keeps the first.
|
|
assert shape["first_seen_commit"] == "main" and shape["last_seen_commit"] == "c2"
|
|
timeline = [(e["event"], e["status"], e["snippet_id"], e["commit"]) for e in history["events"]]
|
|
assert timeline == [
|
|
("classified", "instance", sid, "c1"),
|
|
("drifted", "instance", sid, "c2"),
|
|
("classified", "variant", sid, "c2"),
|
|
("vanished", "variant", sid, "c2"),
|
|
]
|
|
assert history["events"][2]["reason"] == "debug flag is deliberate here"
|
|
assert history["events"][0]["classified_by"] == "agent"
|
|
# Directory-wide read works (the empty sync also vanished the seeded
|
|
# Config and helper rows under src/ — two more events); an outsider
|
|
# reads nothing.
|
|
assert len((await shape_history(owner, pid, "src"))["events"]) == 6
|
|
assert await shape_history(other, pid, "src/app.py") == {}
|