Skip to content

Session contracts

session_contracts

SessionContractRegistry

Thread-safe registry of write contracts keyed by session identity.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class SessionContractRegistry:
    """Thread-safe registry of write contracts keyed by session identity."""

    def __init__(
        self,
        *,
        clock: Callable[[], float],
        ttl_seconds: float = 3_600.0,
    ) -> None:
        self._clock = clock
        self._ttl_seconds = ttl_seconds
        self._bindings: dict[str, tuple[WriteContract, float]] = {}
        self._lock = Lock()

    def bind(self, session_id: str, contract: WriteContract) -> None:
        """Bind a write contract to one session."""
        with self._lock:
            self._bindings[session_id] = (contract, self._clock())

    def resolve(self, session_id: str) -> WriteContract:
        """Return the contract bound to a session."""
        with self._lock:
            binding = self._bindings.get(session_id)
            if binding is None:
                raise UnboundSessionError(
                    f"no write contract bound to session {session_id!r}"
                )
            return binding[0]

    def release(self, session_id: str) -> None:
        """Drop a session binding if it exists."""
        with self._lock:
            self._bindings.pop(session_id, None)

    def purge_expired(self, now: float) -> None:
        """Drop bindings whose age is greater than the configured TTL."""
        with self._lock:
            expired = [
                session_id
                for session_id, (_, bound_at) in self._bindings.items()
                if now - bound_at > self._ttl_seconds
            ]
            for session_id in expired:
                del self._bindings[session_id]
bind(session_id, contract)

Bind a write contract to one session.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
67
68
69
70
def bind(self, session_id: str, contract: WriteContract) -> None:
    """Bind a write contract to one session."""
    with self._lock:
        self._bindings[session_id] = (contract, self._clock())
purge_expired(now)

Drop bindings whose age is greater than the configured TTL.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
87
88
89
90
91
92
93
94
95
96
def purge_expired(self, now: float) -> None:
    """Drop bindings whose age is greater than the configured TTL."""
    with self._lock:
        expired = [
            session_id
            for session_id, (_, bound_at) in self._bindings.items()
            if now - bound_at > self._ttl_seconds
        ]
        for session_id in expired:
            del self._bindings[session_id]
release(session_id)

Drop a session binding if it exists.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
82
83
84
85
def release(self, session_id: str) -> None:
    """Drop a session binding if it exists."""
    with self._lock:
        self._bindings.pop(session_id, None)
resolve(session_id)

Return the contract bound to a session.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
72
73
74
75
76
77
78
79
80
def resolve(self, session_id: str) -> WriteContract:
    """Return the contract bound to a session."""
    with self._lock:
        binding = self._bindings.get(session_id)
        if binding is None:
            raise UnboundSessionError(
                f"no write contract bound to session {session_id!r}"
            )
        return binding[0]

UnboundSessionError

Bases: RuntimeError

Raised when no write contract is bound to a session.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
19
20
class UnboundSessionError(RuntimeError):
    """Raised when no write contract is bound to a session."""

WriteContract dataclass

Validated wire representation of a session filesystem write scope.

A prefix listed in markdown_only_prefixes carries a nature on top of its path: it grants Markdown sidecars only. Without it, a documentation prefix would have to be narrowed by whoever produces the contract, and every producer would reinvent that filter — the divergence this field exists to prevent. Such a prefix must also appear in allowed_prefixes; one that does not is dropped rather than silently granting a wider path.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@dataclass(frozen=True)
class WriteContract:
    """Validated wire representation of a session filesystem write scope.

    A prefix listed in ``markdown_only_prefixes`` carries a **nature** on top
    of its path: it grants Markdown sidecars only. Without it, a documentation
    prefix would have to be narrowed by whoever *produces* the contract, and
    every producer would reinvent that filter — the divergence this field
    exists to prevent. Such a prefix must also appear in ``allowed_prefixes``;
    one that does not is dropped rather than silently granting a wider path.
    """

    execution_root: str
    allowed_prefixes: tuple[str, ...] = ()
    markdown_only_prefixes: tuple[str, ...] = ()

    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> WriteContract:
        """Validate and normalize a transported contract mapping."""
        raw_root = raw.get("execution_root")
        if not isinstance(raw_root, str) or not raw_root.strip():
            raise ValueError("execution_root must be a non-empty string")
        root = os.path.realpath(raw_root.strip())
        prefixes = _normalized_prefixes(raw, "allowed_prefixes", root)
        markdown_only = _normalized_prefixes(raw, "markdown_only_prefixes", root)
        return cls(
            execution_root=root,
            allowed_prefixes=prefixes,
            markdown_only_prefixes=tuple(
                prefix for prefix in markdown_only if prefix in prefixes
            ),
        )

    @classmethod
    def from_json(cls, raw: str) -> WriteContract:
        """Decode, validate and normalize a JSON transport payload."""
        decoded = json.loads(raw)
        if not isinstance(decoded, Mapping):
            raise ValueError("write contract must be a JSON object")
        return cls.from_mapping(decoded)

    def resolve(self, base: str, candidate: str) -> str:
        """Resolve a candidate location relative to an explicit base."""
        return _absolute_location(base, candidate)

    def permits(self, location: str) -> bool:
        """Return whether a resolved location is under an allowed prefix.

        A location granted *only* by Markdown-restricted prefixes must itself
        be Markdown: those prefixes carry a nature, not merely a path. A
        location also covered by an unrestricted prefix keeps that grant.
        """
        granting = [
            prefix
            for prefix in self.allowed_prefixes
            if location == prefix or location.startswith(prefix + os.sep)
        ]
        if not granting:
            return False
        if any(prefix not in self.markdown_only_prefixes for prefix in granting):
            return True
        return location.casefold().endswith(_MARKDOWN_SUFFIXES)

    def contains(self, location: str) -> bool:
        """Return whether a resolved location is under the execution root."""
        root = self.execution_root
        return location == root or location.startswith(root + os.sep)
contains(location)

Return whether a resolved location is under the execution root.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
126
127
128
129
def contains(self, location: str) -> bool:
    """Return whether a resolved location is under the execution root."""
    root = self.execution_root
    return location == root or location.startswith(root + os.sep)
from_json(raw) classmethod

Decode, validate and normalize a JSON transport payload.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
 96
 97
 98
 99
100
101
102
@classmethod
def from_json(cls, raw: str) -> WriteContract:
    """Decode, validate and normalize a JSON transport payload."""
    decoded = json.loads(raw)
    if not isinstance(decoded, Mapping):
        raise ValueError("write contract must be a JSON object")
    return cls.from_mapping(decoded)
from_mapping(raw) classmethod

Validate and normalize a transported contract mapping.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> WriteContract:
    """Validate and normalize a transported contract mapping."""
    raw_root = raw.get("execution_root")
    if not isinstance(raw_root, str) or not raw_root.strip():
        raise ValueError("execution_root must be a non-empty string")
    root = os.path.realpath(raw_root.strip())
    prefixes = _normalized_prefixes(raw, "allowed_prefixes", root)
    markdown_only = _normalized_prefixes(raw, "markdown_only_prefixes", root)
    return cls(
        execution_root=root,
        allowed_prefixes=prefixes,
        markdown_only_prefixes=tuple(
            prefix for prefix in markdown_only if prefix in prefixes
        ),
    )
permits(location)

Return whether a resolved location is under an allowed prefix.

A location granted only by Markdown-restricted prefixes must itself be Markdown: those prefixes carry a nature, not merely a path. A location also covered by an unrestricted prefix keeps that grant.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def permits(self, location: str) -> bool:
    """Return whether a resolved location is under an allowed prefix.

    A location granted *only* by Markdown-restricted prefixes must itself
    be Markdown: those prefixes carry a nature, not merely a path. A
    location also covered by an unrestricted prefix keeps that grant.
    """
    granting = [
        prefix
        for prefix in self.allowed_prefixes
        if location == prefix or location.startswith(prefix + os.sep)
    ]
    if not granting:
        return False
    if any(prefix not in self.markdown_only_prefixes for prefix in granting):
        return True
    return location.casefold().endswith(_MARKDOWN_SUFFIXES)
resolve(base, candidate)

Resolve a candidate location relative to an explicit base.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
104
105
106
def resolve(self, base: str, candidate: str) -> str:
    """Resolve a candidate location relative to an explicit base."""
    return _absolute_location(base, candidate)

WriteContractHeaderError

Bases: ValueError

Raised when the X-AXM-Write-Contract header cannot be decoded.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
32
33
class WriteContractHeaderError(ValueError):
    """Raised when the X-AXM-Write-Contract header cannot be decoded."""

parse_write_contract_header(raw)

Decode and validate an X-AXM-Write-Contract header value.

Source code in packages/axm-mcp/src/axm_mcp/session_contracts.py
Python
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def parse_write_contract_header(raw: str) -> WriteContract:
    """Decode and validate an X-AXM-Write-Contract header value."""
    try:
        payload = _WRITE_CONTRACT_ADAPTER.validate_json(raw)
    except ValidationError as exc:
        raise WriteContractHeaderError(
            f"invalid X-AXM-Write-Contract header: {exc}"
        ) from exc
    return WriteContract(
        execution_root=payload["execution_root"],
        allowed_prefixes=cast("tuple[str, ...]", payload["allowed_prefixes"]),
        markdown_only_prefixes=cast(
            "tuple[str, ...]", payload["markdown_only_prefixes"]
        ),
    )