feat(forge): adapter seam + Gitea implementation — optional read access to the operator's forge (#2689)

Step 4 of milestone 288 (decision #2686). services/forge.py defines the
contract steps 5-7 consume — read_file (content + last_commit_sha, the
provenance stamp), default_branch, resolve_repo, check — with GiteaForge
as the first implementation over the REST contents/repo/version/user
endpoints. Repo identity reuses normalize_repo_key: the host segment
selects whether this forge serves a recorded repo, the remainder is the
API path, so no new identity scheme exists. Read-only by construction;
errors never carry the token; first outbound-HTTP timeout convention
(5s total, no retries — the consumer's fallback is the retry policy).

OPTIONAL per instance (rule #115): get_forge() returns None when
unconfigured and every consumer treats None as today's behavior. Config
lives in admin settings (Settings → Config → Git Forge: kind/base
URL/token, save + test-connection probe reporting version + identity),
with FORGE_* env / Docker-secret fallbacks; DB wins so a UI edit can't
silently lose to an env var. Token treatment follows the smtp_password
convention (masked on read, mask-sentinel skipped on write, absent from
audit details) — and wiring it surfaced that the generic GET/PUT
/api/settings dump bypassed that masking for the owning admin's raw KV
rows, so secret keys are now masked there too (fixes the same exposure
for smtp_password).

Contract tests run against httpx.MockTransport as the fake forge — the
reference behaviors the GitHub adapter (step 8) must reproduce — plus
the off-by-default gate, partial-config-is-off, env-vs-DB precedence,
and route/mask structural checks. Also: the step-2 definition detector
learned to skip dunders after flagging __init__ as 'already defined in
4 files' on this step's own build — guaranteed noise for a hint that
must stay trustworthy.

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

src/scribe/routes/admin.py

@@ -19,6 +19,15 @@ 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 (
@@ -157,6 +166,75 @@ async def test_smtp():
         return jsonify({"error": str(e)}), 500
 
 
+_TOKEN_MASK = "********"
+
+
+@admin_bp.route("/forge", methods=["GET"])
+@admin_required
+async def get_forge_settings():
+    cfg = await forge_config()
+    return jsonify({
+        "kind": cfg["kind"],
+        "base_url": cfg["base_url"],
+        # The token itself never leaves the server — the smtp_password
+        # convention: masked when set, empty when not.
+        "token": _TOKEN_MASK if cfg["token"] else "",
+        "configured": bool(await get_forge()),
+        "kinds": list(FORGE_KINDS),
+    })
+
+
+@admin_bp.route("/forge", 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))
+    # The token 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},
+    )
+    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/settings.py

@@ -16,13 +16,27 @@ logger = logging.getLogger(__name__)
 
 settings_bp = Blueprint("settings", __name__, url_prefix="/api/settings")
 
+# Keys whose values are credentials. The admin endpoints that own them mask on
+# 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"})
+_SECRET_MASK = "********"
+
+
+def _masked(settings: dict) -> dict:
+    return {
+        k: (_SECRET_MASK if k in _SECRET_KEYS and v else v)
+        for k, v in settings.items()
+    }
+
 
 @settings_bp.route("", methods=["GET"])
 @login_required
 async def get_settings_route():
     uid = get_current_user_id()
     settings = await get_all_settings(uid)
-    return jsonify(settings)
+    return jsonify(_masked(settings))
 
 
 @settings_bp.route("", methods=["PUT"])
@@ -36,6 +50,10 @@ async def update_settings_route():
     to_save = {}
     for k, v in data.items():
         str_v = str(v)
+        # A masked secret round-tripping through a client is "unchanged", not
+        # a request to store the mask over the real credential.
+        if k in _SECRET_KEYS and str_v == _SECRET_MASK:
+            continue
         if not str_v:
             await delete_setting(uid, k)
         else:
@@ -45,7 +63,7 @@ async def update_settings_route():
         await set_settings_batch(uid, to_save)
 
     settings = await get_all_settings(uid)
-    return jsonify(settings)
+    return jsonify(_masked(settings))
 
 
 @settings_bp.route("/search", methods=["GET"])