fix(runner): always emit MESSAGES_SNAPSHOT to prevent compaction failures (#1693)

## Summary

- **C1 — `grpc_transport.py`**: Skip gRPC messages with empty payloads
instead of building an empty-content message. An empty-content message
causes `process_messages` to return `""`, which triggers an early
`RunFinishedEvent` before `_stream_claude_sdk` ever runs — so
`run_messages` stays empty and `MESSAGES_SNAPSHOT` is never emitted.
Affects both Operator-managed and control-plane-reconciled runner
sessions.

- **C2 — `run.py`**: Assign UUIDs to message dicts missing an `id` in
`to_run_agent_input`. Without ids, `upsert_message` falls back to append
on every call, creating duplicates in the snapshot for multi-turn
sessions going through the gRPC dict message path.

- **C3 — `adapter.py`**: Remove the `if run_messages:` guard around
`MESSAGES_SNAPSHOT` emission. Runs that produce no assistant output
(interrupted, halted on frontend tool, state-management-tool-only turns)
still need a snapshot so `compactFinishedRun` can succeed. When
`run_messages` is empty the snapshot contains the stamped input history,
which is sufficient. Without a snapshot, compaction logs "session
corrupted, keeping raw events" and chat history is lost after the
tail-read optimization (ffe4a21).

## Root cause

`compactFinishedRun` (backend `agui_store.go`) requires
`MESSAGES_SNAPSHOT` to be present in the JSONL to perform compaction. If
absent, it aborts and leaves raw streaming delta events in the file. On
the next large-session tail-read (`8e3bac3`), those deltas may be
missing from the head scan and the user sees blank chat history on
reconnect.

The `MESSAGES_SNAPSHOT` gap was introduced when `ambient-control-plane`
added a second reconciled runner path: control-plane sessions reach the
runner via gRPC (`GRPCSessionListener`) rather than HTTP, and the gRPC
path has different message-shape guarantees that exposed all three gaps
simultaneously.

## Test plan

- [ ] Operator-managed session: send multiple messages, navigate away
and back — chat history intact
- [ ] Control-plane-reconciled session: same navigation test
- [ ] Session where Claude calls only state-management tools (no text
output): navigate away, verify prior history still present on return
- [ ] Session interrupted mid-run (stop button): reconnect, verify
history from previous complete turns is intact
- [ ] Check runner logs: `MESSAGES_SNAPSHOT` log line should appear for
every `RUN_FINISHED`/`RUN_ERROR` turn
- [ ] Check backend logs: `compaction` log should show `X raw events → Y
snapshot events` (not "session corrupted")

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

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

## Release Notes

* **Bug Fixes**
* Improved handling of empty or malformed messages to prevent processing
failures
* Enhanced message snapshot construction to be more reliable across
scenarios
  * Added automatic message ID generation to ensure data consistency
* Strengthened validation for incoming message payloads with better
error logging

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Co-authored-by: Claude <noreply@anthropic.com>

Author: markturansky

components/runners/ambient-runner/ag_ui_claude_sdk/adapter.py

@@ -1310,55 +1310,62 @@ def flush_pending_msg():
         # Emit MESSAGES_SNAPSHOT with input messages + new messages from this run.
         # Enrich tool result messages with tool names so the frontend can
         # reconstruct parent-child hierarchy with proper display names.
-        if run_messages:
-            enriched: list[Any] = []
-            for msg in run_messages:
-                # Check if this is a tool result message that needs a name
-                msg_role = getattr(msg, "role", None)
-                msg_tcid = getattr(msg, "tool_call_id", None)
-                if msg_role == "tool" and msg_tcid and msg_tcid in tool_name_by_id:
-                    # Convert to dict so we can add the name field
-                    if hasattr(msg, "model_dump"):
-                        d = msg.model_dump(exclude_none=True)
-                    elif hasattr(msg, "dict"):
-                        d = msg.dict(exclude_none=True)
-                    else:
-                        d = {
-                            "id": getattr(msg, "id", ""),
-                            "role": msg_role,
-                            "content": getattr(msg, "content", ""),
-                            "tool_call_id": msg_tcid,
-                        }
-                    d["name"] = tool_name_by_id[msg_tcid]
-                    enriched.append(d)
-                else:
-                    enriched.append(msg)
-
-            # Stamp input messages with the run-start timestamp so they
-            # survive a page refresh (the frontend's local timestamp is
-            # lost when reconnecting to the SSE stream).
-            run_start_iso = (
-                datetime.fromtimestamp(run_start_ts / 1000, tz=timezone.utc).isoformat()
-                if run_start_ts
-                else None
-            )
-            stamped_inputs: list[Any] = []
-            for msg in (input_data.messages if input_data else None) or []:
+        #
+        # Always emit MESSAGES_SNAPSHOT regardless of whether run_messages is
+        # populated — compactFinishedRun requires it to succeed. Runs that
+        # produce no assistant output (interrupted, state-tool-only, halted)
+        # still need a snapshot so the JSONL can be compacted and prior
+        # history is not lost. When run_messages is empty the snapshot is
+        # just the stamped input history.
+        enriched: list[Any] = []
+        for msg in run_messages:
+            # Check if this is a tool result message that needs a name
+            msg_role = getattr(msg, "role", None)
+            msg_tcid = getattr(msg, "tool_call_id", None)
+            if msg_role == "tool" and msg_tcid and msg_tcid in tool_name_by_id:
+                # Convert to dict so we can add the name field
                 if hasattr(msg, "model_dump"):
                     d = msg.model_dump(exclude_none=True)
-                elif isinstance(msg, dict):
-                    d = dict(msg)
+                elif hasattr(msg, "dict"):
+                    d = msg.dict(exclude_none=True)
                 else:
                     d = {
                         "id": getattr(msg, "id", ""),
-                        "role": getattr(msg, "role", ""),
+                        "role": msg_role,
                         "content": getattr(msg, "content", ""),
+                        "tool_call_id": msg_tcid,
                     }
-                if "timestamp" not in d and run_start_iso:
-                    d["timestamp"] = run_start_iso
-                stamped_inputs.append(d)
+                d["name"] = tool_name_by_id[msg_tcid]
+                enriched.append(d)
+            else:
+                enriched.append(msg)
+
+        # Stamp input messages with the run-start timestamp so they
+        # survive a page refresh (the frontend's local timestamp is
+        # lost when reconnecting to the SSE stream).
+        run_start_iso = (
+            datetime.fromtimestamp(run_start_ts / 1000, tz=timezone.utc).isoformat()
+            if run_start_ts
+            else None
+        )
+        stamped_inputs: list[Any] = []
+        for msg in (input_data.messages if input_data else None) or []:
+            if hasattr(msg, "model_dump"):
+                d = msg.model_dump(exclude_none=True)
+            elif isinstance(msg, dict):
+                d = dict(msg)
+            else:
+                d = {
+                    "id": getattr(msg, "id", ""),
+                    "role": getattr(msg, "role", ""),
+                    "content": getattr(msg, "content", ""),
+                }
+            if "timestamp" not in d and run_start_iso:
+                d["timestamp"] = run_start_iso
+            stamped_inputs.append(d)
 
-            all_messages = stamped_inputs + enriched
+        all_messages = stamped_inputs + enriched
+        if all_messages:
             logger.debug(
                 f"MESSAGES_SNAPSHOT: {len(all_messages)} msgs ({message_count} SDK messages processed)"
             )

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

@@ -294,6 +294,12 @@ async def _handle_user_message(self, msg: Any) -> None:
         try:
             runner_input = RunnerInput.model_validate_json(msg.payload)
         except Exception:
+            if not msg.payload or not msg.payload.strip():
+                logger.warning(
+                    "[GRPC LISTENER] Empty payload for seq=%d, skipping to avoid empty-run with no MESSAGES_SNAPSHOT",
+                    msg.seq,
+                )
+                return
             runner_input = RunnerInput(
                 messages=[
                     {"id": str(uuid.uuid4()), "role": "user", "content": msg.payload}

components/runners/ambient-runner/ambient_runner/endpoints/run.py

@@ -41,11 +41,19 @@ def to_run_agent_input(self) -> RunAgentInput:
         parent_run_id = self.parentRunId or self.parent_run_id
         context_list = self.context if isinstance(self.context, list) else []
 
+        # Ensure every message dict has an id so upsert_message deduplication
+        # works correctly in the adapter for multi-turn sessions.
+        messages = []
+        for m in self.messages:
+            if isinstance(m, dict) and not m.get("id"):
+                m = {**m, "id": str(uuid.uuid4())}
+            messages.append(m)
+
         return RunAgentInput(
             thread_id=thread_id,
             run_id=run_id,
             parent_run_id=parent_run_id,
-            messages=self.messages,
+            messages=messages,
             state=self.state or {},
             tools=self.tools or [],
             context=context_list,