Files
FabledScribe/src/scribe/mcp/tools/repos.py
T
bvandeusenandClaude Fable 5 1209e1c2d9
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 8s
CI & Build / integration (push) Failing after 25s
CI & Build / TypeScript typecheck (push) Canceled after 30s
CI & Build / Python tests (push) Canceled after 30s
CI & Build / Build & push image (push) Canceled after 0s
feat(ledger): a repo binding names the branch its ledger follows — bind_repo(ref=) (#2873, milestone 294)
Project 2 is bound to main, so every consolidation of the 2026-08 audit was
invisible to the ledger until the dev→main merge; the operator works on dev
(rule 1). repo_bindings.ref (migration 0082, nullable) is the branch the
coverage refresh reads; NULL keeps the forge default branch. set_binding takes
ref (name sets, "" clears, None leaves standing); bindings_for_project feeds
the refresh; bind_repo exposes ref ("-" clears). to_dict carries it.

Operator decision on #2873 (2026-08-21): per-binding ref, chosen at bind time,
default the repo default branch.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-21 15:10:54 -04:00

84 lines
3.4 KiB
Python

"""Repo -> project binding MCP tools.
Let the operator map the working git repository to a Scribe project so the
SessionStart hook auto-loads that project's context — without pinning a project
id in plugin config (which would force a single project for every repo). Run
`bind_repo` once per repo; the binding is keyed on the normalized git remote, so
it survives re-clones and ssh/https URL differences.
"""
from __future__ import annotations
from scribe.mcp._context import current_user_id
from scribe.services import projects as projects_svc
from scribe.services import repo_bindings as repo_bindings_svc
async def bind_repo(repo_url: str, project_id: int, ref: str = "") -> dict:
"""Bind a git repository to a Scribe project for session-start context.
After this, any session started in that repo auto-loads the project's
context (the SessionStart hook sends the repo's remote; the server resolves
it here). Idempotent — re-binding the same repo updates the target project.
The binding is also what the shape ledger reads (refresh_pattern_coverage):
`ref` names the branch it follows. Default (""): the repo's default branch
— which means the ledger only sees work after a merge. A dev-first project
(rule 1: dev is home) should bind with ref="dev" so classification follows
the push, not the merge. Re-binding with ref="" keeps the standing ref;
pass ref="-" to clear it back to the default branch.
Args:
repo_url: the repo's git remote (e.g. the output of
`git remote get-url origin` — ssh or https form, both work).
project_id: the Scribe project this repo represents.
ref: branch the ledger follows ("" = leave as is / default branch on
a new binding; "-" = clear to the default branch).
"""
uid = current_user_id()
project = await projects_svc.get_project(uid, project_id)
if project is None:
raise ValueError(f"project {project_id} not found")
ref_arg = None if not ref else ("" if ref.strip() == "-" else ref)
binding = await repo_bindings_svc.set_binding(uid, repo_url, project_id, ref_arg)
follows = binding.ref or "the default branch"
return {
"repo_key": binding.repo_key,
"project_id": binding.project_id,
"project_title": project.title,
"ref": binding.ref,
"message": (
f"Bound `{binding.repo_key}` -> {project.title} (id {project.id}); "
f"the ledger follows {follows}."
),
}
async def list_repo_bindings() -> dict:
"""List every repo -> project binding for the current user."""
uid = current_user_id()
bindings = await repo_bindings_svc.list_bindings(uid)
return {"bindings": [b.to_dict() for b in bindings]}
async def unbind_repo(repo_url: str) -> dict:
"""Remove a repo's binding. Sessions there fall back to standing rules only.
Args:
repo_url: the repo's git remote (same form used to bind it).
"""
uid = current_user_id()
removed = await repo_bindings_svc.delete_binding(uid, repo_url)
key = repo_bindings_svc.normalize_repo_key(repo_url)
return {
"removed": removed,
"repo_key": key,
"message": (
f"Unbound `{key}`." if removed else f"No binding found for `{key}`."
),
}
def register(mcp) -> None:
for fn in (bind_repo, list_repo_bindings, unbind_repo):
mcp.tool(name=fn.__name__)(fn)