fix(action): add create-only signed replay

README.md

@@ -95,6 +95,7 @@ See [`examples/agent.yaml`](./examples/agent.yaml) for a complete workflow with
 | `prompt_path` | `.github/prompts` | Path to prompts directory |
 | `prompt_vars` | - | JSON object for template substitution |
 | `github_token` | `github.token` | GitHub token for API access |
+| `replay_new_branch_only` | `false` | Signed replay creates a new remote branch only; existing branches and concurrent name collisions fail without updating the ref. Does not restrict the agent's own GitHub commands. |
 | `opencode_version` | `latest` | OpenCode version to install |
 | `oh_my_opencode_version` | `latest` | oh-my-opencode version to install |
 | `config_json` | - | Full opencode.json content (advanced) |

action.yaml

@@ -99,6 +99,11 @@ inputs:
     required: false
     default: ${{ github.token }}
 
+  replay_new_branch_only:
+    description: Reject existing remote branches during signed commit replay
+    required: false
+    default: "false"
+
   opencode_version:
     description: OpenCode version to install
     required: false
@@ -702,6 +707,7 @@ runs:
       shell: bash
       env:
         GH_TOKEN: ${{ inputs.github_token }}
+        REPLAY_NEW_BRANCH_ONLY: ${{ inputs.replay_new_branch_only }}
       run: |
         python3 "${{ github.action_path }}/scripts/replay_commits.py" \
           "${{ steps.git.outputs.start_sha }}" \

scripts/replay_commits.py

@@ -160,15 +160,16 @@ def get_commit_subject(sha: str) -> str:
 
 
 def get_changed_files() -> list[str]:
-    output = git("diff", "--cached", "--name-only")
-    return [f for f in output.split("\n") if f]
+    output = subprocess.check_output(
+        ["git", "diff", "--cached", "--name-only", "--no-renames", "-z"]
+    )
+    return output.decode("utf-8").split("\0")[:-1]
 
 
-def create_blob(repo: str, file_path: str) -> str | None:
-    path = Path(file_path)
-    if not path.exists():
-        return None
-    content = base64.b64encode(path.read_bytes()).decode("ascii")
+def create_blob(repo: str, object_sha: str) -> str:
+    content = base64.b64encode(
+        subprocess.check_output(["git", "cat-file", "blob", object_sha])
+    ).decode("ascii")
     response = gh_api(
         f"repos/{repo}/git/blobs",
         method="POST",
@@ -177,22 +178,26 @@ def create_blob(repo: str, file_path: str) -> str | None:
     return json.loads(response)["sha"]
 
 
-def get_file_mode(file_path: str) -> str:
-    path = Path(file_path)
-    if path.exists() and os.access(path, os.X_OK):
-        return "100755"
-    return "100644"
-
-
 def create_tree(repo: str, parent_sha: str, files: list[str]) -> str:
     parent_tree = gh_api(f"repos/{repo}/git/commits/{parent_sha}", jq=".tree.sha")
 
     tree_entries: list[dict] = []
     for file_path in files:
-        blob_sha = create_blob(repo, file_path)
+        staged = subprocess.check_output(
+            ["git", "ls-files", "--stage", "-z", "--", f":(literal){file_path}"]
+        ).decode("utf-8")
+        mode, blob_sha = "100644", None
+        for record in staged.split("\0")[:-1]:
+            metadata, path = record.split("\t", 1)
+            if path != file_path:
+                continue
+            mode, object_sha, stage = metadata.split()
+            if stage != "0" or mode not in ("100644", "100755", "120000"):
+                raise ValueError(f"Unsupported index entry: {file_path}")
+            blob_sha = create_blob(repo, object_sha)
         entry: dict = {
             "path": file_path,
-            "mode": get_file_mode(file_path),
+            "mode": mode,
             "type": "blob",
             "sha": blob_sha,
         }
@@ -256,7 +261,18 @@ def replay_commit(repo: str, original_sha: str, parent_sha: str) -> str | None:
     commit_sha = create_commit(repo, message, tree_sha, parent_sha)
     print(f"    Signed: {commit_sha[:7]}")
 
-    git("fetch", "origin", commit_sha, check=False)
+    server = os.environ.get("GITHUB_SERVER_URL", "https://github.com").rstrip("/") + "/"
+    git(
+        "-c",
+        f"http.{server}.extraheader=",
+        "-c",
+        "credential.helper=",
+        "-c",
+        "credential.helper=!gh auth git-credential",
+        "fetch",
+        "origin",
+        commit_sha,
+    )
     git("reset", "--hard", commit_sha)
 
     return commit_sha
@@ -432,6 +448,12 @@ def main() -> int:
     start_branch = sys.argv[2]
     issue_number = sys.argv[3] if len(sys.argv) > 3 else ""
 
+    policy = os.environ.get("REPLAY_NEW_BRANCH_ONLY", "false")
+    if policy not in ("true", "false"):
+        print("REPLAY_NEW_BRANCH_ONLY must be true or false", file=sys.stderr)
+        return 1
+    new_branch_only = policy == "true"
+
     repo = os.environ.get("GITHUB_REPOSITORY")
     if not repo:
         print("GITHUB_REPOSITORY not set", file=sys.stderr)
@@ -469,6 +491,9 @@ def main() -> int:
         return 1
 
     remote_sha = get_remote_branch_sha(repo, current_branch)
+    if new_branch_only and remote_sha:
+        print(f"Remote branch already exists: {current_branch}", file=sys.stderr)
+        return 1
     if remote_sha:
         print(f"Remote branch exists at {remote_sha[:7]}")
         git("fetch", "origin", current_branch, check=False)
@@ -497,7 +522,8 @@ def main() -> int:
         if new_sha:
             parent_sha = new_sha
 
-    is_new_branch = not branch_exists_on_remote(repo, current_branch)
+    # POST creates atomically and rejects a branch claimed during replay.
+    is_new_branch = new_branch_only or not branch_exists_on_remote(repo, current_branch)
 
     if is_new_branch:
         print(f"\nCreating ref {current_branch} -> {parent_sha[:7]}")

tests/test_replay_commits.py

@@ -1,12 +1,169 @@
+import base64
+import json
+import subprocess
 import sys
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import Mock, patch
+
+import pytest
 
 sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
 
+import replay_commits
 from replay_commits import body_has_issue_reference, is_commit_signed
 
 
+@pytest.fixture
+def replay(monkeypatch):
+    monkeypatch.setenv("GITHUB_REPOSITORY", "owner/repo")
+    monkeypatch.setenv("REPLAY_NEW_BRANCH_ONLY", "true")
+    monkeypatch.setattr(sys, "argv", ["replay_commits.py", "start", "main"])
+    results = {
+        "get_current_branch": "feature",
+        "get_default_branch": "main",
+        "get_commits": ["local"],
+        "get_remote_branch_sha": None,
+        "branch_exists_on_remote": False,
+        "replay_commit": "signed",
+        "create_pull_request": "https://github.com/owner/repo/pull/1",
+        "git": "",
+        "gh_api": "",
+    }
+    mocks = {name: Mock(return_value=value) for name, value in results.items()}
+    for name, mock in mocks.items():
+        monkeypatch.setattr(replay_commits, name, mock)
+    return mocks
+
+
+class TestNewBranchOnly:
+    def test_creates_new_ref(self, replay):
+        assert replay_commits.main() == 0
+        replay["gh_api"].assert_called_once_with(
+            "repos/owner/repo/git/refs",
+            method="POST",
+            input_data={"ref": "refs/heads/feature", "sha": "signed"},
+        )
+
+    def test_rejects_existing_branch_before_replay(self, replay):
+        replay["get_remote_branch_sha"].return_value = "existing"
+        assert replay_commits.main() == 1
+        replay["git"].assert_not_called()
+        replay["replay_commit"].assert_not_called()
+        replay["gh_api"].assert_not_called()
+
+    def test_branch_created_during_replay_never_updates_ref(self, replay):
+        replay["branch_exists_on_remote"].return_value = True
+        replay["gh_api"].side_effect = subprocess.CalledProcessError(1, "gh")
+        with pytest.raises(subprocess.CalledProcessError):
+            replay_commits.main()
+        assert replay["gh_api"].call_args.kwargs["method"] == "POST"
+        replay["gh_api"].assert_called_once()
+        replay["create_pull_request"].assert_not_called()
+
+    @pytest.mark.parametrize("value", ["", "tru", "FALSE"])
+    def test_invalid_policy_fails_closed(self, replay, monkeypatch, value):
+        monkeypatch.setenv("REPLAY_NEW_BRANCH_ONLY", value)
+        assert replay_commits.main() == 1
+        replay["git"].assert_not_called()
+        replay["gh_api"].assert_not_called()
+
+    def test_default_preserves_existing_branch_mode(self, replay, monkeypatch):
+        monkeypatch.delenv("REPLAY_NEW_BRANCH_ONLY")
+        replay["branch_exists_on_remote"].return_value = True
+        assert replay_commits.main() == 0
+        assert replay["gh_api"].call_args.kwargs["method"] == "PATCH"
+
+
+class TestReplayFetch:
+    @patch("replay_commits.get_commit_subject", return_value="test")
+    @patch("replay_commits.get_commit_message", return_value="test")
+    @patch("replay_commits.get_changed_files", return_value=["file"])
+    @patch("replay_commits.create_tree", return_value="tree")
+    @patch("replay_commits.create_commit", return_value="signed")
+    @patch("replay_commits.git")
+    def test_fetch_auth_and_failure(self, git, *_):
+        def command(*args, **kwargs):
+            if "fetch" in args:
+                assert args == (
+                    "-c",
+                    "http.https://github.com/.extraheader=",
+                    "-c",
+                    "credential.helper=",
+                    "-c",
+                    "credential.helper=!gh auth git-credential",
+                    "fetch",
+                    "origin",
+                    "signed",
+                )
+                assert kwargs.get("check", True)
+                raise subprocess.CalledProcessError(1, "git fetch")
+            return ""
+
+        git.side_effect = command
+        with pytest.raises(subprocess.CalledProcessError):
+            replay_commits.replay_commit("owner/repo", "local", "start")
+        assert not any(
+            call.args == ("reset", "--hard", "signed") for call in git.call_args_list
+        )
+
+
+class TestReplayTree:
+    @pytest.mark.parametrize("link_name", ["link", " link\n\r"])
+    def test_uses_index_bytes_and_modes(self, tmp_path, monkeypatch, link_name):
+        monkeypatch.chdir(tmp_path)
+        monkeypatch.setenv("GIT_CONFIG_GLOBAL", "/dev/null")
+        monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
+        replay_commits.git("init", "-q")
+        target = tmp_path / "outside-marker"
+        target.write_text("DO_NOT_PUBLISH")
+        paths = {
+            link_name: ("120000", str(target).encode()),
+            "dangling": ("120000", b"missing"),
+            "executable": ("100755", b"staged\x00\xff\n"),
+            "dir/regular": ("100644", b" regular\n"),
+        }
+        for name, (mode, content) in paths.items():
+            path = tmp_path / name
+            path.parent.mkdir(exist_ok=True)
+            if mode == "120000":
+                path.symlink_to(content.decode())
+            else:
+                path.write_bytes(content)
+                path.chmod(0o755 if mode == "100755" else 0o644)
+            replay_commits.git("add", "--", name)
+        (tmp_path / "executable").write_text("UNSTAGED_DO_NOT_PUBLISH")
+        (tmp_path / "executable").chmod(0o644)
+        files = replay_commits.get_changed_files()
+        assert set(files) == set(paths)
+        blobs, tree = [], []
+
+        def api(endpoint, **kwargs):
+            data = kwargs.get("input_data", {})
+            if endpoint.endswith("/blobs"):
+                blobs.append(base64.b64decode(data["content"]))
+                return json.dumps({"sha": str(len(blobs))})
+            if endpoint.endswith("/trees"):
+                tree.extend(data["tree"])
+                return '{"sha":"tree"}'
+            return "parent-tree"
+
+        monkeypatch.setattr(replay_commits, "gh_api", api)
+        assert (
+            replay_commits.create_tree(
+                "owner/repo", "parent", files + ["deleted", "dir"]
+            )
+            == "tree"
+        )
+        for entry in tree:
+            if entry["path"] in ("deleted", "dir"):
+                assert entry["sha"] is None
+            else:
+                mode, content = paths[entry["path"]]
+                assert entry["mode"] == mode
+                assert blobs[int(entry["sha"]) - 1] == content
+        assert b"DO_NOT_PUBLISH" not in blobs
+
+
 class TestIsCommitSigned:
     @patch("replay_commits.gh_api")
     def test_signed_commit_returns_true(self, mock_gh_api):