diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3733689..4241405 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,9 +4,11 @@ summary: Chronological history of repository and skill changes.
 
 # Changelog
 
-## 2026-07-21 — Carve-changesets operating contract
+## 2026-07-21 — Carve-changesets identity and operating contract
 
+- feat: add self-describing changeset identity
 - fix: clarify published terminal evidence
+  (`edeb2f5f5f7b4cfa4e73e8289d34157b192f92ab`)
 - docs: define the carve-changesets operating contract
   (`77865c25190e7205142318229f17c1d3f18e1fef`)
 
diff --git a/justfile b/justfile
index 5bab059..c65f7b2 100644
--- a/justfile
+++ b/justfile
@@ -58,6 +58,9 @@ eval-implement-ticket-claude:
 test-implement-epic:
   python3 -m unittest discover -s {{skills_dir}}/implement-epic/scripts/tests -p 'test_*.py'
 
+test-carve-changesets:
+  python3 -m unittest discover -s {{skills_dir}}/carve-changesets/scripts/tests -p 'test_*.py'
+
 test-prepare-changesets:
   python3 -m unittest discover -s {{skills_dir}}/prepare-changesets/scripts/tests -p 'test_*.py'
 
diff --git a/skills/carve-changesets/scripts/metadata.py b/skills/carve-changesets/scripts/metadata.py
new file mode 100644
index 0000000..047fe74
--- /dev/null
+++ b/skills/carve-changesets/scripts/metadata.py
@@ -0,0 +1,216 @@
+"""Durable identity carried by changeset commits and pull requests."""
+
+from __future__ import annotations
+
+import json
+import re
+import subprocess
+from dataclasses import dataclass
+from typing import Callable, Sequence
+
+TRAILER_SLUG = "Changeset-Slug"
+TRAILER_INDEX = "Changeset-Index"
+TRAILER_SOURCE = "Changeset-Source"
+METADATA_MARKER = "carve-changesets:metadata:v1"
+
+_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
+_BLOCK_RE = re.compile(
+    rf"<!--\s*{re.escape(METADATA_MARKER)}\s*\n(?P<payload>.*?)\n\s*-->",
+    re.DOTALL,
+)
+
+
+class MetadataError(ValueError):
+    """Raised when durable changeset identity is absent or contradictory."""
+
+
+GitRunner = Callable[[Sequence[str], str], str]
+
+
+@dataclass(frozen=True)
+class ChangesetMetadata:
+    """Identity shared by a changeset commit and its pull request."""
+
+    slug: str
+    index: int
+    source_branch: str
+    source_sha: str
+
+    def __post_init__(self) -> None:
+        if not self.slug.strip():
+            raise MetadataError("Changeset slug must not be empty.")
+        if self.index < 1:
+            raise MetadataError("Changeset index must be a positive integer.")
+        if not self.source_branch.strip():
+            raise MetadataError("Changeset source branch must not be empty.")
+        if " @ " in self.source_branch:
+            raise MetadataError("Changeset source branch must not contain ' @ '.")
+        if not _SHA_RE.fullmatch(self.source_sha):
+            raise MetadataError(
+                "Changeset source SHA must be a full lowercase 40-character SHA."
+            )
+
+    @property
+    def source_trailer(self) -> str:
+        return f"{self.source_branch} @ {self.source_sha}"
+
+
+def _run_git_interpret_trailers(args: Sequence[str], input_text: str) -> str:
+    try:
+        result = subprocess.run(
+            ["git", "interpret-trailers", *args],
+            input=input_text,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise MetadataError("Git is required to manage changeset trailers.") from exc
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        raise MetadataError(f"git interpret-trailers failed: {detail}")
+    return result.stdout
+
+
+def stamp_commit_message(
+    message: str,
+    metadata: ChangesetMetadata,
+    *,
+    runner: GitRunner = _run_git_interpret_trailers,
+) -> str:
+    """Return a commit message stamped by ``git interpret-trailers``."""
+
+    if not message.strip():
+        raise MetadataError("Commit message must not be empty.")
+    return runner(
+        (
+            "--trailer",
+            f"{TRAILER_SLUG}: {metadata.slug}",
+            "--trailer",
+            f"{TRAILER_INDEX}: {metadata.index}",
+            "--trailer",
+            f"{TRAILER_SOURCE}: {metadata.source_trailer}",
+        ),
+        message.rstrip() + "\n",
+    )
+
+
+def parse_commit_message(
+    message: str,
+    *,
+    runner: GitRunner = _run_git_interpret_trailers,
+) -> ChangesetMetadata:
+    """Parse required identity trailers from a changeset commit message."""
+
+    parsed = runner(("--parse",), message)
+    values: dict[str, list[str]] = {}
+    for line in parsed.splitlines():
+        key, separator, value = line.partition(":")
+        if separator:
+            values.setdefault(key.strip(), []).append(value.strip())
+
+    required = (TRAILER_SLUG, TRAILER_INDEX, TRAILER_SOURCE)
+    missing = [key for key in required if not values.get(key)]
+    if missing:
+        raise MetadataError(
+            "Missing required changeset trailer(s): " + ", ".join(missing)
+        )
+    duplicates = [key for key in required if len(values[key]) != 1]
+    if duplicates:
+        raise MetadataError(
+            "Ambiguous duplicate changeset trailer(s): " + ", ".join(duplicates)
+        )
+
+    index_text = values[TRAILER_INDEX][0]
+    try:
+        index = int(index_text)
+    except ValueError as exc:
+        raise MetadataError(
+            f"Changeset-Index must be a positive integer, got {index_text!r}."
+        ) from exc
+
+    source_text = values[TRAILER_SOURCE][0]
+    source_branch, separator, source_sha = source_text.rpartition(" @ ")
+    if not separator:
+        raise MetadataError(
+            "Changeset-Source must use '<source-branch> @ <source-sha>'."
+        )
+    return ChangesetMetadata(
+        slug=values[TRAILER_SLUG][0],
+        index=index,
+        source_branch=source_branch,
+        source_sha=source_sha,
+    )
+
+
+def render_pr_metadata(metadata: ChangesetMetadata) -> str:
+    """Render deterministic, human-invisible pull request metadata."""
+
+    payload = json.dumps(
+        {
+            "index": metadata.index,
+            "slug": metadata.slug,
+            "source_branch": metadata.source_branch,
+            "source_sha": metadata.source_sha,
+        },
+        separators=(",", ":"),
+        sort_keys=True,
+    )
+    return f"<!-- {METADATA_MARKER}\n{payload}\n-->"
+
+
+def embed_pr_metadata(body: str, metadata: ChangesetMetadata) -> str:
+    """Append or replace the metadata block without changing human prose."""
+
+    matches = list(_BLOCK_RE.finditer(body))
+    if len(matches) > 1:
+        raise MetadataError("PR body contains multiple changeset metadata blocks.")
+    block = render_pr_metadata(metadata)
+    if matches:
+        match = matches[0]
+        return body[: match.start()] + block + body[match.end() :]
+    separator = "\n" if not body or body.endswith("\n") else "\n\n"
+    return body + separator + block + "\n"
+
+
+def parse_pr_metadata(body: str) -> ChangesetMetadata:
+    """Parse metadata while tolerating arbitrary edits outside its block."""
+
+    matches = list(_BLOCK_RE.finditer(body))
+    if not matches:
+        if "carve-changesets:metadata" in body:
+            raise MetadataError(
+                "Malformed carve-changesets PR metadata block; restore the v1 delimiters."
+            )
+        raise MetadataError("PR body is missing the carve-changesets metadata block.")
+    if len(matches) > 1:
+        raise MetadataError("PR body contains multiple changeset metadata blocks.")
+    try:
+        payload = json.loads(matches[0].group("payload"))
+    except json.JSONDecodeError as exc:
+        raise MetadataError(
+            f"PR metadata block contains invalid JSON: {exc.msg}."
+        ) from exc
+    if not isinstance(payload, dict):
+        raise MetadataError("PR metadata block must contain a JSON object.")
+    expected = {"slug", "index", "source_branch", "source_sha"}
+    missing = sorted(expected - payload.keys())
+    extra = sorted(payload.keys() - expected)
+    if missing:
+        raise MetadataError("PR metadata is missing field(s): " + ", ".join(missing))
+    if extra:
+        raise MetadataError("PR metadata has unknown field(s): " + ", ".join(extra))
+    if not isinstance(payload["slug"], str):
+        raise MetadataError("PR metadata field 'slug' must be a string.")
+    if not isinstance(payload["index"], int) or isinstance(payload["index"], bool):
+        raise MetadataError("PR metadata field 'index' must be an integer.")
+    if not isinstance(payload["source_branch"], str):
+        raise MetadataError("PR metadata field 'source_branch' must be a string.")
+    if not isinstance(payload["source_sha"], str):
+        raise MetadataError("PR metadata field 'source_sha' must be a string.")
+    return ChangesetMetadata(
+        slug=payload["slug"],
+        index=payload["index"],
+        source_branch=payload["source_branch"],
+        source_sha=payload["source_sha"],
+    )
diff --git a/skills/carve-changesets/scripts/rehydrate.py b/skills/carve-changesets/scripts/rehydrate.py
new file mode 100644
index 0000000..a083c32
--- /dev/null
+++ b/skills/carve-changesets/scripts/rehydrate.py
@@ -0,0 +1,245 @@
+"""Reconstruct changeset topology from live git refs and GitHub PR records."""
+
+from __future__ import annotations
+
+import re
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, Sequence
+
+from metadata import (
+    ChangesetMetadata,
+    MetadataError,
+    parse_commit_message,
+    parse_pr_metadata,
+)
+
+
+class RehydrationError(RuntimeError):
+    """Raised when live evidence cannot identify one unambiguous chain."""
+
+
+@dataclass(frozen=True)
+class PullRequestRecord:
+    """GitHub fields supplied by the consolidated CLI's gh chokepoint."""
+
+    number: int
+    head_branch: str
+    head_sha: str
+    base_branch: str
+    state: str
+    body: str
+
+
+@dataclass(frozen=True)
+class ChangesetRecord:
+    metadata: ChangesetMetadata
+    branch: str
+    head: str
+    base: str
+    pr_number: int | None = None
+    pr_state: str | None = None
+
+
+@dataclass(frozen=True)
+class Chain:
+    base_branch: str
+    source_branch: str
+    source_sha: str
+    changesets: tuple[ChangesetRecord, ...]
+
+
+def _git(cwd: Path, *args: str) -> str:
+    try:
+        result = subprocess.run(
+            ["git", *args],
+            cwd=cwd,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise RehydrationError(
+            "Git is required to rehydrate a changeset chain."
+        ) from exc
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        raise RehydrationError(f"git {' '.join(args)} failed: {detail}")
+    return result.stdout
+
+
+def _discover_heads(
+    cwd: Path, source_branch: str, remote: str
+) -> dict[int, tuple[str, str]]:
+    output = _git(
+        cwd,
+        "for-each-ref",
+        "--format=%(refname)%00%(objectname)",
+        "refs/heads",
+        f"refs/remotes/{remote}",
+    )
+    prefix = re.escape(source_branch)
+    local_pattern = re.compile(
+        rf"^refs/heads/(?P<branch>{prefix}-(?P<index>[1-9][0-9]*))$"
+    )
+    remote_pattern = re.compile(
+        rf"^refs/remotes/{re.escape(remote)}/(?P<branch>{prefix}-(?P<index>[1-9][0-9]*))$"
+    )
+    candidates: dict[int, dict[str, tuple[str, str]]] = {}
+    for line in output.splitlines():
+        ref, separator, head = line.partition("\0")
+        if not separator:
+            continue
+        match = remote_pattern.fullmatch(ref)
+        kind = "remote"
+        if match is None:
+            match = local_pattern.fullmatch(ref)
+            kind = "local"
+        if match is None:
+            continue
+        index = int(match.group("index"))
+        candidates.setdefault(index, {})[kind] = (match.group("branch"), head)
+
+    heads: dict[int, tuple[str, str]] = {}
+    for index, variants in candidates.items():
+        local = variants.get("local")
+        published = variants.get("remote")
+        if local and published and local[1] != published[1]:
+            raise RehydrationError(
+                f"Changeset branch {local[0]} is ambiguous: local head {local[1]} "
+                f"differs from {remote} head {published[1]}."
+            )
+        heads[index] = published or local  # type: ignore[assignment]
+    return heads
+
+
+def _pr_by_branch(
+    pull_requests: Iterable[PullRequestRecord], source_branch: str
+) -> dict[str, PullRequestRecord]:
+    pattern = re.compile(rf"^{re.escape(source_branch)}-[1-9][0-9]*$")
+    grouped: dict[str, list[PullRequestRecord]] = {}
+    for pr in pull_requests:
+        if pattern.fullmatch(pr.head_branch):
+            grouped.setdefault(pr.head_branch, []).append(pr)
+    duplicates = {branch: prs for branch, prs in grouped.items() if len(prs) > 1}
+    if duplicates:
+        detail = ", ".join(
+            f"{branch} -> PRs {', '.join(f'#{pr.number}' for pr in prs)}"
+            for branch, prs in sorted(duplicates.items())
+        )
+        raise RehydrationError(
+            f"Multiple PRs claim the same changeset branch: {detail}."
+        )
+    return {branch: prs[0] for branch, prs in grouped.items()}
+
+
+def rehydrate_chain(
+    *,
+    source_branch: str,
+    pull_requests: Sequence[PullRequestRecord] = (),
+    base_branch: str | None = None,
+    cwd: Path | str = Path.cwd(),
+    remote: str = "origin",
+) -> Chain:
+    """Reconstruct an ordered chain without consulting local plan or state files."""
+
+    if not source_branch.strip():
+        raise RehydrationError("Source branch must not be empty.")
+    repo = Path(cwd)
+    heads = _discover_heads(repo, source_branch, remote)
+    if not heads:
+        raise RehydrationError(
+            f"No changeset branches named {source_branch}-N were found locally or on {remote}."
+        )
+    found = sorted(heads)
+    expected = list(range(1, found[-1] + 1))
+    if found != expected:
+        missing = sorted(set(expected) - set(found))
+        raise RehydrationError(
+            "Changeset branch sequence has gap(s): missing index "
+            + ", ".join(str(index) for index in missing)
+            + "."
+        )
+
+    prs = _pr_by_branch(pull_requests, source_branch)
+    if base_branch is None:
+        first_pr = prs.get(f"{source_branch}-1")
+        if first_pr is None:
+            raise RehydrationError(
+                "Base branch is required when changeset 1 has no PR relationship."
+            )
+        base_branch = first_pr.base_branch
+    if not base_branch.strip():
+        raise RehydrationError("Base branch must not be empty.")
+
+    records: list[ChangesetRecord] = []
+    source_sha: str | None = None
+    slugs: set[str] = set()
+    for index in found:
+        branch, head = heads[index]
+        message = _git(repo, "show", "-s", "--format=%B", head)
+        try:
+            metadata = parse_commit_message(message)
+        except MetadataError as exc:
+            raise RehydrationError(f"Changeset branch {branch}: {exc}") from exc
+        if metadata.index != index:
+            raise RehydrationError(
+                f"Changeset branch {branch} has Changeset-Index {metadata.index}; expected {index}."
+            )
+        if metadata.source_branch != source_branch:
+            raise RehydrationError(
+                f"Changeset branch {branch} names source {metadata.source_branch!r}; "
+                f"expected {source_branch!r}."
+            )
+        if source_sha is None:
+            source_sha = metadata.source_sha
+        elif metadata.source_sha != source_sha:
+            raise RehydrationError(
+                f"Changeset branch {branch} names source SHA {metadata.source_sha}; "
+                f"expected {source_sha}."
+            )
+        if metadata.slug in slugs:
+            raise RehydrationError(
+                f"Duplicate changeset slug {metadata.slug!r} in chain."
+            )
+        slugs.add(metadata.slug)
+
+        expected_base = base_branch if index == 1 else f"{source_branch}-{index - 1}"
+        pr = prs.get(branch)
+        if pr is not None:
+            if pr.head_sha != head:
+                raise RehydrationError(
+                    f"PR #{pr.number} head {pr.head_sha} disagrees with branch {branch} head {head}."
+                )
+            if pr.base_branch != expected_base:
+                raise RehydrationError(
+                    f"PR #{pr.number} base {pr.base_branch!r} conflicts with expected "
+                    f"base {expected_base!r} for changeset {index}."
+                )
+            try:
+                pr_metadata = parse_pr_metadata(pr.body)
+            except MetadataError as exc:
+                raise RehydrationError(f"PR #{pr.number}: {exc}") from exc
+            if pr_metadata != metadata:
+                raise RehydrationError(
+                    f"PR #{pr.number} metadata disagrees with commit trailers for {branch}."
+                )
+        records.append(
+            ChangesetRecord(
+                metadata=metadata,
+                branch=branch,
+                head=head,
+                base=expected_base,
+                pr_number=pr.number if pr else None,
+                pr_state=pr.state.upper() if pr else None,
+            )
+        )
+
+    assert source_sha is not None
+    return Chain(
+        base_branch=base_branch,
+        source_branch=source_branch,
+        source_sha=source_sha,
+        changesets=tuple(records),
+    )
diff --git a/skills/carve-changesets/scripts/status.py b/skills/carve-changesets/scripts/status.py
new file mode 100644
index 0000000..9be57b2
--- /dev/null
+++ b/skills/carve-changesets/scripts/status.py
@@ -0,0 +1,56 @@
+"""Render live changeset chain status without local recordkeeping files."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Sequence
+
+from rehydrate import Chain, PullRequestRecord, rehydrate_chain
+
+
+def render_status(chain: Chain) -> str:
+    """Render branches, heads, PRs, bases, and merge state as a compact table."""
+
+    rows = [("INDEX", "SLUG", "BRANCH", "HEAD", "PR", "BASE", "STATE")]
+    for changeset in chain.changesets:
+        pr = f"#{changeset.pr_number}" if changeset.pr_number is not None else "-"
+        state = changeset.pr_state or "MATERIALIZED"
+        rows.append(
+            (
+                str(changeset.metadata.index),
+                changeset.metadata.slug,
+                changeset.branch,
+                changeset.head[:12],
+                pr,
+                changeset.base,
+                state,
+            )
+        )
+    widths = [max(len(row[column]) for row in rows) for column in range(len(rows[0]))]
+    return "\n".join(
+        "  ".join(
+            value.ljust(widths[column]) for column, value in enumerate(row)
+        ).rstrip()
+        for row in rows
+    )
+
+
+def status_from_live(
+    *,
+    source_branch: str,
+    pull_requests: Sequence[PullRequestRecord] = (),
+    base_branch: str | None = None,
+    cwd: Path | str = Path.cwd(),
+    remote: str = "origin",
+) -> str:
+    """Rehydrate and render status using only supplied GitHub records and git refs."""
+
+    return render_status(
+        rehydrate_chain(
+            source_branch=source_branch,
+            pull_requests=pull_requests,
+            base_branch=base_branch,
+            cwd=cwd,
+            remote=remote,
+        )
+    )
diff --git a/skills/carve-changesets/scripts/tests/helpers.py b/skills/carve-changesets/scripts/tests/helpers.py
new file mode 100644
index 0000000..ce18c15
--- /dev/null
+++ b/skills/carve-changesets/scripts/tests/helpers.py
@@ -0,0 +1,58 @@
+"""Test helpers for carve-changesets identity and rehydration."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+SCRIPTS_DIR = Path(__file__).resolve().parents[1]
+if str(SCRIPTS_DIR) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS_DIR))
+
+
+def run(cwd: Path, *args: str, input_text: str | None = None) -> str:
+    result = subprocess.run(
+        list(args),
+        cwd=cwd,
+        input=input_text,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise AssertionError(
+            f"Command failed ({result.returncode}): {' '.join(args)}\n"
+            f"stdout: {result.stdout}\nstderr: {result.stderr}"
+        )
+    return result.stdout.strip()
+
+
+def commit(cwd: Path, message: str) -> str:
+    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as message_file:
+        message_file.write(message)
+        message_file.flush()
+        run(cwd, "git", "commit", "-F", message_file.name)
+    return run(cwd, "git", "rev-parse", "HEAD")
+
+
+def init_repo(root: Path) -> tuple[Path, Path, str]:
+    bare = root / "remote.git"
+    repo = root / "builder"
+    run(root, "git", "init", "--bare", str(bare))
+    run(root, "git", "init", "-b", "main", str(repo))
+    run(repo, "git", "config", "user.name", "Carve Tests")
+    run(repo, "git", "config", "user.email", "carve@example.test")
+    (repo / "base.txt").write_text("base\n")
+    run(repo, "git", "add", "base.txt")
+    commit(repo, "initial")
+    run(repo, "git", "remote", "add", "origin", str(bare))
+    run(repo, "git", "push", "-u", "origin", "main")
+
+    run(repo, "git", "checkout", "-b", "feature/report")
+    (repo / "source.txt").write_text("complete source\n")
+    run(repo, "git", "add", "source.txt")
+    source_sha = commit(repo, "source result")
+    run(repo, "git", "push", "-u", "origin", "feature/report")
+    return repo, bare, source_sha
diff --git a/skills/carve-changesets/scripts/tests/test_metadata.py b/skills/carve-changesets/scripts/tests/test_metadata.py
new file mode 100644
index 0000000..313b732
--- /dev/null
+++ b/skills/carve-changesets/scripts/tests/test_metadata.py
@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import unittest
+
+import helpers  # noqa: F401
+from metadata import (
+    ChangesetMetadata,
+    MetadataError,
+    embed_pr_metadata,
+    parse_commit_message,
+    parse_pr_metadata,
+    render_pr_metadata,
+    stamp_commit_message,
+)
+
+
+class MetadataTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.metadata = ChangesetMetadata(
+            slug="api-foundation",
+            index=2,
+            source_branch="feature/report",
+            source_sha="a" * 40,
+        )
+
+    def test_commit_message_round_trips_through_git_interpret_trailers(self) -> None:
+        message = stamp_commit_message("feat: add API foundation", self.metadata)
+
+        self.assertIn("Changeset-Slug: api-foundation", message)
+        self.assertEqual(self.metadata, parse_commit_message(message))
+
+    def test_parse_commit_message_rejects_missing_trailer(self) -> None:
+        with self.assertRaisesRegex(MetadataError, "Changeset-Source"):
+            parse_commit_message(
+                "feat: incomplete\n\nChangeset-Slug: incomplete\nChangeset-Index: 1\n"
+            )
+
+    def test_pr_metadata_survives_human_body_edits(self) -> None:
+        body = embed_pr_metadata("## Summary\n\nOriginal prose.\n", self.metadata)
+        edited = "Reviewer context added.\n\n" + body.replace("Original", "Improved")
+
+        self.assertEqual(self.metadata, parse_pr_metadata(edited))
+
+    def test_embedding_replaces_one_existing_block(self) -> None:
+        old = ChangesetMetadata("old", 1, "feature/report", "b" * 40)
+        body = f"Human prose.\n\n{render_pr_metadata(old)}\n"
+
+        updated = embed_pr_metadata(body, self.metadata)
+
+        self.assertEqual(1, updated.count("carve-changesets:metadata:v1"))
+        self.assertEqual(self.metadata, parse_pr_metadata(updated))
+
+    def test_pr_metadata_rejects_multiple_blocks(self) -> None:
+        block = render_pr_metadata(self.metadata)
+        with self.assertRaisesRegex(MetadataError, "multiple"):
+            parse_pr_metadata(f"{block}\n{block}\n")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/skills/carve-changesets/scripts/tests/test_rehydrate.py b/skills/carve-changesets/scripts/tests/test_rehydrate.py
new file mode 100644
index 0000000..f0783dc
--- /dev/null
+++ b/skills/carve-changesets/scripts/tests/test_rehydrate.py
@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+import helpers
+from metadata import ChangesetMetadata, embed_pr_metadata, stamp_commit_message
+from rehydrate import PullRequestRecord, RehydrationError, rehydrate_chain
+from status import status_from_live
+
+
+class RehydrationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp_dir = Path(tempfile.mkdtemp())
+        self.repo, self.bare, self.source_sha = helpers.init_repo(self.temp_dir)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.temp_dir)
+
+    def _materialize(
+        self, indices: tuple[int, ...] = (1, 2)
+    ) -> tuple[dict[int, str], list[PullRequestRecord]]:
+        heads: dict[int, str] = {}
+        prs: list[PullRequestRecord] = []
+        previous = "main"
+        for index in indices:
+            branch = f"feature/report-{index}"
+            helpers.run(self.repo, "git", "checkout", "-b", branch, previous)
+            (self.repo / f"changeset-{index}.txt").write_text(f"changeset {index}\n")
+            helpers.run(self.repo, "git", "add", f"changeset-{index}.txt")
+            metadata = ChangesetMetadata(
+                slug=f"part-{index}",
+                index=index,
+                source_branch="feature/report",
+                source_sha=self.source_sha,
+            )
+            heads[index] = helpers.commit(
+                self.repo,
+                stamp_commit_message(f"feat: changeset {index}", metadata),
+            )
+            helpers.run(self.repo, "git", "push", "-u", "origin", branch)
+            base = "main" if index == 1 else f"feature/report-{index - 1}"
+            prs.append(
+                PullRequestRecord(
+                    number=100 + index,
+                    head_branch=branch,
+                    head_sha=heads[index],
+                    base_branch=base,
+                    state="MERGED" if index == 1 else "OPEN",
+                    body=embed_pr_metadata(
+                        f"## Overall Feature\n\nReport API\n\n## This Changeset ({index} of 2)\n",
+                        metadata,
+                    ),
+                )
+            )
+            previous = branch
+        return heads, prs
+
+    def _fresh_clone(self) -> Path:
+        clone = self.temp_dir / "fresh"
+        helpers.run(self.temp_dir, "git", "clone", str(self.bare), str(clone))
+        helpers.run(clone, "git", "fetch", "--prune", "origin")
+        return clone
+
+    def test_rehydrates_full_chain_after_local_state_is_deleted(self) -> None:
+        heads, prs = self._materialize()
+        state_dir = self.repo / ".carve-changesets"
+        state_dir.mkdir()
+        (state_dir / "plan.json").write_text("{}\n")
+        shutil.rmtree(state_dir)
+        clone = self._fresh_clone()
+
+        chain = rehydrate_chain(
+            source_branch="feature/report", pull_requests=prs, cwd=clone
+        )
+
+        self.assertEqual("main", chain.base_branch)
+        self.assertEqual(self.source_sha, chain.source_sha)
+        self.assertEqual(
+            ["part-1", "part-2"], [item.metadata.slug for item in chain.changesets]
+        )
+        self.assertEqual([heads[1], heads[2]], [item.head for item in chain.changesets])
+        self.assertEqual([101, 102], [item.pr_number for item in chain.changesets])
+        self.assertEqual(
+            ["main", "feature/report-1"], [item.base for item in chain.changesets]
+        )
+
+    def test_status_is_rendered_from_rehydration_without_local_files(self) -> None:
+        _, prs = self._materialize()
+        clone = self._fresh_clone()
+        output = status_from_live(
+            source_branch="feature/report", pull_requests=prs, cwd=clone
+        )
+
+        self.assertIn("feature/report-1", output)
+        self.assertIn("#101", output)
+        self.assertIn("MERGED", output)
+        self.assertIn("feature/report-2", output)
+        self.assertIn("OPEN", output)
+
+    def test_trailers_survive_propagation_rebase(self) -> None:
+        _, _ = self._materialize()
+        helpers.run(self.repo, "git", "checkout", "feature/report-1")
+        (self.repo / "upstream.txt").write_text("upstream\n")
+        helpers.run(self.repo, "git", "add", "upstream.txt")
+        helpers.commit(self.repo, "feat: update first changeset")
+        helpers.run(self.repo, "git", "checkout", "feature/report-2")
+        helpers.run(self.repo, "git", "rebase", "feature/report-1")
+        message = helpers.run(self.repo, "git", "show", "-s", "--format=%B", "HEAD")
+
+        from metadata import parse_commit_message
+
+        parsed = parse_commit_message(message)
+        self.assertEqual(2, parsed.index)
+        self.assertEqual("part-2", parsed.slug)
+        self.assertEqual(self.source_sha, parsed.source_sha)
+
+    def test_missing_branch_index_fails_closed(self) -> None:
+        self._materialize(indices=(1, 3))
+        clone = self._fresh_clone()
+
+        with self.assertRaisesRegex(RehydrationError, "missing index 2"):
+            rehydrate_chain(
+                source_branch="feature/report", base_branch="main", cwd=clone
+            )
+
+    def test_missing_commit_trailer_fails_closed(self) -> None:
+        helpers.run(self.repo, "git", "checkout", "-b", "feature/report-1", "main")
+        (self.repo / "plain.txt").write_text("plain\n")
+        helpers.run(self.repo, "git", "add", "plain.txt")
+        helpers.commit(self.repo, "feat: no trailers")
+        helpers.run(self.repo, "git", "push", "-u", "origin", "feature/report-1")
+        clone = self._fresh_clone()
+
+        with self.assertRaisesRegex(
+            RehydrationError, "Missing required changeset trailer"
+        ):
+            rehydrate_chain(
+                source_branch="feature/report", base_branch="main", cwd=clone
+            )
+
+    def test_conflicting_pr_base_fails_closed(self) -> None:
+        _, prs = self._materialize()
+        clone = self._fresh_clone()
+        conflicting = [
+            prs[0],
+            PullRequestRecord(**{**prs[1].__dict__, "base_branch": "main"}),
+        ]
+
+        with self.assertRaisesRegex(RehydrationError, "conflicts with expected base"):
+            rehydrate_chain(
+                source_branch="feature/report", pull_requests=conflicting, cwd=clone
+            )
+
+    def test_trailer_and_pr_metadata_disagreement_fails_closed(self) -> None:
+        _, prs = self._materialize()
+        clone = self._fresh_clone()
+        wrong = ChangesetMetadata("wrong", 2, "feature/report", self.source_sha)
+        conflicting = [
+            prs[0],
+            PullRequestRecord(
+                **{**prs[1].__dict__, "body": embed_pr_metadata(prs[1].body, wrong)}
+            ),
+        ]
+
+        with self.assertRaisesRegex(RehydrationError, "metadata disagrees"):
+            rehydrate_chain(
+                source_branch="feature/report", pull_requests=conflicting, cwd=clone
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()