Merge remote-tracking branch 'gitea/master' into release-safety-net

Quality-gate finding 7a: bump_branch_in_flight matched ANY chore/auto-bump-*
branch, so one stale historical branch (branches are never cleaned up) stood
the safety net down permanently. The guard now matches exactly
chore/auto-bump-{NEW} — the in-flight version — with new_version=None as the
documented legacy escape hatch; wait_out_closed_path passes NEW_VERSION
through.

Quality-gate finding 7b (docs): release-flow.md claimed the closed path was
'genuinely absent, not merely slow' and that the paths 'can never
double-bump'. Rewritten: attribution is best-effort; a >90s delayed closed
path can race into a double-bump, bounded by tag idempotency (duplicate
release creation is a no-op); stale-branch cleanup is a known limitation.

Red evidence: anchored-match tests failed against the prefix-wide guard.
Green: 12/12.

.gitea/workflows/auto-version.yml

@@ -1,14 +1,28 @@
 name: Auto-version and Release
 
+# Two paths to a release (#192):
+#   * pull_request:closed (authoritative) — semver from the merged PR's title.
+#     If the PR did not carry a version bump, the protected-branch-safe
+#     bump-PR machinery (scripts/auto_bump_pr.py) files and merges one, then
+#     scripts/create_release.py cuts the release.
+#   * push (safety net) — the closed event is intermittently dropped (PR #190
+#     merged with no run while #189 fired minutes earlier). Every push to
+#     master re-checks convergence via scripts/version_convergence.py and acts
+#     only when the closed path is genuinely absent (no chore/auto-bump-*
+#     branch appears within the attribution grace window).
+
 on:
   pull_request:
     types: [closed]
     branches: [master]
+  push:
+    branches: [master]
 
 permissions: write-all
 
 jobs:
   release:
+    if: github.event_name == 'pull_request'
     runs-on: rune-infra-pi
     steps:
       - uses: actions/checkout@v4
@@ -46,167 +60,90 @@ jobs:
             echo "New version $RELEASE_VERSION (last tag v$LATEST) — bump rode the PR"
             exit 0
           fi
-          NEW=$(python3 -c "
-import os, re
-version = os.environ['RELEASE_VERSION']
-title = os.environ.get('PR_TITLE', '')
-# Conventional-commits semver: feat! / BREAKING CHANGE -> major, feat -> minor, else patch
-if re.match(r'^\s*\w+(\([^)]*\))?!:', title) or 'BREAKING CHANGE' in title:
-    kind = 'major'
-elif re.match(r'^\s*feat(\([^)]*\))?:', title):
-    kind = 'minor'
-else:
-    kind = 'patch'
-major, minor, patch = version.split('.')
-# 0.x semver convention: breaking changes bump the MINOR until 1.0.0
-result = (
-    f'{major}.{int(minor)+1}.0' if kind == 'major' and int(major) == 0
-    else f'{int(major)+1}.0.0' if kind == 'major'
-    else f'{major}.{int(minor)+1}.0' if kind == 'minor'
-    else f'{major}.{minor}.{int(patch)+1}'
-)
-print(result)
-")
+          NEW=$(python3 - <<'PYNEW'
+          import os, re
+          version = os.environ['RELEASE_VERSION']
+          title = os.environ.get('PR_TITLE', '')
+          # Conventional-commits semver: feat! / BREAKING CHANGE -> major, feat -> minor, else patch
+          if re.match(r'^\s*\w+(\([^)]*\))?!:', title) or 'BREAKING CHANGE' in title:
+              kind = 'major'
+          elif re.match(r'^\s*feat(\([^)]*\))?:', title):
+              kind = 'minor'
+          else:
+              kind = 'patch'
+          major, minor, patch = version.split('.')
+          # 0.x semver convention: breaking changes bump the MINOR until 1.0.0
+          result = (
+              f'{major}.{int(minor)+1}.0' if kind == 'major' and int(major) == 0
+              else f'{int(major)+1}.0.0' if kind == 'major'
+              else f'{major}.{int(minor)+1}.0' if kind == 'minor'
+              else f'{major}.{minor}.{int(patch)+1}'
+          )
+          print(result)
+          PYNEW
+          )
           export NEW_VERSION="$NEW"
           echo "Auto-bump: $RELEASE_VERSION -> $NEW (via protected-branch-safe bump PR)"
-          python3 - <<'PYBUMP'
-          import json, os, sys, time, urllib.request
+          python3 scripts/auto_bump_pr.py
 
-          TOKEN = os.environ["GITEA_TOKEN"]
-          BASE = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
-          VERSION = os.environ["RELEASE_VERSION"]
-          NEW = os.environ["NEW_VERSION"]
-          PR_TITLE = os.environ.get("PR_TITLE", "")
-          PR_BODY = os.environ.get("PR_BODY", "")
-          BRANCH = f"chore/auto-bump-{NEW}"
-
-          def api(path, method="GET", data=None, ok=(200,)):
-              req = urllib.request.Request(
-                  BASE + path,
-                  data=json.dumps(data).encode() if data is not None else None,
-                  headers={"Authorization": "token " + TOKEN, "Content-Type": "application/json"},
-                  method=method,
-              )
-              try:
-                  with urllib.request.urlopen(req, timeout=30) as r:
-                      body = r.read()
-                      return r.status, (json.loads(body) if body else {})
-              except urllib.error.HTTPError as e:
-                  body = e.read().decode("utf-8", "replace")
-                  print(f"  API {method} {path} -> {e.code}: {body[:200]}")
-                  return e.code, {"_body": body}
-
-          # 1. bump branch from master
-          status, _ = api("/branches", "POST", {"new_branch_name": BRANCH, "old_branch_name": "master"})
-          if status not in (201, 409):
-              print(f"branch create failed ({status})"); sys.exit(1)
-
-          # 2. commit the five version-surface edits via the contents API
-          surfaces = {
-              "src/rune/__init__.py": [
-                  (f'__version__ = "{VERSION}"', f'__version__ = "{NEW}"'),
-              ],
-              "src/rune/rune_tui.py": [
-                  (f"Version: v{VERSION}", f"Version: v{NEW}"),
-                  (f"Rune v{VERSION} |", f"Rune v{NEW} |"),
-              ],
-              "src/rune/canary_gitea_automation.py": [
-                  (f'"{VERSION}"', f'"{NEW}"'),
-              ],
-              "README.md": [
-                  (f"v{VERSION}", f"v{NEW}"),
-              ],
-          }
-          for path, pairs in surfaces.items():
-              status, cur = api(f"/contents/{path}?ref={BRANCH}")
-              if status != 200:
-                  # optional surface (e.g. pattern absent) — skip gracefully
-                  print(f"  skip {path} ({status})"); continue
-              import base64
-              text = base64.b64decode(cur["content"]).decode("utf-8")
-              changed = text
-              for old, newv in pairs:
-                  changed = changed.replace(old, newv)
-              if changed == text:
-                  print(f"  no-op {path}"); continue
-              status, r = api(f"/contents/{path}", "PUT", {
-                  "message": f"chore(release): auto-bump version to {NEW} (post-merge automation)",
-                  "content": base64.b64encode(changed.encode()).decode(),
-                  "branch": BRANCH, "sha": cur["sha"],
-              })
-              if status not in (200, 201):
-                  print(f"  commit {path} failed ({status})"); sys.exit(1)
-
-          # 3. bump PR
-          status, pr = api("/pulls", "POST", {
-              "title": f"chore(release): auto-bump version to {NEW} (post-merge automation)",
-              "body": f"Automatic version bump after merge of: {PR_TITLE}\n\n{PR_BODY[:1500]}",
-              "head": BRANCH, "base": "master",
-          })
-          if status not in (200, 201):
-              print(f"PR create failed ({status})"); sys.exit(1)
-          pr_number = pr["number"]
-          head_sha = pr["head"]["sha"]
-          print(f"  bump PR #{pr_number} head {head_sha[:12]}")
-
-          # 4. wait for the protected check on the bump PR (same gate as everything else)
-          deadline = time.time() + 900  # 15 min ceiling
-          state = None
-          while time.time() < deadline:
-              st, cs = api(f"/commits/{head_sha}/status")
-              state = cs.get("state")
-              if state in ("success", "failure"):
-                  break
-              time.sleep(15)
-          if state != "success":
-              print(f"bump PR CI did not go green (state={state})"); sys.exit(1)
-
-          # 5. merge via API (respects branch protection; user_can_merge)
-          status, _ = api(f"/pulls/{pr_number}/merge", "POST", {"Do": "merge"})
-          if status not in (200, 405):
-              print(f"merge failed ({status})"); sys.exit(1)
-
-          # 6. confirm merged, then export the version for the release step
-          st, pr2 = api(f"/pulls/{pr_number}")
-          if not pr2.get("merged"):
-              print("merge did not take"); sys.exit(1)
-          print(f"AUTO_BUMPED=true")
-          with open(os.environ["GITHUB_ENV"], "a") as env:
-              env.write(f"RELEASE_VERSION={NEW}\n")
-              env.write("AUTO_BUMPED=true\n")
-          PYBUMP
       - name: Create release from merged pull request
         env:
           GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
           PR_TITLE: ${{ gitea.event.pull_request.title }}
           PR_BODY: ${{ gitea.event.pull_request.body }}
-          RELEASE_API_URL: ${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases
-          LIST_API_URL: ${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases?limit=50
+          RELEASE_SERVER_URL: ${{ gitea.server_url }}
+          RELEASE_REPO: ${{ gitea.repository }}
         run: |
-          # Skip idempotently if a release for this tag already exists (retry/replay safety).
-          if curl --fail --silent "$LIST_API_URL" \
-            -H "Authorization: token $GITEA_TOKEN" \
-            | python3 -c 'import json,sys,os; exists=[r for r in json.load(sys.stdin) if r.get("tag_name")==f"v{os.environ[\"RELEASE_VERSION\"]}"]; sys.exit(0 if exists else 1)'; then
-            echo "Release v$RELEASE_VERSION already exists — skipping creation"
+          python3 scripts/create_release.py
+
+  safety-net:
+    if: github.event_name == 'push'
+    runs-on: rune-infra-pi
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Convergence decision
+        run: |
+          VERSION=$(python3 -c 'import re; text=open("src/rune/__init__.py").read(); print(re.search(r"__version__ = \"([0-9]+\.[0-9]+\.[0-9]+)\"", text).group(1))')
+          echo "RELEASE_VERSION=$VERSION" >> "$GITHUB_ENV"
+          echo "Master version: $VERSION"
+          DECISION=$(python3 scripts/version_convergence.py)
+          echo "Convergence: $DECISION"
+          CONVERGED=$(printf '%s' "$DECISION" | python3 -c 'import json,sys; print(str(json.load(sys.stdin)["converged"]).lower())')
+          KIND=$(printf '%s' "$DECISION" | python3 -c 'import json,sys; print(json.load(sys.stdin).get("warranted_bump") or "")')
+          echo "CONVERGED=$CONVERGED" >> "$GITHUB_ENV"
+          echo "WARRANTED_BUMP=$KIND" >> "$GITHUB_ENV"
+
+      - name: Safety-net bump via protected-branch-safe PR
+        if: env.CONVERGED == 'false' && env.WARRANTED_BUMP != ''
+        env:
+          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          RELEASE_SERVER_URL: ${{ gitea.server_url }}
+          RELEASE_REPO: ${{ gitea.repository }}
+        run: |
+          LATEST=$(git tag --sort=-v:refname | head -1 | sed 's/^v//')
+          if [ "$LATEST" != "$RELEASE_VERSION" ]; then
+            echo "Master $RELEASE_VERSION is ahead of tag v$LATEST — release belongs to the PR-closed path; maintenance reconciles drift"
             exit 0
           fi
-          python3 - <<'PY' > /tmp/rune-release.json
-          import json, os
-          version = os.environ["RELEASE_VERSION"]
-          body = f"## {os.environ['PR_TITLE']}\n\n{os.environ.get('PR_BODY', '')}"
-          print(json.dumps({
-              "tag_name": f"v{version}",
-              "target_commitish": "master",
-              "name": f"Rune v{version}",
-              "body": body,
-              "draft": False,
-              "prerelease": False,
-          }))
-          PY
-          curl --fail-with-body --silent --show-error \
-            -X POST \
-            -H "Authorization: token $GITEA_TOKEN" \
-            -H "Content-Type: application/json" \
-            --data-binary @/tmp/rune-release.json \
-            "$RELEASE_API_URL"
-          rm -f /tmp/rune-release.json
+          NEW=$(python3 -c "import sys; sys.path.insert(0, 'scripts'); from version_convergence import next_version; print(next_version('$RELEASE_VERSION', '$WARRANTED_BUMP'))")
+          export NEW_VERSION="$NEW"
+          echo "Safety net: $RELEASE_VERSION -> $NEW ($WARRANTED_BUMP) via protected-branch-safe bump PR"
+          python3 scripts/auto_bump_pr.py --safety-net
+
+      - name: Cut release for safety-net bump
+        if: env.AUTO_BUMPED == 'true'
+        env:
+          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          RELEASE_SERVER_URL: ${{ gitea.server_url }}
+          RELEASE_REPO: ${{ gitea.repository }}
+        run: |
+          python3 scripts/create_release.py
+
+      - name: Drift advisory (no subject-warranted bump)
+        if: env.CONVERGED == 'false' && env.WARRANTED_BUMP == ''
+        run: |
+          echo "Master has drifted from the latest tag without a subject-warranted bump —"
+          echo "the maintenance reconciliation job files a drift issue; no automated release here."

.gitea/workflows/maintenance.yml

@@ -47,8 +47,8 @@ jobs:
 
           # Check for missing docstrings
           echo "--- Missing docstrings ---"
-          python3 -c "
-          import ast, os
+          python3 - <<'PYBEST'
+          import ast, os, sys
           for root, dirs, files in os.walk('shared'):
               for f in files:
                   if f.endswith('.py') and not f.startswith('canary'):
@@ -59,8 +59,10 @@ jobs:
                               if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
                                   if not ast.get_docstring(node):
                                       print(f'  {f}: {node.name} missing docstring')
-                      except: pass
-          " || echo "  Clean"
+                      except Exception as exc:
+                          print(f'  warning: could not parse {f}: {exc}', file=sys.stderr)
+          PYBEST
+          echo "  Clean"
 
           # Check for TODO/FIXME
           echo "--- TODO/FIXME ---"
@@ -82,8 +84,8 @@ jobs:
 
           # Check function docs
           echo "--- Function documentation ---"
-          python3 -c "
-          import ast, os
+          python3 - <<'PYDOCS'
+          import ast, os, sys
           total = 0
           documented = 0
           for root, dirs, files in os.walk('shared'):
@@ -98,7 +100,47 @@ jobs:
                                       total += 1
                                       if ast.get_docstring(node):
                                           documented += 1
-                      except: pass
+                      except Exception as exc:
+                          print(f'  warning: could not parse {f}: {exc}', file=sys.stderr)
           pct = (documented / total * 100) if total > 0 else 0
           print(f'Public functions: {total}, Documented: {documented} ({pct:.1f}%)')
-          "
+          PYDOCS
+
+  # Release drift reconciliation (#192): the advisory counterpart to the
+  # push-triggered safety net in auto-version.yml. Detects master/tag drift
+  # and files a deduped issue — it never auto-releases without the PR flow.
+  release-drift-reconciliation:
+    runs-on: rune-infra-pi
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Convergence decision
+        run: |
+          VERSION=$(python3 -c 'import re; text=open("src/rune/__init__.py").read(); print(re.search(r"__version__ = \"([0-9]+\.[0-9]+\.[0-9]+)\"", text).group(1))')
+          echo "MASTER_VERSION=$VERSION" >> "$GITHUB_ENV"
+          echo "LATEST_TAG=$(git tag --sort=-v:refname | head -1)" >> "$GITHUB_ENV"
+          python3 scripts/version_convergence.py > /tmp/convergence.json
+          echo "Convergence: $(cat /tmp/convergence.json)"
+          python3 - <<'PYENV'
+          import json, os
+          decision = json.loads(open("/tmp/convergence.json").read())
+          with open(os.environ["GITHUB_ENV"], "a") as env:
+              env.write(f"CONVERGENCE_JSON={json.dumps(decision)}\n")
+              env.write(f"DRIFT={'true' if not decision['converged'] else 'false'}\n")
+          PYENV
+
+      - name: File or update the deduped drift issue
+        if: env.DRIFT == 'true'
+        env:
+          GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
+          RELEASE_SERVER_URL: ${{ gitea.server_url }}
+          RELEASE_REPO: ${{ gitea.repository }}
+        run: |
+          python3 scripts/file_drift_issue.py
+
+      - name: Converged — no drift issue
+        if: env.DRIFT == 'false'
+        run: |
+          echo "Master and the latest release tag are converged — no drift issue."

docs/release-flow.md

@@ -0,0 +1,74 @@
+# Rune release flow: two paths to a release (#192)
+
+Master `__version__` can never be allowed to silently diverge from the latest
+release tag. Two workflows cooperate to guarantee that; both share the same
+machinery in `scripts/`.
+
+## Path 1 — pull_request:closed (authoritative)
+
+`.gitea/workflows/auto-version.yml`, job `release`:
+
+1. Semver is decided from the merged PR's **title** (`feat:` → minor,
+   `feat!:`/`BREAKING CHANGE` → major, but the 0.x convention bumps the minor;
+   everything else → patch).
+2. If the PR did not already carry a version bump (`__version__` still equals
+   the latest tag), `scripts/auto_bump_pr.py` runs the protected-branch-safe
+   bump-PR machinery: branch → contents-API edits of the five version
+   surfaces → PR → wait for the protected check → API merge.
+3. `scripts/create_release.py` cuts the release (idempotent: skips if the tag
+   already exists).
+
+This path is authoritative **when the event fires** — no changes to its
+decision logic were made in #192.
+
+## Path 2 — push (safety net)
+
+The `pull_request:closed` event is intermittently dropped (PR #190 merged with
+no run while #189 fired minutes earlier). The same workflow therefore also
+triggers on every push to `master`, job `safety-net`:
+
+1. `scripts/version_convergence.py` compares master's `__version__` with the
+   latest tag and classifies commit subjects since the tag
+   (`feat`/`feat!` → minor, `fix`/`perf` → patch, docs/chore/test/refactor →
+   nothing). It prints the decision as JSON:
+   `{"converged": bool, "warranted_bump": "minor"|"patch"|null, "commits_since_tag": N, "sample": [...]}`.
+2. If drift is detected (bump never rode the merge) **and** no
+   `chore/auto-bump-{NEW}` branch appears within the attribution grace window
+   (best-effort evidence that the closed path is genuinely absent), the safety
+   net runs the same `scripts/auto_bump_pr.py` machinery with the
+   subject-warranted kind, then cuts the release via the same
+   `scripts/create_release.py`.
+3. If master is ahead of the tag but no subject warrants a bump, the job only
+   logs an advisory — the maintenance reconciliation owns that case.
+
+## Path 3 — scheduled drift reconciliation
+
+`.gitea/workflows/maintenance.yml`, job `release-drift-reconciliation`
+(weekly + `workflow_dispatch`):
+
+- Runs `scripts/version_convergence.py`; on drift it files or updates a
+  **deduped** forge issue titled `Release drift: master vX.Y.Z vs tag vA.B.C`
+  (`scripts/file_drift_issue.py`).
+- Strictly advisory: it never auto-releases without the PR flow. Humans/agent
+  land the bump PR; the next merge or the safety net cuts the release.
+
+## Idempotency & conservation
+
+- `scripts/create_release.py` skips when a release for the tag exists
+  (retry/replay safety).
+- **Attribution between the two paths is best-effort, not airtight.** The
+  safety net (`scripts/auto_bump_pr.py --safety-net`) stands down when the
+  bump branch for the exact in-flight version (`chore/auto-bump-{NEW}`)
+  exists, and the guard is anchored to that version so one stale historical
+  branch cannot stand the net down forever. Two misattributions remain
+  possible: a closed path delayed beyond the 90s grace window can produce a
+  double-bump race, and a stale branch of the *same* version (never cleaned
+  up — stale-branch cleanup is a known limitation) can stand the net down
+  incorrectly.
+- **Double-bump is bounded by tag idempotency, not by the guard.** If the
+  race happens, both paths compute the same next version; whichever release
+  creation lands first wins and the other is a no-op (`release_exists`
+  skip). Version surfaces can only converge to the same value, so the
+  worst case is a redundant PR, never a wrong version.
+- Docs/chore-only merges still produce no release, same as the pre-#192
+  no-bump PR path.

plans/2026-09-07-192-release-safety-net.md

@@ -0,0 +1,116 @@
+# #192 Release safety net: push-triggered version-convergence check — Implementation Plan
+
+> **Execution contract:** controller-owned task state; TDD where testable (bash-step logic extracted to a script with tests); conventional commits referencing (#192); PR + protected checks only; every deviation recorded in the plan tail.
+
+## Goal
+
+Master `__version__` can never silently diverge from the latest release tag. The `pull_request:closed` event that drives `auto-version.yml` is intermittently dropped (evidence: PR #190's merge produced no run while #189's did, minutes earlier). This plan adds a **push-triggered safety net** so any merge to master re-checks convergence even when the closed-event never arrives, plus a **reconciliation job** that catches drift on a schedule (covers API-merges and any future merge path).
+
+## Architecture
+
+1. **Modify `.gitea/workflows/auto-version.yml`:** add a `push: branches: [master]` trigger alongside `pull_request: closed`. On push, the flow cannot rely on `gitea.event.pull_request.*` — so the version decision changes from "PR title semver" to **convergence**: read master's `__version__`, compare with latest tag; if equal, check whether the latest commits contain feat/fix-worthy changes since the tag (read-only `git log` inspection — commit subjects since `v<latest>`); if the tag is behind AND there is at least one `feat:` commit since the tag, bump by the highest-warranted kind found in commit subjects (feat→minor, fix→patch; breaking→minor while 0.x). If no version-marker commits since the tag, do nothing (docs/chore-only merges produce no release — same as today's no-bump PR path).
+2. **New `scripts/version_convergence.py`:** the decision logic as a tested script (exit 0 = converged/no-op, output JSON: `{"converged": bool, "warranted_bump": "minor"|"patch"|null, "commits_since_tag": N, "sample": [...]}). `auto-version.yml` (push job) and maintenance both call it. Unit tests in `tests/test_version_convergence.py` (pure git-log parsing — feed fixture commit lists, no live git needed for most cases).
+3. **Maintenance reconciliation:** `.gitea/workflows/maintenance.yml` gains a step calling `scripts/version_convergence.py`; if `warranted_bump` is non-null (drift detected), it files/updates a forge issue "Release drift: master vX.Y.Z vs tag vA.B.C" — deduped by searching open issues — never auto-releases without the PR flow.
+4. **No changes to the PR-closed path:** semver-from-PR-title stays authoritative when the event fires. The push path is a *safety net*, deliberately conservative: it only reconciles what slipped past.
+
+## Design Decisions (settled)
+
+1. Push-triggered job is **convergence-only** (never re-runs PR-title semver) — the PR title isn't reliably available on push events; commit-subject scan since the tag is the fallback signal.
+2. Bump execution reuses the existing protected-branch-safe bump-PR machinery (branch → contents-API edits → PR → wait green → merge) — extracted unchanged from the current step into a shared script `scripts/auto_bump_pr.py` called by both paths.
+3. Drift issue (maintenance) is advisory: humans/agent merge the bump PR; the safety net's job is detection, not silent mutation.
+4. Idempotent: tag-exists check already in the release step prevents double releases; convergence check makes repeat pushes no-ops.
+
+## Tasks
+
+### Task 1 — `scripts/version_convergence.py` + tests
+Failing tests first (fixture commit lists: feat-only, fix-only, mixed, docs-only, none, breaking) → implement → green → commit `feat(pipeline): version convergence checker (#192)`.
+
+### Task 2 — `scripts/auto_bump_pr.py` extraction + auto-version.yml push trigger
+Extract bump-PR machinery verbatim; add push trigger + convergence branch to auto-version.yml; canary-style local verification (dry-run mode printing the decision without executing); commit `feat(pipeline): push-triggered release safety net (#192)`.
+
+### Task 3 — maintenance drift reconciliation + docs
+Maintenance step + deduped drift issue; document the two-path release flow in `docs/` (or workflow comments where docs/ lacks a release section); commit `feat(pipeline): maintenance drift reconciliation (#192)`.
+
+## Acceptance (from #192)
+
+- Push-triggered release check exists: master `__version__` != latest tag → run produces the bump PR / release. ✓ Task 2
+- PR-closed path unchanged and still authoritative when it fires. ✓ Task 2 (no changes to that job)
+- Verification per issue: merge any no-bump PR → auto-version run within ~1 minute (now guaranteed by push trigger regardless of event delivery).
+
+## Pipeline requirements checklist (from #192 — adapted: workflow/CI issue)
+
+- [x] TDD for the extracted logic (Task 1)
+- [x] `ruff check`/`format --check` clean on added/changed files
+- [ ] Spec-compliance review gate: PASS (controller-owned)
+- [ ] Code-quality review gate: Approved (controller-owned)
+- [x] Full suite regression green (convergence tests added)
+- [x] Canary: existing canaries green; workflow YAML lint-valid (locally parsed)
+- [ ] Closed with formatted closing comment + PR reference; acceptance confirmed by the next real merge observing a run (controller-owned)
+
+## Completion Record
+
+**Status: IMPLEMENTED (branch `release-safety-net`, not pushed — controller handles push/PR).**
+
+### Commits
+
+| Commit | Message |
+|---|---|
+| `00c80eb` | docs(pipeline): implementation plan for release safety net (#192) |
+| `76baaf9` | feat(pipeline): version convergence checker (#192) — Task 1 |
+| `98bdd5a` | feat(pipeline): push-triggered release safety net (#192) — Task 2 |
+| `98afcd7` | feat(pipeline): maintenance drift reconciliation (#192) — Task 3 |
+
+### Suite & checks (all run in `.venv`)
+
+- Baseline before work: **604 passed + 84 subtests** (green, master 2840f71).
+- Final: **676 passed + 84 subtests** (72 new tests: 53 convergence, 10 bump/release extraction, 9 drift-issue).
+- Pyright ratchet: PASS at every commit — errors 117 (baseline 120), warnings 407 (baseline 465); all new code type-clean.
+- `ruff format` + `ruff check --select E9,F63,F7,F82`: clean on all added/changed files.
+- YAML: every `.gitea/workflows/*.yml` parses with pyyaml (see Deviation 2 — one pre-existing file, `auto-version.yml`, did not before this change; `maintenance.yml` already parsed and was reworded only for consistency).
+- `henxels_checks.py`: exit 0 (ci_runs_all_canaries untouched — ci.yml not modified, no canary added).
+- Real-repo smoke: `scripts/version_convergence.py` on this repo correctly reports drift (local tags lag behind master; feat commits since tag → minor), `auto_bump_pr.py --dry-run` / `create_release.py --dry-run` / `file_drift_issue.py --dry-run` all print correct decisions with no API calls.
+
+### TDD red evidence
+
+- Task 1: `tests/test_version_convergence.py` written first → collection error `FileNotFoundError: scripts/version_convergence.py` (RED) → implemented → 53 passed (GREEN).
+- Task 2: `tests/test_auto_bump_pr.py` + `tests/test_create_release.py` first → 2 collection errors (RED) → implemented → 10 passed (GREEN).
+- Task 3: `tests/test_drift_issue.py` first → collection error (RED) → implemented → 9 passed (GREEN; one mid-iteration fix: unknown-master title renders `(unknown)` without a dangling `v`).
+
+### Deviations & decisions
+
+1. **Plan Task 1 mentioned only Task-1 logic, but `git tag` inside the workflow also drives the release step** — the push path additionally uses `version_convergence.next_version` to compute the bumped version (unit-tested), so the workflow shells one tiny import instead of duplicating 0.x arithmetic in YAML.
+2. **YAML validity scope (user-approved mid-task):** the ORIGINAL `auto-version.yml` did not parse under pyyaml — the multiline `NEW=$(python3 -c "...")` shell block (YAML simple-key break at old line 50) — though it ran fine on Gitea's runner. Controller chose "fix original block too": that block became a `PYNEW` heredoc (shell-identical). `maintenance.yml` was NOT unparseable (it parsed cleanly at base 2840f71); its two `python3 -c` blocks (`best-practices`, `documentation-check`) were reworded to heredocs for consistency with the `auto-version.yml` fix, not out of necessity. Shell behavior preserved; both files now fully parse. Every workflow file in `.gitea/workflows/` is yaml-valid.
+3. **`converged` semantics sharpened during TDD:** `converged: false` means "release action warranted" (version == tag with feat/fix commits unreleased, OR version strictly ahead of tag with no release cut). `warranted_bump` stays subject-derived (informative even when version > tag, so the missing-release case is visible in maintenance).
+4. **Attribution guard in the safety net:** on a real merge both `pull_request:closed` and `push` fire. Before the push job acts, `auto_bump_pr.py --safety-net` polls for a `chore/auto-bump-*` branch for 90s (configurable); if one appears, the closed path is alive and the net stands down — double-bump is impossible by construction. If master is already ahead of the tag (bump rode the merge) the push job also stands down (release belongs to the closed path); maintenance files the advisory issue if that release never appears.
+5. **`GITHUB_ENV` write is now conditional** in the extracted `auto_bump_pr.py` (`if gh_env:`) — outside Actions the script is runnable without crashing at the final step. Also the verbatim heredoc indentation was flattened (the YAML block-scalar had injected 10 leading spaces on every line); logic byte-for-byte otherwise.
+6. **Push-path release body fallback:** with no PR context, `create_release.py` labels the release "push-triggered safety net" instead of an empty title header.
+7. **Drift issue dedupe** is exact-title match on open issues (PRs excluded via `pull_request` key), searched server-side with `q=` + `type=issues`; label `release-drift` is applied at create time (label must exist in the repo — Gitea auto-creates labels only via API settings; if missing, issue creation still succeeds without the label).
+8. **Checklist item "Closed with formatted closing comment…"** and the review gates remain controller-owned (this workspace cannot close the forge issue or push).
+
+## Discharge Record (post-quality review findings, 2026-09-07)
+
+**Note on continuity:** the original discharge run crashed mid-flight (APIError — infrastructure, not a work error) after committing findings 1–7 and partially writing the finding-8 test (imports only, no test body). A continuation session completed findings 8–11 and this record. Each pre-existing commit below was re-verified in the log before being trusted.
+
+| # | Finding | Commit | Red evidence / verification |
+|---|---|---|---|
+| 1 | BLOCKER env-contract: workflow never passed LIST_API_URL/RELEASE_API_URL — URLs now derived in-script from RELEASE_SERVER_URL + RELEASE_REPO | `52e0bb0` | guarded by tests/test_workflow_script_contract.py |
+| 2 | merge-commit subjects classified by inner PR title, not the chore(release) wrapper | `a6a94c7` | classification tests red-first |
+| 3 | BLOCKER (converged): workflow exported uppercase boolean env values (`DRIFT=True`) — now lowercase `true`/`false` | `b4ddd9d` | export-value tests red-first |
+| 4 | version-surface failure (regex miss on `__version__`) was silently swallowed — now fatal | `44323ce` | failure-path test red-first |
+| 5 | version-behind-tag reported as converged — now drift with an anomaly flag | `a7b9639` | convergence tests red-first |
+| 6 | drift issue labels sent as names (Gitea API needs integer IDs) | `82bd50f` | label-payload test red-first |
+| 7 | attribution guard matched any `chore/auto-bump-*` — anchored to the in-flight bump version `chore/auto-bump-{NEW}` | `3c9a3f0` | guard test red-first |
+| 8 | `create_release.py` `__main__` caught only `HTTPError`; a refused connection (URLError) escaped as a traceback | `1798930` | RED: subprocess run of the real entry point against a bound-but-unlistening port → `urllib.error.URLError: <urlopen error [Errno 111] Connection refused>` traceback, exit 1 with no `create_release: API failed` prefix. GREEN after catching `urllib.error.URLError` alongside `HTTPError` (consistent `create_release: API failed` path, exit 1) — 5 passed in tests/test_create_release.py |
+| 9 | pointless f-string `print(f"AUTO_BUMPED=true")` in auto_bump_pr.py → plain string | `fa29a70` | no test (print literal); full suite run instead |
+| 10 | Completion Record inaccuracy: claimed maintenance.yml was pre-existingly unparseable — verified at base 2840f71 that ONLY auto-version.yml failed pyyaml (ScannerError, old line 50); maintenance.yml parsed cleanly. Record corrected (YAML bullet + Deviation 2) | `73af931` | verified by parsing both base-commit blobs with pyyaml |
+| 11 | maintenance.yml:62,102 bare `except: pass` (pre-existing, moved verbatim) → `except Exception as exc:` with a warning printed to stderr (log-and-continue) | `1a05634` | embedded heredoc Python compiles; both workflow files parse with pyyaml |
+
+### Discharge verification battery (all in `.venv`)
+
+- Full suite: **730 passed + 84 subtests** (676 baseline + 54 discharge tests: 53 from findings 1–7 + 1 new finding-8 test) — green.
+- Pyright ratchet: PASS — errors 117 (baseline 120), warnings 407 (baseline 465); not increased.
+- `ruff format --check` + `ruff check --select E9,F63,F7,F82`: clean on the Python files touched this discharge (tests/test_create_release.py, scripts/create_release.py, scripts/auto_bump_pr.py). The .md/.yml touched files are non-Python; their validity is covered by the pyyaml check below and markdown needs none.
+- pyyaml: `.gitea/workflows/auto-version.yml` and `maintenance.yml` both parse; all three embedded `python3 -` heredocs in maintenance.yml compile.
+- `henxels_checks.py`: exit 0.
+- Branch `release-safety-net` at `1a05634`, 11 commits ahead of base, not pushed (controller handles push/PR).
+

scripts/auto_bump_pr.py

@@ -0,0 +1,291 @@
+#!/usr/bin/env python3
+"""Protected-branch-safe bump-PR machinery for the Rune release pipeline (#192).
+
+Extracted verbatim from the 'Auto-bump' step heredoc in
+``.gitea/workflows/auto-version.yml`` so both the pull_request:closed path and
+the push-triggered safety net drive the same code: branch create -> contents-API
+edits of the five version surfaces -> PR -> wait for the protected check ->
+API merge -> confirm.
+
+Env-driven: GITEA_TOKEN, RELEASE_SERVER_URL, RELEASE_REPO, RELEASE_VERSION,
+NEW_VERSION, PR_TITLE, PR_BODY. When running inside Actions it writes
+RELEASE_VERSION/AUTO_BUMPED back to GITHUB_ENV for the release step.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import os
+import sys
+import time
+import urllib.error
+import urllib.request
+from typing import Any
+
+BUMP_BRANCH_PREFIX = "chore/auto-bump-"
+
+# The version surface is NOT optional (spec-gate finding 4, #192): a GET
+# failure here would produce a bump PR that silently bumps nothing. A GET
+# failure on any other surface stays optional (pattern absent — skip).
+VERSION_SURFACE = "src/rune/__init__.py"
+
+
+def is_surface_fatal(path: str) -> bool:
+    return path == VERSION_SURFACE
+
+
+def apply_surface_edits(
+    branch: str,
+    version: str,
+    new_version: str,
+    api: Any,
+    on_commit: Any = None,
+) -> None:
+    """Commit the version-surface edits onto ``branch`` via the contents API.
+
+    ``api`` is the request callable; ``on_commit`` (optional) is called with
+    each committed path — test seam only.
+    """
+    surfaces = {
+        "src/rune/__init__.py": [
+            (f'__version__ = "{version}"', f'__version__ = "{new_version}"'),
+        ],
+        "src/rune/rune_tui.py": [
+            (f"Version: v{version}", f"Version: v{new_version}"),
+            (f"Rune v{version} |", f"Rune v{new_version} |"),
+        ],
+        "src/rune/canary_gitea_automation.py": [
+            (f'"{version}"', f'"{new_version}"'),
+        ],
+        "README.md": [
+            (f"v{version}", f"v{new_version}"),
+        ],
+    }
+    for path, pairs in surfaces.items():
+        status, cur = api(f"/contents/{path}?ref={branch}")
+        if status != 200:
+            if is_surface_fatal(path):
+                print(f"  FATAL: version surface {path} unreadable ({status}) — refusing to file a no-op bump PR")
+                sys.exit(1)
+            # genuinely optional surface (e.g. pattern absent) — skip gracefully
+            print(f"  skip {path} ({status})")
+            continue
+        text = base64.b64decode(cur["content"]).decode("utf-8")
+        changed = text
+        for old, newv in pairs:
+            changed = changed.replace(old, newv)
+        if changed == text:
+            print(f"  no-op {path}")
+            continue
+        status, r = api(
+            f"/contents/{path}",
+            "PUT",
+            {
+                "message": f"chore(release): auto-bump version to {new_version} (post-merge automation)",
+                "content": base64.b64encode(changed.encode()).decode(),
+                "branch": branch,
+                "sha": cur["sha"],
+            },
+        )
+        if status not in (200, 201):
+            print(f"  commit {path} failed ({status})")
+            sys.exit(1)
+        if on_commit is not None:
+            on_commit(path)
+
+
+def run_bump() -> None:
+    TOKEN = os.environ["GITEA_TOKEN"]
+    BASE = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
+    VERSION = os.environ["RELEASE_VERSION"]
+    NEW = os.environ["NEW_VERSION"]
+    PR_TITLE = os.environ.get("PR_TITLE", "")
+    PR_BODY = os.environ.get("PR_BODY", "")
+    BRANCH = f"chore/auto-bump-{NEW}"
+
+    def api(
+        path: str,
+        method: str = "GET",
+        data: dict[str, Any] | None = None,
+        ok: tuple[int, ...] = (200,),
+    ) -> tuple[int, dict[str, Any]]:
+        req = urllib.request.Request(
+            BASE + path,
+            data=json.dumps(data).encode() if data is not None else None,
+            headers={"Authorization": "token " + TOKEN, "Content-Type": "application/json"},
+            method=method,
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=30) as r:
+                body = r.read()
+                return r.status, (json.loads(body) if body else {})
+        except urllib.error.HTTPError as e:
+            body = e.read().decode("utf-8", "replace")
+            print(f"  API {method} {path} -> {e.code}: {body[:200]}")
+            return e.code, {"_body": body}
+
+    # 1. bump branch from master
+    status, _ = api("/branches", "POST", {"new_branch_name": BRANCH, "old_branch_name": "master"})
+    if status not in (201, 409):
+        print(f"branch create failed ({status})")
+        sys.exit(1)
+
+    # 2. commit the version-surface edits via the contents API
+    apply_surface_edits(BRANCH, VERSION, NEW, api)
+
+    # 3. bump PR
+    status, pr = api(
+        "/pulls",
+        "POST",
+        {
+            "title": f"chore(release): auto-bump version to {NEW} (post-merge automation)",
+            "body": f"Automatic version bump after merge of: {PR_TITLE}\n\n{PR_BODY[:1500]}",
+            "head": BRANCH,
+            "base": "master",
+        },
+    )
+    if status not in (200, 201):
+        print(f"PR create failed ({status})")
+        sys.exit(1)
+    pr_number = pr["number"]
+    head_sha = pr["head"]["sha"]
+    print(f"  bump PR #{pr_number} head {head_sha[:12]}")
+
+    # 4. wait for the protected check on the bump PR (same gate as everything else)
+    deadline = time.time() + 900  # 15 min ceiling
+    state = None
+    while time.time() < deadline:
+        st, cs = api(f"/commits/{head_sha}/status")
+        state = cs.get("state")
+        if state in ("success", "failure"):
+            break
+        time.sleep(15)
+    if state != "success":
+        print(f"bump PR CI did not go green (state={state})")
+        sys.exit(1)
+
+    # 5. merge via API (respects branch protection; user_can_merge)
+    status, _ = api(f"/pulls/{pr_number}/merge", "POST", {"Do": "merge"})
+    if status not in (200, 405):
+        print(f"merge failed ({status})")
+        sys.exit(1)
+
+    # 6. confirm merged, then export the version for the release step
+    st, pr2 = api(f"/pulls/{pr_number}")
+    if not pr2.get("merged"):
+        print("merge did not take")
+        sys.exit(1)
+    print("AUTO_BUMPED=true")
+    gh_env = os.environ.get("GITHUB_ENV")
+    if gh_env:
+        with open(gh_env, "a") as env:
+            env.write(f"RELEASE_VERSION={NEW}\n")
+            env.write("AUTO_BUMPED=true\n")
+
+
+def list_branch_names(base: str, token: str) -> list[str]:
+    req = urllib.request.Request(
+        base + "/branches?limit=50",
+        headers={"Authorization": "token " + token, "Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return [b.get("name", "") for b in json.loads(r.read())]
+
+
+def bump_branch_in_flight(names: list[str], new_version: str | None = None) -> bool:
+    """True when the bump branch for THIS merge's in-flight version exists —
+    evidence the PR-closed path is mid-bump.
+
+    Spec-gate finding 7a (#192): the guard used to match ANY chore/auto-bump-*
+    branch, so one stale historical branch (they are never cleaned up) stood
+    the net down permanently. It now matches exactly chore/auto-bump-{NEW};
+    pass new_version=None to restore the legacy prefix-wide behavior.
+    """
+    if new_version is None:
+        return any(name.startswith(BUMP_BRANCH_PREFIX) for name in names)
+    return f"{BUMP_BRANCH_PREFIX}{new_version}" in names
+
+
+def wait_out_closed_path(
+    base: str, token: str, grace_seconds: int, poll_seconds: int = 15, new_version: str | None = None
+) -> bool:
+    """Wait for evidence that the PR-closed path is handling this merge.
+
+    Both events fire from the same merge, so a living closed path creates its
+    chore/auto-bump-{new_version} branch within seconds of job start. Returns
+    True (safety net stands down) as soon as it appears, or False (closed
+    event genuinely absent — the #190/#192 defect) once the grace window
+    elapses. Best-effort attribution only (see docs/release-flow.md): a stale
+    branch of the same version, or a >grace delayed closed path, can still
+    misattribute — double-bump risk is bounded by release tag idempotency.
+    """
+    deadline = time.time() + grace_seconds
+    while True:
+        if bump_branch_in_flight(list_branch_names(base, token), new_version):
+            return True
+        if time.time() >= deadline:
+            return False
+        time.sleep(poll_seconds)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Protected-branch-safe bump-PR machinery (#192)")
+    parser.add_argument("--dry-run", action="store_true", help="print the decision without executing any API call")
+    parser.add_argument(
+        "--safety-net",
+        action="store_true",
+        help="push path: first wait out a possibly-alive PR-closed path before acting",
+    )
+    parser.add_argument(
+        "--attribution-grace-seconds",
+        type=int,
+        default=90,
+        help="how long --safety-net waits for the PR-closed path's bump branch to appear",
+    )
+    args = parser.parse_args(argv)
+
+    if args.dry_run:
+        version = os.environ.get("RELEASE_VERSION", "")
+        new = os.environ.get("NEW_VERSION", "")
+        print(
+            json.dumps(
+                {
+                    "dry_run": True,
+                    "release_version": version,
+                    "new_version": new,
+                    "branch": f"{BUMP_BRANCH_PREFIX}{new}",
+                    "pr_title": f"chore(release): auto-bump version to {new} (post-merge automation)",
+                }
+            )
+        )
+        return 0
+
+    if args.safety_net:
+        base = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
+        token = os.environ["GITEA_TOKEN"]
+        # Attribute against exactly this merge's would-be bump branch: a stale
+        # chore/auto-bump-* branch from an older version must not stand the
+        # net down (spec-gate finding 7a, #192).
+        new_version = os.environ.get("NEW_VERSION") or None
+        try:
+            closed_path_alive = wait_out_closed_path(
+                base, token, args.attribution_grace_seconds, new_version=new_version
+            )
+        except (urllib.error.HTTPError, urllib.error.URLError, OSError) as exc:
+            print(f"attribution check failed ({exc}) — refusing to act (conservative)")
+            return 1
+        if closed_path_alive:
+            print("PR-closed path is mid-bump (bump branch for this version exists) — safety net stands down")
+            return 0
+        print(
+            f"no bump branch after {args.attribution_grace_seconds}s — closed event genuinely absent; safety net acts"
+        )
+
+    run_bump()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/create_release.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+"""Idempotent release creation for the Rune release pipeline (#192).
+
+Extracted from the 'Create release from merged pull request' step in
+``.gitea/workflows/auto-version.yml`` so both the pull_request:closed path and
+the push-triggered safety net cut releases through the same code. Keeps the
+tag-exists idempotency check (retry/replay safety).
+
+Env-driven: GITEA_TOKEN, RELEASE_VERSION, PR_TITLE/PR_BODY (optional — the
+push safety net has no PR context and falls back to a generic body),
+RELEASE_SERVER_URL + RELEASE_REPO. The release API URLs are derived in-script
+from those base coordinates (spec-gate finding 1, #192: the workflow never
+passed LIST_API_URL/RELEASE_API_URL — KeyError at runtime on both paths), so
+the script is self-sufficient; explicit LIST_API_URL/RELEASE_API_URL env
+values still override the derived defaults for direct/manual use.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from typing import Any
+
+# Env-contract (#192 spec-gate finding 1): derived in-script, never required
+# from the caller. Guarded by tests/test_workflow_script_contract.py.
+DERIVED_ENV: dict[str, str] = {
+    "LIST_API_URL": "RELEASE_SERVER_URL + RELEASE_REPO",
+    "RELEASE_API_URL": "RELEASE_SERVER_URL + RELEASE_REPO",
+}
+
+
+def derive_api_urls(server_url: str, repo: str) -> tuple[str, str]:
+    """Derive (list_api_url, release_api_url) from the workflow base coordinates."""
+    base = server_url.rstrip("/") + "/api/v1/repos/" + repo
+    return base + "/releases?limit=50", base + "/releases"
+
+
+def build_release_payload(version: str, pr_title: str, pr_body: str) -> dict[str, Any]:
+    if pr_title:
+        body = f"## {pr_title}\n\n{pr_body}"
+    else:
+        body = "## Release (push-triggered safety net)\n\nCut automatically by the release safety net (#192) after a merge whose closed-event never arrived."
+    return {
+        "tag_name": f"v{version}",
+        "target_commitish": "master",
+        "name": f"Rune v{version}",
+        "body": body,
+        "draft": False,
+        "prerelease": False,
+    }
+
+
+def release_exists(releases: list[dict[str, Any]], version: str) -> bool:
+    """True when a release for ``v<version>`` already exists (idempotency check)."""
+    return any(r.get("tag_name") == f"v{version}" for r in releases)
+
+
+def list_releases(list_api_url: str, token: str) -> list[dict[str, Any]]:
+    req = urllib.request.Request(list_api_url, headers={"Authorization": "token " + token})
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return json.loads(r.read())
+
+
+def create_release(release_api_url: str, token: str, payload: dict[str, Any]) -> None:
+    req = urllib.request.Request(
+        release_api_url,
+        data=json.dumps(payload).encode(),
+        headers={"Authorization": "token " + token, "Content-Type": "application/json"},
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=60) as r:
+        print(f"release created ({r.status})")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Idempotent Gitea release creation (#192)")
+    parser.add_argument("--dry-run", action="store_true", help="print the payload without executing any API call")
+    args = parser.parse_args(argv)
+
+    version = os.environ["RELEASE_VERSION"]
+    payload = build_release_payload(version, os.environ.get("PR_TITLE", ""), os.environ.get("PR_BODY", ""))
+    if args.dry_run:
+        print(json.dumps(payload))
+        return 0
+
+    token = os.environ["GITEA_TOKEN"]
+
+    # Derived URLs (explicit env still overrides — for direct/manual use).
+    server_url = os.environ["RELEASE_SERVER_URL"]
+    repo = os.environ["RELEASE_REPO"]
+    list_api_url = os.environ.get("LIST_API_URL") or derive_api_urls(server_url, repo)[0]
+    release_api_url = os.environ.get("RELEASE_API_URL") or derive_api_urls(server_url, repo)[1]
+
+    # Skip idempotently if a release for this tag already exists (retry/replay safety).
+    releases = list_releases(list_api_url, token)
+    if release_exists(releases, version):
+        print(f"Release v{version} already exists — skipping creation")
+        return 0
+
+    create_release(release_api_url, token, payload)
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except urllib.error.HTTPError as exc:
+        print(
+            f"create_release: API failed ({exc.code}): {exc.read().decode('utf-8', 'replace')[:300]}", file=sys.stderr
+        )
+        sys.exit(1)
+    except urllib.error.URLError as exc:
+        # Connection refused / DNS failure etc. (HTTPError is a URLError
+        # subclass, so this only fires for transport-level failures).
+        print(f"create_release: API failed (transport error): {exc.reason}", file=sys.stderr)
+        sys.exit(1)

scripts/file_drift_issue.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+"""Maintenance drift reconciliation — file or update a deduped drift issue (#192).
+
+Called by ``.gitea/workflows/maintenance.yml`` when
+``scripts/version_convergence.py`` reports drift. Advisory only: the issue
+tells humans/agent to merge the pending bump PR (or land one); this script
+never auto-releases without the PR flow (plan Design Decision 3).
+
+Env-driven: GITEA_TOKEN, RELEASE_SERVER_URL, RELEASE_REPO, MASTER_VERSION,
+LATEST_TAG, CONVERGENCE_JSON (the checker's stdout).
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from typing import Any
+
+DRIFT_ISSUE_LABEL = "release-drift"
+DRIFT_ISSUE_LABEL_COLOR = "ff0000"
+
+
+def resolve_label_ids(api: Any, label_names: list[str]) -> list[int]:
+    """Resolve label NAMES to the integer IDs Gitea's POST /issues requires
+    (spec-gate finding 6, #192; repo convention per shared/custodian.py).
+
+    A label missing from the repo is created first; if that also fails the
+    label is skipped — the issue still files, just unlabeled.
+    """
+    status, labels = api("/labels?limit=50")
+    by_name = {l.get("name"): l.get("id") for l in (labels if isinstance(labels, list) else [])}
+    ids: list[int] = []
+    for name in label_names:
+        label_id = by_name.get(name)
+        if label_id is None:
+            status, created = api("/labels", "POST", {"name": name, "color": DRIFT_ISSUE_LABEL_COLOR})
+            label_id = created.get("id") if status in (200, 201) and isinstance(created, dict) else None
+            if label_id is None:
+                print(f"label {name!r} unavailable (resolve/create failed, status {status}) — filing without it")
+                continue
+        ids.append(label_id)
+    return ids
+
+
+def drift_title(master_version: str, latest_tag: str) -> str:
+    master = f"v{master_version}" if master_version else "(unknown)"
+    return f"Release drift: master {master} vs tag {latest_tag}"
+
+
+def drift_body(decision: dict[str, Any], master_version: str, latest_tag: str) -> str:
+    subjects = "\n".join(f"- `{s}`" for s in decision.get("sample", []))
+    anomaly = decision.get("anomaly")
+    anomaly_line = ""
+    if anomaly == "version_behind_tag":
+        # Spec-gate finding 3 (#192): master's version is BELOW the latest tag.
+        anomaly_line = (
+            "- anomaly: **version_behind_tag** — master's `__version__` is *below* the latest tag\n"
+            "  (a bump merged without riding a release, or a tag moved backwards).\n"
+            "  The safety net deliberately does not act on this state; reconcile manually.\n"
+        )
+    elif anomaly:
+        anomaly_line = f"- anomaly: **{anomaly}**\n"
+    return (
+        "The weekly maintenance reconciliation detected that master and the latest release tag have diverged.\n\n"
+        f"- master `__version__`: **{master_version or '(unknown)'}**\n"
+        f"- latest tag: **{latest_tag}**\n"
+        f"- commits since tag: **{decision.get('commits_since_tag', 0)}**\n"
+        f"- warranted bump (from commit subjects): **{decision.get('warranted_bump') or 'none'}**\n"
+        f"{anomaly_line}\n"
+        "Commit subjects since the tag:\n\n"
+        f"{subjects}\n\n"
+        "This issue is advisory — the reconciliation never auto-releases without the PR flow.\n"
+        "Next step: land a version-bump PR (or let the auto-version safety net file one),\n"
+        "then cut the release. See docs/release-flow.md for the two-path release flow (#192)."
+    )
+
+
+def find_open_drift_issue(issues: list[dict[str, Any]], title: str) -> dict[str, Any] | None:
+    """First open issue whose title matches exactly (dedupe; PRs are not issues)."""
+    for issue in issues:
+        if issue.get("title") == title and "pull_request" not in issue:
+            return issue
+    return None
+
+
+def main(argv: list[str] | None = None, api: Any = None) -> int:
+    parser = argparse.ArgumentParser(description="File/update the deduped release-drift issue (#192)")
+    parser.add_argument("--dry-run", action="store_true", help="print title/body without any API call")
+    parser.add_argument(
+        "--create-comment", action="store_true", help="comment on the existing issue instead of a no-op update"
+    )
+    args = parser.parse_args(argv)
+
+    decision = json.loads(os.environ["CONVERGENCE_JSON"])
+    master_version = os.environ.get("MASTER_VERSION", "")
+    latest_tag = os.environ.get("LATEST_TAG", "")
+    title = drift_title(master_version, latest_tag)
+    body = drift_body(decision, master_version, latest_tag)
+
+    if args.dry_run:
+        print(f"title: {title}")
+        print("---")
+        print(body)
+        return 0
+
+    token = os.environ["GITEA_TOKEN"]
+    base = os.environ["RELEASE_SERVER_URL"] + "/api/v1/repos/" + os.environ["RELEASE_REPO"]
+
+    if api is None:
+
+        def api(  # noqa: F811 — default client when not injected (tests)
+            path: str, method: str = "GET", data: dict[str, Any] | None = None
+        ) -> tuple[int, dict[str, Any] | list[dict[str, Any]]]:
+            req = urllib.request.Request(
+                base + path,
+                data=json.dumps(data).encode() if data is not None else None,
+                headers={"Authorization": "token " + token, "Content-Type": "application/json"},
+                method=method,
+            )
+            with urllib.request.urlopen(req, timeout=30) as r:
+                payload = r.read()
+                return r.status, (json.loads(payload) if payload else {})
+
+    status, issues = api(f"/issues?state=open&type=issues&limit=50&q={urllib.parse.quote(title)}")
+    if status != 200:
+        print(f"issue search failed ({status})")
+        return 1
+    existing = find_open_drift_issue(issues if isinstance(issues, list) else [], title)
+    if existing is not None:
+        print(f"drift issue #{existing['number']} already open — {'commenting' if args.create_comment else 'no-op'}")
+        if args.create_comment:
+            st, _ = api(f"/issues/{existing['number']}/comments", "POST", {"body": body})
+            if st not in (200, 201):
+                print(f"comment failed ({st})")
+                return 1
+        return 0
+
+    label_ids = resolve_label_ids(api, [DRIFT_ISSUE_LABEL])
+    st, created = api(
+        "/issues",
+        "POST",
+        {
+            "title": title,
+            "body": body,
+            "labels": label_ids,
+        },
+    )
+    if st not in (200, 201):
+        print(f"issue create failed ({st})")
+        return 1
+    number = created.get("number") if isinstance(created, dict) else None
+    print(f"filed drift issue #{number}: {title}")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except (KeyError, json.JSONDecodeError, urllib.error.HTTPError, urllib.error.URLError) as exc:
+        print(f"file_drift_issue: failed ({type(exc).__name__}: {exc})", file=sys.stderr)
+        sys.exit(1)

scripts/version_convergence.py

@@ -0,0 +1,204 @@
+#!/usr/bin/env python3
+"""Version convergence checker — the decision logic for the release safety net (#192).
+
+Reads the repo's git log (via subprocess git) and reports whether master is in
+a released state. Used by the push-triggered job in
+``.gitea/workflows/auto-version.yml`` and by the maintenance drift
+reconciliation job in ``.gitea/workflows/maintenance.yml``.
+
+Decision (see ``assemble_release_decision``):
+
+* ``warranted_bump`` — the highest bump kind warranted by commit subjects since
+  the latest tag: ``feat!``/``BREAKING CHANGE``/``feat`` -> minor (0.x
+  convention, matching the workflow's PR-title rule), ``fix``/``perf`` ->
+  patch, anything else (docs/chore/test/refactor/...) -> null.
+* ``converged`` — true when no release action is warranted: either master's
+  ``__version__`` matches the latest tag and no release-worthy commit sits
+  unreleased since it, or master's version is not ahead of the tag in any
+  actionable way. ``converged: false`` means drift: an unreleased bump or a
+  missing release.
+
+Output: the plan's JSON contract on stdout (exit 0 in both converged and
+drift states; exit 1 only on operational failure):
+
+    {"converged": bool, "warranted_bump": "minor"|"patch"|null,
+     "anomaly": "version_behind_tag"|null, "commits_since_tag": N,
+     "sample": [...]}
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+SAMPLE_LIMIT = 10
+
+_VERSION_RE = re.compile(r'__version__ = "([0-9]+\.[0-9]+\.[0-9]+)"')
+_FEAT_RE = re.compile(r"^\s*feat(\([^)]*\))?:")
+_WARRANTY_RE = re.compile(r"^\s*(fix|perf)(\([^)]*\))?:")
+# Gitea merge-commit subject: the conventional title is quoted inside
+# ("Merge pull request 'feat: x' (#N)"). Spec-gate finding 2 (#192): classify
+# the inner title, not the "Merge ..." wrapper (which anchors as 'other').
+_MERGE_COMMIT_RE = re.compile(r"^Merge pull request '(.+)' \(#\d+\)$")
+
+
+def _git(args: list[str], repo: Path | None = None) -> str:
+    proc = subprocess.run(
+        ["git", *args],
+        cwd=str(repo) if repo else None,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return proc.stdout.strip()
+
+
+def read_version(repo: Path | None = None) -> str | None:
+    """Master's ``__version__`` from src/rune/__init__.py, or None if unreadable."""
+    root = repo if repo else Path.cwd()
+    init_py = root / "src" / "rune" / "__init__.py"
+    try:
+        match = _VERSION_RE.search(init_py.read_text(encoding="utf-8"))
+    except OSError:
+        return None
+    return match.group(1) if match else None
+
+
+def latest_tag(repo: Path | None = None) -> str | None:
+    """Highest version tag (e.g. ``v0.54.0``), or None when the repo has no tags."""
+    tags = _git(["tag", "--sort=-v:refname"], repo).splitlines()
+    return tags[0] if tags else None
+
+
+def subjects_since_tag(tag: str, repo: Path | None = None) -> list[str]:
+    """Commit subjects (``%s``) between ``tag`` and HEAD."""
+    out = _git(["log", f"{tag}..HEAD", "--pretty=%s"], repo)
+    return out.splitlines() if out else []
+
+
+def classify_subject(subject: str) -> str:
+    """Classify one commit subject: ``breaking`` | ``feat`` | ``fix`` | ``other``.
+
+    Mirrors the PR-title semver rule in auto-version.yml: ``feat!``/``BREAKING
+    CHANGE`` are breaking; ``feat`` is a feature; ``fix``/``perf`` warrant a
+    patch; everything else releases nothing.
+
+    Merge commits (``Merge pull request '<title>' (#N)``) are classified by
+    their inner PR title so a no-bump feat PR merged via merge-commit stays
+    visible to the net and to maintenance (spec-gate finding 2, #192).
+    """
+    merge_match = _MERGE_COMMIT_RE.match(subject)
+    if merge_match:
+        subject = merge_match.group(1)
+    if "BREAKING CHANGE" in subject or re.match(r"^\s*\w+(\([^)]*\))?!:", subject):
+        return "breaking"
+    if _FEAT_RE.match(subject):
+        return "feat"
+    if _WARRANTY_RE.match(subject):
+        return "fix"
+    return "other"
+
+
+def highest_warranted_bump(subjects: list[str]) -> str | None:
+    """Highest bump kind warranted by commit subjects: minor > patch > None."""
+    kinds = {classify_subject(s) for s in subjects}
+    if "breaking" in kinds or "feat" in kinds:
+        return "minor"
+    if "fix" in kinds:
+        return "patch"
+    return None
+
+
+def next_version(version: str, kind: str) -> str:
+    """Apply a bump kind to a version string (0.x convention: breaking -> minor)."""
+    major, minor, patch = version.split(".")
+    if kind == "major":
+        return f"{major}.{int(minor) + 1}.0" if int(major) == 0 else f"{int(major) + 1}.0.0"
+    if kind == "minor":
+        return f"{major}.{int(minor) + 1}.0"
+    if kind == "patch":
+        return f"{major}.{minor}.{int(patch) + 1}"
+    raise ValueError(f"unknown bump kind: {kind!r}")
+
+
+def version_greater(a: str, b: str) -> bool:
+    """Numeric semver comparison: is ``a`` strictly greater than ``b``?"""
+    return tuple(int(part) for part in a.split(".")) > tuple(int(part) for part in b.split("."))
+
+
+def assemble_release_decision(
+    version: str | None,
+    latest_tag: str | None,
+    subjects: list[str],
+    commit_count: int,
+) -> dict[str, Any]:
+    """Assemble the plan's JSON decision from raw inputs.
+
+    ``converged`` is false (drift) when either:
+      * master's ``__version__`` equals the latest tag but release-worthy
+        commits sit unreleased since it (the bump never rode the merge), or
+      * master's ``__version__`` is strictly ahead of the latest tag (a bump
+        rode the merge but no release/tag was cut for it), or
+      * master's ``__version__`` is strictly BEHIND the latest tag (spec-gate
+        finding 3, #192: a bump merged without riding a release, or a tag
+        moved backwards). That state is drift with ``anomaly:
+        "version_behind_tag"`` and ``warranted_bump`` forced to None — the
+        net must not act (bumping master cannot converge it with a tag that
+        is already ahead); maintenance files/flags it instead.
+    """
+    warranted = highest_warranted_bump(subjects)
+    tag_version = latest_tag[1:] if latest_tag and latest_tag.startswith("v") else latest_tag
+
+    anomaly: str | None = None
+    converged = True
+    if version is None and tag_version is not None:
+        converged = False  # master unreadable while a release history exists
+    elif version is not None and tag_version is None:
+        converged = True  # no tags: first-release territory, the net never acts
+    elif version is not None and tag_version is not None:
+        if version == tag_version:
+            converged = warranted is None
+        elif version_greater(version, tag_version):
+            converged = False  # bump rode the merge but no release was cut
+        else:
+            converged = False  # version behind tag: drift, never silent
+            warranted = None
+            anomaly = "version_behind_tag"
+
+    return {
+        "converged": converged,
+        "warranted_bump": warranted,
+        "anomaly": anomaly,
+        "commits_since_tag": commit_count,
+        "sample": subjects[:SAMPLE_LIMIT],
+    }
+
+
+def decision_in_repo(repo: Path | None = None) -> dict[str, Any]:
+    """Read the repo's git state and return the release decision."""
+    version = read_version(repo)
+    tag = latest_tag(repo)
+    subjects = subjects_since_tag(tag, repo) if tag else []
+    return assemble_release_decision(version, tag, subjects, len(subjects))
+
+
+def main(argv: list[str] | None = None) -> dict[str, Any]:
+    parser = argparse.ArgumentParser(description="Release convergence decision (JSON on stdout)")
+    parser.add_argument("--repo", type=Path, default=None, help="repo root (default: cwd)")
+    args = parser.parse_args(argv)
+    decision = decision_in_repo(args.repo)
+    print(json.dumps(decision))
+    return decision
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except subprocess.CalledProcessError as exc:
+        print(f"version_convergence: git failed ({exc.returncode}): {exc.stderr}", file=sys.stderr)
+        sys.exit(1)

tests/test_auto_bump_pr.py

@@ -0,0 +1,136 @@
+"""Tests for scripts/auto_bump_pr.py — the protected-branch-safe bump-PR machinery (#192).
+
+Pure tests cover the attribution gate (bump-branch detection). The dry-run mode
+is exercised end-to-end with monkeypatched env (no API calls, no token needed).
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_module(name: str) -> Any:
+    spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py")
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+abp = _load_module("auto_bump_pr")
+
+
+class TestBumpBranchInFlight:
+    @pytest.mark.parametrize(
+        ("names", "expected"),
+        [
+            (["master", "chore/auto-bump-0.55.0"], True),
+            (["chore/auto-bump-0.55.0"], True),
+            (["master", "feat/x", "chore/other"], False),
+            ([], False),
+        ],
+    )
+    def test_detection(self, names: list[str], expected: bool) -> None:
+        assert abp.bump_branch_in_flight(names) is expected
+
+    def test_anchored_to_in_flight_version(self) -> None:
+        # Spec-gate finding 7a (#192): matching ANY chore/auto-bump-* branch let
+        # one stale historical branch stand the net down forever. The guard must
+        # match exactly chore/auto-bump-{NEW} — the in-flight version.
+        assert abp.bump_branch_in_flight(["master", "chore/auto-bump-0.54.0"], "0.56.0") is False
+        assert abp.bump_branch_in_flight(["chore/auto-bump-0.55.0"], "0.56.0") is False
+        assert abp.bump_branch_in_flight(["master", "chore/auto-bump-0.56.0"], "0.56.0") is True
+
+    def test_legacy_prefix_widens_to_any_suffix(self) -> None:
+        # Backward-compat escape hatch: when no in-flight version is given,
+        # the old prefix-wide behavior is preserved (documented as best-effort).
+        assert abp.bump_branch_in_flight(["master", "chore/auto-bump-anything"], None) is True
+
+
+class TestDryRun:
+    def test_prints_decision_without_api_calls(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        monkeypatch.setenv("RELEASE_VERSION", "0.55.0")
+        monkeypatch.setenv("NEW_VERSION", "0.56.0")
+        monkeypatch.delenv("GITEA_TOKEN", raising=False)
+        rc = abp.main(["--dry-run"])
+        assert rc == 0
+        decision = json.loads(capsys.readouterr().out)
+        assert decision["dry_run"] is True
+        assert decision["release_version"] == "0.55.0"
+        assert decision["new_version"] == "0.56.0"
+        assert decision["branch"] == "chore/auto-bump-0.56.0"
+        assert "0.56.0" in decision["pr_title"]
+
+    def test_missing_env_defaults_empty(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        for var in ("RELEASE_VERSION", "NEW_VERSION", "GITEA_TOKEN"):
+            monkeypatch.delenv(var, raising=False)
+        rc = abp.main(["--dry-run"])
+        assert rc == 0
+        decision = json.loads(capsys.readouterr().out)
+        assert decision["release_version"] == ""
+        assert decision["branch"] == "chore/auto-bump-"
+
+
+VERSION_SURFACE = "src/rune/__init__.py"
+OTHER_SURFACE = "src/rune/rune_tui.py"
+
+
+class TestSurfaceFailurePolicy:
+    """Spec-gate finding 4 (#192): the version surface is NOT optional.
+
+    A GET failure on src/rune/__init__.py used to be skipped like an optional
+    surface, producing a bump PR that silently bumps nothing. The failure must
+    be fatal; genuinely optional surfaces (pattern absent) still skip.
+    """
+
+    def test_init_py_get_failure_is_fatal(self) -> None:
+        assert abp.is_surface_fatal(VERSION_SURFACE) is True
+
+    def test_other_surfaces_stay_optional(self) -> None:
+        for surface in ("src/rune/rune_tui.py", "src/rune/canary_gitea_automation.py", "README.md"):
+            assert abp.is_surface_fatal(surface) is False, surface
+
+    def test_bump_aborts_when_init_py_unreadable(self) -> None:
+        calls: list[tuple[str, str]] = []
+
+        def fake_api(path: str, method: str = "GET", data: dict[str, Any] | None = None, ok: tuple[int, ...] = (200,)):
+            calls.append((method, path.split("?")[0]))
+            if path.startswith("/contents/src/rune/__init__.py"):
+                return 404, {"_body": "not found"}
+            if path.startswith("/contents/"):
+                return 200, {"content": "", "sha": "deadbeef"}
+            return 200, {}
+
+        with pytest.raises(SystemExit) as excinfo:
+            abp.apply_surface_edits("chore/auto-bump-0.56.0", "0.55.0", "0.56.0", fake_api)
+        assert excinfo.value.code == 1
+        # abort must happen before any PUT was attempted
+        assert all(m != "PUT" for m, _ in calls)
+
+    def test_optional_surface_failure_still_skips(self) -> None:
+        import base64
+
+        committed: list[str] = []
+        init_content = base64.b64encode(b'__version__ = "0.55.0"\n').decode()
+
+        def fake_api(path: str, method: str = "GET", data: dict[str, Any] | None = None, ok: tuple[int, ...] = (200,)):
+            if path.startswith("/contents/src/rune/__init__.py"):
+                return 200, {"content": init_content, "sha": "deadbeef"}
+            if path.startswith("/contents/"):
+                return 404, {"_body": "not found"}
+            return 200, {}
+
+        abp.apply_surface_edits("chore/auto-bump-0.56.0", "0.55.0", "0.56.0", fake_api, on_commit=committed.append)
+        # only the version surface was committed; optional ones skipped
+        assert committed == [VERSION_SURFACE]

tests/test_create_release.py

@@ -0,0 +1,107 @@
+"""Tests for scripts/create_release.py — idempotent release creation (#192).
+
+Pure tests cover payload construction and the tag-exists check; the dry-run
+mode is exercised end-to-end with monkeypatched env (no API calls). The
+script-level error path (finding 8, #192) runs the real entry point as a
+subprocess against a refused port.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+import socket
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_module() -> Any:
+    spec = importlib.util.spec_from_file_location("create_release", ROOT / "scripts" / "create_release.py")
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+cr = _load_module()
+
+
+class TestBuildReleasePayload:
+    def test_with_pr_context(self) -> None:
+        payload = cr.build_release_payload("0.56.0", "feat: widget", "Some body")
+        assert payload == {
+            "tag_name": "v0.56.0",
+            "target_commitish": "master",
+            "name": "Rune v0.56.0",
+            "body": "## feat: widget\n\nSome body",
+            "draft": False,
+            "prerelease": False,
+        }
+
+    def test_without_pr_context_falls_back_to_safety_net_body(self) -> None:
+        payload = cr.build_release_payload("0.56.0", "", "")
+        assert payload["tag_name"] == "v0.56.0"
+        assert "safety net" in payload["body"]
+        assert payload["draft"] is False
+        assert payload["prerelease"] is False
+
+
+class TestReleaseExists:
+    def test_matches_tag_name(self) -> None:
+        releases = [{"tag_name": "v0.54.0"}, {"tag_name": "v0.55.0"}]
+        assert cr.release_exists(releases, "0.55.0") is True
+        assert cr.release_exists(releases, "0.56.0") is False
+        assert cr.release_exists([], "0.55.0") is False
+
+
+class TestURLErrorCaughtAtScriptLevel:
+    """Finding 8 (#192): __main__ catches only HTTPError, so a refused
+    connection (URLError) escapes and crashes with a traceback instead of
+    the consistent error path. Runs the real entry point as a subprocess."""
+
+    def test_refused_connection_exits_1_with_message_not_traceback(self, tmp_path: Path) -> None:
+        with socket.socket() as s:
+            s.bind(("127.0.0.1", 0))
+            refused_port = s.getsockname()[1]  # bound but never listening → ECONNREFUSED
+        env = dict(os.environ)
+        env.update(
+            RELEASE_VERSION="0.56.0",
+            GITEA_TOKEN="test-token",
+            RELEASE_SERVER_URL=f"http://127.0.0.1:{refused_port}",
+            RELEASE_REPO="rune/rune",
+            LIST_API_URL=f"http://127.0.0.1:{refused_port}/api/v1/repos/rune/rune/releases?limit=50",
+            RELEASE_API_URL=f"http://127.0.0.1:{refused_port}/api/v1/repos/rune/rune/releases",
+        )
+        env.pop("PYTEST_CURRENT_TEST", None)
+        proc = subprocess.run(
+            [sys.executable, str(ROOT / "scripts" / "create_release.py")],
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        assert proc.returncode == 1, f"expected exit 1, stderr:\n{proc.stderr}"
+        assert proc.stderr.startswith("create_release: API failed")
+        assert "Traceback" not in proc.stderr
+
+
+class TestDryRunMain:
+    def test_prints_payload_without_api_calls(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        monkeypatch.setenv("RELEASE_VERSION", "0.56.0")
+        monkeypatch.delenv("GITEA_TOKEN", raising=False)
+        monkeypatch.delenv("PR_TITLE", raising=False)
+        rc = cr.main(["--dry-run"])
+        assert rc == 0
+        payload = json.loads(capsys.readouterr().out)
+        assert payload["tag_name"] == "v0.56.0"
+        assert payload["target_commitish"] == "master"
+        assert "safety net" in payload["body"]

tests/test_drift_issue.py

@@ -0,0 +1,177 @@
+"""Tests for scripts/file_drift_issue.py — maintenance drift reconciliation (#192)."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_module() -> Any:
+    spec = importlib.util.spec_from_file_location("file_drift_issue", ROOT / "scripts" / "file_drift_issue.py")
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+fdi = _load_module()
+
+DECISION = {
+    "converged": False,
+    "warranted_bump": "minor",
+    "commits_since_tag": 3,
+    "sample": ["feat: widget", "fix: race", "docs: notes"],
+}
+
+
+class TestDriftTitle:
+    def test_plan_title_format(self) -> None:
+        assert fdi.drift_title("0.55.0", "v0.54.0") == "Release drift: master v0.55.0 vs tag v0.54.0"
+
+    def test_unknown_master_version(self) -> None:
+        assert fdi.drift_title("", "v0.54.0") == "Release drift: master (unknown) vs tag v0.54.0"
+
+
+class TestDriftBody:
+    def test_contains_decision_facts_and_advisory(self) -> None:
+        body = fdi.drift_body(DECISION, "0.55.0", "v0.54.0")
+        assert "0.55.0" in body
+        assert "v0.54.0" in body
+        assert "minor" in body
+        assert "3" in body
+        for subject in DECISION["sample"]:
+            assert subject in body
+        assert "never auto-releases" in body
+        assert "docs/release-flow.md" in body
+
+    def test_version_behind_tag_anomaly_is_flagged(self) -> None:
+        # Spec-gate finding 3 (#192): the reconciler must surface the
+        # version_behind_tag anomaly in the filed issue, not just generic drift.
+        decision = {**DECISION, "warranted_bump": None, "anomaly": "version_behind_tag"}
+        body = fdi.drift_body(decision, "0.54.0", "v0.55.0")
+        assert "version_behind_tag" in body
+        assert "anomaly" in body.lower()
+
+
+class TestFindOpenDriftIssue:
+    def test_matches_exact_title(self) -> None:
+        issues = [{"number": 7, "title": "Release drift: master v0.55.0 vs tag v0.54.0"}]
+        assert fdi.find_open_drift_issue(issues, "Release drift: master v0.55.0 vs tag v0.54.0") == issues[0]
+
+    def test_ignores_other_titles(self) -> None:
+        issues = [{"number": 7, "title": "Release drift: master v0.56.0 vs tag v0.54.0"}]
+        assert fdi.find_open_drift_issue(issues, "Release drift: master v0.55.0 vs tag v0.54.0") is None
+
+    def test_ignores_pull_requests(self) -> None:
+        issues = [{"number": 9, "title": "Release drift: master v0.55.0 vs tag v0.54.0", "pull_request": {}}]
+        assert fdi.find_open_drift_issue(issues, "Release drift: master v0.55.0 vs tag v0.54.0") is None
+
+    def test_empty_list(self) -> None:
+        assert fdi.find_open_drift_issue([], "Release drift: master v0.55.0 vs tag v0.54.0") is None
+
+
+class TestDryRunMain:
+    def test_prints_title_and_body_without_api_calls(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+    ) -> None:
+        monkeypatch.setenv("CONVERGENCE_JSON", json.dumps(DECISION))
+        monkeypatch.setenv("MASTER_VERSION", "0.55.0")
+        monkeypatch.setenv("LATEST_TAG", "v0.54.0")
+        monkeypatch.delenv("GITEA_TOKEN", raising=False)
+        rc = fdi.main(["--dry-run"])
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Release drift: master v0.55.0 vs tag v0.54.0" in out
+        assert "minor" in out
+
+    def test_missing_convergence_json_fails_loudly(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("CONVERGENCE_JSON", raising=False)
+        with pytest.raises(KeyError):
+            fdi.main(["--dry-run"])
+
+
+class TestLabelResolution:
+    """Spec-gate finding 6 (#192): Gitea POST /issues 'labels' requires integer
+    IDs, not names. 'release-drift' must be resolved (and optionally created)
+    server-side; a missing label must not block filing the issue."""
+
+    DECISION_JSON = json.dumps(DECISION)
+
+    def _run(self, fake_api: Any, monkeypatch: pytest.MonkeyPatch, capsys: Any) -> int:
+        monkeypatch.setenv("CONVERGENCE_JSON", self.DECISION_JSON)
+        monkeypatch.setenv("MASTER_VERSION", "0.55.0")
+        monkeypatch.setenv("LATEST_TAG", "v0.54.0")
+        monkeypatch.setenv("GITEA_TOKEN", "t")
+        monkeypatch.setenv("RELEASE_SERVER_URL", "http://forge")
+        monkeypatch.setenv("RELEASE_REPO", "o/r")
+        return fdi.main([], api=fake_api)
+
+    def test_labels_sent_as_integer_ids(self, monkeypatch: pytest.MonkeyPatch, capsys: Any) -> None:
+        calls: list[tuple[str, str, Any]] = []
+
+        def fake_api(path: str, method: str = "GET", data: Any = None) -> tuple[int, Any]:
+            calls.append((method, path, data))
+            if method == "GET" and path.startswith("/issues?"):
+                return 200, []
+            if method == "GET" and path.startswith("/labels"):
+                return 200, [{"id": 42, "name": "release-drift"}]
+            if method == "POST" and path == "/issues":
+                return 201, {"number": 5}
+            raise AssertionError(f"unexpected call: {method} {path}")
+
+        rc = self._run(fake_api, monkeypatch, capsys)
+        assert rc == 0
+        post = next(c for c in calls if c[0] == "POST" and c[1] == "/issues")
+        assert post[2]["labels"] == [42]
+
+    def test_missing_label_is_created(self, monkeypatch: pytest.MonkeyPatch, capsys: Any) -> None:
+        calls: list[tuple[str, str, Any]] = []
+
+        def fake_api(path: str, method: str = "GET", data: Any = None) -> tuple[int, Any]:
+            calls.append((method, path, data))
+            if method == "GET" and path.startswith("/issues?"):
+                return 200, []
+            if method == "GET" and path.startswith("/labels"):
+                return 200, [{"id": 7, "name": "bug"}]
+            if method == "POST" and path == "/labels":
+                return 201, {"id": 99, "name": "release-drift"}
+            if method == "POST" and path == "/issues":
+                return 201, {"number": 5}
+            raise AssertionError(f"unexpected call: {method} {path}")
+
+        rc = self._run(fake_api, monkeypatch, capsys)
+        assert rc == 0
+        post = next(c for c in calls if c[0] == "POST" and c[1] == "/issues")
+        assert post[2]["labels"] == [99]
+        create = next(c for c in calls if c[0] == "POST" and c[1] == "/labels")
+        assert create[2] == {"name": "release-drift", "color": "ff0000"}
+
+    def test_unresolvable_label_skips_label_issue_still_files(
+        self, monkeypatch: pytest.MonkeyPatch, capsys: Any
+    ) -> None:
+        calls: list[tuple[str, str, Any]] = []
+
+        def fake_api(path: str, method: str = "GET", data: Any = None) -> tuple[int, Any]:
+            calls.append((method, path, data))
+            if method == "GET" and path.startswith("/issues?"):
+                return 200, []
+            if method == "GET" and path.startswith("/labels"):
+                return 200, []
+            if method == "POST" and path == "/labels":
+                return 422, {"_body": "label quota exhausted"}
+            if method == "POST" and path == "/issues":
+                return 201, {"number": 5}
+            raise AssertionError(f"unexpected call: {method} {path}")
+
+        rc = self._run(fake_api, monkeypatch, capsys)
+        assert rc == 0  # the issue still files
+        post = next(c for c in calls if c[0] == "POST" and c[1] == "/issues")
+        assert post[2]["labels"] == []  # label skipped, not a name string
+        out = capsys.readouterr().out
+        assert "release-drift" in out  # the skip is visible in the log

tests/test_version_convergence.py

@@ -0,0 +1,354 @@
+"""Tests for scripts/version_convergence.py — the push-triggered release safety net (#192).
+
+Pure decision-logic tests are fed fixture commit-subject lists (no live git).
+One subprocess-level class exercises the real git pipeline in a temp repo.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _load_module() -> Any:
+    spec = importlib.util.spec_from_file_location("version_convergence", ROOT / "scripts" / "version_convergence.py")
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+vc = _load_module()
+
+
+# ---------------------------------------------------------------------------
+# Subject classification (pure fixtures)
+# ---------------------------------------------------------------------------
+
+
+class TestClassifySubject:
+    @pytest.mark.parametrize(
+        ("subject", "expected"),
+        [
+            ("feat: add widget", "feat"),
+            ("feat(tui): add widget", "feat"),
+            ("feat!: remove legacy bus", "breaking"),
+            ("feat(tui)!: remove legacy bus", "breaking"),
+            ("BREAKING CHANGE: drop legacy bus", "breaking"),
+            ("fix: stop event race", "fix"),
+            ("fix(tui): stop event race", "fix"),
+            ("perf: cache capsule lookups", "fix"),
+            ("perf(tui): cache capsule lookups", "fix"),
+            ("docs: update readme", "other"),
+            ("docs(readme): update", "other"),
+            ("chore: bump version", "other"),
+            ("chore(release): auto-bump version to 0.55.0 (post-merge automation)", "other"),
+            ("test: add convergence tests", "other"),
+            ("refactor: extract helper", "other"),
+            ("style: format code", "other"),
+            ("ci: tweak workflow", "other"),
+            ("build: bump dep", "other"),
+            ("Merge pull request 'x' (#12) from y into master", "other"),
+            ("revert feat: add widget", "other"),
+            ("no prefix at all, just words", "other"),
+            ("", "other"),
+        ],
+    )
+    def test_classification(self, subject: str, expected: str) -> None:
+        assert vc.classify_subject(subject) == expected
+
+    @pytest.mark.parametrize(
+        ("merge_subject", "expected"),
+        [
+            # Gitea/GitHub merge-commit subjects: the conventional title is
+            # quoted inside — classification must anchor at the inner title
+            # (#192 spec-gate finding 2), not at "Merge ...".
+            ("Merge pull request 'feat: add widget' (#12)", "feat"),
+            ("Merge pull request 'feat(tui): add widget' (#3)", "feat"),
+            ("Merge pull request 'fix: stop event race' (#14)", "fix"),
+            ("Merge pull request 'fix(api): stop event race' (#15)", "fix"),
+            ("Merge pull request 'feat!: remove legacy bus' (#16)", "breaking"),
+            ("Merge pull request 'BREAKING CHANGE: drop legacy bus' (#17)", "breaking"),
+            ("Merge pull request 'perf: cache lookups' (#18)", "fix"),
+            ("Merge pull request 'docs: update readme' (#19)", "other"),
+            ("Merge pull request 'chore: tidy' (#20)", "other"),
+            ("Merge pull request 'no prefix at all' (#21)", "other"),
+        ],
+    )
+    def test_merge_commit_inner_title_classification(self, merge_subject: str, expected: str) -> None:
+        assert vc.classify_subject(merge_subject) == expected
+
+    def test_merge_commit_without_quoted_title_stays_other(self) -> None:
+        # Truncated/unusual merge subjects must not crash or misclassify.
+        assert vc.classify_subject("Merge branch 'master' into release") == "other"
+        assert vc.classify_subject("Merge pull request #22 from nick/branch") == "other"
+
+
+class TestHighestWarrantedBump:
+    @pytest.mark.parametrize(
+        ("subjects", "expected"),
+        [
+            # 0.x convention (matches the workflow's existing rule): breaking -> minor
+            (["feat!: remove legacy bus", "fix: race"], "minor"),
+            (["feat: add widget", "fix: race"], "minor"),
+            (["feat(tui): widget", "docs: notes"], "minor"),
+            (["chore: x", "BREAKING CHANGE: api removed"], "minor"),
+            (["fix: race", "perf: cache", "chore: tidy"], "patch"),
+            (["docs: update", "chore: tidy", "test: add", "refactor: split"], None),
+            ([], None),
+        ],
+    )
+    def test_decision(self, subjects: list[str], expected: str | None) -> None:
+        assert vc.highest_warranted_bump(subjects) == expected
+
+
+# ---------------------------------------------------------------------------
+# Version arithmetic helpers (used by the push path to compute the new version)
+# ---------------------------------------------------------------------------
+
+
+class TestNextVersion:
+    @pytest.mark.parametrize(
+        ("version", "kind", "expected"),
+        [
+            ("0.51.0", "minor", "0.52.0"),
+            ("0.51.3", "patch", "0.51.4"),
+            ("0.51.0", "major", "0.52.0"),  # 0.x convention: breaking bumps the minor
+            ("1.2.3", "major", "2.0.0"),
+            ("1.2.3", "minor", "1.3.0"),
+            ("1.2.3", "patch", "1.2.4"),
+        ],
+    )
+    def test_next(self, version: str, kind: str, expected: str) -> None:
+        assert vc.next_version(version, kind) == expected
+
+    def test_rejects_bad_kind(self) -> None:
+        with pytest.raises(ValueError):
+            vc.next_version("0.51.0", "docs")
+
+
+class TestVersionGreater:
+    @pytest.mark.parametrize(
+        ("a", "b", "expected"),
+        [
+            ("0.56.0", "0.55.0", True),
+            ("0.55.0", "0.55.0", False),
+            ("0.9.0", "0.10.0", False),  # numeric, not lexicographic
+            ("1.0.0", "0.99.99", True),
+            ("0.55.1", "0.55.0", True),
+        ],
+    )
+    def test_compare(self, a: str, b: str, expected: bool) -> None:
+        assert vc.version_greater(a, b) is expected
+
+
+# ---------------------------------------------------------------------------
+# Decision assembly — the JSON contract from the plan
+# ---------------------------------------------------------------------------
+
+
+class TestAssembleReleaseDecision:
+    def test_unreleased_feat_work_is_drift(self) -> None:
+        # version == tag with release-worthy commits since: the bump never rode
+        # the merge — exactly the dropped-event scenario from #190/#192.
+        data = vc.assemble_release_decision(
+            version="0.55.0", latest_tag="v0.55.0", subjects=["feat: w", "fix: r"], commit_count=2
+        )
+        assert data == {
+            "converged": False,
+            "warranted_bump": "minor",
+            "commits_since_tag": 2,
+            "sample": ["feat: w", "fix: r"],
+            "anomaly": None,
+        }
+
+    def test_fully_released_is_converged(self) -> None:
+        data = vc.assemble_release_decision(version="0.55.0", latest_tag="v0.55.0", subjects=[], commit_count=0)
+        assert data == {
+            "converged": True,
+            "warranted_bump": None,
+            "commits_since_tag": 0,
+            "sample": [],
+            "anomaly": None,
+        }
+
+    def test_docs_only_since_tag_is_converged(self) -> None:
+        data = vc.assemble_release_decision(
+            version="0.55.0", latest_tag="v0.55.0", subjects=["docs: notes"], commit_count=1
+        )
+        assert data == {
+            "converged": True,
+            "warranted_bump": None,
+            "commits_since_tag": 1,
+            "sample": ["docs: notes"],
+            "anomaly": None,
+        }
+
+    def test_version_ahead_of_tag_is_drift(self) -> None:
+        # Bump rode the merge but the tag never got cut (closed-event dropped on
+        # a bump-carrying PR): a release is missing.
+        data = vc.assemble_release_decision(
+            version="0.56.0", latest_tag="v0.55.0", subjects=["feat: w"], commit_count=1
+        )
+        assert data == {
+            "converged": False,
+            "warranted_bump": "minor",
+            "commits_since_tag": 1,
+            "sample": ["feat: w"],
+            "anomaly": None,
+        }
+
+    def test_version_behind_tag_is_drift_with_anomaly(self) -> None:
+        # Spec-gate finding 3 (#192): master's version BELOW the latest tag is
+        # DRIFT (a bump merged without riding a release, or a tag moved
+        # backwards) — it must not be silently converged. The net must also not
+        # act on it (warranted_bump forced None even when subjects warrant one):
+        # bumping master cannot converge it with a tag that is already ahead.
+        data = vc.assemble_release_decision(
+            version="0.54.0", latest_tag="v0.55.0", subjects=["feat: w"], commit_count=1
+        )
+        assert data["converged"] is False
+        assert data["warranted_bump"] is None
+        assert data["anomaly"] == "version_behind_tag"
+
+    def test_version_behind_tag_with_no_warranting_subjects_still_drifts(self) -> None:
+        data = vc.assemble_release_decision(
+            version="0.54.0", latest_tag="v0.55.0", subjects=["docs: notes"], commit_count=1
+        )
+        assert data["converged"] is False
+        assert data["warranted_bump"] is None
+        assert data["anomaly"] == "version_behind_tag"
+
+    def test_sample_capped_at_ten(self) -> None:
+        subjects = [f"fix: thing {i}" for i in range(14)]
+        data = vc.assemble_release_decision(version="0.55.0", latest_tag="v0.55.0", subjects=subjects, commit_count=14)
+        assert data["sample"] == subjects[:10]
+        assert data["commits_since_tag"] == 14
+        assert data["warranted_bump"] == "patch"
+
+    def test_missing_version_is_drift(self) -> None:
+        data = vc.assemble_release_decision(version=None, latest_tag="v0.55.0", subjects=[], commit_count=0)
+        assert data["converged"] is False
+        assert data["warranted_bump"] is None
+
+    def test_no_tags_is_converged_noop(self) -> None:
+        # Fresh repo / first-release territory: the safety net never acts.
+        data = vc.assemble_release_decision(version="0.1.0", latest_tag=None, subjects=[], commit_count=0)
+        assert data == {
+            "converged": True,
+            "warranted_bump": None,
+            "commits_since_tag": 0,
+            "sample": [],
+            "anomaly": None,
+        }
+
+
+class TestMainOutputShape:
+    def test_main_json_contract(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(vc, "read_version", lambda repo=None: "0.55.0")
+        monkeypatch.setattr(vc, "latest_tag", lambda repo=None: "v0.54.0")
+        monkeypatch.setattr(vc, "subjects_since_tag", lambda tag, repo=None: ["feat: w", "fix: r"])
+        out = vc.main([])
+        assert set(out) == {"converged", "warranted_bump", "anomaly", "commits_since_tag", "sample"}
+        assert out["converged"] is False
+        assert out["warranted_bump"] == "minor"
+        assert out["anomaly"] is None
+        assert out["commits_since_tag"] == 2
+        assert out["sample"] == ["feat: w", "fix: r"]
+
+
+# ---------------------------------------------------------------------------
+# Subprocess-level: real temp git repo (skipped when git is unavailable)
+# ---------------------------------------------------------------------------
+
+git_available = shutil.which("git") is not None
+
+GIT_ENV = {
+    **os.environ,
+    "GIT_AUTHOR_NAME": "t",
+    "GIT_AUTHOR_EMAIL": "t@t",
+    "GIT_COMMITTER_NAME": "t",
+    "GIT_COMMITTER_EMAIL": "t@t",
+}
+
+
+@pytest.mark.skipif(not git_available, reason="git not available")
+class TestRealGitRepo:
+    def _git(self, repo: Path, *args: str) -> None:
+        subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True, env=GIT_ENV)
+
+    def _repo(self, tmp_path: Path, commits: list[str]) -> Path:
+        repo = tmp_path / "repo"
+        repo.mkdir()
+        self._git(repo, "init", "-q")
+        (repo / "src" / "rune").mkdir(parents=True)
+        (repo / "src" / "rune" / "__init__.py").write_text('__version__ = "0.1.0"\n')
+        self._git(repo, "add", "-A")
+        self._git(repo, "commit", "-q", "-m", "feat: base")
+        for subject in commits:
+            self._git(repo, "commit", "-q", "--allow-empty", "-m", subject)
+        return repo
+
+    def test_drift_minor(self, tmp_path: Path) -> None:
+        repo = self._repo(tmp_path, ["feat: add widget", "fix: race"])
+        self._git(repo, "tag", "v0.1.0")
+        self._git(repo, "commit", "-q", "--allow-empty", "-m", "feat: post-tag feature")
+        decision = vc.decision_in_repo(repo)
+        assert decision["converged"] is False
+        assert decision["warranted_bump"] == "minor"
+        assert decision["commits_since_tag"] == 1
+        assert decision["sample"] == ["feat: post-tag feature"]
+
+    def test_converged_docs_only(self, tmp_path: Path) -> None:
+        repo = self._repo(tmp_path, [])
+        self._git(repo, "tag", "v0.1.0")
+        self._git(repo, "commit", "-q", "--allow-empty", "-m", "docs: notes")
+        decision = vc.decision_in_repo(repo)
+        assert decision == {
+            "converged": True,
+            "warranted_bump": None,
+            "commits_since_tag": 1,
+            "sample": ["docs: notes"],
+            "anomaly": None,
+        }
+
+    def test_converged_no_commits_since_tag(self, tmp_path: Path) -> None:
+        repo = self._repo(tmp_path, [])
+        self._git(repo, "tag", "v0.1.0")
+        decision = vc.decision_in_repo(repo)
+        assert decision == {
+            "converged": True,
+            "warranted_bump": None,
+            "commits_since_tag": 0,
+            "sample": [],
+            "anomaly": None,
+        }
+
+    def test_version_behind_tag_is_drift_anomaly(self, tmp_path: Path) -> None:
+        # Real-repo check for spec-gate finding 3: tag ahead of master's
+        # __version__ must surface as drift with the anomaly flag.
+        repo = self._repo(tmp_path, [])
+        self._git(repo, "tag", "v0.2.0")  # tag moved ahead of the 0.1.0 version
+        decision = vc.decision_in_repo(repo)
+        assert decision["converged"] is False
+        assert decision["warranted_bump"] is None
+        assert decision["anomaly"] == "version_behind_tag"
+
+    def test_cli_prints_plan_json(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
+        import json
+
+        repo = self._repo(tmp_path, ["fix: race"])
+        self._git(repo, "tag", "v0.1.0")
+        self._git(repo, "commit", "-q", "--allow-empty", "-m", "fix: post-tag race")
+        out = vc.main(["--repo", str(repo)])
+        printed = json.loads(capsys.readouterr().out)
+        assert printed == out
+        assert out["warranted_bump"] == "patch"

tests/test_workflow_script_contract.py

@@ -0,0 +1,200 @@
+"""Env-contract guard between Gitea workflow steps and scripts/*.py (#192).
+
+Spec-gate finding 1 discharge — the systematic guard: every step that invokes
+a ``scripts/*.py`` entrypoint must provide every env var that script reads
+with a hard ``os.environ[...]`` subscript. A read counts as satisfied when it
+comes from any of:
+
+* the step's own ``env:`` block (plus workflow/job-level ``env:``),
+* a ``$GITHUB_ENV`` export from an earlier step in the same job (shell
+  ``echo "NAME=..." >> "$GITHUB_ENV"`` or heredoc ``env.write(f"NAME=...")``),
+* a plain shell ``export NAME=`` inside the invoking run block,
+* a documented script-derived default (``SCRIPT_DERIVED`` below) — keys the
+  script derives itself from other env instead of requiring them.
+
+Regression history: ``create_release.py`` subscript-read LIST_API_URL /
+RELEASE_API_URL while ``auto-version.yml`` only ever passed
+RELEASE_SERVER_URL/RELEASE_REPO — KeyError at runtime on both release paths.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+from typing import Any, Iterator
+
+import pytest
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKFLOW_DIR = ROOT / ".gitea" / "workflows"
+SCRIPTS_DIR = ROOT / "scripts"
+
+WORKFLOWS = ("auto-version.yml", "maintenance.yml")
+
+# Hard env reads: os.environ["KEY"] / os.environ['KEY'] (KeyError risk).
+REQUIRED_ENV_RE = re.compile(r"""os\.environ\[["']([A-Za-z_][A-Za-z0-9_]*)["']\]""")
+# Bare subscript of an allowlisted derived key — the derivation must be real.
+_SUBSCRIPT_TEMPLATES = ('os.environ["{key}"', "os.environ['{key}'")
+# GITHUB_ENV exports, shell form: echo "NAME=..." >> "$GITHUB_ENV"
+GITHUB_ENV_SHELL_RE = re.compile(r"""([A-Za-z_][A-Za-z0-9_]*)=.*?>>\s*["']?\$\{?GITHUB_ENV\}?["']?""")
+# GITHUB_ENV exports, python form: env.write(f"NAME=...") / env.write("NAME=...")
+GITHUB_ENV_PY_RE = re.compile(r"""env\.write\(f?["']([A-Za-z_][A-Za-z0-9_]*)=""")
+# Shell exports inside the invoking run block: export NAME=...
+SHELL_EXPORT_RE = re.compile(r"^\s*export\s+([A-Za-z_][A-Za-z0-9_]*)=", re.MULTILINE)
+# scripts/<name>.py invocation inside a run block
+SCRIPT_INVOKE_RE = re.compile(r"scripts/([a-z_][a-z0-9_]*)\.py")
+
+# Keys each script derives itself from other env instead of requiring them
+# from the workflow. Guarded below: an allowlisted key must NOT be read with
+# a bare subscript (that would KeyError when the derivation inputs are absent).
+SCRIPT_DERIVED: dict[str, frozenset[str]] = {
+    "create_release": frozenset({"LIST_API_URL", "RELEASE_API_URL"}),
+    "auto_bump_pr": frozenset(),
+    "version_convergence": frozenset(),
+    "file_drift_issue": frozenset(),
+}
+
+
+def _load_workflow(name: str) -> dict[str, Any]:
+    path = WORKFLOW_DIR / name
+    assert path.exists(), f"workflow missing: {path}"
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    assert isinstance(data, dict), f"workflow did not parse to a mapping: {name}"
+    return data
+
+
+def _invocations(workflow_name: str) -> Iterator[tuple[str, str, frozenset[str]]]:
+    """Yield (step_id, script_name, provided_env) per scripts/*.py invocation.
+
+    ``provided_env`` models runtime visibility faithfully: step env + same-run
+    shell exports are visible to the invoking step; GITHUB_ENV exports only
+    become visible to *subsequent* steps, so they are accumulated after the
+    step's own invocations are checked.
+    """
+    workflow = _load_workflow(workflow_name)
+    for job_name, job in (workflow.get("jobs") or {}).items():
+        github_env: set[str] = set()
+        for idx, step in enumerate(job.get("steps") or []):
+            run = step.get("run") or ""
+            step_env = frozenset((step.get("env") or {}).keys())
+            shell_exports = frozenset(SHELL_EXPORT_RE.findall(run))
+            for match in SCRIPT_INVOKE_RE.finditer(run):
+                step_id = f"{workflow_name}:{job_name}/step{idx}"
+                yield step_id, match.group(1), step_env | shell_exports | frozenset(github_env)
+            github_env |= set(GITHUB_ENV_SHELL_RE.findall(run))
+            github_env |= set(GITHUB_ENV_PY_RE.findall(run))
+
+
+def _params() -> list[tuple[str, str, frozenset[str]]]:
+    cases: list[tuple[str, str, frozenset[str]]] = []
+    for workflow_name in WORKFLOWS:
+        cases.extend(_invocations(workflow_name))
+    assert cases, "no scripts/*.py invocations found — guard is vacuous"
+    return cases
+
+
+def _script_source(script: str) -> str:
+    return (SCRIPTS_DIR / f"{script}.py").read_text(encoding="utf-8")
+
+
+@pytest.mark.parametrize(("step_id", "script", "provided"), _params(), ids=lambda v: str(v))
+def test_step_provides_every_required_env_read(step_id: str, script: str, provided: frozenset[str]) -> None:
+    source = _script_source(script)
+    required = set(REQUIRED_ENV_RE.findall(source))
+    missing = required - provided - SCRIPT_DERIVED.get(script, frozenset())
+    assert not missing, (
+        f"{step_id}: scripts/{script}.py hard-reads {sorted(missing)} from os.environ "
+        f"but the workflow provides none of them (step/job env, $GITHUB_ENV exports, "
+        f"shell exports, or a documented script-derived default)"
+    )
+
+
+@pytest.mark.parametrize(("step_id", "script", "provided"), _params(), ids=lambda v: str(v))
+def test_derived_allowlist_is_honest(step_id: str, script: str, provided: frozenset[str]) -> None:
+    """An allowlisted derived key must genuinely be derived: no bare subscript
+    read of it may remain in the script (that would KeyError at runtime)."""
+    source = _script_source(script)
+    for key in SCRIPT_DERIVED.get(script, frozenset()):
+        for template in _SUBSCRIPT_TEMPLATES:
+            assert template.format(key=key) not in source, (
+                f"{step_id}: scripts/{script}.py subscripts {key} directly — "
+                f"either provide it from the workflow or derive it in-script"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Boolean env exports (quality-gate finding 5, #192)
+# ---------------------------------------------------------------------------
+
+
+def _run_steps(workflow_name: str) -> Iterator[tuple[str, str, int]]:
+    """Yield (job_name, run_block, step_index) for every run step."""
+    workflow = _load_workflow(workflow_name)
+    for job_name, job in (workflow.get("jobs") or {}).items():
+        for idx, step in enumerate(job.get("steps") or []):
+            run = step.get("run")
+            if run:
+                yield job_name, run, idx
+
+
+# A GITHUB_ENV export of a boolean-typed key must emit lowercase true/false:
+# workflow conditionals compare against the string 'true'/'false', and Python
+# str(True) is capitalized — auto-version.yml emitted 'True'/'False' from
+# json.load while the steps compared env.CONVERGED == 'false', making the
+# safety-net conditionals unreachable.
+BOOLEAN_ENV_KEYS = frozenset({"CONVERGED", "AUTO_BUMPED", "DRIFT"})
+_LOWERCASE_LITERAL_RE = re.compile(r"""['"](true|false)['"]""")
+_JSON_BOOL_ACCESS_RE = re.compile(r"""json\.load\(|\[\s*['"](?:converged|merged)['"]\s*\]""")
+
+
+def _value_expr(line: str, lines: list[str], key: str) -> str | None:
+    """The value-producing expression behind a boolean key export/capture."""
+    capture = re.search(rf"^\s*{key}=\$\((.*)\)\s*$", line)
+    if capture:
+        return capture.group(1)
+    write = re.search(rf"""env\.write\(f?["']{key}=(.*?)["']\)$""", line.strip())
+    if write:
+        return write.group(1)
+    echo = re.search(rf'echo\s+"{key}=(\$[A-Za-z_][A-Za-z0-9_]*)"', line)
+    if echo:  # indirect: resolve the earlier capture line in the same run block
+        ref = echo.group(1).lstrip("${").rstrip("}")
+        for other in lines:
+            other_capture = re.search(rf"^\s*{re.escape(ref)}=\$\((.*)\)\s*$", other)
+            if other_capture:
+                return other_capture.group(1)
+        return echo.group(1)  # unresolvable — not provably lowercase
+    return None
+
+
+def _emits_lowercase(expr: str) -> bool:
+    if ".lower()" in expr:
+        return True
+    if _LOWERCASE_LITERAL_RE.search(expr):
+        return True  # built from explicit lowercase literals
+    if re.search(r"\b(True|False)\b", expr):
+        return False  # Python bool reprs leak capitalized
+    if _JSON_BOOL_ACCESS_RE.search(expr):
+        return False  # bool-typed json field without .lower()
+    return True  # opaque shell var / workflow expression (rendered lowercase)
+
+
+@pytest.mark.parametrize(
+    ("workflow_name", "job", "run", "idx"),
+    [(w, j, r, i) for w in WORKFLOWS for (j, r, i) in _run_steps(w)],
+)
+def test_boolean_env_exports_are_lowercase(workflow_name: str, job: str, run: str, idx: int) -> None:
+    lines = run.splitlines()
+    offenders = []
+    for key in BOOLEAN_ENV_KEYS:
+        for line in lines:
+            if not (re.search(rf"\b{key}=", line) or f'"{key}=' in line):
+                continue
+            expr = _value_expr(line, lines, key)
+            if expr is not None and not _emits_lowercase(expr):
+                offenders.append(f"[{key}] {line.strip()}")
+    assert not offenders, (
+        f"{workflow_name}:{job}/step{idx}: boolean env exports must force lowercase "
+        f"(str(...).lower()) or use lowercase literals — conditionals compare "
+        f"env.CONVERGED == 'false'-style strings: {offenders}"
+    )