fix(control-plane,runner): prevent gRPC message replay on session restart

Control-plane:
- Set IS_RESUME=true when session.StartTime is non-nil (parity
  with legacy operator)
- Query max message seq via SessionMessages().List(size=1,
  orderBy="seq desc") and pass as RESUME_AFTER_SEQ env var

Runner:
- gRPC listener reads RESUME_AFTER_SEQ env var on startup
- Watches from that seq so historical messages are never replayed
- Zero scanning delay — the control-plane provides the seq directly
- Fresh sessions (no env var) start from seq 0 as before

Root cause: the control-plane never set IS_RESUME, so the runner
re-executed the initial prompt on restart. The gRPC listener also
started from seq=0, replaying all historical user messages and
triggering duplicate bridge.run() calls.

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,14 @@ 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"))
+		maxSeq := r.resolveMaxMessageSeq(ctx, sdk, session.ID)
+		if maxSeq > 0 {
+			env = append(env, envVar("RESUME_AFTER_SEQ", fmt.Sprintf("%d", maxSeq)))
+		}
+	}
+
 	if r.cfg.AnthropicAPIKey != "" {
 		env = append(env, envVar("ANTHROPIC_API_KEY", r.cfg.AnthropicAPIKey))
 	}
@@ -769,6 +777,23 @@ func (r *SimpleKubeReconciler) buildEnv(ctx context.Context, session types.Sessi
 	return env
 }
 
+func (r *SimpleKubeReconciler) resolveMaxMessageSeq(ctx context.Context, sdk *sdkclient.Client, sessionID string) int {
+	opts := &types.ListOptions{
+		Size:    1,
+		Search:  fmt.Sprintf("session_id = '%s'", sessionID),
+		OrderBy: "seq desc",
+	}
+	result, err := sdk.SessionMessages().List(ctx, opts)
+	if err != nil {
+		r.logger.Warn().Err(err).Str("session_id", sessionID).Msg("failed to resolve max message seq")
+		return 0
+	}
+	if len(result.Items) == 0 {
+		return 0
+	}
+	return result.Items[0].Seq
+}
+
 func (r *SimpleKubeReconciler) resolveCredentialIDs(ctx context.Context, sdk *sdkclient.Client, projectID string) (map[string]string, error) {
 	result := map[string]string{}
 

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

@@ -185,8 +185,35 @@ def _watch_in_thread(
                 exc_info=True,
             )
 
+    def _resume_after_seq(self) -> int:
+        """Read RESUME_AFTER_SEQ env var set by the control-plane.
+
+        On resumed sessions, the control-plane queries the API server
+        for the current max message seq and passes it as an env var.
+        The listener starts watching after this seq so historical
+        messages are never replayed.
+        """
+        import os
+        raw = os.getenv("RESUME_AFTER_SEQ", "").strip()
+        if not raw:
+            return 0
+        try:
+            seq = int(raw)
+            logger.info(
+                "[GRPC LISTENER] Resuming after seq=%d for session=%s",
+                seq,
+                self._session_id,
+            )
+            return seq
+        except ValueError:
+            logger.warning(
+                "[GRPC LISTENER] Invalid RESUME_AFTER_SEQ=%r, starting from 0",
+                raw,
+            )
+            return 0
+
     async def _listen_loop(self) -> None:
-        last_seq = 0
+        last_seq = self._resume_after_seq()
         backoff = _BACKOFF_INITIAL
 
         while True: