Add durable native execution fences for task transfer

amplifier_foundation/session/__init__.py

@@ -154,7 +154,9 @@
 from .shared_state import (
     FileStamp,
     HeldSession,
+    HeldTransfer,
     SessionBusyError,
+    SessionTransferFencedError,
     SharedSessionStore,
     file_stamp,
 )
@@ -231,6 +233,8 @@
     "FileStamp",
     "file_stamp",
     "SessionBusyError",
+    "SessionTransferFencedError",
     "SharedSessionStore",
     "HeldSession",
+    "HeldTransfer",
 ]

amplifier_foundation/session/shared_state.py

@@ -21,7 +21,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 try:  # Fail loudly on platforms without the locking primitive.
     import fcntl
@@ -80,6 +80,14 @@ class SharedStateError(RuntimeError):
     """Raised when shared-state storage or a checkpoint is invalid."""
 
 
+class SessionTransferFencedError(RuntimeError):
+    """A durable transfer marker prevents ordinary session execution."""
+
+    def __init__(self, fence: dict[str, Any]) -> None:
+        self.fence = copy.deepcopy(fence)
+        super().__init__("session execution is fenced by a durable transfer")
+
+
 def _default_root() -> Path:
     configured = os.environ.get("AMPLIFIER_SESSION_STATE_HOME")
     if configured:
@@ -156,6 +164,36 @@ def _validate_private_file(path: Path) -> None:
         raise SharedStateError(f"state file {path.name} has unsafe permissions")
 
 
+def _mkdir_private_durable(path: Path) -> None:
+    """Create at most 128 private ancestors, persisting each new directory entry.
+
+    A marker's own parent fsync cannot make a newly created ancestor durable.
+    Build from the existing ancestor outward and sync each containing directory
+    before publishing any descendant. Existing directory modes are untouched.
+    """
+    missing = []
+    current = path
+    while True:
+        try:
+            current.lstat()
+        except FileNotFoundError:
+            if len(missing) >= 128 or current == current.parent:
+                raise SharedStateError("native transfer directory exceeds the creation depth limit") from None
+            missing.append(current)
+            current = current.parent
+        else:
+            if not current.is_dir():
+                raise SharedStateError("native transfer ancestor is not a directory")
+            break
+    for directory in reversed(missing):
+        try:
+            directory.mkdir(mode=0o700)
+        except FileExistsError:
+            pass
+        _validate_private_directory(directory, create=False)
+        _fsync_directory(directory.parent)
+
+
 def _ensure_private_state_path(root: Path, directory: Path) -> None:
     """Create the state-root chain, validating each state-owned component."""
 
@@ -253,19 +291,39 @@ def _atomic_json(path: Path, value: dict[str, Any]) -> None:
 
 def _fsync_directory(directory: Path) -> None:
     """Persist a directory entry where the local POSIX filesystem supports it."""
-
-    try:
-        directory_fd = os.open(directory, os.O_RDONLY)
-    except OSError:
-        return
+    directory_fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
     try:
         os.fsync(directory_fd)
-    except OSError:
-        pass  # Some POSIX filesystems do not support directory fsync.
+    except OSError as exc:
+        # These errors specifically mean directory fsync is unsupported. I/O,
+        # space, permission, bad-descriptor and open errors must prevent ack.
+        unsupported = {errno.EINVAL, errno.ENOSYS, errno.ENOTSUP, errno.EOPNOTSUPP}
+        if exc.errno not in unsupported:
+            raise
     finally:
         os.close(directory_fd)
 
 
+def _fsync_ancestor_chain(directory: Path) -> None:
+    """Re-establish ancestor durability, including after a failed earlier mkdir.
+
+    An existing directory may be the residue of a failed unacknowledged write,
+    so existence cannot stand in for parent durability on a retry. Bound the
+    complete chain before doing I/O, then sync from filesystem root inward.
+    """
+    ancestors = []
+    current = directory.absolute()
+    while True:
+        if len(ancestors) >= 128:
+            raise SharedStateError("native transfer directory exceeds the durability depth limit")
+        ancestors.append(current)
+        if current == current.parent:
+            break
+        current = current.parent
+    for ancestor in reversed(ancestors):
+        _fsync_directory(ancestor)
+
+
 def _process_start_identity(path: Path = Path("/proc/self/stat")) -> str | None:
     """Read Linux procfs field 22, handling process names containing spaces."""
 
@@ -360,6 +418,58 @@ def __init__(self, workspace: str | os.PathLike[str], session_id: str, *, root:
         self.session_id = _session_id(session_id)
         self.root = Path(root).expanduser() if root is not None else _default_root()
         self._directory = self.root / "v1" / _workspace_key(self.workspace) / self.session_id
+        home = Path(os.environ.get("AMPLIFIER_HOME") or Path.home() / ".amplifier").expanduser().absolute()
+        slug = str(self.workspace).replace("/", "-").replace("\\", "-").replace(":", "")
+        self._native_root = home
+        self._native_directory = home / "projects" / (slug if slug.startswith("-") else "-" + slug) / "sessions" / self.session_id
+
+    @property
+    def transfer_fence_path(self) -> Path:
+        """Native-history marker; reading this property creates no files."""
+        return self._native_directory / "transfer-fence.json"
+
+    def _transfer_directory(self, *, create: bool = False) -> None:
+        # Native histories predate private coordination roots, so do not change
+        # existing directory modes. Reject links, non-directories and foreign
+        # owners; the marker itself must always be private regular data.
+        relative = self._native_directory.relative_to(self._native_root)
+        directories = [self._native_root]
+        for part in relative.parts:
+            directories.append(directories[-1] / part)
+        for directory in directories:
+            try:
+                info = directory.lstat()
+            except FileNotFoundError:
+                if not create:
+                    return
+                _mkdir_private_durable(directory)
+                info = directory.lstat()
+            if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid():
+                raise SharedStateError("native session transfer directory is unsafe")
+        _validate_private_file(self.transfer_fence_path)
+
+    def transfer_fence(self) -> dict[str, Any] | None:
+        """Inspect durable transfer state without acquiring or starting work."""
+        self._transfer_directory()
+        value = _read_json(self.transfer_fence_path, missing_ok=True)
+        if value is None:
+            return None
+        required = {"version", "session_id", "transfer_id", "destination_host", "role", "phase"}
+        if (set(value) != required or type(value["version"]) is not int or value["version"] != 1
+                or value["session_id"] != self.session_id
+                or not isinstance(value["transfer_id"], str) or not _ID_RE.fullmatch(value["transfer_id"])
+                or not isinstance(value["destination_host"], str) or not 1 <= len(value["destination_host"]) <= 255
+                or any(ord(char) < 32 for char in value["destination_host"])
+                or not isinstance(value["role"], str) or value["role"] not in {"source", "destination"}
+                or not isinstance(value["phase"], str) or value["phase"] not in {"staged", "committed"}
+                or (value["role"] == "destination" and value["phase"] != "staged")):
+            raise SharedStateError("invalid session transfer fence")
+        return value
+
+    def _save_transfer_fence(self, record: dict[str, Any]) -> None:
+        self._transfer_directory(create=True)
+        _fsync_ancestor_chain(self.transfer_fence_path.parent)
+        _atomic_json(self.transfer_fence_path, record)
 
     @property
     def checkpoint_path(self) -> Path:
@@ -400,6 +510,34 @@ def read(self) -> dict[str, Any] | None:
 
     def acquire(self, *, app: str, **diagnostics: Any) -> "HeldSession":
         """Acquire the stable OS lock and publish advisory owner diagnostics."""
+        return self._acquire(app=app, diagnostics=diagnostics)
+
+    def acquire_transfer(self, transfer_id: str, *, app: str, **diagnostics: Any) -> "HeldTransfer":
+        """Acquire only a matching staged fence for explicit adapter recovery.
+
+        This does not grant execution or validate remote release evidence. The
+        application must authenticate the transfer before clearing/committing.
+        Committed source markers cannot be acquired through this API.
+        """
+        _session_id(transfer_id)
+        return cast("HeldTransfer", self._acquire(app=app, diagnostics=diagnostics, transfer_id=transfer_id))
+
+    def confirm_transfer_commit(self, transfer_id: str, *, app: str, **diagnostics: Any) -> dict[str, Any]:
+        """Durably confirm an exact committed source after an uncertain ack.
+
+        This returns evidence only, never an execution or clearing capability.
+        It cannot commit a staged marker or change the recorded destination.
+        """
+        _session_id(transfer_id)
+        held = cast("HeldTransfer", self._acquire(app=app, diagnostics=diagnostics,
+                    transfer_id=transfer_id, committed_only=True))
+        try:
+            return held.commit_transfer()
+        finally:
+            held.release()
+
+    def _acquire(self, *, app: str, diagnostics: dict[str, Any], transfer_id: str | None = None,
+                 committed_only: bool = False) -> "HeldSession":
         _ensure_supported()
         owner = _owner_details(app, diagnostics, self.workspace, self.session_id, self.root)
         _ensure_private_state_path(self.root, self._directory)
@@ -424,13 +562,25 @@ def acquire(self, *, app: str, **diagnostics: Any) -> "HeldSession":
                     pass
                 raise SharedStateError("cannot acquire stable session lock") from exc
             try:
+                fence = self.transfer_fence()
+                if transfer_id is None:
+                    if fence is not None:
+                        raise SessionTransferFencedError(fence)
+                elif committed_only:
+                    if (fence is None or fence["transfer_id"] != transfer_id
+                            or fence["phase"] != "committed" or fence["role"] != "source"):
+                        raise SharedStateError("commit confirmation requires the exact committed source fence")
+                elif fence is None or fence["transfer_id"] != transfer_id or fence["phase"] != "staged":
+                    raise SharedStateError("transfer recovery requires the exact staged fence")
                 _atomic_json(self._owner_path, owner)
             except BaseException:
                 fcntl.flock(fd, fcntl.LOCK_UN)
                 os.close(fd)
                 raise
             _PROCESS_LOCKS.add(lock_name)
             _LOCK_FDS.add(fd)
+        if transfer_id is not None:
+            return HeldTransfer(self, fd, owner, transfer_id)
         return HeldSession(self, fd, owner)
 
     @classmethod
@@ -482,11 +632,41 @@ def active(self) -> bool:
         return self._active and self._pid == os.getpid()
 
     def check(self) -> None:
+        self._check_active()
+        fence = self._store.transfer_fence()
+        if fence is not None:
+            raise SessionTransferFencedError(fence)
+
+    def _check_active(self) -> None:
         if self._pid != os.getpid():
             raise RuntimeError("HeldSession belongs to a different process")
         if not self._active:
             raise RuntimeError("HeldSession is no longer active")
 
+    def fence_transfer(self, transfer_id: str, destination_host: str, *, role: str = "source") -> dict[str, Any]:
+        """Fence this saved, quiescent writer before releasing it for transfer.
+
+        The host must settle work and persist native history first. The marker
+        does not stop a runtime or prove any remote outcome. Once written this
+        handle cannot authorize further execution or checkpoint writes.
+        """
+        _session_id(transfer_id)
+        if (not isinstance(destination_host, str) or not 1 <= len(destination_host) <= 255
+                or any(ord(char) < 32 for char in destination_host) or role not in {"source", "destination"}):
+            raise ValueError("invalid transfer destination or role")
+        record = {"version": 1, "session_id": self._store.session_id, "transfer_id": transfer_id,
+                  "destination_host": destination_host, "role": role, "phase": "staged"}
+        with self._mutex:
+            self._check_active()
+            existing = self._store.transfer_fence()
+            if existing is not None:
+                if existing != record:
+                    raise SharedStateError("a different transfer fence already exists")
+            # Even an identical marker can be left by a failed fsync. Repeat
+            # file and ancestor durability before acknowledging that request.
+            self._store._save_transfer_fence(record)
+            return copy.deepcopy(record)
+
     def read(self) -> dict[str, Any] | None:
         self.check()
         return self._store.read()
@@ -562,3 +742,49 @@ def release(self) -> None:
                         _PROCESS_LOCKS.discard(lock_name)
                         _LOCK_FDS.discard(self._fd)
                     os.close(self._fd)
+
+
+class HeldTransfer(HeldSession):
+    """A receipt-matched lock for resolving a staged fence, never execution."""
+
+    def __init__(self, store: SharedSessionStore, fd: int, owner: dict[str, Any], transfer_id: str) -> None:
+        super().__init__(store, fd, owner)
+        self._transfer_id = transfer_id
+
+    def check(self) -> None:
+        self._check_active()
+        raise SharedStateError("transfer recovery capability cannot authorize execution")
+
+    def fence_transfer(self, transfer_id: str, destination_host: str, *, role: str = "source") -> dict[str, Any]:
+        raise SharedStateError("transfer recovery capability cannot replace a fence")
+
+    def _staged(self) -> dict[str, Any]:
+        self._check_active()
+        record = self._store.transfer_fence()
+        if record is None or record["transfer_id"] != self._transfer_id or record["phase"] != "staged":
+            raise SharedStateError("transfer recovery requires the exact staged fence")
+        return record
+
+    def commit_transfer(self) -> dict[str, Any]:
+        """Permanently fence the source after the host verifies its transfer."""
+        with self._mutex:
+            self._check_active()
+            record = self._store.transfer_fence()
+            if record is None or record["transfer_id"] != self._transfer_id:
+                raise SharedStateError("transfer commit requires the exact source fence")
+            if record["role"] != "source":
+                raise SharedStateError("only a source transfer fence can be committed")
+            record["phase"] = "committed"
+            self._store._save_transfer_fence(record)
+            return record
+
+    def clear_transfer(self) -> None:
+        """Explicitly cancel a staged source or activate a staged destination.
+
+        Authenticate the corresponding cancellation or release receipt before
+        calling. Release this capability and acquire normally before execution.
+        """
+        with self._mutex:
+            self._staged()
+            self._store.transfer_fence_path.unlink()
+            _fsync_directory(self._store.transfer_fence_path.parent)

docs/API_REFERENCE.md

@@ -116,6 +116,7 @@ Portable same-host checkpoints live in `session/shared_state.py`. They are suppo
 | `SharedSessionStore` | `session/shared_state.py` | Addresses a checkpoint; `acquire(app=...)` returns the exclusive writer capability. |
 | `HeldSession` | `session/shared_state.py` | Process-bound, non-copyable writer; atomically writes messages, portable bundle reference, and credential-safe metadata. Its live `delete_checkpoint()` safely removes only the authoritative checkpoint. |
 | `SessionBusyError` | `session/shared_state.py` | Contention error with advisory, bounded owner diagnostics. |
+| `SessionTransferFencedError`, `HeldTransfer` | `session/shared_state.py` | Durable native-history admission fence and restricted receipt-matched resolution capability; see [session transfer fences](SESSION_TRANSFER_FENCE.md). |
 
 Import these names from `amplifier_foundation.session`. Store construction, `read`, `stamp`, and `list_ids` never create state directories. `acquire` alone creates or validates private state directories and the stable `session.lock`; it rejects symlinked, foreign-owned, or group/world-accessible state paths. `release` never rewrites the authoritative checkpoint. See [SHARED_SESSION_STATE.md](SHARED_SESSION_STATE.md) for the same-host participant contract, warm-reuse guidance, and safe checkpoint deletion.