Merge pull request 'Add fail-closed branch-identity proofs for author commit/push workflow (Issue #177)' (#181) from feat/issue-177-branch-drift-proofs into master

This commit was merged in pull request #181.
This commit is contained in:
2026-07-05 15:13:24 -05:00
4 changed files with 492 additions and 9 deletions
+217
View File
@@ -0,0 +1,217 @@
"""Fail-closed branch-identity proofs for author workflows (#177).
Author-side counterpart of the reviewer proofs in ``review_proofs.py``
(#173). During the #173 implementation itself, a commit landed on local
``master`` because the shared checkout's branch moved mid-session (origin
incident of #177). These helpers turn that from an after-the-fact repair
into a fail-closed gate: an author workflow must prove its local git state
before staging, committing, or pushing.
The helpers are pure (no git calls): the workflow gathers the raw facts
(``git branch --show-current``, ``git rev-parse HEAD``, the push refspec,
the branch named in the issue claim) and passes them in, so the same logic
works from prompts, harness assertions, and tests. Shared-worktree branch
switches by other sessions are treated as expected events to detect, not
exceptional ones. Nothing here weakens the review/merge/permission gates.
"""
PROTECTED_BRANCHES = frozenset(
{"master", "main", "develop", "development", "dev"}
)
def _clean(name):
return (name or "").strip()
def verify_branch_for_commit(current_branch, intended_branch):
"""Required behavior 1: prove the branch before staging/committing.
Proven only when both names are present, the intended branch is not a
protected branch, and the current branch equals the intended one (which
also rules out being on any protected branch). Returns {'proven',
'block', 'reasons', 'current_branch', 'intended_branch'}.
"""
reasons = []
current = _clean(current_branch)
intended = _clean(intended_branch)
if not current:
reasons.append(
"current branch unknown (detached HEAD or state not read); "
"fail closed"
)
if not intended:
reasons.append("intended feature branch not stated; fail closed")
if intended and intended in PROTECTED_BRANCHES:
reasons.append(
f"intended branch '{intended}' is a protected branch; author "
"work must target a feature branch"
)
if current and current in PROTECTED_BRANCHES:
reasons.append(
f"current branch '{current}' is a protected branch; committing "
"here is blocked"
)
if current and intended and current != intended:
reasons.append(
f"current branch '{current}' is not the intended feature branch "
f"'{intended}'; stop before staging/committing"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"current_branch": current or None,
"intended_branch": intended or None,
}
def detect_branch_drift(branch_at_validation, head_at_validation,
current_branch, current_head):
"""Required behaviors 23: stop when branch or HEAD moved mid-session.
Compares the branch name and HEAD SHA captured at validation time with
the state observed immediately before commit/push. Any difference —
including an external branch switch in a shared worktree — is drift and
blocks until reconciled. Missing state fails closed.
"""
reasons = []
branch_then = _clean(branch_at_validation)
branch_now = _clean(current_branch)
head_then = _clean(head_at_validation).lower()
head_now = _clean(current_head).lower()
if not branch_then or not head_then:
reasons.append("validation-time branch/HEAD not recorded; fail closed")
if not branch_now or not head_now:
reasons.append("current branch/HEAD not read; fail closed")
if branch_then and branch_now and branch_then != branch_now:
reasons.append(
f"branch changed from '{branch_then}' to '{branch_now}' since "
"validation — possible external branch switch in a shared "
"worktree; stop and reconcile before committing"
)
if head_then and head_now and head_then != head_now:
reasons.append(
"HEAD moved since validation; re-validate on the current HEAD "
"before committing"
)
drifted = bool(reasons)
return {"drifted": drifted, "block": drifted, "reasons": reasons}
def verify_push_target(current_branch, remote_target_branch, intended_branch):
"""Acceptance: a push needs local, remote, and intended branches to match.
Proven only when all three names are present, equal, and not a
protected branch — a feature-branch workflow never pushes a protected
branch, and never pushes to a refspec other than its own branch.
"""
reasons = []
current = _clean(current_branch)
remote_target = _clean(remote_target_branch)
intended = _clean(intended_branch)
if not current:
reasons.append("current branch unknown; fail closed")
if not remote_target:
reasons.append("remote target branch not stated; fail closed")
if not intended:
reasons.append("intended feature branch not stated; fail closed")
for label, name in (("current", current), ("remote target", remote_target),
("intended", intended)):
if name and name in PROTECTED_BRANCHES:
reasons.append(
f"{label} branch '{name}' is a protected branch; author "
"pushes to protected branches are blocked"
)
if current and remote_target and current != remote_target:
reasons.append(
f"push target '{remote_target}' does not match the local branch "
f"'{current}'"
)
if current and intended and current != intended:
reasons.append(
f"local branch '{current}' does not match the intended feature "
f"branch '{intended}'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
}
def assess_protected_branch_commit(commit_branch, pushed=False,
repair_reported=True):
"""Required behavior 4: handle an accidental protected-branch commit.
If a commit landed on a protected branch: it must never be pushed, a
repair is required, and the repair must be *reported* — silently
continuing after (or without) repair is a violation, as is having
pushed the accident.
"""
branch = _clean(commit_branch)
accident = branch in PROTECTED_BRANCHES
violations = []
if accident:
if pushed:
violations.append(
f"accidental commit on protected branch '{branch}' was "
"pushed; protected-branch pushes are forbidden"
)
if not repair_reported:
violations.append(
"protected-branch commit repair was not reported; the "
"workflow must surface the accident and the repair steps, "
"never silently continue"
)
return {
"accident": accident,
"must_not_push": accident,
"repair_required": accident,
"violations": violations,
}
def build_commit_push_report(commit_proof, drift, push_proof, accident=None):
"""Acceptance: final report carries branch proof before commit and push.
Combines the individual proofs; any failed proof, detected drift, or
accident violation makes the status 'blocked' — the workflow stops and
reports instead of continuing.
"""
accident = accident or {"accident": False, "violations": []}
violations = list(accident.get("violations", []))
blocked = (
not commit_proof.get("proven")
or drift.get("drifted")
or not push_proof.get("proven")
or bool(violations)
)
return {
"status": "blocked" if blocked else "ok",
"branch_proof_before_commit": bool(commit_proof.get("proven")),
"branch_proof_before_push": bool(push_proof.get("proven")),
"drift_detected": bool(drift.get("drifted")),
"protected_branch_accident": bool(accident.get("accident")),
"violations": violations,
"reasons": (
list(commit_proof.get("reasons", []))
+ list(drift.get("reasons", []))
+ list(push_proof.get("reasons", []))
),
}
+27 -6
View File
@@ -150,12 +150,33 @@ Worktree folder = branch with `/` replaced by `-`
6. Implement the narrow scope only — no unrelated refactors or formatting churn.
7. Add/update focused tests when behavior changes.
8. Run the checks (tests, compile/lint, `git diff --check`, secret scan).
9. Commit with an issue-linked message.
10. Push the branch.
11. Open a PR to `master`.
12. **If you are the author, stop before review/merge.**
13. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
14. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
Record the branch name and `HEAD` SHA at validation time — the drift
check in step 9 compares against exactly this state.
9. **Branch proof before commit (#177):** prove and state, immediately
before staging/committing (`author_proofs.verify_branch_for_commit`,
`author_proofs.detect_branch_drift`):
- current branch (`git branch --show-current`) equals the intended
feature branch from the issue claim
- current branch is not `master`, `main`, `develop`, `development`, or
`dev`
- branch and `HEAD` have not changed since validation (step 8) — in a
shared checkout another session may switch branches mid-session;
treat that as expected and **stop before committing** when detected
If any check fails, stop and reconcile; do not commit.
10. Commit with an issue-linked message.
11. **Branch proof before push (#177):** prove that the local branch, the
push target branch, and the intended issue branch all match, and that
none of them is a protected branch
(`author_proofs.verify_push_target`). If a commit accidentally landed
on a protected branch, do **not** push: report the accident and the
exact repair steps (`author_proofs.assess_protected_branch_commit`) —
never silently continue after a repair.
12. Push the branch.
13. Open a PR to `master`. The final report must include the branch proofs
from steps 9 and 11 (`author_proofs.build_commit_push_report`).
14. **If you are the author, stop before review/merge.**
15. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism.
16. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include:
- why the PR merge path could not be used
- exact commits pushed
- PR metadata state
@@ -26,8 +26,19 @@ Steps:
cd branches/<type>-issue-<n>-<slug>
6. Implement the narrow scope only; add/update focused tests if behavior changes.
7. Checks: run the test suite, compile/lint changed files, git diff --check,
and scan the diff for secrets.
8. Commit (issue-linked message), push the branch, open a PR to master.
and scan the diff for secrets. Record the branch name and HEAD SHA at
validation time.
8. Branch proof before commit (#177) — prove and state:
- git branch --show-current == the intended issue branch from step 5
- the branch is NOT master/main/develop/development/dev
- branch and HEAD unchanged since step 7 (another session can switch a
shared checkout mid-session; if drift is detected, STOP and reconcile
before committing)
If a commit accidentally lands on a protected branch: do NOT push;
report the accident and the exact repair steps — never silently continue.
9. Commit (issue-linked message). Branch proof before push (#177): local
branch == push target branch == intended issue branch, none protected.
Then push the branch and open a PR to master.
*The PR body MUST use closing keywords like `Closes #N` or `Fixes #N` to close the issue; do NOT use `Implements #N` or `Refs #N` for closing, as Gitea will not auto-close it.*
Include an "LLM Handoff Metadata" block in the PR body (attribution only;
never an eligibility input — docs/llm-agent-sha.md):
@@ -40,7 +51,7 @@ Steps:
- Branch: <branch>
- Worktree: <worktree path>
- Self-review allowed: no
9. Stop before review/merge — you are the author.
10. Stop before review/merge — you are the author.
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (compact; long form only on the high-risk triggers), including the author
+234
View File
@@ -0,0 +1,234 @@
"""Tests for author-side branch-identity proofs (Issue #177).
Issue #177 (author-side counterpart of the #173 reviewer proofs) requires
author workflows to *prove* local git state before staging, committing, or
pushing, instead of discovering drift after the fact:
1. The current branch equals the intended feature branch and is never a
protected branch (master/main/develop/development/dev).
2. Branch or HEAD drift between validation and commit — including external
branch switches in a shared worktree — stops the workflow.
3. A push requires local branch, remote target branch, and intended issue
branch to all match.
4. An accidental commit on a protected branch must not be pushed and its
repair must be reported, never silently continued.
These are the harness assertions from the issue's Required behavior 5.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from author_proofs import ( # noqa: E402
PROTECTED_BRANCHES,
assess_protected_branch_commit,
build_commit_push_report,
detect_branch_drift,
verify_branch_for_commit,
verify_push_target,
)
FEATURE = "feat/issue-177-branch-drift-proofs"
HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd"
HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0"
class TestProtectedBranches(unittest.TestCase):
def test_known_protected_names(self):
for name in ("master", "main", "develop", "development", "dev"):
self.assertIn(name, PROTECTED_BRANCHES)
class TestVerifyBranchForCommit(unittest.TestCase):
"""Required behavior 1: prove the branch before staging/committing."""
def test_on_intended_feature_branch_is_proven(self):
proof = verify_branch_for_commit(FEATURE, FEATURE)
self.assertTrue(proof["proven"])
self.assertFalse(proof["block"])
def test_commit_attempted_while_on_master_is_blocked(self):
# Harness assertion (behavior 5, bullet 1).
proof = verify_branch_for_commit("master", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
self.assertTrue(any("master" in r for r in proof["reasons"]))
def test_every_protected_branch_is_blocked_as_current(self):
for name in PROTECTED_BRANCHES:
proof = verify_branch_for_commit(name, FEATURE)
self.assertTrue(proof["block"], name)
def test_intended_branch_may_not_be_protected(self):
proof = verify_branch_for_commit("master", "master")
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_wrong_feature_branch_is_blocked(self):
proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_missing_current_branch_fails_closed(self):
proof = verify_branch_for_commit("", FEATURE)
self.assertTrue(proof["block"])
def test_missing_intended_branch_fails_closed(self):
proof = verify_branch_for_commit(FEATURE, None)
self.assertTrue(proof["block"])
class TestBranchDrift(unittest.TestCase):
"""Required behaviors 2 + 3: drift between validation and commit stops
the workflow."""
def test_no_drift_when_branch_and_head_unchanged(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1)
self.assertFalse(drift["drifted"])
self.assertFalse(drift["block"])
def test_branch_drift_between_validation_and_commit_is_blocked(self):
# Harness assertion (behavior 5, bullet 2).
drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
def test_shared_worktree_branch_switch_is_detected(self):
# Harness assertion (behavior 5, bullet 4): an external session
# switching the shared checkout to another branch (e.g. master)
# must be detected as drift, not treated as exceptional noise.
drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
self.assertTrue(any("switch" in r.lower() for r in drift["reasons"]))
def test_head_moved_since_validation_is_blocked(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
self.assertTrue(any("HEAD" in r for r in drift["reasons"]))
def test_missing_state_fails_closed(self):
drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None)
self.assertTrue(drift["drifted"])
self.assertTrue(drift["block"])
class TestVerifyPushTarget(unittest.TestCase):
"""Required behavior 1 (push leg) + acceptance: push needs proof that
local, remote, and intended branches all match."""
def test_matching_local_remote_and_intended_is_proven(self):
proof = verify_push_target(FEATURE, FEATURE, FEATURE)
self.assertTrue(proof["proven"])
self.assertFalse(proof["block"])
def test_push_target_mismatch_is_blocked(self):
# Harness assertion (behavior 5, bullet 3).
proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE)
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_local_branch_differs_from_intended_is_blocked(self):
proof = verify_push_target("feat/other", FEATURE, FEATURE)
self.assertTrue(proof["block"])
def test_pushing_a_protected_branch_is_blocked(self):
proof = verify_push_target("master", "master", "master")
self.assertFalse(proof["proven"])
self.assertTrue(proof["block"])
def test_missing_remote_target_fails_closed(self):
proof = verify_push_target(FEATURE, "", FEATURE)
self.assertTrue(proof["block"])
class TestProtectedBranchAccident(unittest.TestCase):
"""Required behavior 4: accidental protected-branch commits must not be
pushed and their repair must be reported."""
def test_feature_branch_commit_is_not_an_accident(self):
result = assess_protected_branch_commit(FEATURE)
self.assertFalse(result["accident"])
self.assertEqual(result["violations"], [])
def test_commit_on_master_is_an_accident_and_must_not_push(self):
result = assess_protected_branch_commit(
"master", pushed=False, repair_reported=True
)
self.assertTrue(result["accident"])
self.assertTrue(result["must_not_push"])
self.assertEqual(result["violations"], [])
self.assertTrue(result["repair_required"])
def test_pushing_the_accident_is_a_violation(self):
result = assess_protected_branch_commit(
"master", pushed=True, repair_reported=True
)
self.assertTrue(any("push" in v.lower() for v in result["violations"]))
def test_silent_repair_is_a_violation(self):
# Harness assertion (behavior 5, bullet 5): the repair path must not
# silently continue without reporting.
result = assess_protected_branch_commit(
"master", pushed=False, repair_reported=False
)
self.assertTrue(any("report" in v.lower() for v in result["violations"]))
class TestCommitPushReport(unittest.TestCase):
"""Acceptance criteria: the final report includes branch proof before
commit and before push, and blocks instead of continuing."""
def _report(self, **overrides):
kwargs = {
"commit_proof": verify_branch_for_commit(FEATURE, FEATURE),
"drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1),
"push_proof": verify_push_target(FEATURE, FEATURE, FEATURE),
"accident": assess_protected_branch_commit(FEATURE),
}
kwargs.update(overrides)
return build_commit_push_report(**kwargs)
def test_fully_proven_report_is_ok(self):
report = self._report()
self.assertEqual(report["status"], "ok")
self.assertTrue(report["branch_proof_before_commit"])
self.assertTrue(report["branch_proof_before_push"])
self.assertFalse(report["drift_detected"])
self.assertEqual(report["violations"], [])
def test_commit_proof_failure_blocks(self):
report = self._report(
commit_proof=verify_branch_for_commit("master", FEATURE)
)
self.assertEqual(report["status"], "blocked")
self.assertFalse(report["branch_proof_before_commit"])
def test_drift_blocks(self):
report = self._report(
drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1)
)
self.assertEqual(report["status"], "blocked")
self.assertTrue(report["drift_detected"])
def test_push_proof_failure_blocks(self):
report = self._report(
push_proof=verify_push_target(FEATURE, "feat/other", FEATURE)
)
self.assertEqual(report["status"], "blocked")
self.assertFalse(report["branch_proof_before_push"])
def test_accident_violations_block(self):
report = self._report(
accident=assess_protected_branch_commit(
"master", pushed=False, repair_reported=False
)
)
self.assertEqual(report["status"], "blocked")
self.assertTrue(report["violations"])
if __name__ == "__main__":
unittest.main()