"""Family canon MCP tools — the promotion engine's agent door. Thin wrappers over services/family.py. The agent is the decider here: no person approves a promotion, so the criteria are stated in these docstrings and enforced by the service, and every decision is logged with its reason. """ from __future__ import annotations from scribe.mcp._context import current_user_id from scribe.services import family as family_svc from scribe.services import family_adoption as adoption_svc async def list_family_ideas( status: str = "", platform: str = "", limit: int = 50, offset: int = 0, ) -> dict: """Family ideas — records that every project on a platform shares — with their state: `candidate` (waiting to be evaluated, or held back by a criterion), `canon` (promoted; every project on its platforms answers it) or `retired`. Args: status: candidate | canon | retired. Empty = all. platform: a platform slug (list_platforms). Empty = all. limit / offset: page through the list. """ ideas = await family_svc.list_ideas( current_user_id(), status=status or None, platform=platform or None, limit=max(1, min(limit, 200)), offset=max(0, offset), ) return {"ideas": ideas, "limit": limit, "offset": offset} async def get_family_idea(note_id: int) -> dict: """One family idea with what an evaluation needs: its state, platforms, decision history (brief — each reason once; list_family_decisions has the full rows), ledger counts, THE THREE CRITERIA, the `ledger` (every project's answer, with `needs_recheck`), and the `precedents` — the decisions on the ideas nearest this one by meaning. Read the precedents before deciding, and decide consistently with them unless this idea differs in a way you can name. Args: note_id: the idea's record id. """ uid = current_user_id() idea = await family_svc.get_idea(uid, note_id) if idea is None: raise ValueError(f"#{note_id} is not a family idea you can read") idea["precedents"] = await family_svc.precedents(uid, note_id) idea["ledger"] = await adoption_svc.list_adoptions(uid, idea_id=note_id) return idea async def propose_family_idea(note_id: int, reason: str, applies_when: str = "") -> dict: """Record a note, snippet or lesson as a family-idea CANDIDATE: something you believe every project on some platform will face. It reaches no project's ledger until it is promoted. Proposing is cheap and does not promote. When the record already meets the three criteria, call promote_family_idea directly instead. Args: note_id: the record that carries the idea. reason: why this looks like a family idea. applies_when: optional draft of when it applies, in platform terms. """ idea, created = await family_svc.propose( current_user_id(), note_id, reason=reason, applies_when=applies_when or None, ) return {"idea": idea, "created": created} async def promote_family_idea( note_id: int, applies_when: str, platforms: list[str], platform_terms: str, platform_problem: str, proven: str, evidence: list[str], reason: str, precedent_ids: list[int] | None = None, ) -> dict: """Evaluate a record against the three criteria and, if all hold, promote it to family canon. YOU decide; nobody approves. The record of why is what makes the decision reviewable and the next one consistent. An idea is promoted only when ALL THREE hold, and each is answered with your reasoning: 1. platform_terms — its 'when it applies' is stated in terms of a platform (what a project is built on or ships as), not one app's domain. Any project on the platform could read it and know whether it applies. 2. platform_problem — it answers a problem the platform itself causes, or a stance the operator holds across projects. One app's preference does not qualify. 3. proven — it has worked for real at least once (CI green, verified on a device, shipped). Name that in `evidence`. Any criterion left blank — or `applies_when`, `platforms` or `evidence` left empty — VETOES the promotion on its own. A veto is not an error: the idea stays a candidate and the veto is logged, so the next evaluation of something like it sees why this one was held. That is also how you record "evaluated, and it does not qualify": leave the failing criterion blank and say why in `reason`. Before deciding, read get_family_idea's `precedents`. The engine also finds the nearest earlier decisions itself and stores them as consulted. On promotion every project that is on one of `platforms` and that you can write gets an `unassessed` row in its adoption ledger. If the idea was canon before, its version moves, so earlier answers read as needing a recheck. Args: note_id: the record that carries the idea. applies_when: when it applies, in platform terms. platforms: platform slugs it is for (list_platforms). platform_terms: why criterion 1 holds (blank = it does not). platform_problem: why criterion 2 holds (blank = it does not). proven: why criterion 3 holds (blank = it does not). evidence: what proved it — a CI run, a task, a commit, a device check. reason: the decision in one or two sentences. precedent_ids: earlier family decisions you followed, if any. """ return await family_svc.promote( current_user_id(), note_id, applies_when=applies_when, platforms=platforms or [], criteria={ "platform_terms": platform_terms, "platform_problem": platform_problem, "proven": proven, }, evidence=evidence or [], reason=reason, precedent_ids=precedent_ids or [], ) async def retire_family_idea(note_id: int, reason: str) -> dict: """Demote a family idea — it no longer applies, or a better one replaces it. Its history and every judged ledger answer are kept; rows nobody had judged yet are closed. Undoable with undo_family_decision. Args: note_id: the idea. reason: why it is retired. """ return await family_svc.retire(current_user_id(), note_id, reason=reason) async def undo_family_decision(decision_id: int, reason: str) -> dict: """Reverse a family decision — a promotion, a revision, a retirement, a proposal, or one project's assessment — restoring the state it recorded as `before`. Only the latest decision on an idea (or on one project's answer to it) can be undone; the undo is logged too, so the history keeps both. Undoing an owed answer closes its task; undoing back to owed reopens it. Args: decision_id: the decision (list_family_decisions). reason: why it is undone. """ return await family_svc.undo(current_user_id(), decision_id, reason=reason) async def list_family_decisions( note_id: int = 0, project_id: int = 0, limit: int = 50, offset: int = 0, ) -> dict: """The family decision log, newest first: every proposal, promotion, veto, revision, retirement, assessment and undo, with its reason, evidence and the precedents it followed. `undoable` marks the decision an undo would reverse — per idea, and per project's answer. Args: note_id: only this idea's decisions. 0 = all. project_id: only this project's assessments. 0 = all. limit / offset: page through the log. """ rows = await family_svc.list_decisions( current_user_id(), idea_id=note_id or None, project_id=project_id or None, limit=max(1, min(limit, 200)), offset=max(0, offset), ) return {"decisions": rows, "limit": limit, "offset": offset} async def revise_family_idea( note_id: int, reason: str, applies_when: str = "", platforms: list[str] | None = None, evidence: list[str] | None = None, ) -> dict: """Record that a canon idea's SUBSTANCE changed — you rewrote its note's approach, its traps or its checklist, or its applicability or platforms moved. The canon version moves, so every project's answer given against the old version reads `needs_recheck` until it is assessed again. A typo fix is not a revision. A change a project that adopted the old version would need to act on is. Args: note_id: the canon idea. reason: what changed, in a sentence. applies_when: a new 'when it applies'. Empty = keep the current one. platforms: new platform slugs. Omit = keep the current ones. evidence: what prompted the change, if anything. """ return await family_svc.revise( current_user_id(), note_id, reason=reason, applies_when=applies_when or None, platforms=platforms, evidence=evidence, ) async def get_family_adoption(project_id: int, idea_id: int) -> dict: """Everything one assessment needs: the idea (its 'when it applies', platforms and version), this project's current answer, THE FOUR OUTCOMES, the `precedents` — this idea's answers in other projects and this project's answers to the nearest ideas — and the reference implementations beside this project's languages. Read it before assess_family_adoption. Args: project_id: the project answering. idea_id: the canon idea. """ return await adoption_svc.get_adoption(current_user_id(), project_id, idea_id) async def list_family_adoptions( project_id: int = 0, idea_id: int = 0, status: str = "", needs_recheck: bool = False, platform: str = "", ) -> dict: """The adoption ledger: each project's answer to each canon idea that reaches it — `unassessed`, `adopted`, `variant`, `exempt` or `owed`, with its reason, the canon version it was given against, `needs_recheck` when the idea has moved on since, and the owed task. Args: project_id: one project's answers. 0 = all you can read. idea_id: one idea's answers. 0 = all canon. status: one outcome. Empty = all. needs_recheck: only answers given against an older canon version. platform: only ideas for this platform slug. """ rows = await adoption_svc.list_adoptions( current_user_id(), project_id=project_id or None, idea_id=idea_id or None, status=status or None, recheck_only=needs_recheck, platform=platform or None, ) return {"adoptions": rows} async def assess_family_adoption( project_id: int, idea_id: int, outcome: str, reason: str, evidence: list[str] | None = None, precedent_ids: list[int] | None = None, ) -> dict: """Answer one canon family idea for one project. YOU decide; nobody approves. Judge in this order and stop at the first that holds: 1. exempt — the idea's 'when it applies' is false for this project. `reason` names the fact about the project that makes it false. 2. variant — it applies, and the project departs for a reason that names a FACT about itself the canon did not account for. A preference, a taste, or "we already did it another way" is not a reason: that is owed — or, if this project's way is better rather than different, a conflict (resolve_family_conflict). 3. adopted — it applies and the project does it. `evidence` names where (a file, a commit, a task, a CI run). 4. owed — none of the above. A task is filed in THIS project naming the gap and the reference implementation for its language, filed under the project's System matching the idea's area (retag it with update_task if that guess is wrong). Nothing edits another repository; the project picks the task up itself. Read get_family_adoption first: answer consistently with its precedents unless this project differs in a way you can name in `reason`. The engine also records the precedents it found itself. The owed task follows the answer: adopted closes it as done, exempt or variant cancels it, owed again reopens it. The same answer given twice records nothing the second time. When the project's code is in the shape ledger, prefer answering THERE: classify_shapes the shape against the idea (idea_id=…) as an instance or a variant, and this answer moves with it. An answer the shapes already give otherwise is refused — the two ledgers never disagree; reclassify the shapes if they are wrong. Args: project_id: the project answering. idea_id: the canon idea. outcome: exempt | variant | adopted | owed. reason: why — required for every outcome. evidence: where it is done (required for adopted), or what you checked. precedent_ids: earlier family decisions you followed, if any. """ return await adoption_svc.assess( current_user_id(), project_id, idea_id, outcome=outcome, reason=reason, evidence=evidence, precedent_ids=precedent_ids, ) async def resolve_family_conflict( idea_id: int, canon_project_id: int, other_project_id: int, ground: str, fold: str, reason: str, grounds_checked: dict | None = None, evidence: list[str] | None = None, conditions: dict | None = None, precedent_ids: list[int] | None = None, ) -> dict: """Settle two projects that solve the same canon idea differently, each for reasons it believes. YOU decide, by THE CONFLICT ORDER — the first ground that applies wins, and for every ground above it you say in `grounds_checked` why it did not decide: 1. operator_stance — one side follows a stance the operator stated (a rule, a preference, a recorded decision). Name it in `evidence`. 2. covers_failure — one side covers a recorded failure (an incident, an issue, a lesson) the other does not. Name it in `evidence`. The other side becomes owed. 3. split_by_condition — both are right under different conditions. The canon splits: `conditions={"canon": …, "other": …}`, and each side is canon where its condition holds. 4. most_recent_complete — none of the above: the side verified most recently and covering the most wins. Name the verification. `canon_project_id` is the side whose approach the idea's note states after this. If the note says something else now, rewrite it first (update_note) — the note is the canon. The losing side's reasoning, `fold`, is appended to the idea's note as a trap (ground 2), an alternative (1, 4) or the branch for its condition (3). It is never dropped. The version moves, so every other project's answer reads `needs_recheck`. The canon side is answered adopted; the other owed (with a task in its project) or, in a split, adopted. Args: idea_id: the canon idea. canon_project_id: the side whose approach is canon after this. other_project_id: the side that loses, or the split's other branch. ground: operator_stance | covers_failure | split_by_condition | most_recent_complete. fold: the other side's reasoning, as it should read in the note. reason: the decision in one or two sentences. grounds_checked: {earlier ground: why it did not decide}. evidence: the stance, the failure or the verification. conditions: for a split, {"canon": condition, "other": condition}. precedent_ids: earlier family decisions you followed, if any. """ return await adoption_svc.resolve_conflict( current_user_id(), idea_id, canon_project_id=canon_project_id, other_project_id=other_project_id, ground=ground, grounds_checked=grounds_checked, fold=fold, evidence=evidence, reason=reason, conditions=conditions, precedent_ids=precedent_ids, ) async def set_family_references(note_id: int, snippet_ids: list[int]) -> dict: """Set a family idea's reference implementations — the snippets an owed task points a project at, ideally one per language. Replaces the list. The idea is what transfers; a reference is where to start, not code to copy. Args: note_id: the family idea. snippet_ids: snippets implementing it. [] clears the list. """ refs = await adoption_svc.set_references(current_user_id(), note_id, snippet_ids or []) return {"references": refs} async def set_family_topic(note_id: int, topic_id: int) -> dict: """Link a family idea to the rule topic that holds its binding norms — the rules every project on its platforms follows. The note says when the idea applies and why; the topic's rules say what must hold. One idea per topic. get_family_idea and get_family_adoption then list those rules, so an assessment judges against them. Use it when an idea already has rules written for it, or when you write rules that exist to carry one idea. The version does not move: linking changes where the norms are found, not what they say. Args: note_id: the family idea. topic_id: a topic in a rulebook you own (list_rulebooks, list_topics). 0 unlinks. """ return await family_svc.set_topic(current_user_id(), note_id, topic_id) def register(mcp) -> None: for fn in ( list_family_ideas, get_family_idea, propose_family_idea, promote_family_idea, retire_family_idea, undo_family_decision, list_family_decisions, revise_family_idea, get_family_adoption, list_family_adoptions, assess_family_adoption, resolve_family_conflict, set_family_references, set_family_topic, ): mcp.tool(name=fn.__name__)(fn)