feat(forge): per-user forge connections — keyring, host-keyed resolution, project pin (#2778)

A forge token is a user's credential, not an instance's. The single
admin-settings config is replaced by per-user keyring rows (one per forge
host), and every server-side forge read runs on the PROJECT OWNER's keyring:

- forge_connections table + projects.forge_connection_id pin (migration 0078,
  which also carries the existing admin config into the first admin's row and
  deletes the old setting keys — no legacy dual-read)
- get_forge() replaced by get_forges(owner_id, project_id) -> ForgeSelector;
  resolve(repo) picks the connection whose host serves the repo. A pinned
  project uses ONLY its pinned connection; a stale pin (ownership moved) is
  ignored, never honored across users
- env FORGE_* config survives as an implicit entry for admin owners only;
  a stored row for the same host beats it
- consumers threaded: pull-time freshness (owner of the note), coverage
  (owner of the project), coverage routes' configured flag
- routes: /api/settings/forge-connections CRUD + per-connection test
  (own-rows only, tokens never returned); /api/admin/forge shrinks to
  /api/admin/forge-webhook (secret only); PUT /api/projects/<id>/forge pins,
  owner-or-admin asking, owner's connections only
- UI: Git Forges card moves to Settings -> Integrations as a connection
  list; webhook secret stays in the admin Config tab; owner-only forge
  select on the project coverage card
- backups exclude forge_connections (credentials, api_keys precedent) and
  the pin, so restores fall back to keyring resolution

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/config.py

@@ -60,11 +60,12 @@ class Config:
     # the MCP layer doesn't proxy web search (Claude has its own).
     SEARXNG_URL: str = os.environ.get("SEARXNG_URL", "")
 
-    # Git forge integration (#2689) — optional read access to the operator's
-    # forge so snippet bodies can be fetched/verified server-side. Normally
-    # configured in Settings → Config (stored as admin settings); these env
-    # fallbacks exist so a deployment can keep the token in a Docker secret
-    # instead of the database. DB value wins when both are set.
+    # Git forge integration (#2689) — optional read access to a git forge so
+    # snippet bodies can be fetched/verified server-side. Connections are
+    # per-user keyring rows (#2778, Settings → Git forges); these env values
+    # survive as an implicit keyring entry for ADMIN users' projects only, so
+    # a deployment can keep the operator's token in a Docker secret instead
+    # of the database. A stored row for the same host wins over the env entry.
     FORGE_KIND: str = os.environ.get("FORGE_KIND", "")
     FORGE_BASE_URL: str = os.environ.get("FORGE_BASE_URL", "").rstrip("/")
     FORGE_TOKEN: str = _read_secret("FORGE_TOKEN", "FORGE_TOKEN_FILE", "")

src/scribe/models/__init__.py

@@ -43,5 +43,6 @@ from scribe.models.rulebook import (  # noqa: E402, F401
     Rulebook, RulebookTopic, Rule, project_rulebook_subscriptions,
 )
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
+from scribe.models.forge_connection import ForgeConnection  # noqa: E402, F401
 from scribe.models.system import System, RecordSystem  # noqa: E402, F401
 from scribe.models.design_system import DesignSystem, DesignToken  # noqa: E402, F401

src/scribe/models/forge_connection.py

@@ -0,0 +1,45 @@
+from sqlalchemy import ForeignKey, Integer, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import TimestampMixin
+
+
+class ForgeConnection(Base, TimestampMixin):
+    """One user's read-only credential for one git forge host (#2778).
+
+    The keyring model: a user owns a set of connections and every server-side
+    forge read for a project runs on the PROJECT OWNER's set, resolved by the
+    repo's host. One row per (user, host) — the repo's host picks the
+    connection deterministically, so there is no "default forge" pointer to
+    maintain or tie-break.
+
+    `host` is derived from `base_url` at write time and stored because it is
+    the lookup key; the service layer keeps the two in step. The token is a
+    secret: to_dict never includes it, and no route may return it.
+    """
+
+    __tablename__ = "forge_connections"
+    __table_args__ = (
+        UniqueConstraint("user_id", "host", name="uq_forge_connections_user_host"),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
+    )
+    kind: Mapped[str] = mapped_column(Text, nullable=False)
+    base_url: Mapped[str] = mapped_column(Text, nullable=False)
+    host: Mapped[str] = mapped_column(Text, nullable=False)
+    token: Mapped[str] = mapped_column(Text, nullable=False)
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "user_id": self.user_id,
+            "kind": self.kind,
+            "base_url": self.base_url,
+            "host": self.host,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }

src/scribe/models/project.py

@@ -27,6 +27,15 @@ class Project(Base, TimestampMixin, SoftDeleteMixin):
     design_system_id: Mapped[int | None] = mapped_column(
         BigInteger, ForeignKey("design_systems.id", ondelete="SET NULL"), nullable=True
     )
+    # The per-project forge pin (#2778). NULL is the ordinary state: forge
+    # reads resolve against the owner's keyring by repo host. When set, the
+    # project's forge reads use ONLY this connection — an explicit, auditable
+    # choice, constrained by the service layer to a connection the project
+    # OWNER holds (never a collaborator's token).
+    forge_connection_id: Mapped[int | None] = mapped_column(
+        BigInteger, ForeignKey("forge_connections.id", ondelete="SET NULL"),
+        nullable=True,
+    )
 
     def to_dict(self) -> dict:
         return {
@@ -38,6 +47,7 @@ class Project(Base, TimestampMixin, SoftDeleteMixin):
             "status": self.status,
             "color": self.color,
             "design_system_id": self.design_system_id,
+            "forge_connection_id": self.forge_connection_id,
             "created_at": self.created_at.isoformat(),
             "updated_at": self.updated_at.isoformat(),
         }

src/scribe/routes/admin.py

@@ -19,15 +19,6 @@ from scribe.services.backup import (
     restore_full_backup,
 )
 from scribe.services.email import SMTP_SETTING_KEYS, get_base_url, get_smtp_config, is_smtp_configured, send_test_email
-from scribe.services.forge import (
-    FORGE_BASE_URL_KEY,
-    FORGE_KIND_KEY,
-    FORGE_KINDS,
-    FORGE_TOKEN_KEY,
-    ForgeError,
-    forge_config,
-    get_forge,
-)
 from scribe.services.logging import get_logs, get_log_stats, log_audit
 from scribe.services.notifications import send_invitation_email
 from scribe.services.settings import (
@@ -169,85 +160,48 @@ async def test_smtp():
 _TOKEN_MASK = "********"
 
 
-@admin_bp.route("/forge", methods=["GET"])
+# The forge CONFIG moved to per-user keyring rows (#2778, Settings → Git
+# forges); what stays admin is the webhook secret, because the push endpoint
+# is one URL per instance and authenticates deliveries, not users.
+
+
+@admin_bp.route("/forge-webhook", methods=["GET"])
 @admin_required
-async def get_forge_settings():
+async def get_forge_webhook_settings():
     from scribe.config import Config
     from scribe.routes.webhooks import FORGE_WEBHOOK_SECRET_KEY
 
-    cfg = await forge_config()
     webhook_secret = (
         await get_admin_setting(FORGE_WEBHOOK_SECRET_KEY, "")
         or Config.FORGE_WEBHOOK_SECRET
     )
     return jsonify({
-        "kind": cfg["kind"],
-        "base_url": cfg["base_url"],
         # Secrets never leave the server — the smtp_password convention:
         # masked when set, empty when not.
-        "token": _TOKEN_MASK if cfg["token"] else "",
         "webhook_secret": _TOKEN_MASK if webhook_secret else "",
-        "configured": bool(await get_forge()),
-        "kinds": list(FORGE_KINDS),
     })
 
 
-@admin_bp.route("/forge", methods=["PUT"])
+@admin_bp.route("/forge-webhook", methods=["PUT"])
 @admin_required
-async def update_forge_settings():
-    data = await request.get_json() or {}
-    uid = get_current_user_id()
-
-    kind = str(data.get("kind", "")).strip().lower()
-    if kind and kind not in FORGE_KINDS:
-        return jsonify({"error": f"Unknown forge kind {kind!r}"}), 400
-    base_url = str(data.get("base_url", "")).strip().rstrip("/")
-    if base_url and not base_url.startswith(("http://", "https://")):
-        return jsonify({"error": "Forge base URL must use http or https"}), 400
-
-    await set_admin_setting(FORGE_KIND_KEY, kind)
-    await set_admin_setting(FORGE_BASE_URL_KEY, base_url)
-    token = data.get("token")
-    # The mask coming back means "unchanged" — the form round-trips what GET
-    # showed it, and storing the mask would silently break the integration.
-    if token is not None and token != _TOKEN_MASK:
-        await set_admin_setting(FORGE_TOKEN_KEY, str(token))
+async def update_forge_webhook_settings():
     from scribe.routes.webhooks import FORGE_WEBHOOK_SECRET_KEY
 
+    data = await request.get_json() or {}
+    uid = get_current_user_id()
     webhook_secret = data.get("webhook_secret")
+    # The mask coming back means "unchanged" — the form round-trips what GET
+    # showed it, and storing the mask would silently break the integration.
     if webhook_secret is not None and webhook_secret != _TOKEN_MASK:
         await set_admin_setting(FORGE_WEBHOOK_SECRET_KEY, str(webhook_secret))
-    # The token is deliberately absent from the audit detail.
+    # The secret is deliberately absent from the audit detail.
     await log_audit(
-        "forge_config", user_id=uid, username=g.user.username,
-        ip_address=request.remote_addr,
-        details={"kind": kind, "base_url": base_url},
+        "forge_webhook_config", user_id=uid, username=g.user.username,
+        ip_address=request.remote_addr, details={},
     )
     return jsonify({"status": "ok"})
 
 
-@admin_bp.route("/forge/test", methods=["POST"])
-@admin_required
-async def test_forge():
-    """Probe the SAVED forge config: reachability and token acceptance in one
-    press, so a misconfiguration is visible now rather than as silent
-    fallbacks later (#2663's lesson, applied to integrations)."""
-    uid = get_current_user_id()
-    forge = await get_forge()
-    if forge is None:
-        return jsonify({"error": "Forge is not configured — save kind, base URL and token first"}), 400
-    try:
-        result = await forge.check()
-    except ForgeError as e:
-        return jsonify({"error": str(e)}), 502
-    await log_audit(
-        "forge_test", user_id=uid, username=g.user.username,
-        ip_address=request.remote_addr,
-        details={"ok": True, "username": result.get("username", "")},
-    )
-    return jsonify(result)
-
-
 @admin_bp.route("/logs", methods=["GET"])
 @admin_required
 async def list_logs():

src/scribe/routes/projects.py

@@ -1,7 +1,7 @@
 """Project management routes."""
 import logging
 
-from quart import Blueprint, jsonify, request
+from quart import Blueprint, g, jsonify, request
 
 from scribe.auth import login_required, get_current_user_id
 from scribe.routes.utils import not_found, parse_pagination
@@ -130,7 +130,7 @@ async def get_coverage_route(project_id: int):
     push or an explicit refresh).
     """
     from scribe.services.coverage import cached_coverage
-    from scribe.services.forge import get_forge
+    from scribe.services.forge import get_forges
 
     uid = get_current_user_id()
     result = await get_project_for_user(uid, project_id)
@@ -139,7 +139,9 @@ async def get_coverage_route(project_id: int):
     project, _ = result
     owner_uid = project.user_id or uid
     return jsonify({
-        "configured": await get_forge() is not None,
+        # The OWNER's keyring (#2778) — whether refresh could do anything,
+        # regardless of who is looking.
+        "configured": (await get_forges(owner_uid, project_id)).configured,
         "coverage": await cached_coverage(owner_uid, project_id),
     })
 
@@ -153,7 +155,7 @@ async def refresh_coverage_route(project_id: int):
     and wants the new number, and the forge timeout bounds the wait.
     """
     from scribe.services.coverage import refresh_coverage
-    from scribe.services.forge import ForgeError, get_forge
+    from scribe.services.forge import ForgeError, get_forges
 
     uid = get_current_user_id()
     result = await get_project_for_user(uid, project_id)
@@ -161,20 +163,56 @@ async def refresh_coverage_route(project_id: int):
         return not_found("Project")
     project, _ = result
     owner_uid = project.user_id or uid
-    if await get_forge() is None:
-        return jsonify({"error": "No git forge is configured (Settings → Config → Git Forge)"}), 400
+    selector = await get_forges(owner_uid, project_id)
+    if not selector.configured:
+        return jsonify({
+            "error": "The project owner has no forge connection "
+            "(Settings → Git forges)"
+        }), 400
     try:
-        coverage = await refresh_coverage(owner_uid, project_id)
+        coverage = await refresh_coverage(owner_uid, project_id, selector=selector)
     except ForgeError as exc:
         return jsonify({"error": str(exc)}), 502
     if coverage is None:
         return jsonify({
-            "error": "No bound repo is served by the configured forge — "
-            "bind the project's repo (bind_repo) on a remote the forge hosts"
+            "error": "No bound repo is served by the owner's forge connections — "
+            "bind the project's repo (bind_repo) on a remote a connection hosts"
         }), 400
     return jsonify({"coverage": coverage})
 
 
+@projects_bp.route("/<int:project_id>/forge", methods=["PUT"])
+@login_required
+async def set_project_forge_route(project_id: int):
+    """Pin the project to one forge connection, or clear the pin (#2778).
+
+    Body: {"connection_id": <id> | null}. Owner-or-admin may ask; either way
+    the pin can only reference a connection the project OWNER holds — the
+    service enforces that, so a collaborator's token can never end up serving
+    someone else's project.
+    """
+    from scribe.services.forge_connections import set_project_pin
+
+    uid = get_current_user_id()
+    result = await get_project_for_user(uid, project_id)
+    if result is None:
+        return not_found("Project")
+    project, permission = result
+    if permission != "owner" and g.user.role != "admin":
+        return jsonify({"error": "Only the project owner can change its forge"}), 403
+
+    data = await request.get_json() or {}
+    raw = data.get("connection_id")
+    if raw is not None and not isinstance(raw, int):
+        return jsonify({"error": "connection_id must be an integer or null"}), 400
+    owner_uid = project.user_id or uid
+    if not await set_project_pin(owner_uid, project_id, raw):
+        return jsonify({
+            "error": "That connection does not belong to the project owner"
+        }), 400
+    return jsonify({"forge_connection_id": raw})
+
+
 @projects_bp.route("/<int:project_id>/notes", methods=["GET"])
 @login_required
 async def get_project_notes_route(project_id: int):

src/scribe/routes/settings.py

@@ -20,7 +20,8 @@ settings_bp = Blueprint("settings", __name__, url_prefix="/api/settings")
 # read and skip the mask on write; this generic KV surface has to apply the
 # same treatment, or it silently un-masks what those endpoints masked — the
 # rows live on the admin's own user_id, so the plain GET returned them raw.
-_SECRET_KEYS = frozenset({"smtp_password", "forge_token", "forge_webhook_secret"})
+# (forge_token left with 0078: forge credentials are keyring rows now, #2778.)
+_SECRET_KEYS = frozenset({"smtp_password", "forge_webhook_secret"})
 _SECRET_MASK = "********"
 
 
@@ -73,3 +74,103 @@ async def test_search():
     if not Config.searxng_enabled():
         return jsonify({"configured": False, "results": [], "searxng_url": ""})
     return jsonify({"configured": True, "results": [], "searxng_url": Config.SEARXNG_URL})
+
+
+# --- forge connections (#2778) ------------------------------------------------
+# The user's keyring: read-only forge credentials, one per host, resolved by
+# repo host for every server-side forge read on the user's projects. Strictly
+# own-rows — a connection is a credential, and there is no admin view of
+# another user's keyring. Tokens never leave the server: the model's to_dict
+# omits them, and the routes never echo the submitted value back.
+
+
+@settings_bp.route("/forge-connections", methods=["GET"])
+@login_required
+async def list_forge_connections_route():
+    from scribe.services.forge import FORGE_KINDS
+    from scribe.services.forge_connections import list_connections
+
+    uid = get_current_user_id()
+    rows = await list_connections(uid)
+    return jsonify({
+        "connections": [r.to_dict() for r in rows],
+        "kinds": list(FORGE_KINDS),
+    })
+
+
+@settings_bp.route("/forge-connections", methods=["POST"])
+@login_required
+async def create_forge_connection_route():
+    from scribe.services.forge_connections import create_connection
+
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    try:
+        row = await create_connection(
+            uid,
+            kind=str(data.get("kind", "")),
+            base_url=str(data.get("base_url", "")),
+            token=str(data.get("token", "")),
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+    return jsonify(row.to_dict()), 201
+
+
+@settings_bp.route("/forge-connections/<int:connection_id>", methods=["PUT"])
+@login_required
+async def update_forge_connection_route(connection_id: int):
+    from scribe.services.forge_connections import update_connection
+
+    uid = get_current_user_id()
+    data = await request.get_json() or {}
+    token = str(data.get("token", ""))
+    # The mask coming back means "unchanged" — the form round-trips what the
+    # list showed, and storing the mask would silently break the connection.
+    if token == _SECRET_MASK:
+        token = ""
+    try:
+        row = await update_connection(
+            uid, connection_id,
+            kind=str(data.get("kind", "")),
+            base_url=str(data.get("base_url", "")),
+            token=token,
+        )
+    except ValueError as exc:
+        return jsonify({"error": str(exc)}), 400
+    if row is None:
+        return jsonify({"error": "Connection not found"}), 404
+    return jsonify(row.to_dict())
+
+
+@settings_bp.route("/forge-connections/<int:connection_id>", methods=["DELETE"])
+@login_required
+async def delete_forge_connection_route(connection_id: int):
+    from scribe.services.forge_connections import delete_connection
+
+    uid = get_current_user_id()
+    if not await delete_connection(uid, connection_id):
+        return jsonify({"error": "Connection not found"}), 404
+    return "", 204
+
+
+@settings_bp.route("/forge-connections/<int:connection_id>/test", methods=["POST"])
+@login_required
+async def test_forge_connection_route(connection_id: int):
+    """Probe the SAVED connection: reachability and token acceptance in one
+    press, so a misconfiguration is visible now rather than as silent
+    fallbacks later (#2663's lesson, applied per keyring row)."""
+    from scribe.services.forge import ForgeError, build_adapter
+    from scribe.services.forge_connections import get_connection
+
+    uid = get_current_user_id()
+    row = await get_connection(uid, connection_id)
+    if row is None:
+        return jsonify({"error": "Connection not found"}), 404
+    adapter = build_adapter(row.kind, row.base_url, row.token)
+    if adapter is None:
+        return jsonify({"error": "Connection is not usable — check kind and base URL"}), 400
+    try:
+        return jsonify(await adapter.check())
+    except ForgeError as exc:
+        return jsonify({"error": str(exc)}), 502

src/scribe/services/backup.py

@@ -72,6 +72,12 @@ _NOT_INCLUDED = [
     "api_keys", "note_embeddings", "app_logs", "notifications",
     "invitation_tokens", "password_reset_tokens", "user_profiles",
     "retrieval_logs",
+    # Sensitive credentials, same reasoning as api_keys: a backup that carries
+    # forge tokens is a token-exfiltration file. Users re-add connections
+    # after a restore; the per-project pin (projects.forge_connection_id) is
+    # deliberately not exported either, so restored projects fall back to
+    # keyring-by-host resolution — the documented unpinned behavior (#2778).
+    "forge_connections",
 ]
 
 

src/scribe/services/coverage.py

@@ -33,7 +33,7 @@ import re
 import tarfile
 from datetime import datetime, timezone
 
-from scribe.services.forge import ForgeAdapter, get_forge
+from scribe.services.forge import ForgeSelector, get_forges
 from scribe.services.repo_bindings import keys_for_project
 from scribe.services.settings import get_setting, set_setting
 
@@ -252,27 +252,32 @@ async def _recorded_locations(user_id: int, project_id: int) -> list[tuple[str,
 
 
 async def compute_coverage(
-    user_id: int, project_id: int, *, forge: ForgeAdapter | None = None
+    user_id: int, project_id: int, *, selector: ForgeSelector | None = None
 ) -> dict | None:
     """Measure a project's pattern-library coverage against its bound repos.
 
-    None means "nothing to measure" — no forge configured, or none of the
-    project's bound repos is served by it. That is the ordinary state for a
-    forge-less install and every caller treats it as silence, not failure.
+    None means "nothing to measure" — the owner's keyring serves none of the
+    project's bound repos (#2778). That is the ordinary state for a
+    forge-less user and every caller treats it as silence, not failure.
     Forge errors (unreachable, bad token) RAISE — the two callers are a
     refresh button and a background task, and both want to know.
+
+    ``user_id`` is the project OWNER's id: the cache lives there, and the
+    keyring resolved here must be the same one every other read uses.
     """
-    forge = forge if forge is not None else await get_forge()
-    if forge is None:
+    if selector is None:
+        selector = await get_forges(user_id, project_id)
+    if not selector.configured:
         return None
 
     repos: list[dict] = []
     matched_all: list[tuple[str, str, str, bool]] = []
     recorded = await _recorded_locations(user_id, project_id)
     for key in await keys_for_project(user_id, project_id):
-        api_repo = forge.resolve_repo(key)
-        if api_repo is None:
-            continue  # bound to a host this forge doesn't serve
+        hit = selector.resolve(key)
+        if hit is None:
+            continue  # bound to a host no connection serves
+        forge, api_repo = hit
         ref = await forge.default_branch(api_repo)
         shapes = shapes_from_archive(await forge.archive(api_repo, ref))
         matched = match_shapes(shapes, recorded)
@@ -299,10 +304,10 @@ async def compute_coverage(
 
 
 async def refresh_coverage(
-    user_id: int, project_id: int, *, forge: ForgeAdapter | None = None
+    user_id: int, project_id: int, *, selector: ForgeSelector | None = None
 ) -> dict | None:
     """Compute and cache. The only writer of the cache key."""
-    coverage = await compute_coverage(user_id, project_id, forge=forge)
+    coverage = await compute_coverage(user_id, project_id, selector=selector)
     if coverage is not None:
         await set_setting(
             user_id, f"{_CACHE_KEY_PREFIX}{project_id}", json.dumps(coverage)

src/scribe/services/forge.py

@@ -8,10 +8,12 @@ webhooks (step 6), and coverage can be measured (step 7).
 
 Design constraints, in force everywhere below:
 
-  - OPTIONAL per instance (rule #115). `get_forge()` returns None when nothing
-    is configured, and every consumer must treat None as "keep today's
-    behavior". An install that never configures a forge is not degraded — it
-    is the baseline.
+  - OPTIONAL per user (rule #115, sharpened by #2778). Connections are
+    per-user keyring rows resolved by repo host on the PROJECT OWNER's
+    keyring; `get_forges()` returns an empty selector when the owner has
+    nothing configured, and every consumer must treat that as "keep today's
+    behavior". A user who never configures a forge is not degraded — that is
+    the baseline.
   - READ-ONLY by construction. The adapter exposes reads; there is no write
     method to misuse. The token an operator mints for it only ever needs read
     scope, and the docs say so.
@@ -40,18 +42,14 @@ from dataclasses import dataclass
 from urllib.parse import quote, urlsplit
 
 import httpx
+from sqlalchemy import select
 
 from scribe.config import Config
 from scribe.services.repo_bindings import normalize_repo_key
-from scribe.services.settings import get_admin_setting
 
 logger = logging.getLogger(__name__)
 
-FORGE_KIND_KEY = "forge_kind"
-FORGE_BASE_URL_KEY = "forge_base_url"
-FORGE_TOKEN_KEY = "forge_token"
-
-# Kinds an instance can configure. Matches _FORGE_CLASSES below.
+# Kinds a connection can use. Matches _FORGE_CLASSES below.
 FORGE_KINDS = ("gitea", "github")
 
 # Total budget per forge call. Consumers either have a cache to fall back to
@@ -86,7 +84,8 @@ class ForgeFile:
     path: str
 
 
-def _host_of(url: str) -> str:
+def host_of(url: str) -> str:
+    """The lowercase hostname of a URL — the keyring's lookup key (#2778)."""
     return (urlsplit(url).hostname or "").lower()
 
 
@@ -111,7 +110,7 @@ class ForgeAdapter:
 
     @property
     def host(self) -> str:
-        return _host_of(self.base_url)
+        return host_of(self.base_url)
 
     def resolve_repo(self, repo_or_url: str) -> str | None:
         """The forge-API repo path for a recorded repo — or None if this forge
@@ -359,39 +358,124 @@ _FORGE_CLASSES: dict[str, type[ForgeAdapter]] = {
 }
 
 
-async def forge_config() -> dict:
-    """The instance's forge configuration, DB-first with env fallback.
+def build_adapter(
+    kind: str, base_url: str, token: str, *, transport=None
+) -> ForgeAdapter | None:
+    """One validated adapter from raw connection values, or None.
 
-    The env channel exists so a deployment can keep the token out of the
-    database entirely (Docker secret via FORGE_TOKEN_FILE) — the DB value wins
-    when both are present because the admin UI writes there, and a UI edit
-    that silently loses to an env var would look exactly like a broken form.
+    None means "this connection cannot serve reads" — the same contract the
+    old instance-wide lookup had, applied per keyring row. Misconfigurations
+    are logged, never raised: a bad row must not break the reads the good
+    rows can still serve.
     """
-    return {
-        "kind": (await get_admin_setting(FORGE_KIND_KEY, "") or Config.FORGE_KIND)
-        .strip()
-        .lower(),
-        "base_url": (
-            await get_admin_setting(FORGE_BASE_URL_KEY, "") or Config.FORGE_BASE_URL
-        ).rstrip("/"),
-        "token": await get_admin_setting(FORGE_TOKEN_KEY, "") or Config.FORGE_TOKEN,
-    }
-
-
-async def get_forge(*, transport=None) -> ForgeAdapter | None:
-    """The configured forge adapter, or None — and None means "behave exactly
-    as if this module did not exist", which every consumer must honor."""
-    cfg = await forge_config()
-    cls = _FORGE_CLASSES.get(cfg["kind"])
+    kind = (kind or "").strip().lower()
+    base_url = (base_url or "").rstrip("/")
+    cls = _FORGE_CLASSES.get(kind)
     if cls is None:
-        if cfg["kind"]:
+        if kind:
             # A kind we don't implement is a misconfiguration, not "off" —
             # say so once per lookup rather than silently reading as absent.
-            logger.warning("unknown forge kind %r configured — forge disabled", cfg["kind"])
+            logger.warning("unknown forge kind %r configured — connection disabled", kind)
         return None
-    if not cfg["base_url"] or not cfg["token"]:
+    if not base_url or not token:
         return None
-    if not cfg["base_url"].startswith(("http://", "https://")):
-        logger.warning("forge base URL %r has no http(s) scheme — forge disabled", cfg["base_url"])
+    if not base_url.startswith(("http://", "https://")):
+        logger.warning("forge base URL %r has no http(s) scheme — connection disabled", base_url)
         return None
-    return cls(cfg["base_url"], cfg["token"], transport=transport)
+    return cls(base_url, token, transport=transport)
+
+
+@dataclass(frozen=True)
+class ForgeSelector:
+    """The forge reads available to one project owner (#2778).
+
+    Consumers ask it to serve a REPO, not to hand over "the forge": resolve()
+    walks the owner's adapters and returns the (adapter, api_repo) pair for
+    the first one whose host serves the repo — or None, which every consumer
+    treats exactly as the old "no forge configured" state. An empty selector
+    IS rule #115's baseline.
+    """
+
+    adapters: tuple[ForgeAdapter, ...] = ()
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.adapters)
+
+    def resolve(self, repo_or_url: str) -> tuple[ForgeAdapter, str] | None:
+        for adapter in self.adapters:
+            repo = adapter.resolve_repo(repo_or_url)
+            if repo is not None:
+                return adapter, repo
+        return None
+
+
+async def get_forges(
+    owner_id: int, project_id: int | None = None, *, transport=None
+) -> ForgeSelector:
+    """The forge selector for reads on behalf of ``owner_id``'s records.
+
+    The keyring model (#2778): every server-side forge read for a record runs
+    on the PROJECT OWNER's connections, resolved by repo host — a forge token
+    is a user's credential, and one user's reads must never ride another
+    user's token. Pass the record's ``project_id`` so the per-project pin
+    applies: a pinned project uses ONLY its pinned connection (explicit and
+    auditable); a pin that no longer belongs to the owner (ownership moved) is
+    ignored with a warning rather than honored across users.
+
+    The env config (FORGE_KIND/FORGE_BASE_URL/FORGE_TOKEN) survives as an
+    implicit keyring entry for ADMIN owners only — it is the operator's
+    token, so it must not serve other users' reads — and a stored row for the
+    same host beats it, because the UI writes rows.
+    """
+    from scribe.models import async_session
+    from scribe.models.forge_connection import ForgeConnection
+    from scribe.models.project import Project
+    from scribe.models.user import User
+
+    async with async_session() as session:
+        pinned_id = None
+        if project_id:
+            pinned_id = (
+                await session.execute(
+                    select(Project.forge_connection_id).where(Project.id == project_id)
+                )
+            ).scalar_one_or_none()
+        rows = list(
+            (
+                await session.execute(
+                    select(ForgeConnection)
+                    .where(ForgeConnection.user_id == owner_id)
+                    .order_by(ForgeConnection.id)
+                )
+            ).scalars().all()
+        )
+        role = ""
+        if Config.FORGE_KIND and Config.FORGE_BASE_URL and Config.FORGE_TOKEN:
+            role = (
+                await session.execute(select(User.role).where(User.id == owner_id))
+            ).scalar_one_or_none() or ""
+
+    if pinned_id:
+        pin = next((r for r in rows if r.id == pinned_id), None)
+        if pin is not None:
+            adapter = build_adapter(pin.kind, pin.base_url, pin.token, transport=transport)
+            return ForgeSelector((adapter,) if adapter is not None else ())
+        logger.warning(
+            "project %s pins forge connection %s the owner (%s) does not hold — pin ignored",
+            project_id, pinned_id, owner_id,
+        )
+
+    adapters: list[ForgeAdapter] = []
+    for row in rows:
+        adapter = build_adapter(row.kind, row.base_url, row.token, transport=transport)
+        if adapter is not None:
+            adapters.append(adapter)
+    if role == "admin":
+        env = build_adapter(
+            Config.FORGE_KIND, Config.FORGE_BASE_URL, Config.FORGE_TOKEN,
+            transport=transport,
+        )
+        if env is not None and all(a.host != env.host for a in adapters):
+            adapters.append(env)
+    return ForgeSelector(tuple(adapters))

src/scribe/services/forge_connections.py

@@ -0,0 +1,194 @@
+"""User-level forge connection CRUD — the keyring rows get_forges reads (#2778).
+
+A connection is a user's read-only credential for one forge host; one row per
+(user, host) keeps host-keyed resolution deterministic with no default-pointer
+machinery. Everything here is own-rows-only: a connection is a credential, and
+no caller — admin included — reads or edits another user's. The token never
+leaves the server (model.to_dict omits it; routes mask "set/unset").
+
+Validation matches what build_adapter will accept, checked here so a bad
+value is a 400 at the form instead of a silently dead keyring row.
+"""
+from __future__ import annotations
+
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.forge_connection import ForgeConnection
+from scribe.models.project import Project
+from scribe.services.forge import FORGE_KINDS, host_of
+
+
+def validate_connection(kind: str, base_url: str) -> str | None:
+    """The error a connection's non-secret values would earn, or None."""
+    if kind not in FORGE_KINDS:
+        return f"Unknown forge kind {kind!r} (one of: {', '.join(FORGE_KINDS)})"
+    if not base_url.startswith(("http://", "https://")):
+        return "Forge base URL must use http or https"
+    if not host_of(base_url):
+        return "Forge base URL carries no hostname"
+    return None
+
+
+async def list_connections(user_id: int) -> list[ForgeConnection]:
+    async with async_session() as session:
+        rows = await session.execute(
+            select(ForgeConnection)
+            .where(ForgeConnection.user_id == user_id)
+            .order_by(ForgeConnection.host)
+        )
+        return list(rows.scalars().all())
+
+
+async def get_connection(user_id: int, connection_id: int) -> ForgeConnection | None:
+    async with async_session() as session:
+        return (
+            await session.execute(
+                select(ForgeConnection).where(
+                    ForgeConnection.id == connection_id,
+                    ForgeConnection.user_id == user_id,
+                )
+            )
+        ).scalar_one_or_none()
+
+
+async def create_connection(
+    user_id: int, *, kind: str, base_url: str, token: str
+) -> ForgeConnection:
+    """Create a keyring row. Raises ValueError on bad values or a host the
+    user already holds — one row per (user, host) IS the resolution model,
+    so a second token for the same host is an update, not a create."""
+    kind = (kind or "").strip().lower()
+    base_url = (base_url or "").strip().rstrip("/")
+    token = token or ""
+    error = validate_connection(kind, base_url)
+    if error is None and not token:
+        error = "A token is required (read scope is enough)"
+    if error:
+        raise ValueError(error)
+    host = host_of(base_url)
+    async with async_session() as session:
+        existing = (
+            await session.execute(
+                select(ForgeConnection).where(
+                    ForgeConnection.user_id == user_id,
+                    ForgeConnection.host == host,
+                )
+            )
+        ).scalar_one_or_none()
+        if existing is not None:
+            raise ValueError(
+                f"You already have a connection for {host} — edit that one; "
+                "resolution is by host, so a second row could never be reached"
+            )
+        row = ForgeConnection(
+            user_id=user_id, kind=kind, base_url=base_url, host=host, token=token
+        )
+        session.add(row)
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+
+async def update_connection(
+    user_id: int,
+    connection_id: int,
+    *,
+    kind: str = "",
+    base_url: str = "",
+    token: str = "",
+) -> ForgeConnection | None:
+    """Update own row; empty string = leave unchanged (the settings-form
+    sentinel convention). None when the row isn't the caller's."""
+    async with async_session() as session:
+        row = (
+            await session.execute(
+                select(ForgeConnection).where(
+                    ForgeConnection.id == connection_id,
+                    ForgeConnection.user_id == user_id,
+                )
+            )
+        ).scalar_one_or_none()
+        if row is None:
+            return None
+        new_kind = (kind or "").strip().lower() or row.kind
+        new_base = (base_url or "").strip().rstrip("/") or row.base_url
+        error = validate_connection(new_kind, new_base)
+        if error:
+            raise ValueError(error)
+        new_host = host_of(new_base)
+        if new_host != row.host:
+            clash = (
+                await session.execute(
+                    select(ForgeConnection.id).where(
+                        ForgeConnection.user_id == user_id,
+                        ForgeConnection.host == new_host,
+                        ForgeConnection.id != row.id,
+                    )
+                )
+            ).scalar_one_or_none()
+            if clash is not None:
+                raise ValueError(
+                    f"You already have a connection for {new_host} — edit that one"
+                )
+        row.kind = new_kind
+        row.base_url = new_base
+        row.host = new_host
+        if token:
+            row.token = token
+        await session.commit()
+        await session.refresh(row)
+        return row
+
+
+async def delete_connection(user_id: int, connection_id: int) -> bool:
+    """Delete own row. Project pins pointing at it go NULL (FK SET NULL) —
+    those projects fall back to keyring resolution, which is the documented
+    unpinned behavior, not a surprise."""
+    async with async_session() as session:
+        row = (
+            await session.execute(
+                select(ForgeConnection).where(
+                    ForgeConnection.id == connection_id,
+                    ForgeConnection.user_id == user_id,
+                )
+            )
+        ).scalar_one_or_none()
+        if row is None:
+            return False
+        await session.delete(row)
+        await session.commit()
+        return True
+
+
+async def set_project_pin(
+    owner_id: int, project_id: int, connection_id: int | None
+) -> bool:
+    """Point a project at one of its OWNER's connections, or clear the pin.
+
+    The caller settles WHO may ask (routes check owner-or-admin); this
+    settles WHOSE connection is eligible: only the project owner's — pinning
+    a collaborator's token to someone else's project is the confused-deputy
+    channel this feature exists to close. False = project or connection not
+    eligible.
+    """
+    async with async_session() as session:
+        project = (
+            await session.execute(select(Project).where(Project.id == project_id))
+        ).scalar_one_or_none()
+        if project is None or (project.user_id or 0) != owner_id:
+            return False
+        if connection_id:
+            held = (
+                await session.execute(
+                    select(ForgeConnection.id).where(
+                        ForgeConnection.id == connection_id,
+                        ForgeConnection.user_id == owner_id,
+                    )
+                )
+            ).scalar_one_or_none()
+            if held is None:
+                return False
+        project.forge_connection_id = connection_id or None
+        await session.commit()
+        return True

src/scribe/services/snippets.py

@@ -941,10 +941,10 @@ async def record_verification(
 
 # --- pull-time freshness (#2690) ---------------------------------------------
 # A pull is the moment freshness matters: the reader is about to trust the
-# cached body. When the instance has a forge configured, the pull fetches the
-# recorded file and answers the one mechanically-answerable question — does
-# the cached code still appear in the source, verbatim after whitespace
-# normalization? The body is a FRAGMENT of the file, so "serve the fetched
+# cached body. When the record owner's keyring serves a forge, the pull
+# fetches the recorded file and answers the one mechanically-answerable
+# question — does the cached code still appear in the source, verbatim after
+# whitespace normalization? The body is a FRAGMENT of the file, so "serve the fetched
 # file" would clobber the record; confirmation + provenance refresh is what
 # fetching can honestly deliver, and divergence is reported, not overwritten.
 #
@@ -990,7 +990,8 @@ async def _refresh_provenance(note, commit_sha: str) -> None:
 async def attach_live_body(note, data: dict) -> None:
     """Decorate a PULL response with forge-checked freshness (#2690).
 
-    Adds, when (and only when) a forge is configured:
+    Adds, when (and only when) the record owner's keyring serves a forge
+    (#2778):
       - ``body_source``: "forge" (confirmed against the source just now) or
         "cache" (the stored body, for whatever reason follows)
       - ``body_freshness``: "current" | "diverged" | "missing" |
@@ -1004,14 +1005,17 @@ async def attach_live_body(note, data: dict) -> None:
     attention state verify_snippet uses.
     """
     from scribe.services.background import spawn
-    from scribe.services.forge import ForgeError, ForgeNotFound, get_forge
+    from scribe.services.forge import ForgeError, ForgeNotFound, get_forges
 
     try:
-        forge = await get_forge()
+        # The OWNER's keyring, honoring the project pin (#2778) — freshness
+        # for a record is checked with its owner's credential, never the
+        # reader's.
+        selector = await get_forges(note.user_id, getattr(note, "project_id", None))
     except Exception:
         logger.warning("forge lookup failed during pull", exc_info=True)
         return
-    if forge is None:
+    if not selector.configured:
         return
 
     fields = data.get("snippet") if isinstance(data.get("snippet"), dict) else None
@@ -1033,18 +1037,19 @@ async def attach_live_body(note, data: dict) -> None:
     # address a forge API — the project's repo BINDING is the identity that
     # can (#2691). Try the location string first (it may be a real remote),
     # then fall back to the bindings of the snippet's project.
-    repo = forge.resolve_repo(loc["repo"])
-    if repo is None and getattr(note, "project_id", None):
+    resolved = selector.resolve(loc["repo"])
+    if resolved is None and getattr(note, "project_id", None):
         from scribe.services.repo_bindings import keys_for_project
 
         for key in await keys_for_project(note.user_id, note.project_id):
-            repo = forge.resolve_repo(key)
-            if repo is not None:
+            resolved = selector.resolve(key)
+            if resolved is not None:
                 break
-    if repo is None:
+    if resolved is None:
         data["body_source"] = "cache"
         data["body_freshness"] = "repo-not-on-this-forge"
         return
+    forge, repo = resolved
 
     stored_prov_sha = (fields.get("provenance") or {}).get("commit_sha") or ""