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
51
52
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
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
65
66
67
68
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
85
86
87
88
89
90
91
92
93
94
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
80
81
82
83
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
70
71
72
73
74
75
76
77
78
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
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
@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, ...] = ()
    dependency_environment: DependencyEnvironment | None = None
    read_roots: tuple[str, ...] = ()
    read_path_aliases: tuple[tuple[str, str], ...] = ()
    dependency_environments: tuple[DependencyEnvironment, ...] = ()

    @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)
        reads = _normalized_prefixes(raw, "read_roots", root) or (root,)
        aliases = _read_aliases(raw.get("read_path_aliases", ()), reads)
        markdown_only = _normalized_prefixes(raw, "markdown_only_prefixes", root)
        environment = raw.get("dependency_environment")
        if environment is not None and not isinstance(environment, Mapping):
            raise ValueError("dependency_environment must be an object")
        if isinstance(environment, Mapping):
            read_env = DependencyEnvironment.from_mapping(environment)
            if read_env.package_root != root:
                raise ValueError("dependency package_root must match execution_root")
        extra = raw.get("dependency_environments", ())
        if not isinstance(extra, (tuple, list)) or any(
            not isinstance(env, Mapping) for env in extra
        ):
            raise ValueError("dependency_environments must be a sequence of objects")
        return cls(
            dependency_environments=tuple(
                DependencyEnvironment.from_mapping(env) for env in extra
            ),
            dependency_environment=(
                DependencyEnvironment.from_mapping(environment)
                if isinstance(environment, Mapping)
                else None
            ),
            execution_root=root,
            read_roots=reads,
            read_path_aliases=aliases,
            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 readable(self, location: str) -> bool:
        """Accept local reads and declared sources resolved by the test interpreter."""
        location = os.path.realpath(location)
        if any(
            location == root or location.startswith(root + os.sep)
            for root in (self.read_roots or (self.execution_root,))
        ):
            return True
        environments = self.dependency_environments
        if self.dependency_environment is not None:
            environments = (self.dependency_environment, *environments)
        return any(dependency_sources(env).permits(location) for env in environments)

    def contains(self, location: str) -> bool:
        """Return whether a resolved location is under the execution root."""
        root = os.path.realpath(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
191
192
193
194
def contains(self, location: str) -> bool:
    """Return whether a resolved location is under the execution root."""
    root = os.path.realpath(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
148
149
150
151
152
153
154
@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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@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)
    reads = _normalized_prefixes(raw, "read_roots", root) or (root,)
    aliases = _read_aliases(raw.get("read_path_aliases", ()), reads)
    markdown_only = _normalized_prefixes(raw, "markdown_only_prefixes", root)
    environment = raw.get("dependency_environment")
    if environment is not None and not isinstance(environment, Mapping):
        raise ValueError("dependency_environment must be an object")
    if isinstance(environment, Mapping):
        read_env = DependencyEnvironment.from_mapping(environment)
        if read_env.package_root != root:
            raise ValueError("dependency package_root must match execution_root")
    extra = raw.get("dependency_environments", ())
    if not isinstance(extra, (tuple, list)) or any(
        not isinstance(env, Mapping) for env in extra
    ):
        raise ValueError("dependency_environments must be a sequence of objects")
    return cls(
        dependency_environments=tuple(
            DependencyEnvironment.from_mapping(env) for env in extra
        ),
        dependency_environment=(
            DependencyEnvironment.from_mapping(environment)
            if isinstance(environment, Mapping)
            else None
        ),
        execution_root=root,
        read_roots=reads,
        read_path_aliases=aliases,
        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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
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)
readable(location)

Accept local reads and declared sources resolved by the test interpreter.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
178
179
180
181
182
183
184
185
186
187
188
189
def readable(self, location: str) -> bool:
    """Accept local reads and declared sources resolved by the test interpreter."""
    location = os.path.realpath(location)
    if any(
        location == root or location.startswith(root + os.sep)
        for root in (self.read_roots or (self.execution_root,))
    ):
        return True
    environments = self.dependency_environments
    if self.dependency_environment is not None:
        environments = (self.dependency_environment, *environments)
    return any(dependency_sources(env).permits(location) for env in environments)
resolve(base, candidate)

Resolve a candidate location relative to an explicit base.

Source code in packages/axm/src/axm/tools/write_scope.py
Python
156
157
158
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
36
37
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
40
41
42
43
44
45
46
47
48
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.from_mapping(payload)