"""The promotion engine without a database (milestone 463 step 3). The parts that decide — which citations claim lineage, which criteria veto — are pure and pinned here, beside the doors' wiring: the triggers ride the write tools, fail open, and the criteria the service enforces are the ones the agent is told. The state machine against Postgres is in tests/test_integration_family_promotion.py. """ from __future__ import annotations from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from scribe.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS from scribe.mcp.tools import family as family_tools from scribe.mcp.tools.milestones import update_milestone from scribe.services import family as family_svc from scribe.services.family import CRITERIA_KEYS, lineage_citations, vetoes from tests.helpers import fake_milestone pytestmark = pytest.mark.usefixtures("_bind_user") # --- which citations claim lineage --------------------------------------------- @pytest.mark.parametrize("text", [ "Signed release lane, matching roundtable-android (task #1615).", "Ported from #1615, with the keystore step moved first.", "Same shape as #1615: the APK is baked into the image.", "This mirrors #1615 for the desktop client.", "Modelled on #1615.", ]) def test_a_citation_with_a_lineage_word_names_its_source(text): assert lineage_citations(text) == [1615] @pytest.mark.parametrize("text", [ "See #1615 for context.", "#1615 is related.", "Closes #1615.", # The lineage word is in the PREVIOUS sentence — a different claim. "This is matching the old flow. Unrelated: #1615.", "Matching the old flow\nsee #1615", "", ]) def test_a_citation_without_lineage_is_silent(text): assert lineage_citations(text) == [] def test_lineage_citations_are_in_order_and_unique(): text = "Based on #20; ported from #10. Same pattern as #20." assert lineage_citations(text) == [20, 10] # --- each criterion vetoes on its own ----------------------------------------------- GOOD = dict( applies_when="any app that installs its own updates", platforms=["android-app"], criteria={ "platform_terms": "stated as an Android distribution concern", "platform_problem": "Play Protect and signature checks are the platform's", "proven": "shipped in two apps", }, evidence=["CI run 8274 green", "#4775"], ) def test_all_three_criteria_held_means_no_veto(): assert vetoes(**GOOD) == [] @pytest.mark.parametrize("key", CRITERIA_KEYS) def test_each_criterion_left_unsupported_vetoes_on_its_own(key): case = {**GOOD, "criteria": {**GOOD["criteria"], key: " "}} assert vetoes(**case) == [key] def test_platform_terms_needs_an_applicability_test_and_a_scope(): assert vetoes(**{**GOOD, "applies_when": ""}) == ["platform_terms"] assert vetoes(**{**GOOD, "platforms": []}) == ["platform_terms"] def test_proven_needs_named_evidence(): assert vetoes(**{**GOOD, "evidence": []}) == ["proven"] assert vetoes(**{**GOOD, "evidence": [" "]}) == ["proven"] def test_the_criteria_the_agent_is_told_are_the_ones_enforced(): """Rule 119: the criteria are product text. The promote tool's docstring must name every criterion the service checks, by its parameter name.""" doc = family_tools.promote_family_idea.__doc__ for key in CRITERIA_KEYS: assert key in doc, f"promote_family_idea's docstring never names {key}" assert len(family_svc.CRITERIA) == 3 # --- the doors ---------------------------------------------------------------------- def test_every_family_tool_is_classified(): reads = {"list_family_ideas", "get_family_idea", "list_family_decisions"} writes = {"propose_family_idea", "promote_family_idea", "retire_family_idea", "undo_family_decision"} assert reads <= _READ_ONLY_TOOLS assert writes <= _WRITE_TOOLS def _note(**kw): base = dict(id=7, project_id=2, title="t", body="b", is_task=False) return SimpleNamespace(**{**base, **kw}) async def test_a_citation_hint_wins_and_repeat_is_not_asked(): data: dict = {} repeat = AsyncMock(return_value="repeat") with patch.object(family_svc, "citation_trigger", AsyncMock(return_value="cited")), \ patch.object(family_svc, "repeat_trigger", repeat): await family_svc.attach_family_hint(1, data, _note(), created=True) assert data == {"family_hint": "cited"} repeat.assert_not_awaited() async def test_the_repeat_check_runs_only_on_a_create(): repeat = AsyncMock(return_value="repeat") with patch.object(family_svc, "citation_trigger", AsyncMock(return_value=None)), \ patch.object(family_svc, "repeat_trigger", repeat): edited: dict = {} await family_svc.attach_family_hint(1, edited, _note(), created=False) created: dict = {} await family_svc.attach_family_hint(1, created, _note(), created=True) assert edited == {} assert created == {"family_hint": "repeat"} async def test_a_failing_trigger_never_breaks_the_write(): data: dict = {"id": 7} with patch.object(family_svc, "citation_trigger", AsyncMock(side_effect=RuntimeError("db down"))): await family_svc.attach_family_hint(1, data, _note(), created=True) assert data == {"id": 7} async def test_closing_a_milestone_on_a_platform_carries_the_hint(): closed = fake_milestone(id=5, project_id=3, status="done") with patch("scribe.mcp.tools.milestones.milestones_svc.update_milestone", AsyncMock(return_value=closed)), \ patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=True)), \ patch.object(family_svc, "milestone_trigger", AsyncMock(return_value="evaluate")): out = await update_milestone(project_id=3, milestone_id=5, status="done") assert out["family_hint"] == "evaluate" async def test_re_saving_a_closed_milestone_asks_nothing(): closed = fake_milestone(id=5, project_id=3, status="done") trigger = AsyncMock(return_value="evaluate") with patch("scribe.mcp.tools.milestones.milestones_svc.update_milestone", AsyncMock(return_value=closed)), \ patch.object(family_svc, "milestone_is_open", AsyncMock(return_value=False)), \ patch.object(family_svc, "milestone_trigger", trigger): out = await update_milestone(project_id=3, milestone_id=5, status="done") assert "family_hint" not in out trigger.assert_not_awaited()