Merge abbb10dd00 into 421290d066

agent/dashboard/workflow_approval.py

@@ -44,6 +44,33 @@ async def workflow_push_approved(thread_id: str, fingerprint: str) -> bool:
     return approvals.get(fingerprint, {}).get("status") == WORKFLOW_APPROVAL_APPROVED
 
 
+async def workflow_push_rejected(thread_id: str, fingerprint: str) -> bool:
+    approvals = await get_workflow_push_approvals(thread_id)
+    return approvals.get(fingerprint, {}).get("status") == WORKFLOW_APPROVAL_REJECTED
+
+
+async def find_workflow_push_approval(
+    thread_id: str,
+    *,
+    repo: str,
+    branch: str,
+    files: list[str],
+) -> dict[str, Any] | None:
+    """Return the most recent approved record matching identity-level keys, if any."""
+    approvals = await get_workflow_push_approvals(thread_id)
+    identity = (repo, branch, tuple(sorted(files)))
+    matches = [
+        r
+        for r in approvals.values()
+        if r.get("status") == WORKFLOW_APPROVAL_APPROVED
+        and (r.get("repo"), r.get("branch"), tuple(sorted(r.get("files", [])))) == identity
+    ]
+    if not matches:
+        return None
+    matches.sort(key=lambda r: str(r.get("decided_at", "")), reverse=True)
+    return matches[0]
+
+
 async def ensure_workflow_push_pending(
     thread_id: str,
     *,

agent/middleware/workflow_push_guard.py

@@ -22,8 +22,10 @@ from langgraph.types import Command
 
 from ..dashboard.workflow_approval import (
     ensure_workflow_push_pending,
+    find_workflow_push_approval,
     mark_workflow_push_notified,
     workflow_push_approved,
+    workflow_push_rejected,
 )
 from ..tools.slack_thread_reply import build_workflow_approval_blocks
 from ..utils.github_app import (
@@ -42,6 +44,19 @@ _SHELL_OPERATORS = {";", "|", "||", "&"}
 _REF_NAME = re.compile(r"^[A-Za-z0-9._/@+-]+$")
 _GIT_OBJECT_ID = re.compile(r"^[0-9a-fA-F]{40,64}$")
 _UNSAFE_RAW_COMMAND = re.compile(r"[;|`$<>\n\r]")
+# Command wrappers that can prefix a `git` invocation (e.g. `command git push`,
+# `env FOO=bar git push`, `/usr/bin/git push`). Recognized so a wrapped or path-qualified
+# git push cannot slip past the guard as "not a git command".
+_GIT_WRAPPERS = {"command", "env", "nice", "sudo", "stdbuf", "nohup", "time", "ionice", "setsid"}
+
+
+class _BlockedGitPush:
+    """Sentinel returned by the parser when a git push command is unsafe or unrecognized."""
+
+    __slots__ = ("reason",)
+
+    def __init__(self, reason: str) -> None:
+        self.reason = reason
 
 
 @dataclass(frozen=True)
@@ -58,13 +73,15 @@ class WorkflowPushChange:
     fingerprint: str
     repo: str
     branch: str
-    base_sha: str
-    head_sha: str
     files: list[str]
+    head_sha: str
     remote: str
     local_ref: str
     remote_ref: str
     fixed_command: str
+    base_sha: str = ""
+    blocked: bool = False
+    blocked_reason: str = ""
 
 
 @dataclass(frozen=True)
@@ -143,52 +160,80 @@ def _response_ok(response: Any) -> bool:
     return True
 
 
-def _parse_git_push(command: str) -> ParsedGitPush | None:
+def _parse_git_push(command: str) -> ParsedGitPush | _BlockedGitPush | None:
     stripped = command.strip()
     if _UNSAFE_RAW_COMMAND.search(stripped) or "&" in stripped.replace("&&", ""):
-        return None
+        return _BlockedGitPush("unsafe shell characters in git push command")
     try:
         tokens = shlex.split(stripped)
     except ValueError:
-        return None
+        return _BlockedGitPush("unparseable git push command")
     if not tokens:
         return None
 
     if len(tokens) >= 4 and tokens[0] == "cd" and tokens[2] == "&&":
         if any(token in _SHELL_OPERATORS or token == "&&" for token in tokens[3:]):
-            return None
+            return _BlockedGitPush("chained shell commands in git push")
         return _parse_git_tokens(tokens[3:], repo_dir=tokens[1])
 
     if any(token in _SHELL_OPERATORS or token == "&&" for token in tokens):
-        return None
+        return _BlockedGitPush("chained shell commands")
     return _parse_git_tokens(tokens, repo_dir=None)
 
 
-def _parse_git_tokens(tokens: list[str], *, repo_dir: str | None) -> ParsedGitPush | None:
-    if not tokens or tokens[0] != "git":
+def _parse_git_tokens(
+    tokens: list[str], *, repo_dir: str | None
+) -> ParsedGitPush | _BlockedGitPush | None:
+    """Parse tokens into a supported git push, a block sentinel, or None.
+
+    Recognizes bareword `git`, path invocations (`/usr/bin/git`), and wrapper-prefixed
+    forms (`command`/`env`/`nice`/... git). Anything git-push-shaped that cannot be reduced
+    to `git [-C <dir>] push [-u] origin <refspec>` fails closed to a block rather than a
+    silent pass-through. Non-push git commands and non-git commands return None.
+    """
+    git_idx = next((idx for idx, tok in enumerate(tokens) if tok.rsplit("/", 1)[-1] == "git"), None)
+    if git_idx is None:
         return None
-    i = 1
-    while i < len(tokens) and tokens[i] != "push":
-        if tokens[i] == "-C" and i + 1 < len(tokens):
-            repo_dir = tokens[i + 1]
+    git_args = tokens[git_idx + 1 :]
+    if "push" not in git_args:
+        # A non-push git command (e.g. `git status`, `git pull`).
+        return None
+    # Everything before the `git` executable must be a benign wrapper prefix (a known
+    # wrapper, an option flag, or a NAME=VALUE assignment). An unrecognized leading token
+    # (a value-taking wrapper option like the `10` in `nice -n 10`, an unknown wrapper, or
+    # `git` used as an argument to another program) is ambiguous, so fail closed.
+    if any(
+        not (tok in _GIT_WRAPPERS or tok.startswith("-") or "=" in tok) for tok in tokens[:git_idx]
+    ):
+        return _BlockedGitPush(
+            "unrecognized wrapper before `git push`; use `git push origin <branch>`"
+        )
+    i = 0
+    while i < len(git_args) and git_args[i] != "push":
+        if git_args[i] == "-C" and i + 1 < len(git_args):
+            repo_dir = git_args[i + 1]
             i += 2
             continue
-        return None
-    if i >= len(tokens) or tokens[i] != "push":
-        return None
-    return _parse_push_args(tokens[i + 1 :], repo_dir=repo_dir)
+        # A git push carrying unsupported global options (e.g. `git -c k=v push`,
+        # `git --git-dir=.git push`). We cannot safely normalize it, so fail closed.
+        return _BlockedGitPush(
+            "unsupported git options before `push`; use `git push origin <branch>`"
+        )
+    return _parse_push_args(git_args[i + 1 :], repo_dir=repo_dir)
 
 
-def _parse_push_args(tokens: list[str], *, repo_dir: str | None) -> ParsedGitPush | None:
+def _parse_push_args(
+    tokens: list[str], *, repo_dir: str | None
+) -> ParsedGitPush | _BlockedGitPush | None:
     set_upstream = False
     while tokens and tokens[0] in {"-u", "--set-upstream"}:
         set_upstream = True
         tokens = tokens[1:]
     if len(tokens) != 2 or tokens[0] != "origin":
-        return None
+        return _BlockedGitPush("unrecognized or unsafe git push arguments")
     parsed = _parse_refspec(tokens[1])
     if parsed is None:
-        return None
+        return _BlockedGitPush("unrecognized or unsafe git push refspec")
     local_ref, remote_ref = parsed
     return ParsedGitPush(
         repo_dir=repo_dir,
@@ -282,6 +327,95 @@ def _run_coroutine_sync(coro: Awaitable[ToolMessage | Command]) -> ToolMessage |
     return value
 
 
+def _parse_ls_tree(output: str) -> list[tuple[str, str]]:
+    """Parse `git ls-tree -r` output into a sorted list of (sha, path) tuples."""
+    entries: list[tuple[str, str]] = []
+    for line in output.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        # Format: "<mode> <type> <sha>\t<path>"
+        meta, _, path = line.partition("\t")
+        if not path:
+            continue
+        parts = meta.split()
+        if len(parts) < 3:
+            continue
+        entries.append((parts[2], path))
+    entries.sort(key=lambda item: item[1])
+    return entries
+
+
+def _workflow_tree_at_ref(
+    backend: Any, repo_dir: str | None, ref: str
+) -> tuple[list[tuple[str, str]], str] | None:
+    """Return the sorted workflow tree entries and a stable hash for the given ref.
+
+    Returns an empty list when the ref has no workflow files, so callers can detect
+    additions and deletions against a base ref.
+    """
+    ls_tree = _run_git(backend, repo_dir, f"ls-tree -r {shlex.quote(ref)} -- .github/workflows")
+    if not ls_tree.ok:
+        return None
+    entries = _parse_ls_tree(ls_tree.output)
+    content_hash = _fingerprint({"tree": entries})
+    return entries, content_hash
+
+
+def _fetch_remote_base(backend: Any, repo_dir: str | None, remote: str, branch: str) -> str | None:
+    """Fetch the base ref from the exact URL the push will target and return a local alias.
+
+    The base is fetched from the remote's effective *push* URL (`git remote get-url --push`),
+    not the remote name, because git honors a separate `pushurl`: an untrusted sandbox can
+    point the fetch URL at attacker/local content while the push still lands on the real
+    repo. Fetching the base from the push URL keeps the base and the push destination the
+    same authenticated repo, so a split cannot hide a workflow change. Uses `FETCH_HEAD`
+    (freshly fetched), never a local `refs/remotes/origin/*` ref the sandbox could rewrite.
+
+    Returns None (which makes the caller require approval) if the push destination is
+    ambiguous or rewritten: multiple `pushurl` entries (git pushes to ALL of them, so a
+    single base cannot represent the destination) or any `insteadOf`/`pushInsteadOf` URL
+    rewrite the sandbox could use to make the inspected URL differ from the push target.
+    """
+    rewrites = _run_git(
+        backend, repo_dir, "config --get-regexp " + shlex.quote(r"url\..*\.(push)?insteadof")
+    )
+    if rewrites.ok and rewrites.output.strip():
+        # A URL rewrite means the inspected push URL may not be the real destination.
+        return None
+
+    push_url_res = _run_git(backend, repo_dir, f"remote get-url --push --all {shlex.quote(remote)}")
+    push_urls = (
+        [line.strip() for line in push_url_res.output.splitlines() if line.strip()]
+        if push_url_res.ok
+        else []
+    )
+    # `git push` sends to every configured push URL; if there is not exactly one we cannot
+    # represent the destination with a single base, so fail closed.
+    if len(push_urls) != 1:
+        return None
+    push_url = push_urls[0]
+
+    fetch = _run_git(backend, repo_dir, f"fetch {shlex.quote(push_url)} {shlex.quote(branch)}")
+    if fetch.ok:
+        return "FETCH_HEAD"
+
+    default = _run_git(backend, repo_dir, f"ls-remote --symref {shlex.quote(push_url)} HEAD")
+    default_branch = ""
+    if default.ok:
+        default_branch = _first_line(default.output).removeprefix("ref: refs/heads/").split("\t")[0]
+
+    for fallback in (default_branch, "dev", "main"):
+        if not fallback or fallback == branch:
+            continue
+        fetch = _run_git(
+            backend, repo_dir, f"fetch {shlex.quote(push_url)} {shlex.quote(fallback)}"
+        )
+        if fetch.ok:
+            return "FETCH_HEAD"
+    return None
+
+
 def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPushChange | None:
     root_result = _run_git(backend, parsed.repo_dir, "rev-parse --show-toplevel")
     if not root_result.ok:
@@ -302,42 +436,34 @@ def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPu
     if not head or not _GIT_OBJECT_ID.fullmatch(head):
         return None
 
-    remote_branch = f"refs/remotes/{parsed.remote}/{parsed.remote_ref}"
-    remote_branch_exists = _run_git(
-        backend, root, f"rev-parse --verify {shlex.quote(remote_branch)}"
-    )
-    if remote_branch_exists.ok and _first_line(remote_branch_exists.output):
-        base_ref = remote_branch
-        range_expr = f"{shlex.quote(base_ref)}..{shlex.quote(head)}"
-        base_sha = _first_line(_run_git(backend, root, f"rev-parse {shlex.quote(base_ref)}").output)
-    else:
-        origin_head = _run_git(backend, root, "symbolic-ref --short refs/remotes/origin/HEAD")
-        base_ref = _first_line(origin_head.output) if origin_head.ok else "origin/main"
-        range_expr = f"{shlex.quote(base_ref)}...{shlex.quote(head)}"
-        base_sha = _first_line(
-            _run_git(
-                backend, root, f"merge-base {shlex.quote(head)} {shlex.quote(base_ref)}"
-            ).output
+    head_tree = _workflow_tree_at_ref(backend, root, head)
+    if head_tree is None:
+        # The workflow tree at head could not be read (an empty tree returns ([], hash), so
+        # None means the `ls-tree` read genuinely failed). Fail closed rather than skipping
+        # the guard, mirroring the base-read path.
+        return WorkflowPushChange(
+            fingerprint="",
+            repo="",
+            branch=branch_name,
+            files=[],
+            head_sha=head,
+            remote=parsed.remote,
+            local_ref=parsed.local_ref,
+            remote_ref=parsed.remote_ref,
+            fixed_command="",
+            blocked=True,
+            blocked_reason="could not read the workflow tree at the pushed head; blocking to be safe",
         )
 
-    names = _run_git(
-        backend,
-        root,
-        f"diff --name-only --diff-filter=ACMRTD {range_expr} -- .github/workflows",
-    )
-    if not names.ok:
-        return None
-    files = sorted(
-        line.strip()
-        for line in names.output.splitlines()
-        if line.strip().startswith(_WORKFLOW_PREFIX)
-    )
-    if not files:
-        return None
+    base_ref = _fetch_remote_base(backend, root, parsed.remote, branch_name)
+    if base_ref is not None:
+        base_tree = _workflow_tree_at_ref(backend, root, base_ref)
+        if base_tree is not None and head_tree[0] == base_tree[0]:
+            # No workflow change against the trusted remote base.
+            return None
 
-    diff = _run_git(backend, root, f"diff --binary --full-index {range_expr} -- .github/workflows")
-    if not diff.ok or not diff.output:
-        return None
+    head_entries, head_content_hash = head_tree
+    files = sorted({path for _sha, path in head_entries})
 
     remote = _run_git(backend, root, "config --get remote.origin.url")
     repo = _normalize_remote(_first_line(remote.output)) if remote.ok else ""
@@ -347,25 +473,18 @@ def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPu
         fixed_args.append("--set-upstream")
     fixed_args.extend([parsed.remote, fixed_refspec])
     fixed_command = _git_command(root, " ".join(shlex.quote(arg) for arg in fixed_args))
-    payload = {
+    content_payload = {
         "repo": repo,
         "branch": branch_name,
-        "base_sha": base_sha,
-        "head_sha": head,
         "files": files,
-        "diff": diff.output,
-        "remote": parsed.remote,
-        "local_ref": parsed.local_ref,
-        "remote_ref": parsed.remote_ref,
-        "fixed_refspec": fixed_refspec,
+        "content_hash": head_content_hash,
     }
     return WorkflowPushChange(
-        fingerprint=_fingerprint(payload),
+        fingerprint=_fingerprint(content_payload),
         repo=repo,
         branch=branch_name,
-        base_sha=base_sha,
-        head_sha=head,
         files=files,
+        head_sha=head,
         remote=parsed.remote,
         local_ref=parsed.local_ref,
         remote_ref=parsed.remote_ref,
@@ -373,17 +492,38 @@ def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPu
     )
 
 
-def _blocked_message(change: WorkflowPushChange, *, already_rejected: bool = False) -> ToolMessage:
+def _blocked_message(
+    change: WorkflowPushChange,
+    *,
+    already_rejected: bool = False,
+    stale: bool = False,
+    blocked: bool = False,
+) -> ToolMessage:
     status = "rejected" if already_rejected else "approval_required"
-    content = {
-        "status": "error",
-        "error_type": "WorkflowPushApprovalRequired",
-        "error": (
+    if blocked:
+        error = change.blocked_reason or "This git push command is not recognized as safe."
+        error_type = "WorkflowPushBlocked"
+    elif stale:
+        error = (
+            "This git push includes GitHub workflow file changes. A previous approval "
+            "exists for the same branch and workflow files, but the workflow content "
+            "at the pushed head has changed since that approval (for example, a rebase "
+            "that changed the workflow files or an amend that edited them). The thread "
+            "owner must re-approve the new fingerprint before Open SWE can push it."
+        )
+        error_type = "WorkflowPushApprovalRequired"
+    else:
+        error = (
             "This git push includes GitHub workflow file changes and requires human "
             "approval before Open SWE can push it. Retry the same standalone git push "
-            "after the thread owner approves the workflow diff."
-        ),
-        "workflow_approval_status": status,
+            "after the thread owner approves the workflow files."
+        )
+        error_type = "WorkflowPushApprovalRequired"
+    content = {
+        "status": "error",
+        "error_type": error_type,
+        "error": error,
+        "workflow_approval_status": status if not blocked else "blocked",
         "fingerprint": change.fingerprint,
         "files": change.files,
         "repo": change.repo,
@@ -414,10 +554,12 @@ def _approval_slack_message(change: WorkflowPushChange) -> str:
     branch = change.branch or "the current branch"
     return (
         "*Workflow file approval required*\n"
-        f"Open SWE is trying to push changes to GitHub workflow files in `{repo}` on `{branch}`.\n\n"
-        f"*Files:*\n{files}\n\n"
+        f"Open SWE is trying to push the workflow files below to `{repo}` on `{branch}`.\n\n"
+        f"*Files at the pushed head:*\n{files}\n\n"
         f"*Fingerprint:* `{change.fingerprint}`\n\n"
-        "Approve only if this exact workflow diff is expected. If the workflow files change, "
+        "Approval covers the exact workflow files and content listed above at the pushed head, "
+        "including future rebases or amends that replay the same workflow tree. If the set of "
+        "workflow files, the branch, or the workflow-file content at the pushed head changes, "
         "a new fingerprint will be required."
     )
 
@@ -455,6 +597,23 @@ async def _approval_state(request: ToolCallRequest, change: WorkflowPushChange)
     try:
         if await workflow_push_approved(thread_id, change.fingerprint):
             return "approved"
+        if await workflow_push_rejected(thread_id, change.fingerprint):
+            return "rejected"
+
+        # If the exact identity fingerprint is not approved, check whether a prior
+        # approval covers the same (repo, branch, files) identity. If so, the workflow
+        # tree at the pushed head changed underneath the prior approval (rebase/amend
+        # that edited workflow files), so we surface a loud re-approval message rather
+        # than a fresh silent pending record.
+        prior = await find_workflow_push_approval(
+            thread_id,
+            repo=change.repo,
+            branch=change.branch,
+            files=change.files,
+        )
+        if prior is not None:
+            return "stale_approval"
+
         record, _created = await ensure_workflow_push_pending(
             thread_id,
             fingerprint=change.fingerprint,
@@ -527,6 +686,20 @@ class WorkflowPushGuardMiddleware(AgentMiddleware):
         parsed = _parse_git_push(command)
         if parsed is None:
             return None
+        if isinstance(parsed, _BlockedGitPush):
+            return WorkflowPushChange(
+                fingerprint="",
+                repo="",
+                branch="",
+                files=[],
+                head_sha="",
+                remote="origin",
+                local_ref="",
+                remote_ref="",
+                fixed_command="",
+                blocked=True,
+                blocked_reason=parsed.reason,
+            )
         backend = _backend(_thread_id(request))
         if backend is None:
             return None
@@ -538,13 +711,31 @@ class WorkflowPushGuardMiddleware(AgentMiddleware):
         handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
         change: WorkflowPushChange,
     ) -> ToolMessage | Command:
+        if change.blocked:
+            return _tool_message_for_request(_blocked_message(change, blocked=True), request)
         thread_id = _thread_id(request)
         state = await _approval_state(request, change)
         if state == "approved" and thread_id:
             safe_request = _override_execute_command(request, change.fixed_command)
             return await _run_with_workflow_token(thread_id, request, lambda: handler(safe_request))
+        if state == "stale_approval":
+            record, _created = await ensure_workflow_push_pending(
+                thread_id,
+                fingerprint=change.fingerprint,
+                repo=change.repo,
+                branch=change.branch,
+                base_sha=change.base_sha,
+                head_sha=change.head_sha,
+                files=change.files,
+            )
+            await _post_slack_approval_if_needed(request, change, record)
         return _tool_message_for_request(
-            _blocked_message(change, already_rejected=state == "rejected"), request
+            _blocked_message(
+                change,
+                already_rejected=state == "rejected",
+                stale=state == "stale_approval",
+            ),
+            request,
         )
 
     def wrap_tool_call(

agent/prompt.py

@@ -317,7 +317,7 @@ Steps, in order:
 
 **IMPORTANT: If `git push` or `gh` returns "403", "Permission denied", or another permanent authorization failure, do not retry. Report the error to the user immediately and stop.**
 
-**IMPORTANT: Workflow files (`.github/workflows/`) may be changed only when explicitly requested. Any push that includes workflow-file changes requires human approval of the exact workflow diff fingerprint before it can proceed — do not attempt to bypass it.**
+**IMPORTANT: Workflow files (`.github/workflows/`) may be changed only when explicitly requested. Any push that includes workflow files at the pushed head requires human approval before it can proceed. Approval is keyed to the repo, branch, and the exact workflow files and content present at the pushed head, so rebases or amends that replay the same workflow tree do not require a fresh approval; changing the branch, the set of workflow files, or the workflow-file content at the pushed head does require a new approval. Do not attempt to bypass it.**
 
 4. **Notify the source** immediately after pushing and, when applicable, PR creation/update succeeds. Include a brief summary plus the PR link or branch URL:
    - Linear-triggered: use `linear_comment` with an `@mention` of the user who triggered the task

tests/test_workflow_push_guard.py

@@ -20,25 +20,83 @@ class _Response:
 class _Backend:
     id = "sandbox-id"
 
-    def __init__(self, *, workflow_files: str = ".github/workflows/ci.yml") -> None:
+    def __init__(
+        self,
+        *,
+        workflow_files: str = ".github/workflows/ci.yml",
+        tree_entries: dict[str, str] | None = None,
+        base_tree_entries: dict[str, str] | None = None,
+        remote_branches: set[str] | None = None,
+        default_branch: str = "main",
+        push_url: str = "https://github.com/langchain-ai/open-swe.git",
+        push_urls: list[str] | None = None,
+        url_rewrites: str = "",
+        ls_tree_head_ok: bool = True,
+    ) -> None:
         self.workflow_files = workflow_files
+        self.tree_entries = tree_entries
+        # Default base tree is empty so any head workflow file is treated as a change.
+        self.base_tree_entries = base_tree_entries if base_tree_entries is not None else {}
+        self.remote_branches = remote_branches or {"feature"}
+        self.default_branch = default_branch
+        self.push_url = push_url
+        # Effective push URLs (`git remote get-url --push --all`); defaults to [push_url].
+        self.push_urls = push_urls if push_urls is not None else ([push_url] if push_url else [])
+        self.url_rewrites = url_rewrites
+        self.ls_tree_head_ok = ls_tree_head_ok
         self.commands: list[str] = []
         self.head = "a" * 40
+        self.base_sha = "b" * 40
+
+    def _head_tree(self) -> str:
+        if self.tree_entries is not None:
+            return "".join(
+                f"100644 blob {sha}\t{path}\n" for path, sha in sorted(self.tree_entries.items())
+            )
+        files = [path for path in self.workflow_files.split("\n") if path.strip()]
+        return "".join(f"100644 blob {self.head}\t{path}\n" for path in files)
+
+    def _base_tree(self) -> str:
+        if self.base_tree_entries is not None:
+            return "".join(
+                f"100644 blob {sha}\t{path}\n"
+                for path, sha in sorted(self.base_tree_entries.items())
+            )
+        return self._head_tree()
+
+    def _fetch_branch(self, command: str) -> _Response:
+        # Format: "git [-C /repo] fetch <url> <branch>"
+        idx = command.find(" fetch ")
+        if idx == -1:
+            return _Response("", 1)
+        parts = command[idx + len(" fetch ") :].split()
+        if len(parts) < 2:
+            return _Response("", 1)
+        branch = parts[-1]
+        if branch in self.remote_branches:
+            return _Response("")
+        return _Response("", 1)
 
     def execute(self, command: str, *, timeout: int | None = None) -> _Response:
         self.commands.append(command)
         if "rev-parse --show-toplevel" in command:
             return _Response("/repo\n")
-        if "rev-parse --verify refs/remotes/origin/feature" in command:
-            return _Response("", 1)
-        if "symbolic-ref --short refs/remotes/origin/HEAD" in command:
-            return _Response("origin/main\n")
-        if f"merge-base {self.head} origin/main" in command:
-            return _Response("base-sha\n")
-        if "diff --name-only" in command:
-            return _Response(f"{self.workflow_files}\n" if self.workflow_files else "")
-        if "diff --binary --full-index" in command:
-            return _Response("diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n")
+        if "config --get-regexp" in command:
+            return _Response(f"{self.url_rewrites}\n") if self.url_rewrites else _Response("", 1)
+        if "remote get-url --push" in command:
+            if not self.push_urls:
+                return _Response("", 1)
+            return _Response("".join(f"{url}\n" for url in self.push_urls))
+        if "ls-tree -r" in command:
+            if "FETCH_HEAD" in command:
+                return _Response(self._base_tree())
+            return _Response(self._head_tree()) if self.ls_tree_head_ok else _Response("", 1)
+        if " fetch " in command:
+            return self._fetch_branch(command)
+        if "ls-remote --symref" in command:
+            return _Response(
+                f"ref: refs/heads/{self.default_branch}\tHEAD\n{self.base_sha}\tHEAD\n"
+            )
         if "config --get remote.origin.url" in command:
             return _Response("git@github.com:langchain-ai/open-swe.git\n")
         if "rev-parse --abbrev-ref HEAD" in command:
@@ -93,11 +151,25 @@ def test_parse_git_push_supports_git_c_and_cd() -> None:
         remote_ref="feature",
         set_upstream=True,
     )
-    assert guard._parse_git_push("git status && git push") is None
-    assert guard._parse_git_push("git push origin feature; git push origin evil:feature") is None
 
 
-def test_workflow_change_for_push_fingerprints_workflow_diff() -> None:
+def test_parse_git_push_blocks_unsafe_and_unrecognized_forms() -> None:
+    assert isinstance(guard._parse_git_push("git status && git push"), guard._BlockedGitPush)
+    assert isinstance(
+        guard._parse_git_push("git push origin feature; git push origin evil:feature"),
+        guard._BlockedGitPush,
+    )
+    assert isinstance(
+        guard._parse_git_push("git push --force origin feature"), guard._BlockedGitPush
+    )
+    assert isinstance(guard._parse_git_push("git push origin"), guard._BlockedGitPush)
+    assert isinstance(
+        guard._parse_git_push("git push origin HEAD~1:feature"), guard._BlockedGitPush
+    )
+    assert guard._parse_git_push("git status") is None
+
+
+def test_workflow_change_for_push_fingerprints_head_workflow_tree() -> None:
     backend = _Backend()
     change = guard._workflow_change_for_push(
         backend,
@@ -154,6 +226,12 @@ async def test_unapproved_workflow_push_blocks_and_posts_slack(
     async def fake_approved(thread_id: str, fingerprint: str) -> bool:
         return False
 
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
     async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
         return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
 
@@ -169,6 +247,8 @@ async def test_unapproved_workflow_push_blocks_and_posts_slack(
         posted["notified"] = fingerprint
 
     monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
     monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
     monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
     monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
@@ -205,7 +285,11 @@ async def test_approved_workflow_push_elevates_and_restores(
         refreshed.append(dict(permissions))
         return True
 
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
     monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
     monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh)
 
     pushed_command = ""
@@ -241,7 +325,11 @@ async def test_workflow_push_restoration_falls_back_when_actions_read_unavailabl
         refreshed.append(dict(permissions))
         return "actions" not in permissions
 
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
     monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
     monkeypatch.setattr(guard, "refresh_proxy_token", fake_refresh)
 
     async def handler(_request: Any) -> ToolMessage:
@@ -256,7 +344,13 @@ async def test_workflow_push_restoration_falls_back_when_actions_read_unavailabl
 
 
 async def test_non_workflow_push_runs_without_approval(monkeypatch: pytest.MonkeyPatch) -> None:
-    guard.SANDBOX_BACKENDS["thread-1"] = _Backend(workflow_files="")
+    # Head contains workflow files, but they are identical to the trusted remote base,
+    # so no approval is required for a code-only change.
+    backend = _Backend(
+        tree_entries={".github/workflows/ci.yml": "blob-sha-1"},
+        base_tree_entries={".github/workflows/ci.yml": "blob-sha-1"},
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
     called = False
 
     async def fail_approval(*args: Any, **kwargs: Any) -> bool:
@@ -274,6 +368,270 @@ async def test_non_workflow_push_runs_without_approval(monkeypatch: pytest.Monke
     assert called is True
     assert isinstance(result, ToolMessage)
     assert result.content == "pushed"
+    # The base is fetched from the effective push URL, not the `origin` remote name.
+    assert any("remote get-url --push" in cmd for cmd in backend.commands)
+    assert any(f" fetch {backend.push_url}" in cmd for cmd in backend.commands)
+
+
+async def test_stale_workflow_approval_is_loud_and_blocks(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    guard.SANDBOX_BACKENDS["thread-1"] = _Backend()
+    posted: dict[str, Any] = {}
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return {"fingerprint": "old-fp", "status": "approved", "decided_at": "2024-01-01T00:00:00"}
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(
+        channel_id: str, thread_ts: str, message: str, **kwargs: Any
+    ) -> tuple[str, None]:
+        posted.update(
+            channel_id=channel_id, thread_ts=thread_ts, message=message, blocks=kwargs["blocks"]
+        )
+        return "1700000000.000300", None
+
+    async def fake_notified(thread_id: str, fingerprint: str) -> None:
+        posted["notified"] = fingerprint
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+    assert "changed since that approval" in payload["error"]
+    assert posted["channel_id"] == "C123"
+
+
+async def test_rebased_workflow_push_keeps_fingerprint_for_same_tree(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Fingerprint is based on the workflow tree at head, not on a diff against origin.
+    backend = _Backend(
+        tree_entries={".github/workflows/ci.yml": "blob-sha-1"},
+    )
+    backend.head = "b" * 40
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["fingerprint"]
+    assert payload["files"] == [".github/workflows/ci.yml"]
+
+    # Rebase with a new head but the same workflow tree -> fingerprint stays stable.
+    backend.head = "c" * 40
+    result2 = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+    assert isinstance(result2, ToolMessage)
+    payload2 = json.loads(str(result2.content))
+    assert payload2["fingerprint"] == payload["fingerprint"]
+
+    # A content change to the same workflow file produces a different fingerprint.
+    backend.tree_entries = {".github/workflows/ci.yml": "blob-sha-2"}
+    result3 = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+    assert isinstance(result3, ToolMessage)
+    payload3 = json.loads(str(result3.content))
+    assert payload3["fingerprint"] != payload["fingerprint"]
+
+
+async def test_deleted_workflow_file_requires_approval(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Deleting a workflow file means it is no longer in the head workflow tree. The
+    # guard should still trigger if there are other workflow files at head; if the last
+    # workflow file is deleted, the push is no longer workflow-guarded.
+    backend = _Backend(
+        tree_entries={
+            ".github/workflows/ci.yml": "blob-sha-1",
+            ".github/workflows/other.yml": "blob-sha-2",
+        },
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    # Push with the full workflow tree present requires approval.
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+    assert ".github/workflows/ci.yml" in payload["files"]
+
+    # A push whose head tree has deleted ci.yml but still contains other.yml still
+    # requires approval, with a different fingerprint.
+    backend.tree_entries = {".github/workflows/other.yml": "blob-sha-2"}
+    result2 = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+    assert isinstance(result2, ToolMessage)
+    payload2 = json.loads(str(result2.content))
+    assert payload2["workflow_approval_status"] == "approval_required"
+    assert payload2["files"] == [".github/workflows/other.yml"]
+    assert payload2["fingerprint"] != payload["fingerprint"]
+
+
+async def test_base_poisoning_does_not_bypass_guard(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # The sandbox can rewrite refs/remotes/origin/*; the guard must use the workflow
+    # tree at the pushed head, not a diff against a remote ref. This backend records
+    # whether any command touches the remote-tracking ref.
+    backend = _Backend(
+        tree_entries={
+            ".github/workflows/ci.yml": "blob-sha-1",
+            ".github/workflows/evil.yml": "blob-sha-evil",
+        },
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+    # The evil workflow file present at head must appear in the approval list.
+    assert ".github/workflows/evil.yml" in payload["files"]
+    # No remote-tracking ref should have been consulted.
+    assert not any("refs/remotes/origin" in cmd for cmd in backend.commands)
+
+
+async def test_exact_rejection_checked_before_stale_approval(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    backend = _Backend()
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return True
+
+    # A prior approval for the same repo/branch/files exists; the stale path would
+    # normally match. But because the exact fingerprint is rejected, we must return
+    # "rejected", not "stale_approval".
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return {"fingerprint": "old-fp", "status": "approved", "decided_at": "2024-01-01T00:00:00"}
+
+    async def fake_pending(*args: Any, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        raise AssertionError("should not create a new pending record")
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "rejected"
 
 
 async def test_approved_workflow_push_aborts_when_elevation_fails(
@@ -348,3 +706,211 @@ async def test_approved_workflow_push_runs_on_non_langsmith_providers(
     assert isinstance(result, ToolMessage)
     assert result.content == "pushed"
     assert refresh_calls == []
+
+
+def test_parse_git_push_guards_wrapped_and_optioned_forms() -> None:
+    # Path-qualified and wrapped git invocations must still be recognized as pushes
+    # (so the guard engages), not passed through as "not a git command".
+    for command in (
+        "/usr/bin/git push origin feature",
+        "command git push origin feature",
+        "env FOO=bar git push origin feature",
+    ):
+        parsed = guard._parse_git_push(command)
+        assert isinstance(parsed, guard.ParsedGitPush), command
+        assert parsed.remote_ref == "feature"
+
+    # A push carrying unsupported global options, an unknown wrapper, or a value-taking
+    # wrapper option cannot be normalized, so it is blocked (fail closed) rather than run
+    # unguarded. None of these may return None.
+    for command in (
+        "git -c protocol.version=2 push origin feature",
+        "git --git-dir=.git push origin feature",
+        "git --no-pager push origin feature",
+        "nice -n 10 git push origin feature",
+        "sudo -u ci git push origin feature",
+        "timeout 5 git push origin feature",
+        "ionice -c 2 git push origin feature",
+    ):
+        result = guard._parse_git_push(command)
+        assert result is not None, command
+        assert isinstance(result, guard._BlockedGitPush), command
+
+    # A non-push git command is still ignored.
+    assert guard._parse_git_push("/usr/bin/git status") is None
+
+
+async def test_base_fetched_from_push_url_not_origin_remote(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # A changed workflow file at head (vs the trusted base fetched from the push URL) must
+    # require approval, and the base must be fetched from the push URL, never the bare
+    # `origin` remote name (which the sandbox can repoint via `remote set-url --push`).
+    backend = _Backend(
+        tree_entries={".github/workflows/ci.yml": "head-sha"},
+        base_tree_entries={".github/workflows/ci.yml": "base-sha"},
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+    assert any("remote get-url --push" in cmd for cmd in backend.commands)
+    assert any(f" fetch {backend.push_url}" in cmd for cmd in backend.commands)
+    assert not any(" fetch origin " in cmd for cmd in backend.commands)
+
+
+async def test_multiple_push_urls_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None:
+    # `git push` sends to ALL configured push URLs, but `get-url --push` shows only the
+    # first. If the head workflow tree matches the base fetched from the first (benign)
+    # URL, the guard must still require approval rather than skip, because a second push
+    # URL could land the change on the real repo. Head == base here, but multiple push
+    # URLs force fail-closed (approval required).
+    backend = _Backend(
+        tree_entries={".github/workflows/ci.yml": "same-sha"},
+        base_tree_entries={".github/workflows/ci.yml": "same-sha"},
+        push_urls=[
+            "https://github.com/attacker/local.git",
+            "https://github.com/langchain-ai/open-swe.git",
+        ],
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        raise AssertionError("push must not run without approval when push URLs are ambiguous")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+
+
+async def test_url_rewrite_fails_closed(monkeypatch: pytest.MonkeyPatch) -> None:
+    # An insteadOf/pushInsteadOf rewrite means the inspected push URL may not be the real
+    # destination, so the guard must require approval (never auto-skip).
+    backend = _Backend(
+        tree_entries={".github/workflows/ci.yml": "same-sha"},
+        base_tree_entries={".github/workflows/ci.yml": "same-sha"},
+        url_rewrites="url.https://evil.example/.insteadof https://github.com/",
+    )
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fake_approved(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_rejected(thread_id: str, fingerprint: str) -> bool:
+        return False
+
+    async def fake_find_approval(*args: Any, **kwargs: Any) -> dict[str, Any] | None:
+        return None
+
+    async def fake_pending(thread_id: str, **kwargs: Any) -> tuple[dict[str, Any], bool]:
+        return {"fingerprint": kwargs["fingerprint"], "status": "pending", "notified": False}, True
+
+    async def fake_post(*args: Any, **kwargs: Any) -> tuple[str, None]:
+        return "1700000000.000300", None
+
+    async def fake_notified(*args: Any, **kwargs: Any) -> None:
+        return None
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fake_approved)
+    monkeypatch.setattr(guard, "workflow_push_rejected", fake_rejected)
+    monkeypatch.setattr(guard, "find_workflow_push_approval", fake_find_approval)
+    monkeypatch.setattr(guard, "ensure_workflow_push_pending", fake_pending)
+    monkeypatch.setattr(guard, "post_slack_thread_reply_with_ts", fake_post)
+    monkeypatch.setattr(guard, "mark_workflow_push_notified", fake_notified)
+
+    async def handler(_request: Any) -> ToolMessage:
+        raise AssertionError("push must not run without approval when a URL rewrite exists")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["workflow_approval_status"] == "approval_required"
+
+
+async def test_unreadable_head_tree_blocks(monkeypatch: pytest.MonkeyPatch) -> None:
+    # If `ls-tree` on the head cannot be read, the guard must fail closed (block), not
+    # skip the guard and run the original push.
+    backend = _Backend(ls_tree_head_ok=False)
+    guard.SANDBOX_BACKENDS["thread-1"] = backend
+
+    async def fail_approval(*args: Any, **kwargs: Any) -> bool:
+        raise AssertionError("approval should not be checked for a blocked push")
+
+    monkeypatch.setattr(guard, "workflow_push_approved", fail_approval)
+
+    called = False
+
+    async def handler(_request: Any) -> ToolMessage:
+        nonlocal called
+        called = True
+        return ToolMessage(content="pushed", tool_call_id="call-1")
+
+    result = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
+
+    assert called is False
+    assert isinstance(result, ToolMessage)
+    assert result.status == "error"
+    payload = json.loads(str(result.content))
+    assert payload["error_type"] == "WorkflowPushBlocked"