Accept repository grant v2 with publish scope (ACTIVITY-WP-0046-T02)

Parses Activity Core's publish-capable grant: remote, branch, path
prefixes inside allowed_paths, fast-forward only. Grant id matches
activity-core 83e6446 byte for byte (test vector); v1 ids are unchanged.
Evidence adds remote, branch and fast_forward_only without raw paths.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 269692@bnt-lap001
Assistant-Session: 8077e68a-0985-4b41-898c-2dc602b0d6be

rein_aharness/repository_grant.py

@@ -4,35 +4,118 @@ from __future__ import annotations
 
 import hashlib
 import json
+import re
 from dataclasses import dataclass
 from typing import Any
 
 from rein_aharness.repository_transaction import RepositoryAcceptancePolicy
 
 REPOSITORY_GRANT_VERSION = "1"
+PUBLISH_GRANT_VERSION = "2"
 _GRANT_KEYS = frozenset({"version", "allowed_paths", "commit_count", "publish"})
+_PUBLISH_GRANT_KEYS = _GRANT_KEYS | {"publish_scope"}
 _COMMIT_COUNT_KEYS = frozenset({"min", "max"})
+_SCOPE_KEYS = frozenset({"remote", "branch", "path_prefixes", "fast_forward_only"})
+_MAX_PATH_PREFIXES = 20
+# Same rules as Activity Core's repository grant v2 (ACTIVITY-WP-0046).
+_REMOTE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,199}$")
 
 
 class RepositoryGrantError(ValueError):
     """A repository grant is missing, ambiguous, or unsupported."""
 
 
+@dataclass(frozen=True)
+class PublishScope:
+    """Where a version 2 grant may push: one remote branch, fast-forward only."""
+
+    remote: str
+    branch: str
+    path_prefixes: tuple[str, ...]
+    fast_forward_only: bool = True
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.remote, str) or not _REMOTE_RE.fullmatch(self.remote):
+            raise RepositoryGrantError("publish_scope.remote must be a plain remote name")
+        branch = self.branch
+        if (
+            not isinstance(branch, str)
+            or not _BRANCH_RE.fullmatch(branch)
+            or branch == "HEAD"
+            or branch.startswith("refs/")
+            or ".." in branch
+            or "//" in branch
+            or branch.endswith(("/", ".", ".lock"))
+        ):
+            raise RepositoryGrantError("publish_scope.branch must be a plain branch name")
+        prefixes = self.path_prefixes
+        if (
+            not isinstance(prefixes, tuple)
+            or not 1 <= len(prefixes) <= _MAX_PATH_PREFIXES
+            or any(not isinstance(prefix, str) for prefix in prefixes)
+        ):
+            raise RepositoryGrantError(
+                f"publish_scope.path_prefixes must hold 1-{_MAX_PATH_PREFIXES} strings"
+            )
+        if tuple(sorted(set(prefixes))) != prefixes:
+            raise RepositoryGrantError(
+                "publish_scope.path_prefixes must be unique and in canonical sorted order"
+            )
+        for prefix in prefixes:
+            _validate_path_pattern(prefix)
+        if self.fast_forward_only is not True:
+            raise RepositoryGrantError("publish_scope.fast_forward_only must be true")
+
+    @classmethod
+    def from_mapping(cls, value: Any) -> "PublishScope":
+        if not isinstance(value, dict):
+            raise RepositoryGrantError("publish_scope must be an object")
+        _require_exact_keys(value, _SCOPE_KEYS, "publish_scope")
+        raw = value["path_prefixes"]
+        if not isinstance(raw, list) or any(not isinstance(p, str) for p in raw):
+            raise RepositoryGrantError("publish_scope.path_prefixes must be a string array")
+        if len(set(raw)) != len(raw):
+            raise RepositoryGrantError("publish_scope.path_prefixes must not contain duplicates")
+        return cls(
+            remote=value["remote"],
+            branch=value["branch"],
+            path_prefixes=tuple(sorted(raw)),
+            fast_forward_only=value["fast_forward_only"],
+        )
+
+    def covers_path(self, path: str) -> bool:
+        return any(_covers(prefix, path) for prefix in self.path_prefixes)
+
+    def canonical(self) -> dict[str, Any]:
+        return {
+            "branch": self.branch,
+            "fast_forward_only": self.fast_forward_only,
+            "path_prefixes": list(self.path_prefixes),
+            "remote": self.remote,
+        }
+
+
 @dataclass(frozen=True)
 class RepositoryGrant:
-    """One explicit, local-only repository mutation authority envelope."""
+    """One explicit repository mutation authority envelope.
+
+    Version 1 is local-only. Version 2 additionally grants a fast-forward
+    push of the accepted commits within ``publish_scope``.
+    """
 
     version: str
     allowed_paths: tuple[str, ...]
     min_commits: int
     max_commits: int
     publish: bool
+    publish_scope: PublishScope | None = None
 
     def __post_init__(self) -> None:
-        if self.version != REPOSITORY_GRANT_VERSION:
+        if self.version not in (REPOSITORY_GRANT_VERSION, PUBLISH_GRANT_VERSION):
             raise RepositoryGrantError(
-                f"unsupported version {self.version!r}; "
-                f"expected {REPOSITORY_GRANT_VERSION!r}"
+                f"unsupported version {self.version!r}; expected "
+                f"{REPOSITORY_GRANT_VERSION!r} or {PUBLISH_GRANT_VERSION!r}"
             )
         if (
             not isinstance(self.allowed_paths, tuple)
@@ -50,10 +133,21 @@ class RepositoryGrant:
         _positive_int(self.max_commits, "max_commits")
         if not isinstance(self.publish, bool):
             raise RepositoryGrantError("publish must be a boolean")
-        if self.publish:
-            raise RepositoryGrantError(
-                "version 1 does not grant publication; publish must be false"
-            )
+        if self.version == REPOSITORY_GRANT_VERSION:
+            if self.publish or self.publish_scope is not None:
+                raise RepositoryGrantError(
+                    "version 1 does not grant publication; publish must be false"
+                )
+        else:
+            if not self.publish or not isinstance(self.publish_scope, PublishScope):
+                raise RepositoryGrantError(
+                    "version 2 requires publish: true and a publish_scope"
+                )
+            for prefix in self.publish_scope.path_prefixes:
+                if not any(_covers(allowed, prefix) for allowed in self.allowed_paths):
+                    raise RepositoryGrantError(
+                        "publish path prefix is outside allowed_paths"
+                    )
         try:
             RepositoryAcceptancePolicy(
                 allowed_paths=self.allowed_paths,
@@ -67,13 +161,19 @@ class RepositoryGrant:
     def from_mapping(cls, value: Any) -> "RepositoryGrant":
         if not isinstance(value, dict):
             raise RepositoryGrantError("must be an object")
-        _require_exact_keys(value, _GRANT_KEYS, "repository_grant")
-
+        if "version" not in value:
+            _require_exact_keys(value, _GRANT_KEYS, "repository_grant")
         version = value["version"]
-        if not isinstance(version, str) or version != REPOSITORY_GRANT_VERSION:
+        if version not in (REPOSITORY_GRANT_VERSION, PUBLISH_GRANT_VERSION):
             raise RepositoryGrantError(
-                f"unsupported version {version!r}; expected {REPOSITORY_GRANT_VERSION!r}"
+                f"unsupported version {version!r}; expected "
+                f"{REPOSITORY_GRANT_VERSION!r} or {PUBLISH_GRANT_VERSION!r}"
             )
+        _require_exact_keys(
+            value,
+            _GRANT_KEYS if version == REPOSITORY_GRANT_VERSION else _PUBLISH_GRANT_KEYS,
+            "repository_grant",
+        )
 
         raw_paths = value["allowed_paths"]
         if not isinstance(raw_paths, list) or not raw_paths:
@@ -94,10 +194,11 @@ class RepositoryGrant:
         publish = value["publish"]
         if not isinstance(publish, bool):
             raise RepositoryGrantError("publish must be a boolean")
-        if publish:
-            raise RepositoryGrantError(
-                "version 1 does not grant publication; publish must be false"
-            )
+        scope = (
+            PublishScope.from_mapping(value["publish_scope"])
+            if "publish_scope" in value
+            else None
+        )
 
         return cls(
             version=version,
@@ -105,6 +206,7 @@ class RepositoryGrant:
             min_commits=min_commits,
             max_commits=max_commits,
             publish=publish,
+            publish_scope=scope,
         )
 
     @property
@@ -134,22 +236,31 @@ class RepositoryGrant:
             "min_commits": self.min_commits,
             "max_commits": self.max_commits,
             "publish": self.publish,
+            **(
+                {
+                    "remote": self.publish_scope.remote,
+                    "branch": self.publish_scope.branch,
+                    "fast_forward_only": self.publish_scope.fast_forward_only,
+                    "publish_path_prefix_count": len(self.publish_scope.path_prefixes),
+                }
+                if self.publish_scope is not None
+                else {}
+            ),
         }
 
     def _canonical_json(self) -> str:
-        return json.dumps(
-            {
-                "allowed_paths": self.allowed_paths,
-                "commit_count": {
-                    "max": self.max_commits,
-                    "min": self.min_commits,
-                },
-                "publish": self.publish,
-                "version": self.version,
+        identity: dict[str, Any] = {
+            "allowed_paths": self.allowed_paths,
+            "commit_count": {
+                "max": self.max_commits,
+                "min": self.min_commits,
             },
-            sort_keys=True,
-            separators=(",", ":"),
-        )
+            "publish": self.publish,
+            "version": self.version,
+        }
+        if self.publish_scope is not None:
+            identity["publish_scope"] = self.publish_scope.canonical()
+        return json.dumps(identity, sort_keys=True, separators=(",", ":"))
 
 
 def _require_exact_keys(
@@ -175,3 +286,22 @@ def _positive_int(value: Any, field_name: str) -> int:
     if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
         raise RepositoryGrantError(f"{field_name} must be a positive integer")
     return value
+
+
+def _validate_path_pattern(pattern: str) -> None:
+    parts = pattern.rstrip("/").split("/")
+    if (
+        not pattern
+        or len(pattern) > 256
+        or pattern.startswith("/")
+        or "\\" in pattern
+        or "\0" in pattern
+        or any(part in {"", ".", ".."} for part in parts)
+        or parts[0] == ".git"
+    ):
+        raise RepositoryGrantError("publish path prefix must be repository-relative")
+
+
+def _covers(allowed: str, path: str) -> bool:
+    allowed_parts = allowed.rstrip("/").split("/")
+    return path.rstrip("/").split("/")[: len(allowed_parts)] == allowed_parts

tests/test_repository_grant.py

@@ -145,7 +145,8 @@ def test_repository_grant_drives_transaction_acceptance(tmp_path: Path) -> None:
         (_grant(extra="value"), "unknown: extra"),
         ({1: "not-a-field"}, "field names must be strings"),
         (_grant(version=1), "unsupported version"),
-        (_grant(version="2"), "unsupported version"),
+        (_grant(version="2"), "missing: publish_scope"),
+        (_grant(version="3"), "unsupported version"),
     ],
 )
 def test_repository_grant_rejects_invalid_envelope(
@@ -452,3 +453,80 @@ def _git(repo: Path, *args: str) -> str:
         capture_output=True,
         text=True,
     ).stdout.strip()
+
+
+def _publish_grant(**scope_changes: object) -> dict:
+    scope: dict = {
+        "remote": "origin",
+        "branch": "main",
+        "path_prefixes": ["checkpoints/"],
+        "fast_forward_only": True,
+    }
+    scope.update(scope_changes)
+    return _grant(
+        version="2",
+        allowed_paths=["checkpoints/"],
+        commit_count={"min": 1, "max": 1},
+        publish=True,
+        publish_scope=scope,
+    )
+
+
+def test_v2_grant_id_matches_activity_core() -> None:
+    # Vector computed with activity-core 83e6446 (ACTIVITY-WP-0046).
+    grant = RepositoryGrant.from_mapping(_publish_grant())
+    assert grant.grant_id == "e0a395aa3ae8f810849d02e277b11828"
+    assert grant.acceptance_policy().policy_id == "9dcedd6a46f505008616429a603dd87c"
+    narrower = RepositoryGrant.from_mapping(
+        _publish_grant(path_prefixes=["checkpoints/2026/"])
+    )
+    assert narrower.grant_id != grant.grant_id
+    reordered = RepositoryGrant.from_mapping(
+        _publish_grant(path_prefixes=["checkpoints/b/", "checkpoints/a/"])
+    )
+    assert reordered.grant_id == RepositoryGrant.from_mapping(
+        _publish_grant(path_prefixes=["checkpoints/a/", "checkpoints/b/"])
+    ).grant_id
+
+
+def test_v2_grant_scope_covers_only_prefixed_paths() -> None:
+    scope = RepositoryGrant.from_mapping(_publish_grant()).publish_scope
+    assert scope is not None
+    assert scope.covers_path("checkpoints/2026-10.md")
+    assert not scope.covers_path("checkpoints-other/x.md")
+    assert not scope.covers_path("docs/x.md")
+
+
+def test_v2_grant_evidence_names_scope_without_paths() -> None:
+    evidence = RepositoryGrant.from_mapping(_publish_grant()).evidence()
+    assert evidence["publish"] is True
+    assert evidence["remote"] == "origin"
+    assert evidence["branch"] == "main"
+    assert evidence["fast_forward_only"] is True
+    assert "checkpoints/" not in json.dumps(evidence)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        _publish_grant(fast_forward_only=False),
+        _publish_grant(remote="https://example.org/repo.git"),
+        _publish_grant(branch="refs/heads/main"),
+        _publish_grant(branch="feature/../main"),
+        _publish_grant(branch="-force"),
+        _publish_grant(branch="HEAD"),
+        _publish_grant(path_prefixes=["docs/"]),
+        _publish_grant(path_prefixes=["checkpoints-other/"]),
+        _publish_grant(path_prefixes=[]),
+        _publish_grant(path_prefixes=["checkpoints/"] * 2),
+        _publish_grant(path_prefixes=[f"checkpoints/{i}/" for i in range(21)]),
+        _publish_grant(force=True),
+        {**_publish_grant(), "publish": False},
+        {**_publish_grant(), "version": "1"},
+        _grant(version="2", publish=True),
+        _grant(publish_scope=_publish_grant()["publish_scope"]),
+    ],
+)
+def test_v2_grant_rejects_unsafe_or_inconsistent_publish_scope(value: dict) -> None:
+    with pytest.raises(RepositoryGrantError):
+        RepositoryGrant.from_mapping(value)