fix: fence MCP S3 writes after a Cloud turn is stopped

CHANGELOG.md

@@ -239,6 +239,18 @@ Entries before v0.5.0 were written retroactively as summaries.
 
 ### Fixed
 
+- **Stopping a Cloud turn no longer lets its in-flight tool calls change the deck
+  afterwards.** A tool call the agent had already sent kept running on the MCP runtime and
+  wrote to S3 after the stop (on dev: `run_python` saved 4 slides 1.7 s after a cancel, and
+  `generate_pptx`'s previews and thumbnail landed 1.5 s after the stop button). Each turn
+  now sends a generation number with its Presentation Maker calls (composers included), and
+  a stopped turn raises the deck's write fence (`writeFence` on the deck item) to it through
+  the new `fence_turn` tool. The MCP server checks the fence before every S3 write and skips
+  the writes of a stopped turn (the call returns a "Cancelled" error), and the deck record
+  update after a build is conditional on it. A turn that ends normally fences nothing, so
+  its last previews still land when the next message is sent quickly. Calls without a
+  generation (the local server, other MCP clients) are unchanged.
+
 - **Stopping a Cloud turn no longer leaves it running in the background.** The stop button
   ends the runtime session, and AgentCore then sends the agent's process SIGTERM. The agent
   used to wait for the running turn to finish, so the turn and its composers kept calling

agent/Dockerfile

@@ -27,7 +27,7 @@ RUN uv export --frozen --directory /tmp/workspace --package sdpm-agent --no-dev
 RUN useradd -m -u 1000 bedrock_agentcore
 USER bedrock_agentcore
 
-COPY agent/basic_agent.py agent/cost_logger.py agent/errors.py agent/factory.py agent/composition.py agent/mcp_clients.py agent/message_hooks.py agent/model_profiles.py agent/resilience.py agent/session.py agent/stop_guard.py agent/streaming.py ./
+COPY agent/basic_agent.py agent/cost_logger.py agent/errors.py agent/factory.py agent/composition.py agent/mcp_clients.py agent/message_hooks.py agent/model_profiles.py agent/resilience.py agent/session.py agent/stop_guard.py agent/streaming.py agent/turn_fence.py ./
 COPY agent/prompts/ prompts/
 COPY agent/modes/ modes/
 COPY agent/tools/ tools/

agent/basic_agent.py

@@ -23,6 +23,7 @@
 from errors import BAD_INPUT, AUTH, error_event
 from factory import create_agent
 from streaming import stream_agent
+from turn_fence import TurnFence
 
 logger = logging.getLogger("sdpm.agent")
 
@@ -87,8 +88,11 @@ async def agent_stream(payload, context):
 
     cancel = asyncio.Event()
     _cancel_events[session_id] = cancel
-    # The turn's stop event: set on SIGTERM, observed by Strands and the composers.
-    stop = stop_guard.begin_turn()
+    # The turn's write fence: its generation goes with every Presentation Maker call, and
+    # a stopped turn fences its decks so its in-flight calls no longer write (turn_fence.py).
+    fence = TurnFence()
+    # The turn's stop event: set on SIGTERM (which also fences), observed by Strands and the composers.
+    stop = stop_guard.begin_turn(on_stop=fence.advance)
 
     try:
         os.environ["_CURRENT_SESSION_ID"] = session_id
@@ -102,6 +106,7 @@ async def agent_stream(payload, context):
             jwt_token=jwt_token,
             chat_model_id=requested_chat_model_id,
             create_model_id=requested_create_model_id,
+            fence=fence,
         )
 
         yield {"mcp_status": mcp_status}
@@ -114,6 +119,12 @@ async def agent_stream(payload, context):
         yield error_event(e)
     finally:
         stop_guard.end_turn(stop)
+        # Stopped (cancel, a newer request, the client gone, SIGTERM): fence the turn's decks.
+        # hold: a SIGTERM that follows (StopRuntimeSession after a cancel) waits for the call.
+        if stop.is_set():
+            stop_guard.hold(fence.advance())
+        else:
+            fence.release()
         if _cancel_events.get(session_id) is cancel:
             del _cancel_events[session_id]
 

agent/factory.py

@@ -33,6 +33,7 @@
 from modes.composer import make_compose_slides
 from resilience import LoopGuard, gate_tool_call
 from session import fix_excess_tool_results
+from turn_fence import TurnFence
 from tools.hearing_tool import hearing
 from tools.web_tools import web_fetch
 
@@ -64,7 +65,7 @@ def _resolve_model_id(requested: str | None, default: str) -> str:
 
 
 _MCP_FACTORIES = [
-    lambda jwt_token, session_id="", tool_filters=None: mcp_agentcore_runtime(jwt_token=jwt_token, session_id=session_id, tool_filters=tool_filters),
+    lambda jwt_token, session_id="", tool_filters=None, **kwargs: mcp_agentcore_runtime(jwt_token=jwt_token, session_id=session_id, tool_filters=tool_filters, **kwargs),
     lambda jwt_token, session_id="", tool_filters=None: mcp_aws_knowledge(),
     lambda jwt_token, session_id="", tool_filters=None: mcp_aws_pricing(),
 ]
@@ -221,9 +222,13 @@ def _session_manager(memory_id: str, session_id: str, user_id: str, region: str)
 # Unified factory
 # ---------------------------------------------------------------------------
 
-def create_agent(mode: str, user_id: str, session_id: str, jwt_token: str, chat_model_id: str | None = None, create_model_id: str | None = None) -> tuple[Agent, list[dict]]:
+def create_agent(mode: str, user_id: str, session_id: str, jwt_token: str, chat_model_id: str | None = None, create_model_id: str | None = None, fence: TurnFence | None = None) -> tuple[Agent, list[dict]]:
     """Create a Strands Agent for the given mode.
 
+    ``fence`` (turn_fence.py): its generation goes into the ``_meta`` of every
+    Presentation Maker call (orchestrator and composers), and it learns the decks the
+    turn's tool calls name.
+
     Returns:
         Tuple of (Configured Strands Agent, MCP status list).
     """
@@ -251,6 +256,7 @@ def create_agent(mode: str, user_id: str, session_id: str, jwt_token: str, chat_
 
     # MCP servers — Presentation Maker built per turn (it carries the JWT), AWS
     # Knowledge / AWS Pricing reused from the process; all connected concurrently below.
+    fence_kwargs = {"meta": fence.meta} if fence is not None else {}
     built = []
     names = []
     required_flags = []
@@ -260,7 +266,7 @@ def create_agent(mode: str, user_id: str, session_id: str, jwt_token: str, chat_
         else:
             # Apply tool_filters only to the Presentation Maker server (index 0)
             filters = {"allowed": cfg.allowed_tools} if (i == 0 and cfg.allowed_tools) else None
-            built.append(factory_fn(jwt_token, session_id=session_id, tool_filters=filters))
+            built.append(factory_fn(jwt_token, session_id=session_id, tool_filters=filters, **fence_kwargs))
         names.append(name)
         required_flags.append(required)
 
@@ -311,7 +317,7 @@ def critical_path(required_clients: list) -> None:
         # Composers get their own Presentation Maker client per group; it carries the
         # composer allowlist so a composer sees only the deck tools.
         composer_filters = {"allowed": MODES["composer"].allowed_tools}
-        composer_mcp_factory = lambda mcp_session_id="": mcp_agentcore_runtime(jwt_token=jwt_token, session_id=mcp_session_id, tool_filters=composer_filters)  # noqa: E731
+        composer_mcp_factory = lambda mcp_session_id="": mcp_agentcore_runtime(jwt_token=jwt_token, session_id=mcp_session_id, tool_filters=composer_filters, **fence_kwargs)  # noqa: E731
         compose_slides = make_compose_slides(mcp_servers, composer_model, composer_mcp_factory, extra_tools=[web_fetch], model_id=resolved_create, user_id=user_id, session_id=session_id)
         tools.append(compose_slides)
 
@@ -381,6 +387,11 @@ def critical_path(required_clients: list) -> None:
     # Write gate: no tool runs once the process is stopping (SIGTERM, stop_guard.py)
     agent.hooks.add_callback(BeforeToolCallEvent, gate_tool_call)
 
+    # Write fence: the decks this turn touches, and the client that fences them on a stop
+    if fence is not None:
+        agent.hooks.add_callback(BeforeToolCallEvent, fence.note_tool_call)
+        fence.bind(mcp_servers[0] if mcp_servers else None)
+
     # Cost logger
     agent.hooks.add_callback(AfterInvocationEvent, log_usage)
 

agent/mcp_clients.py

@@ -27,6 +27,7 @@ def mcp_agentcore_runtime(
     jwt_token: str,
     session_id: str = "",
     tool_filters: dict | None = None,
+    meta: dict | None = None,
 ) -> MCPClient:
     """Pattern 1: Amazon Bedrock AgentCore Runtime MCP Server with JWT Bearer authentication.
 
@@ -36,6 +37,9 @@ def mcp_agentcore_runtime(
             routes by that header, so omitting it makes every request land on a
             new microVM — see the note below. Empty string disables the header.
         tool_filters: Optional tool filter dict (e.g. {"allowed": ["tool1", "tool2"]}).
+        meta: Optional ``_meta`` sent with every tool call (the turn's write-fence
+            generation, turn_fence.py). It travels in the JSON-RPC body, which the
+            AgentCore proxy forwards as is; Strands merges trace context into it.
     """
     region = os.environ.get("AWS_REGION", os.environ.get("AWS_DEFAULT_REGION", "us-east-1"))
     runtime_arn = os.environ["MCP_RUNTIME_ARN"]
@@ -62,6 +66,8 @@ def mcp_agentcore_runtime(
     kwargs: dict = {}
     if tool_filters:
         kwargs["tool_filters"] = tool_filters
+    if meta:
+        kwargs["meta"] = meta
 
     return MCPClient(
         lambda: streamablehttp_client(

agent/stop_guard.py

@@ -11,21 +11,26 @@
 The guard, installed through the app's lifespan so it runs before uvicorn's own handler:
   1. closes the write gate (resilience.close_writes): tool calls, compose's direct MCP
      calls and Memory events (GatedMemorySessionManager) are dropped from then on;
-  2. sets every running turn's stop event, which Strands and the composers observe;
+  2. sets every running turn's stop event, which Strands and the composers observe, and
+     runs its stop callback (the turn's fence_turn call, turn_fence.py);
   3. starts a watchdog that flushes OpenTelemetry and exits the process after
      EXIT_AFTER_S, whatever is still running (a composer thread blocked in a read);
   4. closes the process's resident MCP clients (AWS Knowledge, and AWS Pricing's stdio
      subprocess) on a side thread, best effort: the watchdog's exit bounds it.
-basic_agent passes GRACEFUL_SHUTDOWN_S to uvicorn so it stops waiting before that.
+basic_agent passes GRACEFUL_SHUTDOWN_S to uvicorn so it stops waiting before that. The
+lifespan's shutdown then waits, up to the watchdog's exit, for the stop calls still
+running (hold / wait_held), since a turn can end before its fence_turn reaches the server.
 """
 
+import asyncio
 import contextlib
 import logging
 import os
 import signal
 import sys
 import threading
 import time
+from typing import Callable
 
 from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager
 
@@ -39,35 +44,70 @@
 
 # Re-entrant: the SIGTERM handler takes it on the main thread, which may hold it already.
 _lock = threading.RLock()
-_running: set[threading.Event] = set()
+# Running turns: stop event -> what to call when the turn is stopped (or None).
+_running: dict[threading.Event, Callable[[], object] | None] = {}
 _exiting = threading.Event()
+# Calls a stopped turn still has to send (turn_fence's fence_turn), and when the watchdog
+# exits the process. The lifespan's shutdown waits for them until then: right after it,
+# uvicorn re-raises SIGTERM under the default handler, which ends the process at once.
+_held: set[threading.Thread] = set()
+_exit_at: float | None = None
 
 
-def begin_turn() -> threading.Event:
+def begin_turn(on_stop: Callable[[], object] | None = None) -> threading.Event:
     """Register a running turn and return its stop event (set on SIGTERM).
 
+    ``on_stop`` runs when SIGTERM stops the turn (on the signal handler's thread, so it
+    must not block: turn_fence.TurnFence.advance hands its call to a thread and returns
+    it, and the process waits for a returned thread before it exits; see hold()).
     Pair with end_turn() in a finally block.
     """
     stop = threading.Event()
     with _lock:
-        _running.add(stop)
+        _running[stop] = on_stop
     if writes_closed():
         stop.set()  # the process is already stopping
     return stop
 
 
 def end_turn(stop: threading.Event) -> None:
     with _lock:
-        _running.discard(stop)
+        _running.pop(stop, None)
+
+
+def hold(thread: object) -> None:
+    """Keep the process alive for ``thread`` (if it is one) when it shuts down, up to the
+    watchdog's exit."""
+    if not isinstance(thread, threading.Thread):
+        return
+    with _lock:
+        _held.difference_update([t for t in _held if not t.is_alive()])
+        _held.add(thread)
+
+
+def wait_held() -> None:
+    """Wait for the held threads, until the watchdog's exit (EXIT_AFTER_S without SIGTERM)."""
+    with _lock:
+        threads = list(_held)
+        deadline = _exit_at if _exit_at is not None else time.monotonic() + EXIT_AFTER_S
+    for thread in threads:
+        thread.join(max(0.0, deadline - time.monotonic()))
+    if any(t.is_alive() for t in threads):
+        logger.warning("SIGTERM guard: exiting with %d stop call(s) still running", sum(t.is_alive() for t in threads))
 
 
 def stop_all() -> int:
     """Close the write gate and set the stop event of every running turn."""
     close_writes()
     with _lock:
-        running = list(_running)
-    for stop in running:
+        running = list(_running.items())
+    for stop, on_stop in running:
         stop.set()
+        if on_stop is not None:
+            try:
+                hold(on_stop())
+            except Exception as e:  # noqa: BLE001 - the other turns still stop
+                logger.warning("SIGTERM guard: a turn's stop callback failed: %s", e)
     return len(running)
 
 
@@ -118,6 +158,10 @@ def _close_resident_clients() -> None:
 
 def _on_sigterm(previous):
     def handler(signum, frame):
+        global _exit_at
+        if not _exiting.is_set():  # before stop_all, so wait_held already sees the deadline
+            with _lock:
+                _exit_at = time.monotonic() + EXIT_AFTER_S
         stopped = stop_all()
         if not _exiting.is_set():  # signal handlers run on the main thread only: no race here
             _exiting.set()
@@ -139,6 +183,8 @@ def install() -> None:
 
 @contextlib.asynccontextmanager
 async def lifespan(app):
-    """App lifespan: uvicorn has installed its signal handlers by the time this starts."""
+    """App lifespan: uvicorn has installed its signal handlers by the time this starts.
+    Its shutdown waits for the stop calls still running (wait_held)."""
     install()
     yield
+    await asyncio.to_thread(wait_held)