diff --git a/src/agent_env/env/gateway/changes.py b/src/agent_env/env/gateway/changes.py
new file mode 100644
index 00000000..38609bd7
--- /dev/null
+++ b/src/agent_env/env/gateway/changes.py
@@ -0,0 +1,60 @@
+"""Who made each change to an env's state (urn:agentenv:changes/v1): changelog rows joined onto the gateway's write windows."""
+from __future__ import annotations
+
+import heapq
+from typing import Any
+
+
+class ChangeAttribution:
+    """The write windows (changelog ids before and after a call) of the trajectory events logged since start."""
+
+    def __init__(self) -> None:
+        self._calls: dict[str, dict] = {}
+        self._windows: list[tuple[int, int, int, dict]] = []  # (before, after, order, actor)
+
+    def observe(self, event: dict) -> None:
+        """Take one trajectory event as `_log_event` writes it."""
+        kind = event.get("event_type")
+        if kind == "tool_call":
+            self._calls[event["event_id"]] = {"kind": "tool_call", "event_id": event["event_id"],
+                                              "tool": (event.get("tool_call") or {}).get("function_name"),
+                                              **_who(event)}
+        elif kind in ("tool_call_result", "tool_call_failed"):
+            actor = self._calls.pop(event.get("tool_call_event_id"), None)
+            if actor is not None and "changelog_id_before" in event:  # a gateway-native tool's result has no window
+                failed = {"failed": True} if kind == "tool_call_failed" else {}
+                self._add(event["changelog_id_before"], event.get("changelog_id"), {**actor, **failed})
+        elif kind == "internal_tool_call":
+            self._add(event.get("changelog_id_before"), event.get("changelog_id_after"), {
+                "kind": "trigger_action", "event_id": event["event_id"], "trigger_id": event.get("trigger_id"),
+                "action_index": event.get("action_index"), "tool": event.get("tool"), "role": event.get("as")})
+        elif kind == "http_request":
+            self._add(event.get("changelog_id_before"), event.get("changelog_id_after"), {
+                "kind": "request", "event_id": event["event_id"],
+                "request": {k: event.get(k) for k in ("route", "service", "method", "path", "status")},
+                **_who(event)})
+
+    def _add(self, before: Any, after: Any, actor: dict) -> None:
+        before = 0 if before is None else before  # MAX(id) of an empty changelog reads null
+        if isinstance(before, int) and isinstance(after, int) and after > before:
+            self._windows.append((before, after, len(self._windows), actor))
+
+    def actors(self, row_ids: list[int]) -> list[list[dict]]:
+        """For each of ``row_ids`` (ascending), the actors whose ``(before, after]`` window holds it, by finish order."""
+        windows = sorted(self._windows)
+        open_windows: list[tuple[int, int, dict]] = []  # a heap of (after, order, actor)
+        out: list[list[dict]] = []
+        i = 0
+        for row_id in row_ids:
+            while i < len(windows) and windows[i][0] < row_id:
+                before, after, order, actor = windows[i]
+                heapq.heappush(open_windows, (after, order, actor))
+                i += 1
+            while open_windows and open_windows[0][0] < row_id:
+                heapq.heappop(open_windows)
+            out.append([actor for _, _, actor in sorted(open_windows, key=lambda w: w[1])])
+        return out
+
+
+def _who(event: dict) -> dict:
+    return {"role": event.get("role"), **({"caused_by": event["caused_by"]} if event.get("caused_by") else {})}
diff --git a/src/agent_env/env/gateway/constants.py b/src/agent_env/env/gateway/constants.py
index cdc8420c..034d8044 100644
--- a/src/agent_env/env/gateway/constants.py
+++ b/src/agent_env/env/gateway/constants.py
@@ -77,6 +77,11 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 EXT_TRAJECTORY_URI = "urn:agentenv:trajectory/v1"
 EXT_STEP_URI = "urn:agentenv:step/v1"
 EXT_STATE_URI = "urn:agentenv:state/v1"
+EXT_CHANGES_URI = "urn:agentenv:changes/v1"
+
+# Rows per GET /changes page: the default, and the most a client may ask for.
+CHANGES_PAGE_DEFAULT = 1000
+CHANGES_PAGE_MAX = 10000
 
 # Gateway's compose service name — how backing servers reach it to build env_get_time_url (compose path).
 GATEWAY_COMPOSE_HOST = "gateway"
@@ -198,7 +203,7 @@ def random_mcp_server_name() -> str:
     },
     {
         "uri": EXT_TRAJECTORY_URI,
-        "description": "Env trajectory (urn:agentenv:trajectory/v1): the env's append-only JSONL record of tool calls, results, and trigger firings, virtual_time-stamped while the clock is armed.",
+        "description": "Env trajectory (urn:agentenv:trajectory/v1): the env's append-only JSONL record of tool calls, results, trigger firings and the requests that may write through its REST proxy, each write with the changelog ids it spanned, virtual_time-stamped while the clock is armed.",
         "params": {"endpoint": "/trajectory", "methods": {
             "get": {"method": "GET", "endpoint": "/trajectory"}}},
         "required": False,
@@ -217,4 +222,13 @@ def random_mcp_server_name() -> str:
             "get": {"method": "GET", "endpoint": "/state"}}},
         "required": False,
     },
+    {
+        "uri": EXT_CHANGES_URI,
+        "description": "Changes (urn:agentenv:changes/v1): the env's changelog rows after after_id, oldest first, each with the actors whose write window holds it: the role, the tool call, trigger action or request, and the trigger fires it was made for. None for a write the gateway did not serve; several for one written while calls overlapped.",
+        "params": {"endpoint": "/changes", "methods": {
+            "get": {"method": "GET", "endpoint": "/changes", "request": {"type": "object", "properties": {
+                "after_id": {"type": "integer", "minimum": 0},
+                "limit": {"type": "integer", "minimum": 1, "maximum": CHANGES_PAGE_MAX}}}}}},
+        "required": False,
+    },
 ]
diff --git a/src/agent_env/env/gateway/gateway.py b/src/agent_env/env/gateway/gateway.py
index 4a4a4d4d..70ef07da 100644
--- a/src/agent_env/env/gateway/gateway.py
+++ b/src/agent_env/env/gateway/gateway.py
@@ -41,6 +41,8 @@
     AGENT_ENV_ROLE_META_KEY,  # noqa: F401  (the server module serves the whole vocabulary, used in this file or not)
     AGENT_ENV_SESSION_META_KEY,  # noqa: F401
     CARD_FETCH_TIMEOUT_S,
+    CHANGES_PAGE_DEFAULT,
+    CHANGES_PAGE_MAX,
     CHILD_DISCOVERY_TIMEOUT_S,
     DATA_PLANE_LOAD_TIMEOUT_MAX_S,
     DEFAULT_ROLE,
@@ -62,6 +64,7 @@
     tool_name_clashes_text,
 )
 from .caller import Caller
+from .changes import ChangeAttribution
 from .child_transport import child_http_client
 from .clock import Clock, ClockError, _iso
 from .get_time import GET_TIME_TOOL_NAME, build_get_time_tool
@@ -70,6 +73,9 @@
 
 logger = logging.getLogger(__name__)
 
+# RFC 9110's safe methods: a request using one asks for no change, so the proxy measures no write window for it.
+_SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE"})
+
 
 class ToolNameClash(ValueError):
     """More than one source serves a tool under one name, so an agent couldn't tell which it calls: the gateway serves
@@ -139,6 +145,7 @@ def __init__(
 
         self._event_counter = 0
         self._event_lock = asyncio.Lock()
+        self._changes = ChangeAttribution()
         Path(GATEWAY_TRAJECTORY_FILE).parent.mkdir(parents=True, exist_ok=True)
         self._trajectory_file = open(GATEWAY_TRAJECTORY_FILE, "a")
         atexit.register(self._close_trajectory_file)
@@ -291,6 +298,10 @@ async def clock_state(request: Request) -> Response:
         async def env_trajectory(request: Request) -> Response:
             return self._serve_trajectory()
 
+        @self._mcp.custom_route("/changes", methods=["GET"])
+        async def env_changes(request: Request) -> Response:
+            return await self._serve_changes(request)
+
         @self._mcp.custom_route(WELL_KNOWN_PATH, methods=["GET"])
         async def get_env_card(request: Request) -> Response:
             return await self._serve_env_card()
@@ -505,8 +516,31 @@ async def _proxy_rest_request(self, request: Request, url_map: dict[str, str] |
         base_url = url_map.get(service_name)
         if base_url is None:
             return self._json_response({"error": f"Unknown service: {service_name}"}, 404)
-        return await forward(request, base_url, role=self._request_role(request), timeout=self.REST_PROXY_TIMEOUT,
-                             mcp_server=url_map is self.rest_proxy_urls and service_name.startswith("mcp-"))
+        role = self._request_role(request)
+        mcp_server = url_map is self.rest_proxy_urls and service_name.startswith("mcp-")
+        if request.method in _SAFE_METHODS:
+            return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server)
+        changelog_before = self._query_changelog_id()
+        attribution = self._attribution(role)  # at the start: a fire may be gone by the time the response ends
+
+        async def log_write(status: int) -> None:
+            changelog_after = self._query_changelog_id()
+            if changelog_after is None or changelog_after <= (changelog_before or 0):
+                return  # only a request that wrote changelog rows is a write window
+            await self._log_event({
+                "event_type": "http_request",
+                "route": "website" if url_map is self.website_urls else "svc",
+                "service": service_name,
+                "method": request.method,
+                "path": "/" + request.path_params.get("path", ""),
+                "status": status,
+                **attribution,
+                "changelog_id_before": changelog_before,
+                "changelog_id_after": changelog_after,
+            })
+
+        return await forward(request, base_url, role=role, timeout=self.REST_PROXY_TIMEOUT, mcp_server=mcp_server,
+                             on_done=log_write)
 
     @staticmethod
     def _request_role(request: Request) -> str:
@@ -950,7 +984,12 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
         })
         try:
             async with self._tool_call_guard():
-                result = await self._call_child_tool(server_url, tool_name, arguments, caller=Caller(role=role))
+                changelog_before = self._query_changelog_id()
+                try:
+                    result = await self._call_child_tool(server_url, tool_name, arguments, caller=Caller(role=role))
+                except Exception as e:
+                    await self._log_failed_call(tool_call_event_id, changelog_before, e, source="step_api")
+                    raise
                 changelog_id = self._query_changelog_id()
 
             await self._log_event({
@@ -958,6 +997,7 @@ async def _step_call_tool(self, body: dict, role: str) -> Response:
                 "source": "step_api",
                 "tool_call_event_id": tool_call_event_id,
                 "tool_call_result": result.model_dump(mode="json"),
+                "changelog_id_before": changelog_before,
                 "changelog_id": changelog_id,
             })
             barrier_tasks, barrier_timeout_s = self._fire_triggers(role, tool_name, arguments, result)
@@ -1073,13 +1113,19 @@ async def proxy(**kwargs) -> CallToolResult:
             })
 
             async with self._tool_call_guard():
-                result = await self._call_child_tool(server_url, tool_name, kwargs, caller=caller)
+                changelog_before = self._query_changelog_id()
+                try:
+                    result = await self._call_child_tool(server_url, tool_name, kwargs, caller=caller)
+                except Exception as e:
+                    await self._log_failed_call(tool_call_event_id, changelog_before, e)
+                    raise
                 changelog_id = self._query_changelog_id()
 
             await self._log_event({
                 "event_type": "tool_call_result",
                 "tool_call_event_id": tool_call_event_id,
                 "tool_call_result": result.model_dump(mode="json"),
+                "changelog_id_before": changelog_before,
                 "changelog_id": changelog_id,
             })
             return result
@@ -1136,6 +1182,16 @@ async def _log_native_tool_result(self, event_id: str | None, result: Any) -> No
         except Exception:
             logger.exception("Failed to log tool_call_result for a gateway tool")
 
+    async def _log_failed_call(self, tool_call_event_id: str, changelog_before: int | None, error: Exception,
+                               **source: str) -> None:
+        """Close a failed call's write window, which may hold a write it committed first (fail-open)."""
+        try:
+            await self._log_event({"event_type": "tool_call_failed", **source, "tool_call_event_id": tool_call_event_id,
+                                   "error": f"{type(error).__name__}: {str(error)[:300]}",
+                                   "changelog_id_before": changelog_before, "changelog_id": self._query_changelog_id()})
+        except Exception:
+            logger.exception("Failed to log a failed tool call")
+
     def _attribution(self, role: str) -> dict:
         """Who made a tool call, for its trajectory event: the caller's role, and the trigger fires it was made for
         when that role is an executor realizing them."""
@@ -1156,8 +1212,41 @@ async def _log_event(self, event: dict) -> str:
                 event, default=lambda o: o.model_dump(mode="json") if hasattr(o, "model_dump") else str(o),
             ) + "\n")
             self._trajectory_file.flush()
+            self._changes.observe(event)
         return event_id
 
+    async def _serve_changes(self, request: Request) -> Response:
+        """The changelog rows after ``after_id`` (at most ``limit``, oldest first), each with its actors (see ``changes``)."""
+        try:
+            after_id = int(request.query_params.get("after_id", 0))
+            limit = int(request.query_params.get("limit", CHANGES_PAGE_DEFAULT))
+        except ValueError:
+            return self._json_response({"ok": False, "error": "after_id and limit must be integers"}, 400)
+        if after_id < 0 or not 1 <= limit <= CHANGES_PAGE_MAX:
+            return self._json_response(
+                {"ok": False, "error": f"after_id must be >= 0 and limit between 1 and {CHANGES_PAGE_MAX}"}, 400)
+        if self._db_conn is None:
+            return self._json_response({"ok": False, "error": "this env keeps no changelog"}, 404)
+        try:
+            rows = await asyncio.to_thread(self._read_changelog, after_id, limit + 1)
+        except Exception as e:
+            logger.exception("Reading the changelog failed")
+            return self._json_response({"ok": False, "error": f"changelog unavailable: {e}"}, 503)
+        rows, has_more = rows[:limit], len(rows) > limit
+        actors = self._changes.actors([row["id"] for row in rows])
+        return self._json_response(
+            {"changes": [{**row, "actors": who} for row, who in zip(rows, actors)], "has_more": has_more}, 200)
+
+    def _read_changelog(self, after_id: int, count: int) -> list[dict]:
+        with self._db_conn.cursor() as cur:
+            cur.execute('SELECT id, "timestamp", schema_name, table_name, operation, row_id, summary, changed_fields '
+                        "FROM public._changelog WHERE id > %s ORDER BY id LIMIT %s", (after_id, count))
+            names = [column[0] for column in cur.description]
+            rows = [dict(zip(names, values)) for values in cur.fetchall()]
+        for row in rows:
+            row["timestamp"] = row["timestamp"].isoformat()
+        return rows
+
     def _serve_trajectory(self) -> Response:
         """Stream the whole env trajectory as JSONL; only a mid-write tail line can arrive torn."""
         try:
diff --git a/src/agent_env/env/gateway/rest_proxy.py b/src/agent_env/env/gateway/rest_proxy.py
index baf1f473..0114ae04 100644
--- a/src/agent_env/env/gateway/rest_proxy.py
+++ b/src/agent_env/env/gateway/rest_proxy.py
@@ -23,7 +23,7 @@
 import json
 import logging
 import posixpath
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Awaitable, Callable
 from urllib.parse import unquote
 
 import httpx
@@ -52,14 +52,16 @@
 ROLE_HEADER_BYTES = AGENT_ENV_ROLE_HEADER.lower().encode()
 
 
-async def forward(request: Request, base_url: str, *, role: str, timeout: httpx.Timeout, mcp_server: bool) -> Response:
+async def forward(request: Request, base_url: str, *, role: str, timeout: httpx.Timeout, mcp_server: bool,
+                  on_done: Callable[[int], Awaitable[None]] | None = None) -> Response:
     """Forward `request` to the service at `base_url` as `role` and stream the response through. `mcp_server` says
     the service is a backing MCP server, whose own endpoint takes the role in its JSON-RPC `_meta` as well as in the
-    header. A 400 names a request-target the proxy cannot route; a 502 names an upstream that never answered."""
+    header. A 400 names a request-target the proxy cannot route; a 502 names an upstream that never answered.
+    `on_done` is awaited with the status once the response has ended, after its last byte when streamed."""
     try:
         target_url = upstream_url(base_url, request)
     except ValueError as e:  # an escaped slash inside the route prefix; refuse rather than forward a wrong path
-        return _json_response({"error": str(e)}, 400)
+        return await _done(_json_response({"error": str(e)}, 400), on_done)
     content = await request.body()  # /svc and /website request bodies are small (JSON-RPC, forms)
     headers = upstream_headers(request, role)
     if mcp_server and request.method == "POST" and is_child_mcp_path(request.path_params.get("path", "")):
@@ -74,8 +76,14 @@ async def forward(request: Request, base_url: str, *, role: str, timeout: httpx.
     except Exception as e:
         await client.aclose()
         logger.error(f"Proxy to {target_url} failed: {e!r}")
-        return _json_response({"error": f"Proxy request failed: {e!r}"}, 502)
-    return UpstreamResponse(upstream, client, target_url)
+        return await _done(_json_response({"error": f"Proxy request failed: {e!r}"}, 502), on_done)
+    return UpstreamResponse(upstream, client, target_url, on_done)
+
+
+async def _done(response: Response, on_done: Callable[[int], Awaitable[None]] | None) -> Response:
+    if on_done is not None:
+        await on_done(response.status_code)
+    return response
 
 
 def upstream_url(base_url: str, request: Request) -> str:
@@ -161,13 +169,15 @@ class UpstreamResponse(StreamingResponse):
 
     _DROP = HOP_BY_HOP | {b"date", b"server"}  # uvicorn adds the gateway's own; forwarding upstream's duplicates them
 
-    def __init__(self, upstream: httpx.Response, client: httpx.AsyncClient, target_url: str) -> None:
+    def __init__(self, upstream: httpx.Response, client: httpx.AsyncClient, target_url: str,
+                 on_done: Callable[[int], Awaitable[None]] | None = None) -> None:
         super().__init__(self._body(upstream, target_url), status_code=upstream.status_code)
         # Header block forwarded byte for byte: repeated names (Set-Cookie) stay separate, non-latin-1 values
         # never hit a str round-trip, and Content-Length/Content-Encoding stay valid for the undecoded body.
         self.raw_headers = [(k.lower(), v) for k, v in upstream.headers.raw if k.lower() not in self._DROP]
         self._upstream = upstream
         self._client = client
+        self._on_done = on_done
 
     @staticmethod
     async def _body(upstream: httpx.Response, target_url: str) -> AsyncIterator[bytes]:
@@ -186,6 +196,8 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         finally:  # normal end, upstream error, or the downstream client going away
             await self._upstream.aclose()
             await self._client.aclose()
+            if self._on_done is not None:
+                await self._on_done(self.status_code)
 
 
 async def fetch_child_card(client: httpx.AsyncClient, key: str, base_url: str) -> dict | None:
diff --git a/src/agent_env/task_step/task_steps/snapshot_env.py b/src/agent_env/task_step/task_steps/snapshot_env.py
index c78c5042..943ea99f 100644
--- a/src/agent_env/task_step/task_steps/snapshot_env.py
+++ b/src/agent_env/task_step/task_steps/snapshot_env.py
@@ -17,11 +17,11 @@
 import httpx
 
 from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI, FilePart, uploaded_object_path
-from agentenv_protocol.client import find_extension
+from agentenv_protocol.client import find_extension, find_extension_method
 from agentenv_protocol.transfers import WriteNamespaceGrant
 
 from agent_env.a2a_agent.object_transfer import ObjectLimits, namespace_grant
-from agent_env.env.gateway.constants import EXT_TRAJECTORY_URI
+from agent_env.env.gateway.constants import CHANGES_PAGE_MAX, EXT_CHANGES_URI, EXT_TRAJECTORY_URI
 from agent_env.store import get_config
 from agent_env.store.base import GrantUnavailableError
 from agent_env.store.ids import derive_id, is_local_id, validate_local_id
@@ -39,6 +39,7 @@
 _CONNECT_TIMEOUT_SECONDS = 30
 
 _TRAJECTORY_CAPTURE_BUDGET_SECONDS = 300
+_CHANGES_CAPTURE_BUDGET_SECONDS = 300
 
 # Must match the S3 credentials extension URI the server-side credentials mixin advertises.
 _S3_CREDENTIALS_EXTENSION_URI = "urn:agentenv:add-s3-credentials/v1"
@@ -153,6 +154,8 @@ class SnapshotEnvTaskStep(TaskStep):
 
     Output: ``context.metadata["env_snapshotted_universes"][self.id]`` =
     ``{"id", "version"}``. Fails the step if any service fails to export.
+
+    ``include_env_changes`` also stores who changed what (urn:agentenv:changes/v1) under ``metadata["env_changes"]``.
     """
 
     type: ClassVar[str] = "snapshot_env"
@@ -175,6 +178,7 @@ def __init__(
         original_universe_artifact_id: Optional[str] = None,
         export_timeout_seconds: int = DEFAULT_EXPORT_TIMEOUT_SECONDS,
         include_env_trajectory: bool = False,
+        include_env_changes: bool = False,
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
     ):
@@ -187,6 +191,7 @@ def __init__(
         self.original_universe_artifact_id = original_universe_artifact_id
         self.export_timeout_seconds = export_timeout_seconds
         self.include_env_trajectory = include_env_trajectory
+        self.include_env_changes = include_env_changes
 
     def to_dict(self) -> dict:
         base = super().to_dict()
@@ -198,6 +203,7 @@ def to_dict(self) -> dict:
         base["original_universe_artifact_id"] = self.original_universe_artifact_id
         base["export_timeout_seconds"] = self.export_timeout_seconds
         base["include_env_trajectory"] = self.include_env_trajectory
+        base["include_env_changes"] = self.include_env_changes
         return base
 
     @classmethod
@@ -214,6 +220,7 @@ def from_dict(cls, data: dict) -> SnapshotEnvTaskStep:
                 "export_timeout_seconds", cls.DEFAULT_EXPORT_TIMEOUT_SECONDS
             ),
             include_env_trajectory=data.get("include_env_trajectory", False),
+            include_env_changes=data.get("include_env_changes", False),
         )
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
@@ -243,6 +250,17 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 entry["error"] = detail
             context.metadata.setdefault("env_trajectory", {})[env_id] = entry
 
+        if self.include_env_changes:
+            entry = {"captured_at_utc": datetime.now(timezone.utc).isoformat(), "capture_step_id": self.id}
+            try:
+                await asyncio.wait_for(self._capture_env_changes(entry, context, env_id, deployed),
+                                       timeout=_CHANGES_CAPTURE_BUDGET_SECONDS)
+            except Exception as e:
+                detail = f"{type(e).__name__}: {str(e)[:200]}"
+                logger.warning(f"snapshot_env: env changes capture aborted (continuing): {detail}")
+                entry["error"] = detail
+            context.metadata.setdefault("env_changes", {})[env_id] = entry
+
         # Pin the env version the sandbox actually runs, so service enumeration
         # can't drift if the env was re-registered since this sandbox deployed.
         env_version = deployed.env_version if deployed else None
@@ -363,6 +381,40 @@ async def _capture_env_trajectory(self, entry: dict, context: TaskStepContext, e
         entry["object_url"] = object_url
         entry["event_count"] = event_count
 
+    async def _capture_env_changes(self, entry: dict, context: TaskStepContext, env_id: str, deployed) -> None:
+        """Page the gateway's changes into a JSONL upload; fill ``entry`` with its url and attribution counts."""
+        method, url = _changes_route(deployed, self.gateway_url)
+        counts = {"change_count": 0, "attributed_count": 0, "ambiguous_count": 0}
+        timeout = httpx.Timeout(_CHANGES_CAPTURE_BUDGET_SECONDS, connect=_CONNECT_TIMEOUT_SECONDS)
+        tmp = tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False)
+        try:
+            after_id, has_more = 0, True
+            async with httpx.AsyncClient(timeout=timeout) as client:
+                while has_more:
+                    response = await client.request(method, url, params={"after_id": after_id, "limit": CHANGES_PAGE_MAX})
+                    response.raise_for_status()
+                    page = response.json()
+                    for change in page["changes"]:
+                        tmp.write((json.dumps(change, default=str) + "\n").encode())
+                        counts["change_count"] += 1
+                        counts["attributed_count"] += len(change["actors"]) == 1
+                        counts["ambiguous_count"] += len(change["actors"]) > 1
+                        after_id = change["id"]
+                    has_more = page["has_more"]
+            tmp.close()
+            instance_key = context.instance_id or f"adhoc-{uuid.uuid4().hex[:12]}"
+            key = (
+                f"{get_config().get_artifact_key_prefix()}env_changes/"
+                f"instance_id={instance_key}/{env_id}-{uuid.uuid4().hex[:8]}.jsonl"
+            )
+            object_url = await asyncio.to_thread(get_config().get_object_store().put_file, key, tmp.name,
+                                                 "application/jsonl")
+        finally:
+            tmp.close()
+            os.unlink(tmp.name)
+        entry["object_url"] = object_url
+        entry.update(counts)
+
     def _derive_snapshot_id(self, context: TaskStepContext) -> str:
         if self.snapshot_id:
             return self.snapshot_id
@@ -689,6 +741,15 @@ def _register_uploaded_bundle():
         )
 
 
+def _changes_route(deployed, override_url: Optional[str]) -> tuple[str, str]:
+    """The verb and URL of the env's changes: an override keeps `{override}/changes`, else the stored card names them."""
+    if override_url:
+        return "GET", f"{override_url.rstrip('/')}/changes"
+    deployed.require(EXT_CHANGES_URI, "get")
+    op = find_extension_method(deployed.environment_card, EXT_CHANGES_URI, "get")
+    return op["method"], f"{deployed.environment_url}{op['endpoint']}"
+
+
 def _trajectory_route(deployed, override_url: Optional[str]) -> tuple[str, str]:
     """The verb and URL of the env trajectory: an override keeps `{override}/trajectory`, else the stored card names them."""
     if override_url:
diff --git a/tst/integration/env/gateway/gateway_test.py b/tst/integration/env/gateway/gateway_test.py
index 601eb5d5..1756a55a 100644
--- a/tst/integration/env/gateway/gateway_test.py
+++ b/tst/integration/env/gateway/gateway_test.py
@@ -1768,6 +1768,71 @@ async def assert_loaded_via_step(client, gateway_url, tools, tool_names):
         await multi_env.close()
 
 
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_gateway_changes_name_the_role_behind_each_write(mcp_server_envs, email_service_artifact):
+    """Every changelog row a tool call wrote names its role and tool, and the call's result carries the ids it spanned."""
+    email_ref = mcp_server_envs["email"]
+    email_env = Env.get(email_ref.id, version=email_ref.version)
+    try:
+        result = await email_env.deploy()
+        await email_env.load_environment_artifact(email_service_artifact)  # the changelog's triggers go in after a load
+        gw = result.gateway_url
+        async with httpx.AsyncClient() as client:
+            seen = (await client.get(f"{gw}/changes", params={"limit": 10000}, timeout=30)).json()["changes"]
+            after_id = seen[-1]["id"] if seen else 0
+            for role, subject in (("default", "first"), ("alice@example.com", "second")):
+                resp = await client.post(f"{gw}/step", headers={AGENT_ENV_ROLE_HEADER: role}, timeout=60, json={
+                    "action": "call_tool", "tool_name": "send_email",
+                    "arguments": {"recipients": ["dana@example.com"], "subject": subject, "content": "body"}})
+                assert resp.status_code == 200, resp.text
+
+            changes = (await client.get(f"{gw}/changes", params={"after_id": after_id}, timeout=30)).json()
+            assert changes["changes"] and changes["has_more"] is False, changes
+            actors = [c["actors"] for c in changes["changes"]]
+            assert all(len(who) == 1 and who[0]["kind"] == "tool_call" and who[0]["tool"] == "send_email"
+                       for who in actors), actors
+            roles = [who[0]["role"] for who in actors]
+            assert set(roles) == {"default", "alice@example.com"} and roles == sorted(roles, key=lambda r: r != "default")
+
+            events = await _trajectory_events(client, gw)
+            results = [e for e in events if e["event_type"] == "tool_call_result"][-2:]
+            assert all(e["changelog_id"] > (e["changelog_id_before"] or 0) for e in results), results
+    finally:
+        await email_env.close()
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_gateway_changes_name_the_request_behind_a_write_through_the_rest_proxy(mcp_server_envs,
+                                                                                      email_service_artifact):
+    """An MCP client writing through /svc/mcp-email/mcp (an SSE answer) is a write window too, named by its role."""
+    email_ref = mcp_server_envs["email"]
+    email_env = Env.get(email_ref.id, version=email_ref.version)
+    try:
+        result = await email_env.deploy()
+        await email_env.load_environment_artifact(email_service_artifact)
+        gw = result.gateway_url
+        async with httpx.AsyncClient() as client:
+            seen = (await client.get(f"{gw}/changes", params={"limit": 10000}, timeout=30)).json()["changes"]
+            after_id = seen[-1]["id"] if seen else 0
+            async with httpx.AsyncClient(headers={AGENT_ENV_ROLE_HEADER: "carol@example.com"}, timeout=60) as as_carol:
+                async with streamable_http_client(f"{gw}/svc/mcp-email/mcp", http_client=as_carol) as (read, write, _):
+                    async with ClientSession(read, write) as session:
+                        await session.initialize()
+                        sent = await session.call_tool("send_email", {
+                            "recipients": ["dana@example.com"], "subject": "lunch?", "content": "noon"})
+                        assert not sent.isError, sent
+
+            changes = (await client.get(f"{gw}/changes", params={"after_id": after_id}, timeout=30)).json()["changes"]
+            assert changes, "the call wrote no changelog row"
+            assert all([(a["kind"], a["role"], a["request"]["route"], a["request"]["service"], a["request"]["path"])
+                        for a in c["actors"]] == [("request", "carol@example.com", "svc", "mcp-email", "/mcp")]
+                       for c in changes), changes
+    finally:
+        await email_env.close()
+
+
 @pytest.mark.asyncio
 @pytest.mark.integration
 @skip_without_model_endpoint()
diff --git a/tst/unit/env/gateway/changes_test.py b/tst/unit/env/gateway/changes_test.py
new file mode 100644
index 00000000..68180016
--- /dev/null
+++ b/tst/unit/env/gateway/changes_test.py
@@ -0,0 +1,481 @@
+"""Who changed what (urn:agentenv:changes/v1): GET /changes joins changelog rows onto the gateway's write windows."""
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import json
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from urllib.parse import unquote, urlsplit
+
+import httpx
+import pytest
+from mcp.types import CallToolResult, TextContent, Tool as MCPTool
+
+from agent_env.env.gateway import InternalMCPServer
+from agent_env.env.gateway import rest_proxy as rest_proxy_module
+from agent_env.env.gateway.caller import Caller
+from agent_env.env.gateway.changes import ChangeAttribution
+from agent_env.env.gateway.gateway import GATEWAY_TRAJECTORY_FILE, Gateway
+
+URL = "http://backing/mcp"
+TOOL = MCPTool(name="mail_send_message", description="", inputSchema={"type": "object", "properties": {}})
+_RealAsyncClient = httpx.AsyncClient
+
+
+def _call(event_id: str, tool: str, role: str, **extra) -> dict:
+    return {"event_type": "tool_call", "event_id": event_id, "tool_call": {"function_name": tool, "arguments": {}},
+            "role": role, **extra}
+
+
+def _result(call_id: str, before, after) -> dict:
+    return {"event_type": "tool_call_result", "event_id": f"{call_id}-r", "tool_call_event_id": call_id,
+            "changelog_id_before": before, "changelog_id": after}
+
+
+# ---------------------------------------------------------------- the join
+
+def test_a_row_is_attributed_to_the_call_whose_window_holds_it():
+    changes = ChangeAttribution()
+    fire = {"trigger_id": "reply", "action_index": 0, "fire": 1}
+    for event in (
+        _call("event_1", "mail_send_message", "default"), _result("event_1", 0, 2),
+        _call("event_3", "mail_send_message", "dana@example.com", caused_by=[fire]), _result("event_3", 2, 3),
+        {"event_type": "internal_tool_call", "event_id": "event_5", "trigger_id": "mirror", "action_index": 1,
+         "tool": "drive_touch_file", "as": None, "changelog_id_before": 3, "changelog_id_after": 4},
+        {"event_type": "http_request", "event_id": "event_6", "route": "website", "service": "mail", "method": "POST",
+         "path": "/api/drafts", "status": 201, "role": "default", "changelog_id_before": 4, "changelog_id_after": 5},
+    ):
+        changes.observe(event)
+
+    assert changes.actors([1, 2, 3, 4, 5, 6]) == [
+        [{"kind": "tool_call", "event_id": "event_1", "tool": "mail_send_message", "role": "default"}],
+        [{"kind": "tool_call", "event_id": "event_1", "tool": "mail_send_message", "role": "default"}],
+        [{"kind": "tool_call", "event_id": "event_3", "tool": "mail_send_message", "role": "dana@example.com",
+          "caused_by": [fire]}],
+        [{"kind": "trigger_action", "event_id": "event_5", "trigger_id": "mirror", "action_index": 1,
+          "tool": "drive_touch_file", "role": None}],
+        [{"kind": "request", "event_id": "event_6", "role": "default", "request": {
+            "route": "website", "service": "mail", "method": "POST", "path": "/api/drafts", "status": 201}}],
+        [],
+    ]
+
+
+def test_overlapping_calls_both_name_a_row_written_between_them():
+    changes = ChangeAttribution()
+    changes.observe(_call("event_1", "a", "default"))
+    changes.observe(_call("event_2", "b", "carol@example.com"))
+    changes.observe(_result("event_2", 0, 2))
+    changes.observe(_result("event_1", 0, 3))
+    assert [[a["event_id"] for a in who] for who in changes.actors([1, 2, 3])] == [
+        ["event_2", "event_1"], ["event_2", "event_1"], ["event_1"]]
+
+
+def test_calls_without_a_write_window_attribute_nothing():
+    changes = ChangeAttribution()
+    for event in (
+        _call("event_1", "read_only", "default"), _result("event_1", 4, 4),
+        _call("event_3", "no_changelog", "default"), _result("event_3", None, None),
+        _call("event_5", "list_website_urls", "default"),
+        {"event_type": "tool_call_result", "event_id": "event_6", "tool_call_event_id": "event_5", "tool_call_result": {}},
+        _result("event_unknown", 0, 9),
+        _call("event_7", "raised", "default"),
+    ):
+        changes.observe(event)
+    assert changes.actors([1, 4, 5, 9]) == [[], [], [], []]
+
+
+# ---------------------------------------------------------------- the gateway records the windows
+
+class _Changelog:
+    """The env's changelog as the gateway reads it: MAX(id) through `_query_changelog_id`, rows through a cursor."""
+
+    def __init__(self):
+        self.rows: list[dict] = []
+        self.queries: list[tuple[str, tuple]] = []
+
+    def write(self, table: str, row_id: str) -> None:
+        self.rows.append({"id": len(self.rows) + 1, "timestamp": datetime(2026, 10, 14, 8, len(self.rows), tzinfo=timezone.utc),
+                          "schema_name": "mail", "table_name": table, "operation": "INSERT", "row_id": row_id,
+                          "summary": f"Inserted: {row_id}", "changed_fields": {"id": row_id}})
+
+    def max_id(self) -> int | None:
+        return len(self.rows) or None
+
+    def cursor(self):
+        log = self
+
+        class _Cursor:
+            description = [(name,) for name in ("id", "timestamp", "schema_name", "table_name", "operation",
+                                                "row_id", "summary", "changed_fields")]
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *exc):
+                return False
+
+            def execute(self, sql, params):
+                log.queries.append((sql, params))
+                after_id, count = params
+                self._rows = [r for r in log.rows if r["id"] > after_id][:count]
+
+            def fetchall(self):
+                return [tuple(r[name] for (name,) in self.description) for r in self._rows]
+
+        return _Cursor()
+
+
+class _WritingSession:
+    """The child server: each call writes one changelog row, then raises `failure` when one is set."""
+
+    def __init__(self, changelog: _Changelog, failure: Exception | None = None):
+        self.changelog = changelog
+        self.failure = failure
+
+    async def list_tools(self):
+        return SimpleNamespace(tools=[TOOL])
+
+    async def call_tool(self, name, arguments, *args, meta=None, **kwargs):
+        self.changelog.write("messages", f"m{len(self.changelog.rows) + 1}")
+        if self.failure is not None:
+            raise self.failure
+        return CallToolResult(content=[TextContent(type="text", text="ok")], isError=False)
+
+
+def _gateway(changelog: _Changelog, failure: Exception | None = None, **kw) -> Gateway:
+    gw = Gateway(host="127.0.0.1", port=0, server_name="t", internal_mcp_servers=[InternalMCPServer("mail", URL)], **kw)
+    session = _WritingSession(changelog, failure)
+
+    async def _open(stack):
+        return {URL: session}
+
+    gw._open_persistent_sessions = _open  # type: ignore[method-assign]
+    gw._query_changelog_id = changelog.max_id  # type: ignore[method-assign]
+    gw._db_conn = changelog
+    gw._create_and_register_proxy_tool(URL, TOOL)
+    gw._tool_server_urls[TOOL.name] = URL
+    return gw
+
+
+def _trajectory() -> list[dict]:
+    with open(GATEWAY_TRAJECTORY_FILE) as f:
+        return [json.loads(line) for line in f]
+
+
+async def _http(gw: Gateway, method: str, target: str, *, headers: dict | None = None, body: bytes = b"") -> tuple[int, bytes]:
+    """One request through the gateway's ASGI app, its scope as uvicorn builds it."""
+    url = urlsplit(target)
+    scope = {
+        "type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, "http_version": "1.1",
+        "method": method, "scheme": "http", "path": unquote(url.path), "raw_path": url.path.encode(),
+        "query_string": url.query.encode(), "root_path": "", "server": ("gateway", 80), "client": ("c", 1),
+        "headers": [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()],
+    }
+    done = asyncio.Event()
+    received = False
+    sent: list[dict] = []
+
+    async def receive():
+        nonlocal received
+        if not received:
+            received = True
+            return {"type": "http.request", "body": body, "more_body": False}
+        await done.wait()
+        return {"type": "http.disconnect"}
+
+    async def send(message):
+        sent.append(message)
+
+    try:
+        await gw._mcp.streamable_http_app()(scope, receive, send)
+    finally:
+        done.set()
+    status = next(m["status"] for m in sent if m["type"] == "http.response.start")
+    return status, b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")
+
+
+@pytest.mark.asyncio
+async def test_tool_calls_record_the_changelog_ids_they_spanned_and_changes_names_their_role():
+    changelog = _Changelog()
+    gw = _gateway(changelog)
+    before = len(_trajectory())
+    try:
+        resp = await gw._step_call_tool({"tool_name": TOOL.name, "arguments": {"to": "dana@example.com"}}, "default")
+        assert resp.status_code == 200
+        ctx = contextvars.copy_context()
+        ctx.run(Caller(role="carol@example.com").bind)
+        await ctx.run(asyncio.create_task, gw._mcp._tool_manager.call_tool(TOOL.name, {}))
+    finally:
+        await gw._close_child_sessions()
+
+    results = [e for e in _trajectory()[before:] if e["event_type"] == "tool_call_result"]
+    assert [(e["changelog_id_before"], e["changelog_id"]) for e in results] == [(None, 1), (1, 2)]
+    status, body = await _http(gw, "GET", "/changes")
+    changes = json.loads(body)
+    assert status == 200 and changes["has_more"] is False
+    assert [(c["id"], c["row_id"], [(a["kind"], a["tool"], a["role"]) for a in c["actors"]]) for c in changes["changes"]] == [
+        (1, "m1", [("tool_call", TOOL.name, "default")]),
+        (2, "m2", [("tool_call", TOOL.name, "carol@example.com")]),
+    ]
+    assert changes["changes"][0]["timestamp"] == "2026-10-14T08:00:00+00:00"
+    assert changes["changes"][0]["changed_fields"] == {"id": "m1"}
+
+
+def test_a_result_that_measured_no_window_claims_no_row():
+    changes = ChangeAttribution()
+    changes.observe(_call("event_1", "list_website_urls", "default"))
+    changes.observe({"event_type": "tool_call_result", "event_id": "event_2", "tool_call_event_id": "event_1",
+                     "changelog_id": 5})
+    assert changes.actors([1, 5]) == [[], []]
+
+
+@pytest.mark.asyncio
+async def test_a_gateway_tool_called_through_step_after_a_write_claims_none_of_it():
+    changelog = _Changelog()
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"})
+    try:
+        assert (await gw._step_call_tool({"tool_name": TOOL.name, "arguments": {}}, "default")).status_code == 200
+        assert (await gw._step_call_tool({"tool_name": "list_website_urls", "arguments": {}},
+                                         "carol@example.com")).status_code == 200
+    finally:
+        await gw._close_child_sessions()
+
+    _, body = await _http(gw, "GET", "/changes")
+    assert [[(a["tool"], a["role"]) for a in c["actors"]] for c in json.loads(body)["changes"]] == [
+        [(TOOL.name, "default")]]
+
+
+@pytest.mark.asyncio
+async def test_a_call_that_writes_then_fails_still_names_its_write_and_keeps_its_error():
+    changelog = _Changelog()
+    gw = _gateway(changelog, failure=RuntimeError("lost the connection after committing"))
+    try:
+        resp = await gw._step_call_tool({"tool_name": TOOL.name, "arguments": {}}, "default")
+        ctx = contextvars.copy_context()
+        ctx.run(Caller(role="carol@example.com").bind)
+        with pytest.raises(Exception, match="lost the connection after committing"):
+            await ctx.run(asyncio.create_task, gw._mcp._tool_manager.call_tool(TOOL.name, {}))
+    finally:
+        await gw._close_child_sessions()
+
+    assert resp.status_code == 500
+    assert json.loads(resp.body)["error"] == "Tool execution failed (RuntimeError): lost the connection after committing"
+    _, body = await _http(gw, "GET", "/changes")
+    assert [[(a["role"], a.get("failed")) for a in c["actors"]] for c in json.loads(body)["changes"]] == [
+        [("default", True)], [("carol@example.com", True)]]
+
+
+@pytest.mark.asyncio
+async def test_a_request_keeps_the_trigger_fire_it_started_under(monkeypatch):
+    changelog = _Changelog()
+    cause = {"trigger_id": "reply", "action_index": 0, "fire": 1}
+    gw = _gateway(changelog, rest_proxy_urls={"mail": "http://mail-backend"})
+    gw._trigger_engine._in_flight["bob@example.com"] = [cause]
+
+    async def body():
+        yield b"a"
+        gw._trigger_engine._in_flight.pop("bob@example.com")
+        changelog.write("drafts", "d1")
+        yield b"b"
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    before = len(_trajectory())
+
+    await _http(gw, "POST", "/svc/mail/api/drafts", headers={"AgentEnv-Role": "bob@example.com"})
+
+    [logged] = [e for e in _trajectory()[before:] if e["event_type"] == "http_request"]
+    assert logged["caused_by"] == [cause]
+
+
+@pytest.mark.asyncio
+async def test_a_write_through_the_rest_proxy_is_a_logged_window_and_a_read_is_not(monkeypatch):
+    changelog = _Changelog()
+
+    async def body():
+        yield b'{"ok": true}'
+
+    def website(request: httpx.Request) -> httpx.Response:
+        if request.method != "GET":
+            changelog.write("drafts", "d1")
+        return httpx.Response(201 if request.method == "POST" else 200, content=body())
+
+    transport = httpx.MockTransport(website)
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, website_urls={"mail": "http://mail-frontend"}, rest_proxy_urls={"mail": "http://mail-backend"})
+    gw._trigger_engine._in_flight["bob@example.com"] = [{"trigger_id": "reply", "action_index": 0, "fire": 2}]
+    before = len(_trajectory())
+
+    assert (await _http(gw, "GET", "/website/mail/inbox"))[0] == 200
+    assert (await _http(gw, "POST", "/website/mail/api/drafts", headers={"AgentEnv-Role": "bob@example.com"}))[0] == 201
+    assert (await _http(gw, "PATCH", "/svc/mail/api/drafts/d1"))[0] == 200
+
+    requests = [e for e in _trajectory()[before:] if e["event_type"] == "http_request"]
+    assert [{k: e[k] for k in ("route", "service", "method", "path", "status", "role", "changelog_id_before",
+                               "changelog_id_after")} for e in requests] == [
+        {"route": "website", "service": "mail", "method": "POST", "path": "/api/drafts", "status": 201,
+         "role": "bob@example.com", "changelog_id_before": None, "changelog_id_after": 1},
+        {"route": "svc", "service": "mail", "method": "PATCH", "path": "/api/drafts/d1", "status": 200,
+         "role": "default", "changelog_id_before": 1, "changelog_id_after": 2},
+    ]
+    assert requests[0]["caused_by"] == [{"trigger_id": "reply", "action_index": 0, "fire": 2}]
+    _, body_bytes = await _http(gw, "GET", "/changes")
+    [first, second] = json.loads(body_bytes)["changes"]
+    assert first["actors"] == [{"kind": "request", "event_id": requests[0]["event_id"], "role": "bob@example.com",
+                                "caused_by": [{"trigger_id": "reply", "action_index": 0, "fire": 2}],
+                                "request": {"route": "website", "service": "mail", "method": "POST",
+                                            "path": "/api/drafts", "status": 201}}]
+    assert second["actors"][0]["request"]["route"] == "svc"
+
+
+@pytest.mark.asyncio
+async def test_a_write_made_while_the_response_streams_is_inside_the_window(monkeypatch):
+    """Headers first, then a write while the body streams (an MCP SSE answer): the window closes at the body's end."""
+    changelog = _Changelog()
+
+    async def body():
+        yield b"event: message\n"
+        changelog.write("messages", "m1")
+        yield b'data: {"result": {}}\n\n'
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, rest_proxy_urls={"mcp-mail": "http://mail:8000"})
+    before = len(_trajectory())
+
+    status, body_bytes = await _http(gw, "POST", "/svc/mcp-mail/mcp", headers={"AgentEnv-Role": "carol@example.com"},
+                                     body=b'{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {}}')
+
+    assert status == 200 and body_bytes.endswith(b'{"result": {}}\n\n')
+    [logged] = [e for e in _trajectory()[before:] if e["event_type"] == "http_request"]
+    assert (logged["path"], logged["changelog_id_before"], logged["changelog_id_after"]) == ("/mcp", None, 1)
+    _, changes = await _http(gw, "GET", "/changes")
+    assert [a["role"] for a in json.loads(changes)["changes"][0]["actors"]] == ["carol@example.com"]
+
+
+@pytest.mark.asyncio
+async def test_a_write_window_does_not_hold_the_stream_back(monkeypatch):
+    changelog = _Changelog()
+    gate = asyncio.Event()
+
+    async def body():
+        yield b"a" * 1024
+        await gate.wait()
+        changelog.write("messages", "m1")
+        yield b"b" * 1024
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, rest_proxy_urls={"mcp-mail": "http://mail:8000"})
+    before = len(_trajectory())
+    first_chunk = asyncio.Event()
+    received = False
+
+    async def receive():
+        nonlocal received
+        if not received:
+            received = True
+            return {"type": "http.request", "body": b"{}", "more_body": False}
+        await asyncio.sleep(3600)
+
+    async def send(message):
+        if message["type"] == "http.response.body" and message.get("body"):
+            first_chunk.set()
+
+    scope = {"type": "http", "asgi": {"version": "3.0", "spec_version": "2.3"}, "http_version": "1.1", "method": "POST",
+             "scheme": "http", "path": "/svc/mcp-mail/mcp", "raw_path": b"/svc/mcp-mail/mcp", "query_string": b"",
+             "root_path": "", "server": ("g", 80), "client": ("c", 1), "headers": []}
+    task = asyncio.create_task(gw._mcp.streamable_http_app()(scope, receive, send))
+    await asyncio.wait_for(first_chunk.wait(), 5)
+    assert not [e for e in _trajectory()[before:] if e["event_type"] == "http_request"]
+    gate.set()
+    await asyncio.wait_for(task, 5)
+    [logged] = [e for e in _trajectory()[before:] if e["event_type"] == "http_request"]
+    assert (logged["changelog_id_before"], logged["changelog_id_after"]) == (None, 1)
+
+
+@pytest.mark.asyncio
+async def test_a_mutating_request_that_wrote_nothing_is_not_logged(monkeypatch):
+    changelog = _Changelog()
+    changelog.write("messages", "m1")
+
+    async def body():
+        yield b'{"jsonrpc": "2.0", "id": 1, "result": {"parts": []}}'
+
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, content=body()))
+    monkeypatch.setattr(rest_proxy_module.httpx, "AsyncClient", lambda **kw: _RealAsyncClient(transport=transport, **kw))
+    gw = _gateway(changelog, rest_proxy_urls={"mail": "http://mail-backend"})
+    before = len(_trajectory())
+
+    for method, path in (("POST", "/svc/mail/agentenv"), ("PUT", "/svc/mail/api/drafts/d1"), ("DELETE", "/svc/mail/x")):
+        assert (await _http(gw, method, path, body=b'{"method": "data/get"}'))[0] == 200
+
+    assert [e for e in _trajectory()[before:] if e["event_type"] == "http_request"] == []
+
+
+@pytest.mark.asyncio
+async def test_changes_pages_by_after_id_and_limit():
+    changelog = _Changelog()
+    for i in range(5):
+        changelog.write("messages", f"m{i + 1}")
+    gw = _gateway(changelog)
+
+    status, body = await _http(gw, "GET", "/changes?after_id=1&limit=2")
+    page = json.loads(body)
+    assert status == 200 and [c["id"] for c in page["changes"]] == [2, 3] and page["has_more"] is True
+    assert all(c["actors"] == [] for c in page["changes"])
+    _, body = await _http(gw, "GET", "/changes?after_id=3&limit=2")
+    assert [c["id"] for c in json.loads(body)["changes"]] == [4, 5] and json.loads(body)["has_more"] is False
+    assert changelog.queries[-1][1] == (3, 3)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("query,status,fragment", [
+    ("?after_id=x", 400, "must be integers"),
+    ("?after_id=-1", 400, "after_id must be >= 0"),
+    ("?limit=0", 400, "limit between 1 and 10000"),
+    ("?limit=10001", 400, "limit between 1 and 10000"),
+])
+async def test_changes_refuses_a_bad_page(query, status, fragment):
+    got, body = await _http(_gateway(_Changelog()), "GET", f"/changes{query}")
+    assert got == status and fragment in json.loads(body)["error"]
+
+
+@pytest.mark.asyncio
+async def test_an_env_without_a_changelog_says_so():
+    gw = Gateway(host="127.0.0.1", port=0, server_name="t", internal_mcp_servers=[])
+    status, body = await _http(gw, "GET", "/changes")
+    assert status == 404 and json.loads(body)["error"] == "this env keeps no changelog"
+
+
+def test_an_empty_changelog_still_reads_null():
+    queries = []
+
+    class _Conn:
+        def cursor(self):
+            class _Cursor:
+                def __enter__(self):
+                    return self
+
+                def __exit__(self, *exc):
+                    return False
+
+                def execute(self, sql):
+                    queries.append(sql)
+
+                def fetchone(self):
+                    return (None,)
+            return _Cursor()
+
+    gw = Gateway(host="127.0.0.1", port=0, server_name="t", internal_mcp_servers=[])
+    gw._db_conn = _Conn()
+    assert gw._query_changelog_id() is None
+    assert queries == ["SELECT MAX(id) FROM public._changelog"]
+
+
+def test_a_first_write_on_an_empty_changelog_still_has_a_window():
+    changes = ChangeAttribution()
+    changes.observe(_call("event_1", "mail_send_message", "default"))
+    changes.observe(_result("event_1", None, 2))
+    assert [[a["event_id"] for a in who] for who in changes.actors([1, 2, 3])] == [["event_1"], ["event_1"], []]
diff --git a/tst/unit/env/gateway/gateway_card_test.py b/tst/unit/env/gateway/gateway_card_test.py
index dff2d166..5acecd8c 100644
--- a/tst/unit/env/gateway/gateway_card_test.py
+++ b/tst/unit/env/gateway/gateway_card_test.py
@@ -201,11 +201,12 @@ def test_rewrite_child_card_drops_the_childs_interfaces():
     assert _gateway({"mcp-slack": "http://slack:18765"})._rewrite_child_card("mcp-slack", child)["additionalInterfaces"] == []
 
 
-def test_gateway_card_lists_its_seven_extensions():
+def test_gateway_card_lists_its_eight_extensions():
     assert _gateway({})._gateway_extensions() is gateway_constants.GATEWAY_EXTENSIONS
     assert [e["uri"] for e in _gateway({})._gateway_extensions()] == [
         "urn:agentenv:disable-tool/v1", "urn:agentenv:enable-tool/v1", "urn:agentenv:triggers/v1",
         "urn:agentenv:clock/v1", "urn:agentenv:trajectory/v1", "urn:agentenv:step/v1", "urn:agentenv:state/v1",
+        "urn:agentenv:changes/v1",
     ]
 
 
diff --git a/tst/unit/env/test_env.py b/tst/unit/env/test_env.py
index 7c52efd9..57d8982a 100644
--- a/tst/unit/env/test_env.py
+++ b/tst/unit/env/test_env.py
@@ -24,7 +24,7 @@
     EnvNeedsSandbox, gateway_url_of, require_gateway_url, require_sandbox,
 )
 from agent_env.env.gateway.constants import (
-    EXT_CLOCK_URI, EXT_DISABLE_TOOL_URI, EXT_ENABLE_TOOL_URI, EXT_STATE_URI, EXT_STEP_URI, EXT_TRAJECTORY_URI, EXT_TRIGGERS_URI,
+    EXT_CHANGES_URI, EXT_CLOCK_URI, EXT_DISABLE_TOOL_URI, EXT_ENABLE_TOOL_URI, EXT_STATE_URI, EXT_STEP_URI, EXT_TRAJECTORY_URI, EXT_TRIGGERS_URI,
     WELL_KNOWN_PATH,
 )
 from agent_env.env.gateway.gateway import Gateway
@@ -137,8 +137,9 @@ def test_a_leaf_card_is_its_own_only_child_env():
     (EXT_CLOCK_URI, "state", None, "GET", "/clock/state"),
     (EXT_TRIGGERS_URI, "state", None, "GET", "/triggers/state"),
     (EXT_TRAJECTORY_URI, "get", None, "GET", "/trajectory"),
+    (EXT_CHANGES_URI, "get", {"after_id": "7", "limit": "100"}, "GET", "/changes"),
 ], ids=["clock-set_time", "triggers-register", "step", "state", "tools-disable", "tools-enable", "clock-state",
-        "triggers-state", "trajectory-get"])
+        "triggers-state", "trajectory-get", "changes-get"])
 async def test_invoke_calls_the_endpoint_and_verb_the_card_advertises(sent, composed, base, uri, method, params, verb, path):
     await _record(composed, base).invoke(uri, method, params)
     [request] = sent
diff --git a/tst/unit/task_step/test_snapshot_env.py b/tst/unit/task_step/test_snapshot_env.py
index d8d32313..b6eca56b 100644
--- a/tst/unit/task_step/test_snapshot_env.py
+++ b/tst/unit/task_step/test_snapshot_env.py
@@ -4,6 +4,7 @@
 
 import json
 import re
+from types import SimpleNamespace
 
 import pytest
 from agentenv_protocol import FilePart
@@ -60,6 +61,7 @@ def test_dict_roundtrip_and_registry():
         original_universe_artifact_id="base-universe",
         export_timeout_seconds=120,
         include_env_trajectory=True,
+        include_env_changes=True,
     )
     data = step.to_dict()
     assert data["type"] == "snapshot_env"
@@ -69,7 +71,7 @@ def test_dict_roundtrip_and_registry():
     for attr in (
         "env_id", "env_instance_id", "gateway_url", "snapshot_id",
         "original_universe_artifact_id", "export_timeout_seconds",
-        "include_env_trajectory",
+        "include_env_trajectory", "include_env_changes",
     ):
         assert getattr(restored, attr) == getattr(step, attr)
 
@@ -79,6 +81,7 @@ def test_from_dict_defaults():
     assert step.env_id is None
     assert step.export_timeout_seconds == SnapshotEnvTaskStep.DEFAULT_EXPORT_TIMEOUT_SECONDS
     assert step.include_env_trajectory is False
+    assert step.include_env_changes is False
 
 
 # ── env resolution ───────────────────────────────────────────────────────────
@@ -450,6 +453,92 @@ async def test_trajectory_capture_with_a_gateway_url_override_reads_the_override
     assert [(m, u) for m, u, _ in client.requests] == [("GET", "https://override/trajectory")]
 
 
+# ── env changes capture ──────────────────────────────────────────────────────
+
+class _FakeChangesClient(_FakeAsyncClient):
+    """Answers GET /changes from `rows`, paged by after_id and limit as the gateway pages them."""
+
+    def __init__(self, rows=None, error=None, page_size=2):
+        super().__init__(error=error)
+        self.rows = rows or []
+        self.page_size = page_size
+
+    async def request(self, method, url, params=None):
+        self.requests.append((method, url, params))
+        if self.error:
+            raise self.error
+        page = [r for r in self.rows if r["id"] > params["after_id"]]
+        body = {"changes": page[:self.page_size], "has_more": len(page) > self.page_size}
+        return SimpleNamespace(raise_for_status=lambda: None, json=lambda: body)
+
+
+def _change(row_id, *actors):
+    return {"id": row_id, "table_name": "messages", "operation": "INSERT", "row_id": f"m{row_id}",
+            "actors": [{"kind": "tool_call", "event_id": f"event_{a}", "tool": "mail_send_message", "role": "default"}
+                       for a in actors]}
+
+
+@pytest.mark.asyncio
+async def test_changes_capture_pages_every_row_into_jsonl_and_counts_the_attributed(monkeypatch):
+    _wire_fake_snapshot(monkeypatch)
+    rows = [_change(1, 1), _change(2, 3), _change(3), _change(4, 5, 6), _change(5, 7)]
+    client = _FakeChangesClient(rows)
+    store = _wire_fake_capture(monkeypatch, client)
+
+    ctx = _ctx(_deployed())
+    await _step(include_env_changes=True).execute(ctx)
+
+    ((key, (data, content_type)),) = store.uploads.items()
+    assert key.startswith("env_changes/instance_id=task-x-65717eb66eeec083/env-1-") and key.endswith(".jsonl")
+    assert [json.loads(line) for line in data.decode().splitlines()] == rows and content_type == "application/jsonl"
+    entry = ctx.metadata["env_changes"]["env-1"]
+    assert {k: entry[k] for k in ("change_count", "attributed_count", "ambiguous_count")} == {
+        "change_count": 5, "attributed_count": 3, "ambiguous_count": 1}
+    assert entry["object_url"] == f"s3://fake-bucket/{key}" and entry["capture_step_id"] == "snap-1"
+    assert [(m, u, p["after_id"]) for m, u, p in client.requests] == [
+        ("GET", "https://gw-1/changes", 0), ("GET", "https://gw-1/changes", 2), ("GET", "https://gw-1/changes", 4)]
+    assert "env_trajectory" not in ctx.metadata
+
+
+@pytest.mark.asyncio
+async def test_changes_capture_failure_is_fail_open(monkeypatch):
+    _wire_fake_snapshot(monkeypatch)
+    store = _wire_fake_capture(monkeypatch, _FakeChangesClient(error=RuntimeError("gateway unreachable")))
+
+    ctx = _ctx(_deployed())
+    await _step(include_env_changes=True).execute(ctx)
+
+    assert ctx.metadata["env_changes"]["env-1"]["error"] == "RuntimeError: gateway unreachable"
+    assert store.uploads == {} and ctx.metadata["env_snapshotted_universes"]["snap-1"]["version"] == 1
+
+
+@pytest.mark.asyncio
+async def test_changes_capture_records_one_sentence_when_the_card_offers_no_changes(monkeypatch):
+    _wire_fake_snapshot(monkeypatch)
+    client = _FakeChangesClient([_change(1, 1)])
+    _wire_fake_capture(monkeypatch, client)
+    deployed = _deployed()
+    deployed.environment_card = {"capabilities": {"extensions": [e for e in GATEWAY_EXTENSIONS if e["uri"] == EXT_STATE_URI]}}
+
+    ctx = _ctx(deployed)
+    await _step(include_env_changes=True).execute(ctx)
+
+    assert ctx.metadata["env_changes"]["env-1"]["error"] == (
+        "EnvCapabilityUnsupported: env 'env-1' does not offer 'get' on urn:agentenv:changes/v1.")
+    assert client.requests == []
+
+
+@pytest.mark.asyncio
+async def test_changes_capture_with_a_gateway_url_override_reads_the_override(monkeypatch):
+    _wire_fake_snapshot(monkeypatch)
+    client = _FakeChangesClient([_change(1, 1)])
+    _wire_fake_capture(monkeypatch, client)
+
+    await _step(env_id="env-1", gateway_url="https://override/", include_env_changes=True).execute(_ctx())
+
+    assert [(m, u) for m, u, _ in client.requests] == [("GET", "https://override/changes")]
+
+
 # ── core: all-or-nothing + streaming guard ───────────────────────────────────
 
 class _FakeServiceEnv: