Sign execution history for transport admission

Why: Success-shaped receipts and caller-writable history must not authorize apply/admit without controller-originated provenance.

Evidence: Fabricated, unsigned, tampered-history, and moved-ref tests were RED before HMAC binding; 52 controller/history/admission tests now pass.

Impact: Stores a local 0600 attestation key, signs successful history including transport commit SHA, and verifies it before apply/admit.

Refs: independent Codex review of feature/agent-cli-enforcement

scripts/agent_exec.py

@@ -2,18 +2,23 @@
 from __future__ import annotations
 
 import argparse
+import contextlib
+import errno
 import fcntl
 import hashlib
+import hmac
 import json
 import os
 import re
 import select
 import shutil
 import signal
+import stat
 import subprocess
 import sys
 import time
 import uuid
+from collections.abc import Iterator
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -1971,11 +1976,18 @@ def _finish_run(
     )
     history_path: Path | None = None
     try:
-        history_path = _append_execution_history(repo, receipt, secret_values)
-        receipt["history"] = {
+        history_path, attestation = _append_execution_history(
+            repo, receipt, secret_values
+        )
+        recorded: dict[str, Any] = {
             "path": str(history_path),
             "status": "recorded",
+            "record_id": receipt.get("run_id"),
+            "run_id": receipt.get("run_id"),
         }
+        if attestation:
+            recorded["attestation"] = attestation
+        receipt["history"] = recorded
     except Exception as exc:
         receipt["history"] = {
             "path": str(history_path) if history_path is not None else None,
@@ -1995,32 +2007,32 @@ def _append_execution_history(
     repo: Path,
     receipt: dict[str, Any],
     secret_values: set[str],
-) -> Path:
-    common = _git(
-        repo.resolve(),
-        "rev-parse",
-        "--path-format=absolute",
-        "--git-common-dir",
-        check=False,
-    )
-    if common.returncode != 0 or not common.stdout.strip():
-        raise RuntimeError(
-            "unable to resolve git common dir: "
-            + (common.stderr.strip() or common.stdout.strip())
-        )
-    common_dir = Path(common.stdout.strip())
-    if not common_dir.is_absolute():
-        common_dir = (repo.resolve() / common_dir).resolve()
-    history_dir = common_dir / "agent-exec"
+) -> tuple[Path, str | None]:
+    repo = repo.resolve()
+    history_dir = _git_common_dir(repo) / "agent-exec"
     if history_dir.is_symlink():
         raise RuntimeError("agent-exec history directory must not be a symlink")
     history_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+    if history_dir.is_symlink() or not history_dir.is_dir():
+        raise RuntimeError("agent-exec history directory must not be a symlink")
     os.chmod(history_dir, 0o700)
     history_path = history_dir / "execution-history.jsonl"
-    record = _execution_history_record(repo, receipt)
-    line = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
-    line = _redact_text(line, secret_values) + "\n"
-    fd = os.open(history_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
+    if history_path.is_symlink():
+        raise RuntimeError("agent-exec history file must not be a symlink")
+    attestation: str | None = None
+    if receipt.get("terminal_status") == "success":
+        _bind_success_transport_commit(repo, receipt)
+        key = _ensure_attestation_key(repo)
+        record = _execution_history_record(repo, receipt)
+        line, attestation = _signed_history_line(record, key, secret_values)
+    else:
+        record = _execution_history_record(repo, receipt)
+        line = _redact_text(_canonical_json(record), secret_values) + "\n"
+    fd = os.open(
+        history_path,
+        os.O_WRONLY | os.O_CREAT | os.O_APPEND | os.O_NOFOLLOW | os.O_CLOEXEC,
+        0o600,
+    )
     try:
         os.fchmod(fd, 0o600)
         with os.fdopen(fd, "a", encoding="utf-8", closefd=False) as handle:
@@ -2033,7 +2045,7 @@ def _append_execution_history(
                 fcntl.flock(fd, fcntl.LOCK_UN)
     finally:
         os.close(fd)
-    return history_path
+    return history_path, attestation
 
 
 def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, Any]:
@@ -2073,6 +2085,9 @@ def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str,
             "ref": transport.get("ref"),
             "parent": transport.get("parent"),
         }
+        commit = transport.get("commit")
+        if isinstance(commit, str) and commit:
+            safe_transport["commit"] = commit
     terminal = receipt.get("terminal_status")
     return {
         "schema": HISTORY_SCHEMA,
@@ -2338,7 +2353,12 @@ def _valid_transport_ref(ref: object) -> str | None:
     return ref
 
 
-def _history_file(repo: Path) -> Path | None:
+_ATTESTATION_KEY_NAME = "attestation.key"
+_ATTESTATION_KEY_BYTES = 32
+
+
+def _git_common_dir(repo: Path) -> Path:
+    repo = repo.resolve()
     common = _git(
         repo,
         "rev-parse",
@@ -2347,11 +2367,200 @@ def _history_file(repo: Path) -> Path | None:
         check=False,
     )
     if common.returncode != 0 or not common.stdout.strip():
-        raise RuntimeError(common.stderr.strip() or "unable to resolve git common dir")
+        raise RuntimeError(
+            "unable to resolve git common dir: "
+            + (common.stderr.strip() or common.stdout.strip() or "empty")
+        )
     common_dir = Path(common.stdout.strip())
     if not common_dir.is_absolute():
         common_dir = (repo / common_dir).resolve()
-    history_dir = common_dir / "agent-exec"
+    return common_dir
+
+
+def _canonical_json(data: dict[str, Any]) -> str:
+    return json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def _history_attestation(key: bytes, record: dict[str, Any]) -> str:
+    body = {field: value for field, value in record.items() if field != "attestation"}
+    return hmac.new(key, _canonical_json(body).encode("utf-8"), hashlib.sha256).hexdigest()
+
+
+def _history_attestation_ok(key: bytes, record: dict[str, Any]) -> bool:
+    presented = record.get("attestation")
+    if not isinstance(presented, str) or len(presented) != 64:
+        return False
+    return hmac.compare_digest(_history_attestation(key, record), presented)
+
+
+def _signed_history_line(
+    record: dict[str, Any], key: bytes, secret_values: set[str]
+) -> tuple[str, str]:
+    text = _redact_text(_canonical_json(record), secret_values)
+    body = json.loads(text)
+    if not isinstance(body, dict):
+        raise RuntimeError("history record redaction produced non-object JSON")
+    mac = _history_attestation(key, body)
+    body["attestation"] = mac
+    return _canonical_json(body) + "\n", mac
+
+
+def _bind_success_transport_commit(repo: Path, receipt: dict[str, Any]) -> None:
+    transport = receipt.get("transport")
+    if not isinstance(transport, dict):
+        raise RuntimeError("success history requires transport")
+    ref = transport.get("ref")
+    parent = transport.get("parent")
+    if not isinstance(ref, str) or not isinstance(parent, str):
+        raise RuntimeError("success history requires transport ref and parent")
+    commit = _resolve_commit(repo, ref)
+    resolved_parent = _resolve_commit(repo, parent)
+    if commit is None or resolved_parent is None:
+        raise RuntimeError("success transport ref did not resolve")
+    if _commit_parents(repo, commit) != [resolved_parent]:
+        raise RuntimeError("success transport parent does not match the commit")
+    transport["commit"] = commit
+    transport["parent"] = resolved_parent
+
+
+def _symlink_oserror(exc: OSError, message: str) -> OSError | RuntimeError:
+    if exc.errno == errno.ELOOP:
+        return RuntimeError(message)
+    return exc
+
+
+@contextlib.contextmanager
+def _locked_agent_exec_dir(repo: Path, *, create: bool) -> Iterator[int | None]:
+    """Lock the real agent-exec directory. Missing dirs yield None unless creating.
+
+    Callers share this lock so the first key write cannot race another reader.
+    """
+    history_dir = _git_common_dir(repo) / "agent-exec"
+    if history_dir.is_symlink():
+        raise RuntimeError("agent-exec history directory must not be a symlink")
+    if not create and not history_dir.exists():
+        yield None
+        return
+    if create:
+        history_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+        if history_dir.is_symlink() or not history_dir.is_dir():
+            raise RuntimeError("agent-exec history directory must not be a symlink")
+        os.chmod(history_dir, 0o700)
+    elif not history_dir.is_dir():
+        raise RuntimeError("agent-exec history directory must not be a symlink")
+    try:
+        fd = os.open(
+            history_dir,
+            os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
+        )
+    except OSError as exc:
+        raise _symlink_oserror(
+            exc, "agent-exec history directory must not be a symlink"
+        ) from exc
+    try:
+        info = os.fstat(fd)
+        if not stat.S_ISDIR(info.st_mode):
+            raise RuntimeError("agent-exec history directory must not be a symlink")
+        if stat.S_IMODE(info.st_mode) != 0o700:
+            raise RuntimeError("agent-exec history directory mode must be 0700")
+        fcntl.flock(fd, fcntl.LOCK_EX)
+        try:
+            yield fd
+        finally:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+    finally:
+        os.close(fd)
+
+
+def _read_key_at(dir_fd: int) -> bytes:
+    fd = os.open(
+        _ATTESTATION_KEY_NAME,
+        os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC,
+        dir_fd=dir_fd,
+    )
+    try:
+        current = os.fstat(fd)
+        if not stat.S_ISREG(current.st_mode):
+            raise RuntimeError("attestation key must be a regular file")
+        if stat.S_IMODE(current.st_mode) != 0o600:
+            raise RuntimeError("attestation key mode must be 0600")
+        if current.st_size != _ATTESTATION_KEY_BYTES:
+            raise RuntimeError("attestation key must be 32 bytes")
+        os.lseek(fd, 0, os.SEEK_SET)
+        key = os.read(fd, _ATTESTATION_KEY_BYTES)
+        if len(key) != _ATTESTATION_KEY_BYTES:
+            raise RuntimeError("attestation key must be 32 bytes")
+        return key
+    finally:
+        os.close(fd)
+
+
+def _key_is_symlink(dir_fd: int) -> bool:
+    try:
+        info = os.stat(_ATTESTATION_KEY_NAME, dir_fd=dir_fd, follow_symlinks=False)
+    except FileNotFoundError:
+        return False
+    return stat.S_ISLNK(info.st_mode)
+
+
+def _read_attestation_key(repo: Path) -> bytes | None:
+    """Read the local attestation key. Never creates one and never follows links."""
+    with _locked_agent_exec_dir(repo, create=False) as dir_fd:
+        if dir_fd is None:
+            return None
+        if _key_is_symlink(dir_fd):
+            raise RuntimeError("attestation key must not be a symlink")
+        try:
+            return _read_key_at(dir_fd)
+        except FileNotFoundError:
+            return None
+        except OSError as exc:
+            raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc
+
+
+def _ensure_attestation_key(repo: Path) -> bytes:
+    """Create or read the 32-byte key under the Git common dir.
+
+    This binds controller history to later apply/admit checks. It does not
+    protect against the same OS user deliberately reading the key file.
+    """
+    with _locked_agent_exec_dir(repo, create=True) as dir_fd:
+        if dir_fd is None:
+            raise RuntimeError("agent-exec history directory is unavailable")
+        if _key_is_symlink(dir_fd):
+            raise RuntimeError("attestation key must not be a symlink")
+        flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC
+        try:
+            fd = os.open(_ATTESTATION_KEY_NAME, flags, 0o600, dir_fd=dir_fd)
+        except OSError as exc:
+            raise _symlink_oserror(exc, "attestation key must not be a symlink") from exc
+        try:
+            os.fchmod(fd, 0o600)
+            current = os.fstat(fd)
+            if not stat.S_ISREG(current.st_mode):
+                raise RuntimeError("attestation key must be a regular file")
+            if current.st_size == 0:
+                key = os.urandom(_ATTESTATION_KEY_BYTES)
+                written = os.write(fd, key)
+                if written != _ATTESTATION_KEY_BYTES:
+                    raise RuntimeError("attestation key write was short")
+                os.fsync(fd)
+                return key
+            if current.st_size != _ATTESTATION_KEY_BYTES:
+                raise RuntimeError("attestation key must be 32 bytes")
+            os.lseek(fd, 0, os.SEEK_SET)
+            key = os.read(fd, _ATTESTATION_KEY_BYTES)
+            if len(key) != _ATTESTATION_KEY_BYTES:
+                raise RuntimeError("attestation key must be 32 bytes")
+            if stat.S_IMODE(os.fstat(fd).st_mode) != 0o600:
+                raise RuntimeError("attestation key mode must be 0600")
+            return key
+        finally:
+            os.close(fd)
+
+
+def _history_file(repo: Path) -> Path | None:
+    history_dir = _git_common_dir(repo) / "agent-exec"
     history_path = history_dir / "execution-history.jsonl"
     if history_dir.is_symlink() or history_path.is_symlink():
         return None
@@ -2381,7 +2590,10 @@ def _transport_covers(
     *,
     head: str,
     required: set[str],
+    key: bytes | None,
 ) -> bool:
+    if key is None or not _history_attestation_ok(key, record):
+        return False
     if record.get("schema") != HISTORY_SCHEMA:
         return False
     if record.get("terminal_status") != "success":
@@ -2393,20 +2605,70 @@ def _transport_covers(
         return False
     ref = _valid_transport_ref(transport.get("ref"))
     parent = transport.get("parent")
-    if ref is None or not isinstance(parent, str):
+    signed_commit = transport.get("commit")
+    if ref is None or not isinstance(parent, str) or not isinstance(signed_commit, str):
+        return False
+    if _resolve_commit(repo, ref) != signed_commit:
         return False
     if _resolve_commit(repo, parent) != head:
         return False
-    commit = _resolve_commit(repo, ref)
-    if commit is None or _commit_parents(repo, commit) != [head]:
+    if _commit_parents(repo, signed_commit) != [head]:
         return False
     changed = record.get("changed_paths")
     if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed):
         return False
-    diff_paths = _diff_paths(repo, head, commit)
+    diff_paths = _diff_paths(repo, head, signed_commit)
     if _non_exempt(changed) != required or _non_exempt(diff_paths) != required:
         return False
-    return _index_matches_commit(repo, commit, required)
+    return _index_matches_commit(repo, signed_commit, required)
+
+
+def _apply_signed_history_ok(
+    repo: Path,
+    receipt: dict[str, Any],
+    *,
+    commit: str,
+    base_head: str,
+) -> bool:
+    key = _read_attestation_key(repo)
+    if key is None:
+        return False
+    run_id = receipt.get("run_id")
+    transport = receipt.get("transport")
+    changed = receipt.get("changed_paths")
+    if not isinstance(run_id, str) or not run_id or not isinstance(transport, dict):
+        return False
+    ref = transport.get("ref")
+    parent = transport.get("parent")
+    claimed = transport.get("commit")
+    if not isinstance(ref, str) or parent != base_head or claimed != commit:
+        return False
+    if _resolve_commit(repo, ref) != commit:
+        return False
+    if _commit_parents(repo, commit) != [base_head]:
+        return False
+    if not isinstance(changed, list) or any(not isinstance(path, str) for path in changed):
+        return False
+    if _non_exempt(changed) != _non_exempt(_diff_paths(repo, base_head, commit)):
+        return False
+    for record in _load_history_records(repo):
+        if record.get("run_id") != run_id or record.get("terminal_status") != "success":
+            continue
+        if not _same_repo(repo, record.get("repo")):
+            continue
+        record_paths = record.get("changed_paths")
+        if record_paths != changed:
+            continue
+        rec_transport = record.get("transport")
+        if not isinstance(rec_transport, dict):
+            continue
+        if rec_transport.get("ref") != ref or rec_transport.get("parent") != parent:
+            continue
+        if rec_transport.get("commit") != commit:
+            continue
+        if _history_attestation_ok(key, record):
+            return True
+    return False
 
 
 def cmd_apply(repo: Path, receipt_path: Path) -> int:
@@ -2472,6 +2734,15 @@ def cmd_apply(repo: Path, receipt_path: Path) -> int:
     ):
         return _reject("apply: mismatched transport", secrets)
 
+    try:
+        authorized = _apply_signed_history_ok(
+            repo, receipt, commit=commit, base_head=resolved_base
+        )
+    except Exception as exc:
+        return _reject(f"apply: no signed history: {exc}", secrets)
+    if not authorized:
+        return _reject("apply: no signed history", secrets)
+
     merge = _git(repo, "merge", "--ff-only", "--quiet", commit, check=False)
     if merge.returncode != 0:
         return _reject(
@@ -2500,8 +2771,11 @@ def cmd_admit(repo: Path) -> int:
         if history_path is not None and history_path.is_file():
             history_text = history_path.read_text(encoding="utf-8")
             secrets = _secret_set(history_text)
+        key = _read_attestation_key(repo)
         for record in _load_history_records(repo):
-            if _transport_covers(repo, record, head=head, required=required):
+            if _transport_covers(
+                repo, record, head=head, required=required, key=key
+            ):
                 print("admit: ok")
                 return EXIT_SUCCESS
     except Exception as exc:

tests/test_agent_exec.py

@@ -5,6 +5,8 @@ Set RUN_REAL_AGENT_EXEC=1 and AGENT_EXEC_REAL_PROFILE to exactly one profile.
 """
 from __future__ import annotations
 
+import hashlib
+import hmac
 import json
 import os
 import re
@@ -2342,6 +2344,124 @@ class AgentExecHistoryTests(unittest.TestCase):
         self.assertEqual(records[0]["terminal_status"], "invalid-config")
         self.assertEqual(records[0]["attempts"], [])
 
+    def _run_marker_success(self, name: str) -> tuple[Path, dict[str, object]]:
+        _write_fake_worker(
+            self.bindir,
+            name,
+            body=textwrap.dedent(
+                """\
+                workspace="$1"
+                result="$3"
+                mkdir -p "$workspace/src"
+                printf 'ok\\n' > "$workspace/src/marker.txt"
+                printf '%s\\n' '{"terminal_status":"completed","summary":"done","changed_files":["src/marker.txt"],"evidence":[],"scope_expansion":null}' > "$result"
+                exit 0
+                """
+            ),
+        )
+        _write_config(
+            self.repo,
+            profiles={name: {"adapter": "custom", "executable": name}},
+            role_order=[name],
+        )
+        unit = _write_unit(self.root / f"{name}.json", unit_id=f"unit-{name}")
+        receipt_path = self.receipts / f"{name}.json"
+        code = agent_exec.cmd_run(
+            self.repo,
+            role="implementation",
+            unit_path=unit,
+            receipt_path=receipt_path,
+        )
+        receipt_text = (
+            receipt_path.read_text(encoding="utf-8") if receipt_path.is_file() else ""
+        )
+        self.assertEqual(code, agent_exec.EXIT_SUCCESS, receipt_text)
+        return receipt_path, json.loads(receipt_text)
+
+    def test_success_history_is_mode_correct_and_signed(self) -> None:
+        receipt_path, receipt = self._run_marker_success("signedhistory")
+        key_path = self.history_path.parent / "attestation.key"
+        self.assertTrue(key_path.is_file())
+        self.assertFalse(key_path.is_symlink())
+        self.assertFalse(key_path.parent.is_symlink())
+        self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700)
+        self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600)
+        key = key_path.read_bytes()
+        self.assertEqual(len(key), 32)
+        self.assertNotIn(key, receipt_path.read_bytes())
+        self.assertNotIn(key, self.history_path.read_bytes())
+
+        records = [
+            json.loads(line)
+            for line in self.history_path.read_text(encoding="utf-8").splitlines()
+        ]
+        self.assertEqual(len(records), 1)
+        record = records[0]
+        transport = record["transport"]
+        ref = transport["ref"]
+        commit = _git(self.repo, "rev-parse", ref).stdout.strip()
+        self.assertEqual(record["terminal_status"], "success")
+        self.assertEqual(record["run_id"], receipt["run_id"])
+        self.assertEqual(transport["parent"], receipt["transport"]["parent"])
+        self.assertEqual(transport["commit"], commit)
+        self.assertEqual(receipt["transport"]["commit"], commit)
+        self.assertEqual(record["changed_paths"], receipt["changed_paths"])
+        body = {field: value for field, value in record.items() if field != "attestation"}
+        payload = json.dumps(
+            body, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        self.assertEqual(
+            hmac.new(key, payload, hashlib.sha256).hexdigest(),
+            record["attestation"],
+        )
+        identity = receipt["history"]
+        self.assertEqual(identity["status"], "recorded")
+        self.assertEqual(identity["run_id"], receipt["run_id"])
+        self.assertEqual(identity["record_id"], record["record_id"])
+        self.assertEqual(identity["attestation"], record["attestation"])
+        self.assertNotIn(key.hex(), json.dumps(identity))
+
+    def test_attestation_key_symlink_is_not_followed(self) -> None:
+        history_dir = self.history_path.parent
+        history_dir.mkdir(mode=0o700)
+        outside = self.root / "outside-key"
+        outside.write_bytes(b"A" * 32)
+        key_path = history_dir / "attestation.key"
+        key_path.symlink_to(outside)
+        receipt_path, receipt = self._run_marker_success("symlinkkey")
+        self.assertTrue(key_path.is_symlink())
+        self.assertEqual(outside.read_bytes(), b"A" * 32)
+        self.assertEqual(receipt["history"]["status"], "failed")
+        self.assertNotIn(b"A" * 32, receipt_path.read_bytes())
+
+    def test_attestation_key_first_run_race_shares_one_key(self) -> None:
+        found: list[bytes] = []
+        errors: list[BaseException] = []
+        barrier = threading.Barrier(8)
+
+        def load() -> None:
+            try:
+                barrier.wait(timeout=5)
+                found.append(agent_exec._ensure_attestation_key(self.repo))
+            except BaseException as exc:  # noqa: BLE001 - race failures are the assertion
+                errors.append(exc)
+
+        threads = [threading.Thread(target=load) for _ in range(8)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=10)
+            self.assertFalse(thread.is_alive())
+        self.assertEqual(errors, [])
+        self.assertEqual(len(found), 8)
+        self.assertEqual(len({item for item in found}), 1)
+        self.assertEqual(len(found[0]), 32)
+        key_path = self.history_path.parent / "attestation.key"
+        self.assertFalse(key_path.is_symlink())
+        self.assertEqual(key_path.read_bytes(), found[0])
+        self.assertEqual(stat.S_IMODE(key_path.stat().st_mode), 0o600)
+        self.assertEqual(stat.S_IMODE(key_path.parent.stat().st_mode), 0o700)
+
 
 # Opt-in real CLI probes (U5). Default unittest must not contact providers.
 U5_PROFILES = (

tests/test_agent_exec_admission.py

@@ -1,6 +1,8 @@
 """Provenance admission: apply a success transport, or admit exact staged paths."""
 from __future__ import annotations
 
+import hashlib
+import hmac
 import json
 import os
 import shutil
@@ -148,6 +150,8 @@ class AgentExecAdmissionTests(unittest.TestCase):
         reason: str | None = None,
         transport_parent: str | None = None,
         omit_transport: bool = False,
+        run_id: str | None = None,
+        commit: str | None = None,
     ) -> Path:
         data: dict[str, object] = {
             "schema": agent_exec.RECEIPT_SCHEMA,
@@ -161,16 +165,65 @@ class AgentExecAdmissionTests(unittest.TestCase):
             },
             "changed_paths": paths,
         }
+        if run_id is not None:
+            data["run_id"] = run_id
         if not omit_transport:
-            data["transport"] = {
+            transport: dict[str, object] = {
                 "ref": ref,
                 "parent": parent if transport_parent is None else transport_parent,
                 "attempt_id": "attempt-1",
             }
+            if commit is not None:
+                transport["commit"] = commit
+            data["transport"] = transport
         path = self.root / f"receipt-{uuid.uuid4().hex}.json"
         path.write_text(json.dumps(data), encoding="utf-8")
         return path
 
+    def _common_dir(self) -> Path:
+        return Path(
+            _git(
+                self.repo,
+                "rev-parse",
+                "--path-format=absolute",
+                "--git-common-dir",
+            ).stdout.strip()
+        )
+
+    def _history_path(self) -> Path:
+        return self._common_dir() / "agent-exec" / "execution-history.jsonl"
+
+    def _attestation_key(self) -> bytes:
+        dest = self._common_dir() / "agent-exec"
+        dest.mkdir(mode=0o700, exist_ok=True)
+        os.chmod(dest, 0o700)
+        path = dest / "attestation.key"
+        if path.is_symlink():
+            raise RuntimeError("test attestation key must not be a symlink")
+        if path.exists():
+            key = path.read_bytes()
+            if len(key) != 32:
+                raise RuntimeError("test attestation key must be 32 bytes")
+            return key
+        key = os.urandom(32)
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+        try:
+            os.write(fd, key)
+            os.fchmod(fd, 0o600)
+        finally:
+            os.close(fd)
+        return key
+
+    @staticmethod
+    def _sign_history_record(record: dict[str, object], key: bytes) -> dict[str, object]:
+        body = {field: value for field, value in record.items() if field != "attestation"}
+        payload = json.dumps(
+            body, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        ).encode("utf-8")
+        signed = dict(body)
+        signed["attestation"] = hmac.new(key, payload, hashlib.sha256).hexdigest()
+        return signed
+
     def _history(
         self,
         *,
@@ -179,26 +232,39 @@ class AgentExecAdmissionTests(unittest.TestCase):
         paths: list[str],
         status: str = "success",
         repo: str | None = None,
-    ) -> None:
-        record = {
+        commit: str | None = None,
+        run_id: str | None = None,
+        sign: bool | None = None,
+    ) -> dict[str, object]:
+        if commit is None:
+            commit = _git(self.repo, "rev-parse", f"{ref}^{{commit}}").stdout.strip()
+        identity = run_id or uuid.uuid4().hex
+        record: dict[str, object] = {
             "schema": agent_exec.HISTORY_SCHEMA,
             "terminal_status": status,
+            "record_id": identity,
+            "run_id": identity,
             "repo": repo if repo is not None else str(self.repo.resolve()),
             "changed_paths": paths,
-            "transport": {"ref": ref, "parent": parent},
+            "transport": {"ref": ref, "parent": parent, "commit": commit},
         }
-        common = Path(
-            _git(
-                self.repo,
-                "rev-parse",
-                "--path-format=absolute",
-                "--git-common-dir",
-            ).stdout.strip()
-        )
-        dest = common / "agent-exec"
+        if sign is None:
+            sign = status == "success"
+        if sign:
+            record = self._sign_history_record(record, self._attestation_key())
+        dest = self._history_path().parent
         dest.mkdir(mode=0o700, exist_ok=True)
-        with (dest / "execution-history.jsonl").open("a", encoding="utf-8") as handle:
-            handle.write(json.dumps(record, sort_keys=True) + "\n")
+        os.chmod(dest, 0o700)
+        line = json.dumps(
+            record, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+        with self._history_path().open("a", encoding="utf-8") as handle:
+            handle.write(line + "\n")
+        return record
+
+    def _reset_branch(self) -> None:
+        _git(self.repo, "reset", "--hard", self.head)
+        _git(self.repo, "clean", "-fd")
 
     def _stage(self, rel: str, content: str) -> None:
         path = self.repo / rel
@@ -208,9 +274,23 @@ class AgentExecAdmissionTests(unittest.TestCase):
 
     def test_apply_fast_forwards_clean_matching_transport(self) -> None:
         commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
-        receipt = self._receipt(
-            parent=self.head, ref=ref, paths=["src/marker.txt"]
+        run_id = uuid.uuid4().hex
+        signed = self._history(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt"],
+            commit=commit,
+            run_id=run_id,
         )
+        receipt = self._receipt(
+            parent=self.head,
+            ref=ref,
+            paths=["src/marker.txt"],
+            run_id=run_id,
+            commit=commit,
+        )
+        self.assertEqual(signed["transport"]["commit"], commit)
+        self.assertIn("attestation", signed)
         proc = self._cli(
             "apply", "--repo", str(self.repo), "--receipt", str(receipt)
         )
@@ -309,6 +389,64 @@ class AgentExecAdmissionTests(unittest.TestCase):
                 ),
             )
 
+    def test_apply_rejects_fabricated_unsigned_or_tampered_history(self) -> None:
+        commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        run_id = uuid.uuid4().hex
+        paths = ["src/marker.txt"]
+
+        def reject(label: str) -> None:
+            self._reset_branch()
+            before = self._state()
+            receipt = self._receipt(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                run_id=run_id,
+                commit=commit,
+            )
+            proc = self._cli(
+                "apply", "--repo", str(self.repo), "--receipt", str(receipt)
+            )
+            combined = proc.stdout + proc.stderr
+            self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+            self.assertIn("apply: no signed history", proc.stderr)
+            self.assertNotIn("apply: fast-forwarded", proc.stdout)
+            self.assertEqual(self._state(), before, label)
+
+        with self.subTest(label="fabricated receipt"):
+            reject("fabricated receipt")
+
+        with self.subTest(label="unsigned history"):
+            self._history(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                commit=commit,
+                run_id=run_id,
+                sign=False,
+            )
+            reject("unsigned history")
+
+        with self.subTest(label="tampered history"):
+            self._reset_branch()
+            signed = self._history(
+                parent=self.head,
+                ref=ref,
+                paths=paths,
+                commit=commit,
+                run_id=run_id,
+            )
+            history_path = self._history_path()
+            lines = history_path.read_text(encoding="utf-8").splitlines()
+            tampered = json.loads(lines[-1])
+            self.assertEqual(tampered["attestation"], signed["attestation"])
+            tampered["record_id"] = "tampered-record"
+            lines[-1] = json.dumps(
+                tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+            )
+            history_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+            reject("tampered history")
+
     def test_apply_redacts_secret_shaped_receipt_text(self) -> None:
         _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
         secret = "sk-admissionsecret"
@@ -360,6 +498,69 @@ class AgentExecAdmissionTests(unittest.TestCase):
         self.assertEqual(partial.returncode, agent_exec.EXIT_BLOCKED, partial.stderr)
         self.assertIn("admit: no provenance", partial.stderr)
 
+    def test_admit_rejects_tampered_record_and_moved_ref(self) -> None:
+        commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
+        signed = self._history(parent=self.head, ref=ref, paths=["src/marker.txt"], commit=commit)
+        history_path = self._history_path()
+        lines = history_path.read_text(encoding="utf-8").splitlines()
+        tampered = json.loads(lines[-1])
+        self.assertEqual(tampered["attestation"], signed["attestation"])
+        tampered["record_id"] = "tampered-record"
+        lines[-1] = json.dumps(
+            tampered, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+        )
+        history_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+        self._stage("src/marker.txt", "shipped\n")
+        before = self._state()
+        proc = self._cli("admit", "--repo", str(self.repo))
+        combined = proc.stdout + proc.stderr
+        self.assertEqual(proc.returncode, agent_exec.EXIT_BLOCKED, combined)
+        self.assertIn("admit: no provenance", proc.stderr)
+        self.assertNotIn("admit: ok", proc.stdout)
+        self.assertEqual(self._state(), before)
+
+        self._reset_branch()
+        tree = _git(self.repo, "rev-parse", f"{commit}^{{tree}}").stdout.strip()
+        moved = _git(
+            self.repo,
+            "commit-tree",
+            tree,
+            "-p",
+            self.head,
+            "-m",
+            "moved ref",
+        ).stdout.strip()
+        self.assertNotEqual(moved, commit)
+        restored = self._sign_history_record(
+            {
+                "schema": agent_exec.HISTORY_SCHEMA,
+                "terminal_status": "success",
+                "record_id": signed["record_id"],
+                "run_id": signed["run_id"],
+                "repo": str(self.repo.resolve()),
+                "changed_paths": ["src/marker.txt"],
+                "transport": {"commit": commit, "parent": self.head, "ref": ref},
+            },
+            self._attestation_key(),
+        )
+        history_path.write_text(
+            json.dumps(
+                restored, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+            )
+            + "\n",
+            encoding="utf-8",
+        )
+        _git(self.repo, "update-ref", ref, moved)
+        self._stage("src/marker.txt", "shipped\n")
+        moved_before = self._state()
+        moved_proc = self._cli("admit", "--repo", str(self.repo))
+        moved_text = moved_proc.stdout + moved_proc.stderr
+        self.assertEqual(moved_proc.returncode, agent_exec.EXIT_BLOCKED, moved_text)
+        self.assertIn("admit: no provenance", moved_proc.stderr)
+        self.assertNotIn("admit: ok", moved_proc.stdout)
+        self.assertEqual(self._state(), moved_before)
+        self.assertEqual(_git(self.repo, "rev-parse", ref).stdout.strip(), moved)
+
     def test_admit_rejects_different_content_at_transport_path(self) -> None:
         _commit, ref = self._transport_commit({"src/marker.txt": "shipped\n"})
         self._history(parent=self.head, ref=ref, paths=["src/marker.txt"])