From dbbab859ce4812d0459e0df2c45572f6fbd3a6b2 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 1 Oct 2026 14:30:58 -0400 Subject: [PATCH] =?UTF-8?q?feat(lessons):=20soft=20links=20=E2=80=94=20a?= =?UTF-8?q?=20lesson=20and=20a=20rule=20arriving=20together=20in=20distinc?= =?UTF-8?q?t=20situations=20are=20proposed=20as=20a=20link=20(milestone=20?= =?UTF-8?q?440=20step=203,=20#4637)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When one hook response puts a lesson and a rule in front of the reader, the pair is recorded as evidence on a SUGGESTED lesson_rule_links row; once the pair has arrived together in PROPOSE_SITUATIONS (3) distinct situations, the next co-arrival carries one line asking the reader to judge it with judge_lesson_link. Nothing about surfacing changes: a suggested link carries no rule anywhere (that is #4633, confirmed links only). - lesson_rules: co_surfaced (fail-open; only pairs the reader could confirm — a lesson they may write, a rule they own; judged pairs gather nothing; one proposal per response; PROPOSE_COOLDOWN 6h between asks), plus the pure counting rules: situation_key, add_evidence, proposal_due, evidence_summary. A situation is the prompt on /retrieve (word tokens, sorted and de-duplicated, so trivial rewordings count once) and the FILE on /prior-art (every edit to one file is one situation). - rules_for_lessons shows a suggested link's evidence counts. - plugin_context: build_autoinject_hint returns lesson_ids, build_prompt_rule_hint returns shown_rule_ids, build_write_path_hint records its own pair; routes/plugin /retrieve records the prompt pair. Shown lines, repeats included: relevance makes a co-arrival, not the session ledger. - Evidence lives in the existing evidence column — no migration, and backup already carries it. - Tests: counting rules (unit), the recorder against Postgres (bar, repeat, judged pairs, ownership, cooldown, evidence kept on confirm); conftest stubs co_surfaced for unit tests. Co-Authored-By: Claude Opus 5.5 --- src/scribe/routes/plugin.py | 10 +- src/scribe/services/lesson_rules.py | 211 +++++++++++++++++++- src/scribe/services/plugin_context.py | 32 ++- tests/conftest.py | 8 +- tests/test_integration_lesson_rule_links.py | 73 +++++++ tests/test_lesson_rule_soft_links.py | 103 ++++++++++ 6 files changed, 430 insertions(+), 7 deletions(-) create mode 100644 tests/test_lesson_rule_soft_links.py diff --git a/src/scribe/routes/plugin.py b/src/scribe/routes/plugin.py index 90f42b1..12b3ed3 100644 --- a/src/scribe/routes/plugin.py +++ b/src/scribe/routes/plugin.py @@ -12,6 +12,7 @@ from quart import Blueprint, g, jsonify, request from scribe.auth import admin_required, get_current_user_id, login_required from scribe.config import Config +from scribe.services import lesson_rules as lesson_rules_svc from scribe.services import plugin_context as plugin_ctx_svc from scribe.services import repo_bindings as repo_bindings_svc from scribe.services import report_check as report_check_svc @@ -147,7 +148,14 @@ async def autoinject_retrieve(): result = await plugin_ctx_svc.build_autoinject_hint( g.user.id, q, project_id=project_id, exclude_ids=exclude_ids, context=ctx, ) - blocks = [b for b in (rules["context"], result["context"]) if b] + # The one place both arms' lines meet, so the one place a lesson and a + # rule arriving TOGETHER can be counted (#4637). Keyed on the operator's + # prompt — the situation — and fails open to "". + proposal = await lesson_rules_svc.co_surfaced( + g.user.id, result.get("lesson_ids") or [], rules.get("shown_rule_ids") or [], + arm="p", situation=q, project_id=project_id or None, + ) + blocks = [b for b in (rules["context"], result["context"], proposal) if b] result["context"] = "\n\n".join(blocks) result["rule_ids"] = rules["rule_ids"] return jsonify(result) diff --git a/src/scribe/services/lesson_rules.py b/src/scribe/services/lesson_rules.py index 9364581..742d0c2 100644 --- a/src/scribe/services/lesson_rules.py +++ b/src/scribe/services/lesson_rules.py @@ -17,6 +17,14 @@ UNJUDGED, and `list_unjudged` lists those. The three are kept apart because a lesson that stands alone is the raw material for a rule nobody has written yet (#4634), and that is unreadable if "nobody looked" means the same thing. +THE SOFT LINK (#4637). When one hook request puts a lesson and a rule in +front of the reader together, `co_surfaced` records the pair as evidence on a +SUGGESTED link, and once that evidence crosses a bar the next co-arrival asks +the reader to judge it. The evidence counts distinct SITUATIONS, not +arrivals: two records that resemble each other arrive together on every +similar prompt, and counting each time would measure their similarity over +and over rather than any relation between them. + ACL (rule 78). Linking changes what a lesson says about itself, so it needs WRITE on the lesson (`access.can_write_note`, share-aware). It names a rule, so it needs the rule to be one the caller may read — rules are owner-scoped, and @@ -26,10 +34,13 @@ someone must not hand them the titles of its owner's private rules. """ from __future__ import annotations +import hashlib import logging -from datetime import datetime, timezone +import re +from datetime import datetime, timedelta, timezone from sqlalchemy import and_, delete, exists, func, not_, select +from sqlalchemy.exc import IntegrityError from scribe.models import async_session from scribe.models.lesson_rule_link import ( @@ -55,6 +66,25 @@ _NO_RULE_NOTE = "no rule fits: " # What `rule_judgment` reads on a lesson payload — the three answers. LINKED, NO_RULE, UNJUDGED = "linked", "no_rule", "unjudged" +# THE SOFT-LINK EVIDENCE BAR (#4637) — defaults, stated as defaults (rules 32, +# 115): nothing about this install chose them. +# +# Three distinct situations before a pair is put to the reader. One is +# coincidence and two is a pattern only by courtesy; three different prompts +# or files that each brought the same lesson and rule up together is the +# smallest count that says the pairing is not a single resemblance repeated. +PROPOSE_SITUATIONS = 3 +# A proposal the reader passed over is asked again only after this long, so +# one session's run of related prompts does not repeat the same question on +# every turn. +PROPOSE_COOLDOWN = timedelta(hours=6) +# Fingerprints kept per link. The bar needs three; the rest is for the reader +# judging later, and a list that grew with every situation would bloat a row +# whose answer stopped changing long ago. +_SITUATION_CAP = 50 +# A token shorter than this is grammar, not situation — "a", "is", "to". +_MIN_TOKEN = 3 + # How many rules a new lesson is offered to judge against. A handful, not the # fifty `what_might_apply` returns: this is read in the create response, at # the moment of writing, and a long tail there buries the one that fits. @@ -257,10 +287,15 @@ async def rules_for_lessons(user_id: int, lesson_ids) -> dict[int, list[dict]]: .where(_owned_rules_clause(user_id)) )).all() for link, rule in sorted(rows, key=lambda r: (order.get(r[0].state, 9), r[1].id)): - out[link.lesson_id].append({ + item = { "id": rule.id, "title": rule.title, "kind": rule.kind, "state": link.state, "note": link.note or "", - }) + } + if link.state == SUGGESTED: + # What the suggestion rests on, so a reader can judge it from + # here rather than having to go and count. + item["evidence"] = evidence_summary(link.evidence) + out[link.lesson_id].append(item) return out @@ -445,3 +480,173 @@ async def attach_rule_lessons(user_id: int, data: dict, rule_id: int) -> None: if lessons: data["lessons"] = lessons + + +def situation_key(arm: str, text: str) -> str: + """A fingerprint of one situation, so a repeat counts once. + + `arm` is part of the key — "p" for a prompt, "w" for a file being written + — because the two name different kinds of situation and must not collide. + On the prompt arm the text is the prompt: lowercased, split into word + tokens of _MIN_TOKEN or more, de-duplicated and sorted, so the same ask + re-sent with different spacing, punctuation, case or word order is one + situation. On the write arm the text is the PATH, not the code: every + edit to one file is one situation, however much the code differs between + them. Empty when there is nothing to key on. + """ + if arm == "w": + basis = (text or "").strip() + else: + tokens = sorted({t for t in re.findall(r"[a-z0-9]+", (text or "").lower()) + if len(t) >= _MIN_TOKEN}) + basis = " ".join(tokens) + if not basis: + return "" + return f"{arm}:" + hashlib.sha1(basis.encode()).hexdigest()[:16] + + +def evidence_summary(evidence) -> dict: + """The counts a reader judges a suggested link by.""" + ev = evidence if isinstance(evidence, dict) else {} + return { + "situations": len(ev.get("situations") or []), + "projects": len(ev.get("projects") or []), + "co_surfaced": int(ev.get("co_surfaced") or 0), + } + + +def add_evidence(evidence, key: str, project_id: int | None, now: datetime) -> dict: + """A NEW evidence dict with this co-arrival counted. Pure, so the counting + rules are testable without a database; a new dict, so SQLAlchemy sees the + JSONB column change.""" + ev = dict(evidence) if isinstance(evidence, dict) else {} + ev["co_surfaced"] = int(ev.get("co_surfaced") or 0) + 1 + situations = list(ev.get("situations") or []) + if key and key not in situations and len(situations) < _SITUATION_CAP: + situations.append(key) + ev["situations"] = situations + projects = list(ev.get("projects") or []) + if project_id and int(project_id) not in projects: + projects.append(int(project_id)) + ev["projects"] = projects + ev.setdefault("first_at", now.isoformat()) + ev["last_at"] = now.isoformat() + return ev + + +def proposal_due(evidence, now: datetime) -> bool: + """At the bar, and not asked within the cooldown.""" + ev = evidence if isinstance(evidence, dict) else {} + if len(ev.get("situations") or []) < PROPOSE_SITUATIONS: + return False + last = ev.get("proposed_at") + if not last: + return True + try: + return now - datetime.fromisoformat(last) >= PROPOSE_COOLDOWN + except (TypeError, ValueError): + return True + + +def _proposal_line(lesson_id: int, lesson_title: str, rule_id: int, rule_title: str, + kind: str, evidence: dict) -> str: + counts = evidence_summary(evidence) + where = (f" across {counts['projects']} projects" if counts["projects"] > 1 else "") + return ( + f"> Lesson #{lesson_id} \"{lesson_title}\" and {kind} #{rule_id} " + f"\"{rule_title}\" keep arriving together — {counts['situations']} " + f"distinct situations{where}. If the lesson is an instance of the " + f"{kind}, `judge_lesson_link({lesson_id}, {rule_id}, \"confirm\", " + f"note=\"why\")` links them and the {kind} becomes reachable through " + f"the lesson's situation; if it is not, `\"reject\"` with the why " + f"stops the pair being asked about again." + ) + + +async def co_surfaced( + user_id: int, lesson_ids, rule_ids, *, arm: str, situation: str, + project_id: int | None = None, +) -> str: + """Record that these lessons and rules arrived together in one request, + and return a proposal line when a pair has earned one ("" otherwise). + + Called by the hook doors that put both kinds in front of the reader in + the SAME response — `/retrieve` on a prompt, `/prior-art` on a write — so + the pairing is exact: nothing is joined across requests, and no session + identity is needed. + + Only pairs the reader could confirm are recorded: a lesson they may write + and a rule they own, the two checks `judge_link` makes. A pair already + judged (confirmed or rejected) gathers nothing more — a confirmation needs + no evidence, and a rejection is what stops the asking. + + One proposal per request at most, so a menu never becomes a queue of + questions. Fails open: this rides a hook, and a recall aid must never + break the prompt or the write it decorates. + """ + try: + return await _co_surfaced(user_id, lesson_ids, rule_ids, arm, situation, project_id) + except Exception: + logger.warning("lesson/rule co-surfacing could not be recorded", exc_info=True) + return "" + + +async def _co_surfaced(user_id, lesson_ids, rule_ids, arm, situation, project_id) -> str: + from scribe.services.access import can_write_note + from scribe.services.rulebooks import _owned_rules_clause + + lessons = _ids(lesson_ids) + rules = _ids(rule_ids) + key = situation_key(arm, situation) + if not lessons or not rules or not key: + return "" + lessons = [lid for lid in lessons if await can_write_note(user_id, lid)] + if not lessons: + return "" + now = datetime.now(timezone.utc) + proposal = None + async with async_session() as session: + owned = { + r.id: r for r in (await session.execute( + select(Rule).where(Rule.id.in_(rules)).where(_owned_rules_clause(user_id)) + )).scalars().all() + } + if not owned: + return "" + existing = { + (row.lesson_id, row.rule_id): row for row in (await session.execute( + select(LessonRuleLink).where( + LessonRuleLink.lesson_id.in_(lessons), + LessonRuleLink.rule_id.in_(list(owned)), + ) + )).scalars().all() + } + for lid in lessons: + for rid in owned: + row = existing.get((lid, rid)) + if row is not None and row.state != SUGGESTED: + continue + if row is None: + row = LessonRuleLink(lesson_id=lid, rule_id=rid, state=SUGGESTED) + session.add(row) + ev = add_evidence(row.evidence, key, project_id, now) + if proposal is None and proposal_due(ev, now): + ev["proposed_at"] = now.isoformat() + ev["proposed_count"] = int(ev.get("proposed_count") or 0) + 1 + proposal = (lid, rid, ev) + row.evidence = ev + try: + await session.commit() + except IntegrityError: + # Two hook requests created the same pair at once; the other + # one's row stands, and this co-arrival is simply not counted. + await session.rollback() + return "" + if proposal is None: + return "" + lid, rid, ev = proposal + lesson_title = (await session.execute( + select(Note.title).where(Note.id == lid) + )).scalar_one_or_none() or "" + rule = owned[rid] + return _proposal_line(lid, lesson_title, rid, rule.title, rule.kind or "rule", ev) diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py index 3ae2c43..957da98 100644 --- a/src/scribe/services/plugin_context.py +++ b/src/scribe/services/plugin_context.py @@ -33,6 +33,7 @@ from scribe.services.embeddings import ( semantic_search_notes, semantic_search_rules, ) +from scribe.services import lesson_rules as lesson_rules_svc from scribe.services.lessons import LESSON_NOTE_TYPE from scribe.services.note_usage import record_surfaced from scribe.services.rule_usage import record_rule_surfaced @@ -1275,7 +1276,15 @@ async def build_autoinject_hint( project_id=project_id, ) - return {"context": "\n".join(lines), "note_ids": note_ids, "config": cfg} + # The lessons this menu put in front of the reader, repeats included — for + # the soft-link recorder (#4637), which pairs them with the rule arm's + # lines in the same response. Relevance, not the session ledger, is what + # makes a co-arrival, so a `seen` lesson counts. + lesson_ids = [int(n.id) for _s, n in kept if _record_kind(n) == LESSON_NOTE_TYPE] + return { + "context": "\n".join(lines), "note_ids": note_ids, "config": cfg, + "lesson_ids": lesson_ids, + } async def _reserve_slot_for_preference( @@ -1498,6 +1507,10 @@ async def build_prompt_rule_hint( # (#3668). The slot's hit is surfaced under `preference_slot` by the # helper, against that source's own row. rule_ids = [rule.id for _score, rule in fresh] + # Every rule LINE, repeats and the reserved slot included — what the + # reader actually had in front of them, for the soft-link recorder + # (#4637). Distinct from `rule_ids`, which is telemetry's fresh-only cut. + out["shown_rule_ids"] = [rule.id for _score, rule in hits] # RANKED, not ambient: this arm chose what it showed. The name is also # in `rule_usage.RANKED_SOURCES`, and it has to be — a ranked source @@ -2583,6 +2596,7 @@ async def build_write_path_hint( # # Fails open like every other arm: a rule hint must never break a write. rule_ids: list[int] = [] + shown_rule_ids: list[int] = [] checkpoint: dict = {} try: already = set(exclude_rule_ids or []) @@ -2623,6 +2637,9 @@ async def build_write_path_hint( # would inflate pull_through's denominator with a choice this arm never # made. A reference is a RENDERING decision, not a retrieval outcome. rule_ids.extend(rule.id for _score, rule in fresh) + # Every rule line, repeats included — what the soft-link recorder + # pairs with this response's lessons (#4637). + shown_rule_ids = [rule.id for _score, rule in kept] # The stop, beside the lines rather than instead of them — see # `checkpoint_for`. `kept` is passed, not `fresh`: whether a rule # was named earlier this session says nothing about whether this @@ -2691,6 +2708,19 @@ async def build_write_path_hint( except Exception: logger.debug("write-path rule arm failed", exc_info=True) + # A lesson and a rule on this one response (#4637): the soft-link + # recorder counts the pair, keyed on the FILE — every edit to one file is + # one situation — and asks about it once the evidence holds. Fails open. + shown_lessons = [ + int(item["id"]) for _m, item in menu if item.get("kind") == LESSON_NOTE_TYPE + ] + proposal = await lesson_rules_svc.co_surfaced( + user_id, shown_lessons, shown_rule_ids, arm="w", situation=path, + project_id=project_id or None, + ) + if proposal: + lines.append(proposal) + return { "context": "\n".join(lines), "note_ids": note_ids, diff --git a/tests/conftest.py b/tests/conftest.py index 489d199..5ff10bf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -154,7 +154,10 @@ def _no_lesson_rule_links(request): The candidate search a new lesson runs (#4631) is stubbed to "nothing resembles it" for the same reason: it embeds, and the model load would - otherwise ride along with every unit test that creates a lesson. + otherwise ride along with every unit test that creates a lesson. The + co-surfacing recorder (#4637) is stubbed to "no proposal": every hook + builder test that renders a lesson beside a rule would otherwise reach + for the database to count the pair. Skipped for integration tests, which exercise the real links against Postgres (tests/test_integration_lesson_rule_links.py). @@ -164,7 +167,8 @@ def _no_lesson_rule_links(request): return with patch("scribe.services.lesson_rules.attach_lesson_rules", AsyncMock()), \ patch("scribe.services.lesson_rules.attach_rule_lessons", AsyncMock()), \ - patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])): + patch("scribe.services.lesson_rules.rule_candidates", AsyncMock(return_value=[])), \ + patch("scribe.services.lesson_rules.co_surfaced", AsyncMock(return_value="")): yield diff --git a/tests/test_integration_lesson_rule_links.py b/tests/test_integration_lesson_rule_links.py index 539a8c5..aed1f23 100644 --- a/tests/test_integration_lesson_rule_links.py +++ b/tests/test_integration_lesson_rule_links.py @@ -241,3 +241,76 @@ async def test_deleting_the_lesson_takes_its_answer_with_it(world): select(LessonNoRule).where(LessonNoRule.lesson_id == world["lesson"]) )).scalars().all() assert left == [] + + +# ── #4637: soft links from co-surfacing ───────────────────────────────────── + + +async def _link(lesson_id: int, rule_id: int): + async with async_session() as s: + return (await s.execute( + select(LessonRuleLink).where( + LessonRuleLink.lesson_id == lesson_id, LessonRuleLink.rule_id == rule_id, + ) + )).scalar_one_or_none() + + +async def _arrive(world, prompt: str, project: int | None = None, rules=None) -> str: + return await links_svc.co_surfaced( + world["owner"], [world["lesson"]], rules or [world["r1"]], + arm="p", situation=prompt, project_id=project, + ) + + +async def test_distinct_situations_reach_the_bar_and_the_proposal_fires_once(world): + assert await _arrive(world, "the CI run is still going after twenty minutes") == "" + assert await _arrive(world, "deploy job hanging on the image build step") == "" + line = await _arrive(world, "integration lane overran its usual duration", world["mine"]) + assert f"judge_lesson_link({world['lesson']}, {world['r1']}" in line + + link = await _link(world["lesson"], world["r1"]) + assert link.state == "suggested" + assert links_svc.evidence_summary(link.evidence)["situations"] == 3 + # Within the cooldown a fourth situation is counted but not asked again. + assert await _arrive(world, "pipeline stuck waiting on a runner for ages") == "" + link = await _link(world["lesson"], world["r1"]) + assert links_svc.evidence_summary(link.evidence)["situations"] == 4 + + +async def test_the_same_prompt_repeated_never_reaches_the_bar(world): + for _ in range(5): + assert await _arrive(world, "is the CI run still going") == "" + ev = links_svc.evidence_summary((await _link(world["lesson"], world["r1"])).evidence) + assert ev == {"situations": 1, "projects": 0, "co_surfaced": 5} + + +async def test_a_judged_pair_gathers_nothing_and_is_never_proposed(world): + owner = world["owner"] + await links_svc.judge_link(owner, world["lesson"], world["r1"], "reject", "different failure") + await links_svc.judge_link(owner, world["lesson"], world["r2"], "confirm") + for prompt in ("first situation here", "second situation there", "third situation elsewhere"): + assert await _arrive(world, prompt, rules=[world["r1"], world["r2"]]) == "" + assert (await _link(world["lesson"], world["r1"])).evidence is None + assert (await _link(world["lesson"], world["r2"])).evidence is None + + +async def test_a_rule_the_reader_does_not_own_is_not_paired(world): + for prompt in ("first situation here", "second situation there", "third situation elsewhere"): + assert await _arrive(world, prompt, rules=[world["foreign"]]) == "" + assert await _link(world["lesson"], world["foreign"]) is None + + +async def test_a_suggestion_shows_its_evidence_and_leaves_the_lesson_unjudged(world): + await _arrive(world, "the CI run is still going after twenty minutes", world["mine"]) + rules = (await links_svc.rules_for_lessons(world["owner"], [world["lesson"]]))[world["lesson"]] + assert rules[0]["state"] == "suggested" + assert rules[0]["evidence"] == {"situations": 1, "projects": 1, "co_surfaced": 1} + assert (await _judgment(world["lesson"]))["rule_judgment"] == "unjudged" + + +async def test_confirming_a_suggestion_keeps_what_it_rested_on(world): + for prompt in ("first situation here", "second situation there", "third situation elsewhere"): + await _arrive(world, prompt) + out = await links_svc.judge_link(world["owner"], world["lesson"], world["r1"], "confirm", "same failure class") + assert out["state"] == "confirmed" + assert len(out["evidence"]["situations"]) == 3 diff --git a/tests/test_lesson_rule_soft_links.py b/tests/test_lesson_rule_soft_links.py new file mode 100644 index 0000000..5dc0c65 --- /dev/null +++ b/tests/test_lesson_rule_soft_links.py @@ -0,0 +1,103 @@ +"""Soft links — the counting rules (milestone 440, #4637). + +What a suggested link's evidence counts is a design decision with a failure +mode on each side: count every arrival and two similar records "prove" a +relation by resembling each other on every similar prompt; count too little +and nothing ever reaches the bar. These pin the rules as pure functions. The +recorder itself — ACL, judged pairs left alone, the proposal firing once — +is tested against Postgres in tests/test_integration_lesson_rule_links.py. +""" +from __future__ import annotations + +import inspect +from datetime import datetime, timedelta, timezone + +from scribe.services import lesson_rules as links_svc + +NOW = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) + + +def test_the_same_ask_reworded_trivially_is_one_situation(): + a = links_svc.situation_key("p", "Merge dev to main, please.") + assert a == links_svc.situation_key("p", "please MERGE main to dev") + assert a != links_svc.situation_key("p", "merge the feature branch") + + +def test_a_file_is_one_situation_however_its_code_differs(): + """The write arm keys on the path, so a run of edits to one file — each + with different code — is one situation, not one per edit.""" + path = "src/scribe/services/lesson_rules.py" + assert links_svc.situation_key("w", path) == links_svc.situation_key("w", path) + assert links_svc.situation_key("w", path) != links_svc.situation_key("w", "src/other.py") + + +def test_the_two_arms_never_share_a_key(): + assert links_svc.situation_key("p", "routes.py") != links_svc.situation_key("w", "routes.py") + + +def test_nothing_to_key_on_is_no_situation(): + assert links_svc.situation_key("p", "") == "" + assert links_svc.situation_key("p", "a is to") == "" # grammar only + + +def test_a_repeat_counts_as_an_arrival_but_not_a_situation(): + key = links_svc.situation_key("p", "probe the deployed version") + ev = links_svc.add_evidence(None, key, 2, NOW) + ev = links_svc.add_evidence(ev, key, 2, NOW) + assert links_svc.evidence_summary(ev) == {"situations": 1, "projects": 1, "co_surfaced": 2} + + +def test_projects_are_distinct_and_an_unbound_reader_adds_none(): + ev = links_svc.add_evidence(None, "p:1", 2, NOW) + ev = links_svc.add_evidence(ev, "p:2", 5, NOW) + ev = links_svc.add_evidence(ev, "p:3", None, NOW) + assert links_svc.evidence_summary(ev)["projects"] == 2 + + +def test_evidence_is_a_new_dict_so_the_column_change_is_seen(): + before = {"co_surfaced": 1, "situations": ["p:1"], "projects": []} + after = links_svc.add_evidence(before, "p:2", None, NOW) + assert after is not before + assert before["situations"] == ["p:1"] + + +def test_the_proposal_fires_at_the_bar_and_not_before(): + ev = None + for i in range(links_svc.PROPOSE_SITUATIONS - 1): + ev = links_svc.add_evidence(ev, f"p:{i}", None, NOW) + assert not links_svc.proposal_due(ev, NOW) + ev = links_svc.add_evidence(ev, "p:last", None, NOW) + assert links_svc.proposal_due(ev, NOW) + + +def test_a_passed_over_proposal_waits_out_the_cooldown(): + ev = {"situations": ["p:1", "p:2", "p:3"], "proposed_at": NOW.isoformat()} + assert not links_svc.proposal_due(ev, NOW + timedelta(minutes=5)) + assert links_svc.proposal_due(ev, NOW + links_svc.PROPOSE_COOLDOWN) + + +def test_the_proposal_names_the_call_that_answers_it(): + line = links_svc._proposal_line( + 41, "An overrun run usually failed early", 7, "Read the job log first", + "rule", {"situations": ["p:1", "p:2", "p:3"], "projects": [2, 5]}, + ) + assert 'judge_lesson_link(41, 7, "confirm"' in line + assert '"reject"' in line + assert "3 distinct situations across 2 projects" in line + + +def test_evidence_summary_reads_nothing_as_zero_not_an_error(): + assert links_svc.evidence_summary(None) == {"situations": 0, "projects": 0, "co_surfaced": 0} + + +def test_both_hook_doors_count_the_pair_on_their_own_arm(): + """The two responses that carry lessons and rules together. A door that + renders both and does not record the pair is the miss this step exists + to close, and it would produce no error anywhere — only an empty table.""" + from scribe.routes import plugin as plugin_routes + from scribe.services import plugin_context + + prompt_door = inspect.getsource(plugin_routes.autoinject_retrieve) + assert "co_surfaced(" in prompt_door and 'arm="p"' in prompt_door + write_door = inspect.getsource(plugin_context.build_write_path_hint) + assert "co_surfaced(" in write_door and 'arm="w"' in write_door