fix(runner,operator): improve session backup and resume reliability (#1057)

## Summary

Three issues caused data loss and broken session resume when the runner
container OOMed:

- **Repo data lost on OOM**: `backup_git_repos()` only ran on SIGTERM.
Runner OOM meant no SIGTERM → no repo backup. Now runs periodically
every ~5 min.
- **Session resume broken**: CLI session ID was only persisted after
turn completion. OOM during first turn → no session ID in S3 →
`--resume` never passed on restart → agent lost all context. Now
persists immediately on CLI init.
- **Operator didn't detect OOM**: Pod watch predicate only checked pod
phase, which stays `Running` when one container dies (sidecar still
alive). OOMKilled pods sat indefinitely. Now triggers on container
termination too.

## Changes

| File | Change |
|------|--------|
| `state-sync/sync.sh` | Periodic `backup_git_repos` every Nth sync
cycle (configurable `REPO_BACKUP_INTERVAL`, default 5) |
| `session.py` | `on_session_id` callback persists session ID to disk
immediately on CLI init message |
| `bridge.py` | Resume failure detection + removed redundant post-turn
persist |
| `agenticsession_controller.go` | Pod watch predicate triggers on
container termination, not just pod phase |

## Root cause investigation

Diagnosed from live OOMKilled pods in `bug-bash-mturley` namespace on
UAT:
- Confirmed `ambient-code-runner` container OOMKilled (exit 137, 8Gi
limit)
- State-sync kept running but never backed up repos (only on SIGTERM)
- Session JSONL was valid (237 lines, all valid JSON) and `--resume`
works fine on sessions with dangling tool calls (verified via SDK)
- The actual cause: `claude_session_ids.json` never existed in S3
because the runner OOMed during its first turn, before
`_persist_session_ids()` was called
- Operator pod watch predicate filtered out the event because pod phase
stayed `Running`

## Test plan

- [x] Runner tests pass (488 passed, 11 skipped)
- [x] Operator builds clean (`go vet`, `go build`)
- [ ] Deploy to kind cluster and verify periodic repo backup runs
- [ ] Simulate runner OOM and verify operator detects container
termination
- [ ] Verify session resume works after OOM recovery

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Ambient Code Bot <bot@ambient-code.local>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

components/operator/internal/controller/agenticsession_controller.go

@@ -261,8 +261,26 @@ func (r *AgenticSessionReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			if !strings.HasSuffix(e.ObjectNew.Name, "-runner") {
 				return false
 			}
-			// Trigger if phase changed
-			return e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase
+			// Trigger if pod phase changed
+			if e.ObjectOld.Status.Phase != e.ObjectNew.Status.Phase {
+				return true
+			}
+			// Trigger if a container newly terminated (e.g. runner OOM).
+			// Pod phase stays Running when one container dies but
+			// the sidecar is still alive, so we must also check
+			// individual container statuses.
+			oldTerminated := make(map[string]bool, len(e.ObjectOld.Status.ContainerStatuses))
+			for _, cs := range e.ObjectOld.Status.ContainerStatuses {
+				if cs.State.Terminated != nil {
+					oldTerminated[cs.Name] = true
+				}
+			}
+			for _, cs := range e.ObjectNew.Status.ContainerStatuses {
+				if cs.State.Terminated != nil && !oldTerminated[cs.Name] {
+					return true
+				}
+			}
+			return false
 		},
 		DeleteFunc: func(e event.TypedDeleteEvent[*corev1.Pod]) bool {
 			return strings.HasSuffix(e.Object.Name, "-runner")

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

@@ -178,10 +178,21 @@ async def run(
                 async for event in wrapped_stream:
                     yield event
 
-                # Persist session ID after turn completes (for --resume on pod restart)
-                if worker.session_id:
-                    self._session_manager._session_ids[thread_id] = worker.session_id
-                    self._session_manager._persist_session_ids()
+                # Detect resume failure (session ID already persisted
+                # eagerly by the _on_session_id callback at init time).
+                if (
+                    saved_session_id
+                    and worker.session_id
+                    and worker.session_id != saved_session_id
+                ):
+                    logger.warning(
+                        "Session resume failed: requested --resume %s "
+                        "but CLI created new session %s. "
+                        "Previous conversation history was lost "
+                        "(likely caused by ungraceful runner shutdown).",
+                        saved_session_id,
+                        worker.session_id,
+                    )
 
                 # Capture halt state for this thread to avoid race conditions
                 # with concurrent runs modifying the shared adapter's halted flag

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

@@ -31,7 +31,7 @@
 import os
 from contextlib import suppress
 from pathlib import Path
-from typing import Any, AsyncIterator, Optional
+from typing import Any, AsyncIterator, Callable, Optional
 
 logger = logging.getLogger(__name__)
 
@@ -67,10 +67,12 @@ def __init__(
         thread_id: str,
         options: Any,
         api_key: str,
+        on_session_id: Optional[Callable[[str, str], None]] = None,
     ):
         self.thread_id = thread_id
         self._options = options
         self._api_key = api_key
+        self._on_session_id = on_session_id
 
         # Inbound: (prompt, session_id, output_queue) | _SHUTDOWN
         self._input_queue: asyncio.Queue = asyncio.Queue()
@@ -140,6 +142,13 @@ async def _run(self) -> None:
                                 sid = data.get("session_id")
                                 if sid:
                                     self.session_id = sid
+                                    # Persist immediately so the session ID
+                                    # survives even if this turn never completes
+                                    # (e.g. runner OOM during tool execution).
+                                    if self._on_session_id:
+                                        self._on_session_id(
+                                            self.thread_id, sid
+                                        )
 
                         await output_queue.put(msg)
 
@@ -289,7 +298,9 @@ async def get_or_create(
             )
             await self.destroy(thread_id)
 
-        worker = SessionWorker(thread_id, options, api_key)
+        worker = SessionWorker(
+            thread_id, options, api_key, on_session_id=self._on_session_id
+        )
         await worker.start()
         self._workers[thread_id] = worker
         self._locks[thread_id] = asyncio.Lock()
@@ -345,6 +356,15 @@ async def shutdown(self) -> None:
 
     # ── session ID persistence ──
 
+    def _on_session_id(self, thread_id: str, session_id: str) -> None:
+        """Called by workers as soon as the CLI returns a session ID.
+
+        Persists immediately so the mapping survives even if the current
+        turn never completes (e.g. runner OOM during tool execution).
+        """
+        self._session_ids[thread_id] = session_id
+        self._persist_session_ids()
+
     def _session_ids_path(self) -> Path | None:
         if not self._state_dir:
             return None

components/runners/state-sync/sync.sh

@@ -10,6 +10,7 @@ NAMESPACE="${NAMESPACE:-default}"
 SESSION_NAME="${SESSION_NAME:-unknown}"
 SYNC_INTERVAL="${SYNC_INTERVAL:-60}"
 MAX_SYNC_SIZE="${MAX_SYNC_SIZE:-1073741824}"  # 1GB default
+REPO_BACKUP_INTERVAL="${REPO_BACKUP_INTERVAL:-5}"  # Backup repos every Nth sync cycle
 
 # Sanitize inputs to prevent path traversal
 NAMESPACE="${NAMESPACE//[^a-zA-Z0-9-]/}"
@@ -261,6 +262,7 @@ echo "Session: ${NAMESPACE}/${SESSION_NAME}"
 echo "S3 Endpoint: ${S3_ENDPOINT}"
 echo "S3 Bucket: ${S3_BUCKET}"
 echo "Sync interval: ${SYNC_INTERVAL}s"
+echo "Repo backup every: ${REPO_BACKUP_INTERVAL} sync cycles"
 echo "Max sync size: ${MAX_SYNC_SIZE} bytes"
 echo "========================================="
 
@@ -283,8 +285,15 @@ echo "Waiting 30s for workspace to populate..."
 sleep 30
 
 # Main sync loop
+sync_count=0
 while true; do
     check_size || echo "Size check warning (continuing anyway)"
+    # Periodically backup git repos (every Nth cycle) so repo state
+    # is preserved even if the runner container OOMs without SIGTERM
+    sync_count=$((sync_count + 1))
+    if [ $((sync_count % REPO_BACKUP_INTERVAL)) -eq 0 ]; then
+        backup_git_repos || echo "Repo backup had errors (continuing)"
+    fi
     sync_to_s3 || echo "Sync failed, will retry in ${SYNC_INTERVAL}s..."
     sleep ${SYNC_INTERVAL}
 done