"""The adoption ledger against real Postgres (milestone 463 step 4). What the unit lane can't show: an answer moves the row and logs exactly one decision; an owed answer files its task in the OWING project, tagged to the System matching the idea's, and the task follows the answer from there; a version bump leaves every older answer reading as needing a recheck; each branch of the conflict order folds the losing side into the note and settles both rows; and the same assessment given twice records nothing the second time. Precedent search ranks by meaning and the lane has no embedding model, so the search is stubbed to "nothing similar" (`_no_meaning`). The same-idea precedents — another project's answer to this idea — need no embedder, and are tested here. """ from unittest.mock import AsyncMock, patch import pytest import pytest_asyncio from sqlalchemy import select from scribe.models import async_session from scribe.models.canonical_system import CanonicalSystem from scribe.models.family import FamilyAdoption, FamilyDecision, FamilyIdea, Platform, ProjectPlatform from scribe.models.note import Note from scribe.models.project import Project from scribe.models.system import RecordSystem, System from scribe.models.task_log import TaskLog from scribe.models.user import User from scribe.services import family as family_svc from scribe.services import family_adoption as adoption_svc from tests.helpers import ensure_user pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")] OWNER = "family_adoption_owner" OUTSIDER = "family_adoption_outsider" CRITERIA = { "platform_terms": "stated for any Android app that distributes its own APK", "platform_problem": "signature continuity is the platform's rule, not one app's", "proven": "shipped and updated in place on a device", } async def _purge(username: str) -> None: """SETUP ONLY, as the promotion siblings do: a database call after a `yield` in an autouse fixture orphans a pooled connection.""" async with async_session() as s: for user in (await s.execute(select(User).where(User.username == username))).scalars(): for note in (await s.execute(select(Note).where(Note.user_id == user.id))).scalars(): await s.delete(note) for project in (await s.execute(select(Project).where(Project.user_id == user.id))).scalars(): await s.delete(project) await s.commit() @pytest_asyncio.fixture(autouse=True) async def _clean(): await _purge(OWNER) await _purge(OUTSIDER) @pytest.fixture(autouse=True) def _no_meaning(): with patch("scribe.services.embeddings.semantic_search_notes", AsyncMock(return_value=[])): yield async def _platform_id(slug: str) -> int: async with async_session() as s: return await s.scalar(select(Platform.id).where( Platform.slug == slug, Platform.deleted_at.is_(None))) @pytest_asyncio.fixture async def family(): """Two Android apps and a Go service. The idea is a note in the first app, filed under a System mapped to a canonical area; the second app has its own System mapped to the same area, and one that is not. Promoted, so both apps hold an `unassessed` row.""" android, go = await _platform_id("android-app"), await _platform_id("go") async with async_session() as s: canonical = await s.scalar(select(CanonicalSystem.id).where( CanonicalSystem.deleted_at.is_(None)).order_by(CanonicalSystem.id).limit(1)) owner = await ensure_user(s, OWNER) outsider = await ensure_user(s, OUTSIDER) a = Project(user_id=owner.id, title="android one") b = Project(user_id=owner.id, title="android two") c = Project(user_id=owner.id, title="go service") s.add_all([a, b, c]) await s.flush() s.add_all([ ProjectPlatform(project_id=a.id, platform_id=android, state="declared"), ProjectPlatform(project_id=b.id, platform_id=android, state="declared"), ProjectPlatform(project_id=c.id, platform_id=go, state="declared"), ]) a_sys = System(user_id=owner.id, project_id=a.id, name="Release", canonical_id=canonical) b_sys = System(user_id=owner.id, project_id=b.id, name="Shipping", canonical_id=canonical) b_other = System(user_id=owner.id, project_id=b.id, name="Unrelated") pattern = Note(user_id=owner.id, project_id=a.id, title="Signed APK lane", body="One keystore, two channels, in-place update.") s.add_all([a_sys, b_sys, b_other, pattern]) await s.flush() s.add(RecordSystem(note_id=pattern.id, system_id=a_sys.id)) await s.commit() ids = {"owner": owner.id, "outsider": outsider.id, "a": a.id, "b": b.id, "c": c.id, "note": pattern.id, "b_sys": b_sys.id, "b_other": b_other.id} out = await family_svc.promote( ids["owner"], ids["note"], applies_when="any Android app that ships its own APK", platforms=["android-app"], criteria=CRITERIA, evidence=["CI green on the release lane"], reason="proven, stated for the platform", ) assert out["promoted"] is True return ids async def _assess(f, project: str, outcome: str, reason: str = "the gap", **kw): if outcome == "adopted": kw.setdefault("evidence", ["app/build.gradle: the release lane"]) return await adoption_svc.assess(f["owner"], f[project], f["note"], outcome=outcome, reason=reason, **kw) async def _row(f, project: str) -> FamilyAdoption: async with async_session() as s: return (await s.execute(select(FamilyAdoption).where( FamilyAdoption.project_id == f[project], FamilyAdoption.idea_id == f["note"]))).scalars().one() async def _task(task_id: int) -> Note: async with async_session() as s: return await s.get(Note, task_id) async def _decisions(f, project: str) -> list[FamilyDecision]: async with async_session() as s: return list((await s.execute(select(FamilyDecision).where( FamilyDecision.idea_id == f["note"], FamilyDecision.project_id == f[project]) .order_by(FamilyDecision.id))).scalars().all()) # --- each state change ---------------------------------------------------------------- @pytest.mark.parametrize("outcome", ("adopted", "variant", "exempt")) async def test_an_answer_moves_the_row_and_logs_one_decision(family, outcome): out = await _assess(family, "b", outcome, reason="a fact about android two") row = await _row(family, "b") assert (row.status, row.reason, row.canon_version, row.decided_via) == ( outcome, "a fact about android two", 1, "agent") assert row.assessed_at is not None and row.owed_task_id is None [decision] = await _decisions(family, "b") assert decision.action == "assess" assert decision.before == {"status": "unassessed", "reason": "", "canon_version": None} assert decision.after["status"] == outcome assert out["changed"] is True and out["owed_task"] is None async def test_owed_files_a_task_in_the_owing_project_tagged_to_the_matching_system(family): out = await _assess(family, "b", "owed", reason="no in-place update yet") row = await _row(family, "b") assert row.status == "owed" and row.owed_task_id == out["owed_task"]["id"] task = await _task(row.owed_task_id) # Filed where the work is owed — never in the idea's own project. assert task.project_id == family["b"] and task.status == "todo" assert f"#{family['note']}" in task.body and "no in-place update yet" in task.body assert "any Android app that ships its own APK" in task.body async with async_session() as s: tagged = set((await s.execute(select(RecordSystem.system_id).where( RecordSystem.note_id == task.id))).scalars().all()) assert tagged == {family["b_sys"]} async def test_the_owed_task_follows_the_answer(family): await _assess(family, "b", "owed", reason="not built") task_id = (await _row(family, "b")).owed_task_id await _assess(family, "b", "variant", reason="installs through a managed store") assert (await _task(task_id)).status == "cancelled" out = await _assess(family, "b", "owed", reason="the store plan fell through") assert out["owed_task"] == {"id": task_id, "title": (await _task(task_id)).title, "status": "todo", "action": "reopened"} await _assess(family, "b", "adopted", reason="built") assert (await _task(task_id)).status == "done" async with async_session() as s: logs = (await s.execute(select(TaskLog.content).where(TaskLog.task_id == task_id) .order_by(TaskLog.id))).scalars().all() assert [("variant" in logs[0]), ("owed again" in logs[1]), ("adopted" in logs[2])] == [True] * 3 # One task across the whole life of the answer. async with async_session() as s: filed = (await s.execute(select(Note.id).where( Note.project_id == family["b"], Note.title.like("Adopt the family idea%")))).scalars().all() assert filed == [task_id] async def test_the_same_assessment_twice_records_nothing_the_second_time(family): first = await _assess(family, "b", "owed", reason="not built") second = await _assess(family, "b", "owed", reason="not built") assert first["changed"] is True and second["changed"] is False assert second["decision"] is None assert second["owed_task"]["id"] == first["owed_task"]["id"] assert len(await _decisions(family, "b")) == 1 async def test_a_departure_without_a_reason_writes_nothing(family): with pytest.raises(ValueError, match="exempt needs a reason"): await _assess(family, "b", "exempt", reason=" ") with pytest.raises(ValueError, match="adopted needs evidence"): await _assess(family, "b", "adopted", evidence=[]) assert (await _row(family, "b")).status == "unassessed" assert await _decisions(family, "b") == [] async def test_an_idea_that_does_not_reach_the_project_is_refused(family): with pytest.raises(ValueError, match="not on any of"): await _assess(family, "c", "exempt", reason="it is a Go service") async def test_only_canon_is_assessed(family): await family_svc.retire(family["owner"], family["note"], reason="superseded") with pytest.raises(ValueError, match="not family canon"): await _assess(family, "a", "adopted") async def test_someone_who_cannot_write_the_project_cannot_answer_for_it(family): with pytest.raises(ValueError, match="no write access"): await adoption_svc.assess(family["outsider"], family["b"], family["note"], outcome="exempt", reason="not mine to say") async def test_another_projects_answer_is_recorded_as_precedent(family): first = await _assess(family, "a", "adopted", reason="the source app") second = await _assess(family, "b", "owed", reason="not built") assert first["decision"]["id"] in second["decision"]["precedent_ids"] assert second["precedents"][0]["relation"] == "same idea, another project" assert second["precedents"][0]["project_title"] == "android one" # --- recheck -------------------------------------------------------------------------- async def test_a_revision_leaves_every_older_answer_needing_a_recheck(family): await _assess(family, "a", "adopted") await _assess(family, "b", "exempt", reason="ships through a store") out = await family_svc.revise(family["owner"], family["note"], reason="the keystore now rotates", evidence=["incident"]) assert out["idea"]["canon_version"] == 2 rows = await adoption_svc.list_adoptions(family["owner"], idea_id=family["note"]) assert {r["project_title"]: r["needs_recheck"] for r in rows} == { "android one": True, "android two": True} assert [r["project_title"] for r in await adoption_svc.list_adoptions( family["owner"], idea_id=family["note"], recheck_only=True)] == ["android one", "android two"] await _assess(family, "a", "adopted", reason="rotation added") rows = {r["project_title"]: r for r in await adoption_svc.list_adoptions( family["owner"], idea_id=family["note"])} assert rows["android one"]["needs_recheck"] is False assert rows["android one"]["canon_version"] == 2 assert rows["android two"]["needs_recheck"] is True async def test_a_revision_needs_canon_and_a_reason(family): with pytest.raises(ValueError, match="needs a reason"): await family_svc.revise(family["owner"], family["note"], reason="") with pytest.raises(ValueError, match="at least one platform"): await family_svc.revise(family["owner"], family["note"], reason="x", platforms=[]) # --- undo ------------------------------------------------------------------------------ async def test_undoing_an_owed_answer_restores_the_row_and_closes_its_task(family): out = await _assess(family, "b", "owed", reason="not built") task_id = out["owed_task"]["id"] decisions = await family_svc.list_decisions(family["owner"], project_id=family["b"]) assert decisions[0]["undoable"] is True and decisions[0]["project_title"] == "android two" await family_svc.undo(family["owner"], out["decision"]["id"], reason="assessed too early") row = await _row(family, "b") assert (row.status, row.reason, row.canon_version, row.assessed_at) == ( "unassessed", None, None, None) assert (await _task(task_id)).status == "cancelled" undo_row = (await _decisions(family, "b"))[-1] assert undo_row.action == "undo" and undo_row.precedent_ids == [out["decision"]["id"]] async def test_only_the_latest_answer_is_undoable(family): first = await _assess(family, "b", "owed", reason="not built") await _assess(family, "b", "adopted", reason="built") with pytest.raises(ValueError, match="came after it"): await family_svc.undo(family["owner"], first["decision"]["id"], reason="x") # --- the conflict order ------------------------------------------------------------- def _checked(ground: str) -> dict: keys = adoption_svc.CONFLICT_KEYS return {k: f"{k} does not apply here" for k in keys[:keys.index(ground)]} async def _resolve(f, ground: str, **kw): kw.setdefault("evidence", ["named in the record"]) return await adoption_svc.resolve_conflict( f["owner"], f["note"], canon_project_id=f["a"], other_project_id=f["b"], ground=ground, grounds_checked=_checked(ground), fold="Android two signs in CI with a throwaway key.", reason="settled", **kw) @pytest.mark.parametrize("ground,heading,other_outcome", [ ("operator_stance", "### Alternative — android two's approach", "owed"), ("covers_failure", "### Trap — android two's approach", "owed"), ("most_recent_complete", "### Alternative — android two's approach", "owed"), ]) async def test_a_side_that_loses_is_folded_in_and_owes_the_canon(family, ground, heading, other_outcome): out = await _resolve(family, ground) async with async_session() as s: body = (await s.get(Note, family["note"])).body idea = await s.get(FamilyIdea, family["note"]) assert heading in body and "throwaway key" in body assert idea.canon_version == 2 assert out["decision"]["action"] == "revise" assert out["decision"]["evidence"]["conflict"]["ground"] == ground a, b = await _row(family, "a"), await _row(family, "b") assert (a.status, a.canon_version) == ("adopted", 2) assert (b.status, b.canon_version) == (other_outcome, 2) assert (await _task(b.owed_task_id)).project_id == family["b"] # Each answer names the revision as the decision it followed. assert (await _decisions(family, "b"))[-1].precedent_ids == [out["decision"]["id"]] async def test_a_split_makes_each_side_canon_under_its_condition(family): out = await _resolve(family, "split_by_condition", evidence=None, conditions={"canon": "the app updates itself", "other": "a managed store installs it"}) async with async_session() as s: body = (await s.get(Note, family["note"])).body assert "### When a managed store installs it — android two's approach" in body assert "When the app updates itself, android one's approach above is the canon" in body a, b = await _row(family, "a"), await _row(family, "b") assert (a.status, b.status) == ("adopted", "adopted") assert b.owed_task_id is None and out["owed_task"] is None async def test_a_ground_cannot_skip_the_order_and_nothing_is_written(family): async with async_session() as s: body_before = (await s.get(Note, family["note"])).body with pytest.raises(ValueError, match="'operator_stance' comes before 'covers_failure'"): await adoption_svc.resolve_conflict( family["owner"], family["note"], canon_project_id=family["a"], other_project_id=family["b"], ground="covers_failure", grounds_checked={}, fold="x", evidence=["incident"], reason="settled") async with async_session() as s: assert (await s.get(Note, family["note"])).body == body_before assert (await s.get(FamilyIdea, family["note"])).canon_version == 1 # --- the matrix ----------------------------------------------------------------------- async def test_the_matrix_shows_every_member_and_who_has_not_been_asked(family): android = await _platform_id("android-app") async with async_session() as s: late = Project(user_id=family["owner"], title="android three") s.add(late) await s.flush() s.add(ProjectPlatform(project_id=late.id, platform_id=android, state="detected")) await s.commit() late_id = late.id await _assess(family, "b", "owed", reason="not built") matrix = await adoption_svc.adoption_matrix(family["owner"]) cells = {(c["project_title"]): c for c in matrix["cells"] if c["idea_id"] == family["note"]} assert set(cells) == {"android one", "android two", "android three"} assert cells["android two"]["owed_task"]["status"] == "todo" assert cells["android three"]["reached"] is False and cells["android three"]["status"] == "unassessed" assert [o["key"] for o in matrix["outcomes"]] == list(adoption_svc.OUTCOME_KEYS) # A late member's first answer creates its row. await adoption_svc.assess(family["owner"], late_id, family["note"], outcome="exempt", reason="ships through a store") one = await adoption_svc.adoption_matrix(family["owner"], project_id=late_id) assert [c["status"] for c in one["cells"]] == ["exempt"] # Filtered to a platform no canon idea is for, there is nothing to show. assert (await adoption_svc.adoption_matrix(family["owner"], platform="go"))["ideas"] == [] async def test_the_matrix_hides_what_the_caller_cannot_read(family): matrix = await adoption_svc.adoption_matrix(family["outsider"]) assert all(i["note_id"] != family["note"] for i in matrix["ideas"])