fix(runner): prevent gRPC message replay on session restart

The gRPC listener started watching from seq=0 on every startup,
which replayed all historical messages including the original user
prompt. On session restart, this caused the agent to auto-continue
the conversation without the user sending a new message.

Fix: _catch_up_seq() scans existing messages on startup to find
the current max seq, then watches from there. Only messages that
arrive AFTER the listener starts trigger runs.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jsell-rh

components/ambient-control-plane/internal/reconciler/kube_reconciler.go

@@ -719,6 +719,10 @@ func (r *SimpleKubeReconciler) buildEnv(ctx context.Context, session types.Sessi
 		envVar("REQUESTS_CA_BUNDLE", "/etc/pki/ca-trust/extracted/pem/service-ca.crt"),
 	}
 
+	if session.StartTime != nil {
+		env = append(env, envVar("IS_RESUME", "true"))
+	}
+
 	if r.cfg.AnthropicAPIKey != "" {
 		env = append(env, envVar("ANTHROPIC_API_KEY", r.cfg.AnthropicAPIKey))
 	}

components/runners/ambient-runner/ambient_runner/bridges/claude/bridge.py

@@ -495,10 +495,12 @@ async def start_grpc_listener(self, grpc_url: str) -> None:
         from ambient_runner.bridges.claude.grpc_transport import GRPCSessionListener
 
         session_id = self._context.session_id
+        is_resume = os.getenv("IS_RESUME", "").strip().lower() == "true"
         self._grpc_listener = GRPCSessionListener(
             bridge=self,
             session_id=session_id,
             grpc_url=grpc_url,
+            is_resume=is_resume,
         )
         self._grpc_listener.start()
         logger.info(

components/runners/ambient-runner/ambient_runner/bridges/claude/grpc_transport.py

@@ -94,10 +94,12 @@ def __init__(
         bridge: "PlatformBridge",
         session_id: str,
         grpc_url: str,
+        is_resume: bool = False,
     ) -> None:
         self._bridge = bridge
         self._session_id = session_id
         self._grpc_url = grpc_url
+        self._is_resume = is_resume
         self._grpc_client: Optional["AmbientGRPCClient"] = None
         self.ready = asyncio.Event()
         self._task: Optional[asyncio.Task] = None
@@ -185,8 +187,41 @@ def _watch_in_thread(
                 exc_info=True,
             )
 
+    async def _catch_up_seq(self) -> int:
+        """Find the current max seq so we only process NEW messages.
+
+        Only runs on resumed sessions (is_resume=True). Fresh sessions
+        start from seq 0 with no delay.
+        """
+        if not self._is_resume or self._grpc_client is None:
+            return 0
+
+        def _scan() -> int:
+            max_seq = 0
+            try:
+                for msg in self._grpc_client.session_messages.watch(
+                    self._session_id, after_seq=0, timeout=30.0,
+                ):
+                    max_seq = max(max_seq, msg.seq)
+            except Exception as exc:
+                logger.warning(
+                    "[GRPC LISTENER] Catch-up scan error (using seq=%d): %s",
+                    max_seq,
+                    exc,
+                )
+            return max_seq
+
+        loop = asyncio.get_running_loop()
+        max_seq = await loop.run_in_executor(None, _scan)
+        logger.info(
+            "[GRPC LISTENER] Resumed session — caught up to seq=%d for session=%s",
+            max_seq,
+            self._session_id,
+        )
+        return max_seq
+
     async def _listen_loop(self) -> None:
-        last_seq = 0
+        last_seq = await self._catch_up_seq()
         backoff = _BACKOFF_INITIAL
 
         while True: