feat: add POST /bible/recover endpoint (homelab#176)

helpers/bible_bridge.py

@@ -10,6 +10,9 @@
   1.1 — Added version constant and /version endpoint. No functional changes.
   1.2 — ThreadingHTTPServer so a slow /bible/pull (git network op) no longer
         serializes fast /bible reads from concurrent callers.
+  1.3 — Added /bible/recover — destructive git reset --hard to origin/main
+        for the divergence case /bible/pull's --ff-only refuses to touch
+        (homelab#176).
 
 Runs in the open-terminal pod. Exposes git-cloned story bible projects
 over HTTP so the Fiction Writing Filter (in the pipelines pod) can read
@@ -36,6 +39,7 @@
   GET  /bible?project=alpha&task_type=CONTINUITY — read bible files
   POST /bible/write                              — write/append + git commit
   POST /bible/pull                               — trigger git pull on a project
+  POST /bible/recover                            — destructive reset to origin/main (confirm:true required)
   POST /bible/sync                               — stage all + commit + push
   POST /bible/pr                                 — create branch + PR
 
@@ -67,7 +71,7 @@
 )
 log = logging.getLogger("bible_bridge")
 
-VERSION      = "1.2"
+VERSION      = "1.3"
 BIBLE_ROOT   = os.environ.get("BIBLE_ROOT", "/home/u3aa02715/fiction")
 BRIDGE_PORT  = int(os.environ.get("BRIDGE_PORT", "8765"))
 BRIDGE_TOKEN = os.environ.get("BRIDGE_TOKEN", "")
@@ -218,6 +222,89 @@ def git_pull(project_path: str) -> tuple[bool, str]:
     return True, out
 
 
+def git_recover(project_path: str, confirm: bool) -> tuple[bool, dict]:
+    """Destructively reset a diverged local `GIT_BRANCH` to match
+    `GIT_REMOTE/GIT_BRANCH`. The residual failure mode git_pull's
+    --ff-only can't fix: local main has commits origin doesn't (post-
+    squash-merge or rebased-force-push), which needs `git reset --hard`.
+
+    Refuses without `confirm=True` (deliberate footgun-guard — this
+    endpoint destroys local commits) and refuses if there are
+    uncommitted changes (no auto-data-loss). homelab#176.
+
+    Returns (ok, payload) where payload is the exact response body
+    (the HTTP handler forwards it as-is, adding no extra keys).
+    """
+    if not confirm:
+        return False, {"ok": False, "status": "confirm_required"}
+
+    st_ok, status_out = git(project_path, "status", "--porcelain")
+    if st_ok and status_out.strip():
+        msg = (
+            f"recover: working tree has uncommitted changes; refusing to "
+            f"reset --hard. Resolve manually: commit/stash changes, then retry."
+        )
+        log.warning(msg)
+        return False, {"ok": False, "status": "uncommitted_changes_present", "porcelain": status_out}
+
+    cb_ok, current_branch = git(project_path, "branch", "--show-current")
+    current_branch = current_branch.strip() if cb_ok else ""
+
+    head_ok, head_sha = git(project_path, "rev-parse", "--short", "HEAD")
+    head_sha = head_sha.strip() if head_ok else "unknown"
+
+    fetch_ok, fetch_out = git(project_path, "fetch", GIT_REMOTE, GIT_BRANCH)
+    if not fetch_ok:
+        return False, {"ok": False, "status": "fetch_failed", "detail": fetch_out}
+
+    log_ok, log_out = git(
+        project_path, "log", f"{GIT_REMOTE}/{GIT_BRANCH}..HEAD", "--format=%h %s"
+    )
+    commits_dropped = []
+    if log_ok and log_out.strip():
+        for line in log_out.strip().split("\n"):
+            sha, _, subject = line.partition(" ")
+            commits_dropped.append({"sha": sha, "subject": subject})
+
+    if current_branch != GIT_BRANCH:
+        co_ok, co_out = git(project_path, "checkout", GIT_BRANCH)
+        if not co_ok:
+            msg = f"recover: failed to switch to '{GIT_BRANCH}': {co_out}"
+            log.warning(msg)
+            return False, {"ok": False, "status": "checkout_failed", "detail": co_out}
+
+    reset_ok, reset_out = git(
+        project_path, "reset", "--hard", f"{GIT_REMOTE}/{GIT_BRANCH}"
+    )
+    if not reset_ok:
+        log.warning(f"recover: reset --hard failed: {reset_out}")
+        return False, {"ok": False, "status": "reset_failed", "detail": reset_out}
+
+    new_head_ok, new_head_sha = git(project_path, "rev-parse", "--short", "HEAD")
+    new_head_sha = new_head_sha.strip() if new_head_ok else "unknown"
+    subj_ok, new_head_subject = git(project_path, "log", "-1", "--format=%s")
+    new_head_subject = new_head_subject.strip() if subj_ok else ""
+
+    log.info(
+        f"recover: abandoned '{current_branch or GIT_BRANCH}'@{head_sha} "
+        f"({len(commits_dropped)} commit(s) dropped) -> "
+        f"'{GIT_BRANCH}'@{new_head_sha}"
+    )
+    return True, {
+        "ok": True,
+        "abandoned": {
+            "branch": current_branch or GIT_BRANCH,
+            "head_sha": head_sha,
+            "commits_dropped": commits_dropped,
+        },
+        "now_at": {
+            "branch": GIT_BRANCH,
+            "head_sha": new_head_sha,
+            "head_subject": new_head_subject,
+        },
+    }
+
+
 # ── Bible read ────────────────────────────────────────────────────────────────
 
 def read_bible(project_path: str, task_type: str = None) -> dict:
@@ -514,6 +601,21 @@ def do_POST(self):
             ok, out = git_pull(project_path)
             self._send_json(200 if ok else 500, {"ok": ok, "output": out})
 
+        elif parsed.path == "/bible/recover":
+            project_path, err = resolve_project(body.get("project"))
+            if err:
+                self._send_json(400, {"error": err})
+                return
+            confirm = body.get("confirm", False)
+            ok, payload = git_recover(project_path, confirm)
+            if ok:
+                code = 200
+            elif payload.get("status") in ("confirm_required", "uncommitted_changes_present"):
+                code = 400
+            else:
+                code = 500
+            self._send_json(code, payload)
+
         elif parsed.path == "/bible/sync":
             project_path, err = resolve_project(body.get("project"))
             if err:

tests/test_bible_recover.py

@@ -0,0 +1,155 @@
+"""
+Tests for git_recover() in bible_bridge (homelab#176 — /bible/recover
+endpoint, the destructive-reset auto-recovery path for a local main that
+has diverged from origin/main at the SHA level).
+
+Uses REAL git repos (a local bare repo as "origin", a real clone as the
+working tree) rather than mocking subprocess calls — git_recover's whole
+job is orchestrating real git commands correctly, so a mock would just
+encode our assumptions about git's behavior instead of testing it
+(feedback_mocks_encode_assumed_contracts).
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+HELPERS_DIR = Path(__file__).resolve().parent.parent / "helpers"
+sys.path.insert(0, str(HELPERS_DIR))
+
+import bible_bridge  # noqa: E402
+
+
+def _run(cwd: Path, *args: str) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        ["git", "-C", str(cwd)] + list(args),
+        capture_output=True, text=True, check=True,
+    )
+
+
+def _write_commit(repo: Path, filename: str, content: str, message: str) -> str:
+    (repo / filename).write_text(content)
+    _run(repo, "add", filename)
+    _run(repo, "commit", "-m", message)
+    sha = _run(repo, "rev-parse", "--short", "HEAD").stdout.strip()
+    return sha
+
+
+@pytest.fixture
+def repo_pair(tmp_path, monkeypatch):
+    """Real bare 'origin' repo + a real clone as the working tree, both
+    with git identity configured so commits succeed hermetically."""
+    origin = tmp_path / "origin.git"
+    origin.mkdir()
+    subprocess.run(["git", "init", "--bare", "-b", "main", str(origin)],
+                    capture_output=True, text=True, check=True)
+
+    seed = tmp_path / "seed"
+    seed.mkdir()
+    subprocess.run(["git", "init", "-b", "main", str(seed)],
+                    capture_output=True, text=True, check=True)
+    _run(seed, "config", "user.email", "test@example.com")
+    _run(seed, "config", "user.name", "Test")
+    _write_commit(seed, "bible.md", "seed content\n", "seed commit")
+    _run(seed, "remote", "add", "origin", str(origin))
+    _run(seed, "push", "origin", "main")
+
+    work = tmp_path / "work"
+    subprocess.run(["git", "clone", str(origin), str(work)],
+                    capture_output=True, text=True, check=True)
+    _run(work, "config", "user.email", "test@example.com")
+    _run(work, "config", "user.name", "Test")
+
+    monkeypatch.setattr(bible_bridge, "GIT_REMOTE", "origin")
+    monkeypatch.setattr(bible_bridge, "GIT_BRANCH", "main")
+
+    return origin, work
+
+
+def test_recover_requires_confirm(repo_pair):
+    """Without confirm=True, refuse outright — no git commands run at all."""
+    _origin, work = repo_pair
+    ok, payload = bible_bridge.git_recover(str(work), confirm=False)
+    assert ok is False
+    assert payload == {"ok": False, "status": "confirm_required"}
+
+
+def test_recover_refuses_uncommitted_changes(repo_pair):
+    """A dirty working tree is never auto-reset — no data-loss surprise."""
+    _origin, work = repo_pair
+    (work / "bible.md").write_text("uncommitted local edit\n")
+    ok, payload = bible_bridge.git_recover(str(work), confirm=True)
+    assert ok is False
+    assert payload["status"] == "uncommitted_changes_present"
+    assert "bible.md" in payload["porcelain"]
+
+
+def test_recover_happy_path_diverged_main(repo_pair):
+    """The core scenario: local main has a commit origin doesn't (e.g.
+    post-squash-merge). Recovery resets to origin/main and reports what
+    was abandoned."""
+    origin, work = repo_pair
+
+    # Simulate origin moving forward (as if merged elsewhere)...
+    seed2 = origin.parent / "seed2"
+    subprocess.run(["git", "clone", str(origin), str(seed2)],
+                    capture_output=True, text=True, check=True)
+    _run(seed2, "config", "user.email", "test@example.com")
+    _run(seed2, "config", "user.name", "Test")
+    upstream_sha = _write_commit(seed2, "other.md", "upstream content\n", "upstream commit")
+    _run(seed2, "push", "origin", "main")
+
+    # ...while the local work clone independently commits something that
+    # never made it to origin (the exact "diverged main" wedge).
+    local_sha = _write_commit(work, "local.md", "local-only content\n", "local-only commit")
+
+    ok, payload = bible_bridge.git_recover(str(work), confirm=True)
+
+    assert ok is True
+    assert payload["ok"] is True
+    assert payload["abandoned"]["branch"] == "main"
+    assert payload["abandoned"]["head_sha"] == local_sha
+    dropped_shas = [c["sha"] for c in payload["abandoned"]["commits_dropped"]]
+    assert local_sha in dropped_shas
+    assert payload["now_at"]["branch"] == "main"
+    assert payload["now_at"]["head_sha"] == upstream_sha
+
+    # Real filesystem state matches the report — the dropped file is gone,
+    # the upstream-only file is present.
+    assert not (work / "local.md").exists()
+    assert (work / "other.md").exists()
+
+    # Working tree is genuinely clean after the reset.
+    status = _run(work, "status", "--porcelain").stdout
+    assert status.strip() == ""
+
+
+def test_recover_on_a_non_default_branch(repo_pair):
+    """Divergence can also be discovered while sitting on a feature branch
+    (not just main itself) — recovery should still land on GIT_BRANCH."""
+    origin, work = repo_pair
+    _run(work, "checkout", "-b", "some-feature-branch")
+    local_sha = _write_commit(work, "scratch.md", "scratch\n", "scratch commit")
+
+    ok, payload = bible_bridge.git_recover(str(work), confirm=True)
+
+    assert ok is True
+    assert payload["abandoned"]["branch"] == "some-feature-branch"
+    assert payload["abandoned"]["head_sha"] == local_sha
+    assert payload["now_at"]["branch"] == "main"
+    current_branch = _run(work, "branch", "--show-current").stdout.strip()
+    assert current_branch == "main"
+
+
+def test_recover_no_divergence_is_a_noop_success(repo_pair):
+    """If main already matches origin/main exactly, recovery still
+    succeeds — just with an empty commits_dropped list."""
+    _origin, work = repo_pair
+    ok, payload = bible_bridge.git_recover(str(work), confirm=True)
+    assert ok is True
+    assert payload["abandoned"]["commits_dropped"] == []
+    assert payload["now_at"]["head_sha"] == payload["abandoned"]["head_sha"]