Key workflow approval on head workflow tree and block unsafe pushes

Switches the workflow push guard from diffing against a sandbox-writable
remote-tracking ref to hashing the actual workflow tree at the pushed head via
. This closes the confused-deputy
base-poisoning vector and the replay vector: approval is bound to the exact
files and blob SHAs at head, so a different workflow tree requires re-approval.

Also:
- Fails the git push parser CLOSED: unrecognized or unsafe push forms are now
  blocked instead of running unguarded.
- Checks the exact fingerprint's rejected status before the stale-approval path.
- Updates prompt and Slack copy to reflect the new head-tree behavior.
- Adds tests for base-poisoning, replay, deletion, unparsed push blocking, and
  rejection-before-stale ordering.

Refs: 98

agent/dashboard/workflow_approval.py

@@ -44,6 +44,11 @@ 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,
     *,

agent/middleware/workflow_push_guard.py

@@ -25,6 +25,7 @@ from ..dashboard.workflow_approval import (
     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 (
@@ -45,6 +46,15 @@ _GIT_OBJECT_ID = re.compile(r"^[0-9a-fA-F]{40,64}$")
 _UNSAFE_RAW_COMMAND = re.compile(r"[;|`$<>\n\r]")
 
 
+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)
 class ParsedGitPush:
     repo_dir: str | None
@@ -59,13 +69,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)
@@ -144,28 +156,30 @@ 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:
+def _parse_git_tokens(
+    tokens: list[str], *, repo_dir: str | None
+) -> ParsedGitPush | _BlockedGitPush | None:
     if not tokens or tokens[0] != "git":
         return None
     i = 1
@@ -174,22 +188,25 @@ def _parse_git_tokens(tokens: list[str], *, repo_dir: str | None) -> ParsedGitPu
             repo_dir = tokens[i + 1]
             i += 2
             continue
+        # Any non-push git command (e.g. git status) is not a push, so let it pass.
         return None
     if i >= len(tokens) or tokens[i] != "push":
         return None
     return _parse_push_args(tokens[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,
@@ -283,6 +300,40 @@ def _run_coroutine_sync(coro: Awaitable[ToolMessage | Command]) -> ToolMessage |
     return value
 
 
+def _workflow_tree_at_head(
+    backend: Any, repo_dir: str | None, head: str
+) -> tuple[list[str], str] | None:
+    """Return the sorted workflow file paths and a stable hash of the head workflow tree.
+
+    Uses `git ls-tree -r` so the hash binds to the blob SHAs at the pushed head, not to
+    a diff against a sandbox-writable base ref. This prevents a confused deputy where the
+    sandbox rewrites `refs/remotes/origin/*` to hide a malicious workflow file from the
+    approved diff while still deploying it.
+    """
+    ls_tree = _run_git(backend, repo_dir, f"ls-tree -r {shlex.quote(head)} -- .github/workflows")
+    if not ls_tree.ok:
+        return None
+    entries: list[tuple[str, str]] = []
+    for line in ls_tree.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))
+    if not entries:
+        return None
+    entries.sort(key=lambda item: item[1])
+    files = [path for _sha, path in entries]
+    content_hash = _fingerprint({"tree": [(sha, path) for sha, path in entries]})
+    return files, content_hash
+
+
 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:
@@ -303,44 +354,10 @@ 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}"
+    tree = _workflow_tree_at_head(backend, root, head)
-    remote_branch_exists = _run_git(
+    if tree is None:
-        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
-        )
-
-    names = _run_git(
-        backend,
-        root,
-        f"diff --name-only --diff-filter=ACMRTD {range_expr} -- .github/workflows",
-    )
-    if not names.ok:
         return None
-    files = sorted(
+    files, content_hash = tree
-        line.strip()
-        for line in names.output.splitlines()
-        if line.strip().startswith(_WORKFLOW_PREFIX)
-    )
-    if not files:
-        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
-
-    content_hash = _fingerprint({"diff": diff.output})
 
     remote = _run_git(backend, root, "config --get remote.origin.url")
     repo = _normalize_remote(_first_line(remote.output)) if remote.ok else ""
@@ -360,9 +377,8 @@ def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPu
         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,
@@ -371,27 +387,37 @@ def _workflow_change_for_push(backend: Any, parsed: ParsedGitPush) -> WorkflowPu
 
 
 def _blocked_message(
-    change: WorkflowPushChange, *, already_rejected: bool = False, stale: bool = False
+    change: WorkflowPushChange,
+    *,
+    already_rejected: bool = False,
+    stale: bool = False,
+    blocked: bool = False,
 ) -> ToolMessage:
     status = "rejected" if already_rejected else "approval_required"
-    if stale:
+    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 diff has "
+            "exists for the same branch and workflow files, but the workflow content "
-            "changed since that approval (for example, a rebase or amend). The thread "
+            "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."
+            "after the thread owner approves the workflow files."
         )
+        error_type = "WorkflowPushApprovalRequired"
     content = {
         "status": "error",
-        "error_type": "WorkflowPushApprovalRequired",
+        "error_type": error_type,
         "error": error,
-        "workflow_approval_status": status,
+        "workflow_approval_status": status if not blocked else "blocked",
         "fingerprint": change.fingerprint,
         "files": change.files,
         "repo": change.repo,
@@ -422,12 +448,13 @@ 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"Open SWE is trying to push the workflow files below to `{repo}` on `{branch}`.\n\n"
-        f"*Files:*\n{files}\n\n"
+        f"*Files at the pushed head:*\n{files}\n\n"
         f"*Fingerprint:* `{change.fingerprint}`\n\n"
-        "Approval covers the exact workflow-file content diff listed above on this branch, "
+        "Approval covers the exact workflow files and content listed above at the pushed head, "
-        "including future rebases or amends that replay the same diff. If the set of workflow "
+        "including future rebases or amends that replay the same workflow tree. If the set of "
-        "files, the branch, or the workflow-file content changes, a new fingerprint will be required."
+        "workflow files, the branch, or the workflow-file content at the pushed head changes, "
+        "a new fingerprint will be required."
     )
 
 
@@ -464,11 +491,14 @@ 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 diff
+        # approval covers the same (repo, branch, files) identity. If so, the workflow
-        # changed underneath the prior approval (rebase/amend/edit), so we surface a
+        # tree at the pushed head changed underneath the prior approval (rebase/amend
-        # loud re-approval message rather than a fresh silent pending record.
+        # 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,
@@ -550,6 +580,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
@@ -561,6 +605,8 @@ 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:

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 before it can proceed. Approval is keyed to the repo, branch, workflow files, and the exact workflow-file content diff, so rebases or amends that replay the same workflow diff do not require a fresh approval; changing the branch, the set of workflow files, or the workflow-file content does require a new approval. 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

@@ -24,42 +24,31 @@ class _Backend:
         self,
         *,
         workflow_files: str = ".github/workflows/ci.yml",
-        remote_branch: str | None = None,
+        tree_entries: dict[str, str] | None = None,
-        diff_output: str = "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n",
     ) -> None:
         self.workflow_files = workflow_files
-        self.remote_branch = remote_branch
+        self.tree_entries = tree_entries
-        self.diff_output = diff_output
         self.commands: list[str] = []
         self.head = "a" * 40
 
+    def _ls_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 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:
+        if "ls-tree -r" in command:
-            if self.remote_branch is None:
+            return _Response(self._ls_tree())
-                return _Response("", 1)
-            return _Response(f"{self.remote_branch}\n")
-        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(self.diff_output)
-        if command.startswith("git -C /repo show "):
-            content = "workflow content\n"
-            if "+new content" in self.diff_output:
-                content += "new content\n"
-            return _Response(content)
         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:
             return _Response("feature\n")
-        if "rev-parse refs/remotes/origin/feature" in command:
-            return _Response(f"{self.remote_branch}\n")
         if "rev-parse HEAD" in command or "rev-parse feature" in command:
             return _Response(f"{self.head}\n")
         return _Response("")
@@ -110,11 +99,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,
@@ -171,6 +174,9 @@ 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
 
@@ -189,6 +195,7 @@ 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)
@@ -314,6 +321,9 @@ async def test_stale_workflow_approval_is_loud_and_blocks(
     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"}
 
@@ -332,6 +342,7 @@ async def test_stale_workflow_approval_is_loud_and_blocks(
         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)
@@ -350,17 +361,22 @@ async def test_stale_workflow_approval_is_loud_and_blocks(
     assert posted["channel_id"] == "C123"
 
 
-async def test_rebased_workflow_push_keeps_fingerprint_for_same_diff(
+async def test_rebased_workflow_push_keeps_fingerprint_for_same_tree(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    # Remote branch exists, so base_sha is derived from the remote tip and changes on rebase.
+    # Fingerprint is based on the workflow tree at head, not on a diff against origin.
-    backend = _Backend(remote_branch="r" * 40)
+    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
 
@@ -374,6 +390,7 @@ async def test_rebased_workflow_push_keeps_fingerprint_for_same_diff(
         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)
@@ -389,8 +406,7 @@ async def test_rebased_workflow_push_keeps_fingerprint_for_same_diff(
     assert payload["fingerprint"]
     assert payload["files"] == [".github/workflows/ci.yml"]
 
-    # Rebase onto a new base with the same workflow diff -> fingerprint stays stable.
+    # Rebase with a new head but the same workflow tree -> fingerprint stays stable.
-    backend.remote_branch = "s" * 40
     backend.head = "c" * 40
     result2 = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
     assert isinstance(result2, ToolMessage)
@@ -398,9 +414,7 @@ async def test_rebased_workflow_push_keeps_fingerprint_for_same_diff(
     assert payload2["fingerprint"] == payload["fingerprint"]
 
     # A content change to the same workflow file produces a different fingerprint.
-    backend.diff_output = (
+    backend.tree_entries = {".github/workflows/ci.yml": "blob-sha-2"}
-        "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n+new content\n"
-    )
     result3 = await guard.WorkflowPushGuardMiddleware().awrap_tool_call(_Request(), handler)
     assert isinstance(result3, ToolMessage)
     payload3 = json.loads(str(result3.content))
@@ -410,15 +424,23 @@ async def test_rebased_workflow_push_keeps_fingerprint_for_same_diff(
 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(
-        workflow_files=".github/workflows/ci.yml",
+        tree_entries={
-        diff_output="diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\ndeleted file\n",
+            ".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
 
@@ -432,6 +454,68 @@ async def test_deleted_workflow_file_requires_approval(
         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)
@@ -446,7 +530,47 @@ async def test_deleted_workflow_file_requires_approval(
     assert result.status == "error"
     payload = json.loads(str(result.content))
     assert payload["workflow_approval_status"] == "approval_required"
-    assert payload["files"] == [".github/workflows/ci.yml"]
+    # 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(