feat(gateway): add recoverable mobile approvals

agent/agent_init.py

@@ -728,6 +728,12 @@ def init_agent(
     agent._persist_user_message_idx = None
     agent._persist_user_message_override = None
     agent._persist_user_message_timestamp = None
+    # Gateway-only receipt proof tags copied onto the next user message. The
+    # underscore-prefixed field never leaves the process or persists to SQLite;
+    # it binds the in-memory `_db_persisted` proof to an exact mobile mutation.
+    agent._pending_mobile_mutation_receipt_tags: list[str] = []
+    agent._durable_mobile_mutation_receipt_tags: set[str] = set()
+    agent._mobile_mutation_receipt_condition = None
 
     # Cache anthropic image-to-text fallbacks per image payload/URL so a
     # single tool loop does not repeatedly re-run auxiliary vision on the

agent/agent_runtime_helpers.py

@@ -545,6 +545,32 @@ def _is_codex_interim(m: Dict) -> bool:
                     if prev_content and new_content
                     else (prev_content or new_content)
                 )
+                # The gateway binds a mobile receipt to the exact new user
+                # dict that SessionDB successfully persisted. Repair discards
+                # that dict after merging its text into ``prev``. Preserve an
+                # explicit proof-only tag without marking the merged content as
+                # wholly DB-persisted (``prev`` itself may still need a flush).
+                persisted_receipt_tags = list(
+                    msg.get("_mobile_mutation_persisted_receipt_tags") or ()
+                )
+                if msg.get("_db_persisted"):
+                    for tag in msg.get(
+                        "_mobile_mutation_receipt_tags",
+                        (),
+                    ) or ():
+                        if tag not in persisted_receipt_tags:
+                            persisted_receipt_tags.append(tag)
+                if persisted_receipt_tags:
+                    existing_receipt_tags = list(
+                        prev.get("_mobile_mutation_persisted_receipt_tags")
+                        or ()
+                    )
+                    for tag in persisted_receipt_tags:
+                        if tag not in existing_receipt_tags:
+                            existing_receipt_tags.append(tag)
+                    prev["_mobile_mutation_persisted_receipt_tags"] = (
+                        existing_receipt_tags
+                    )
                 repairs += 1
                 continue
         merged.append(msg)

agent/turn_context.py

@@ -313,8 +313,19 @@ def build_turn_context(
             should_review_memory = True
             agent._turns_since_memory = 0
 
-    # Add user message.
+    # Add user message. Gateway mobile receipt tags are process-private proof
+    # tags: transport sanitizers strip underscore-prefixed fields and SessionDB
+    # stores only the canonical message columns, while `_db_persisted` remains
+    # on this exact dict after a successful append for receipt finalization.
     user_msg = {"role": "user", "content": user_message}
+    mobile_mutation_receipt_tags = list(
+        getattr(agent, "_pending_mobile_mutation_receipt_tags", ()) or ()
+    )
+    agent._pending_mobile_mutation_receipt_tags = []
+    if mobile_mutation_receipt_tags:
+        user_msg["_mobile_mutation_receipt_tags"] = (
+            mobile_mutation_receipt_tags
+        )
     messages.append(user_msg)
     current_turn_user_idx = len(messages) - 1
     agent._persist_user_message_idx = current_turn_user_idx
@@ -360,6 +371,36 @@ def build_turn_context(
             agent.session_id or "none",
             exc_info=True,
         )
+    if user_msg.get("_db_persisted") and mobile_mutation_receipt_tags:
+        receipt_condition = getattr(
+            agent,
+            "_mobile_mutation_receipt_condition",
+            None,
+        )
+        if receipt_condition is not None:
+            with receipt_condition:
+                durable_tags = set(
+                    getattr(
+                        agent,
+                        "_durable_mobile_mutation_receipt_tags",
+                        set(),
+                    )
+                    or set()
+                )
+                durable_tags.update(mobile_mutation_receipt_tags)
+                agent._durable_mobile_mutation_receipt_tags = durable_tags
+                receipt_condition.notify_all()
+        else:
+            durable_tags = set(
+                getattr(
+                    agent,
+                    "_durable_mobile_mutation_receipt_tags",
+                    set(),
+                )
+                or set()
+            )
+            durable_tags.update(mobile_mutation_receipt_tags)
+            agent._durable_mobile_mutation_receipt_tags = durable_tags
 
     # ── Preflight context compression ──
     # Gate the (expensive) full token estimate behind a cheap pre-check.

docs/mobile-client-contract.md

@@ -0,0 +1,104 @@
+# Mobile Client Contract
+
+Hermes exposes its additive Mobile Client Contract on the existing
+authenticated `/api/ws` JSON-RPC transport. Clients must negotiate features
+from `gateway.ready`; a contract major or server version alone does not imply a
+capability.
+
+## Revisioned conversation synchronization
+
+When `conversation.sync` version 1 is advertised, `session.create` and
+`session.resume` include the same `synchronization` envelope:
+
+- `snapshot` is authoritative conversation state. It identifies the server
+  process, live event stream, stable conversation lineage, current stored
+  session tip, and process-local live session. It also includes a revision,
+  event watermark, messages, inflight turn, active tool descriptors, pending
+  interactions, and runtime status.
+- `recovery` reports `complete`, `gap`, or `reset`, the cursor at the snapshot
+  watermark, and any replayable events after the supplied cursor.
+- Every session event carries `schema_major`, `stream_id`, and a monotonically
+  increasing `sequence` within that stream.
+
+The replay store is bounded in memory by both the advertised event and byte
+limits. `gap` means an event needed after the supplied cursor was evicted.
+`reset` means the cursor is absent, invalid, from another server process, or
+from another reconstructed live stream. Clients must never treat either
+outcome as complete replay.
+
+Synchronization state and replay retention begin when an authorized mobile
+transport first attaches to a live session. A session that has never negotiated
+sync does not pay the per-delta replay copying cost. If a legacy transport later
+attaches to a previously negotiated session, retention continues for the next
+mobile reconnect, but the legacy response and event shapes remain unchanged.
+
+## Snapshot and event barrier
+
+Hermes serializes snapshot-visible live state, event sequence allocation,
+replay retention, and event transport enqueueing at one per-stream boundary.
+Snapshot capture takes the conversation history lock before that stream
+boundary. The returned watermark therefore covers every state transition
+published before the snapshot.
+
+A client should:
+
+1. Buffer events for the returned `stream_id` while create or resume is in
+   flight.
+2. Install the complete snapshot at its `watermark`.
+3. Discard buffered or replayed events at or below the watermark.
+4. Apply only events for the same server and stream whose sequence is greater
+   than the watermark, in sequence order.
+5. Replace local state from the snapshot whenever recovery is `gap` or
+   `reset`.
+
+Assistant `message.delta` events additionally carry one `turn_id` and an
+absolute `offset`. The `conversation.sync.delta_offsets.unit` capability names
+the unit as `utf8_bytes`. Clients can therefore ignore an overlapping prefix
+instead of duplicating text after replay or transport coalescing.
+
+## Durable consequential mutations
+
+When `mutation.idempotency` version 1 is advertised, its `methods` list is the
+complete set of mobile methods covered by durable receipts. Each covered
+request requires a non-empty `client_request_id`. Prompt submission,
+interruption, and approval response additionally require the stable
+`expected_stored_session_id`; approval response also requires the
+Hermes-issued `approval_id`.
+
+Receipts are scoped to the authenticated provider and subject. Repeating the
+same request identity with equivalent normalized semantics returns the stored
+result with `mutation.deduplicated` set to `true`. Reusing it with different
+semantics returns `mutation_conflict`. A request abandoned after execution may
+have begun is reported as `mutation_outcome_unknown` and is never executed
+again automatically. Clients can inspect a known receipt with the advertised
+`mutation.status` method.
+
+This guarantee applies only to the methods named by the capability on a
+mobile-scoped connection. Existing legacy transports retain their prior
+request shapes and behavior.
+
+## Recoverable approval lifecycle
+
+When `interaction.lifecycle` version 1 names `approval` in `kinds` and
+`approval.respond` in `response_methods`, approval requests use the
+`approval.lifecycle` version 1 schema. Each request carries a stable
+Hermes-owned `approval_id`, server-redacted presentation fields, creation and
+expiry times, current state, and resolution metadata. Pending descriptors are
+part of the authoritative synchronization snapshot and remain addressable
+after reconnect.
+
+Hermes emits `approval.request` for creation and one of `approval.resolved`,
+`approval.expired`, or `approval.stale` for a terminal transition. Every event
+and snapshot descriptor carries the same `approval_id`. A mobile response must
+name that identity, the stable conversation lineage, a choice, and a durable
+client request identity. Identical retries replay the stored mutation result;
+changed semantics conflict. Short-lived terminal tombstones distinguish
+`already_resolved`, `expired`, `stale`, and `not_found` outcomes without
+consuming another pending approval.
+
+Approval payload and resolution metadata redaction is server-owned. Mobile
+clients must not infer authorization from presentation fields: the response is
+available only when the lifecycle capability, mutation coverage,
+`conversation.control` grant, live reconciled state, and valid Hermes resource
+identities all agree. ID-less legacy desktop and stdin responses retain their
+existing FIFO behavior.

gateway/platforms/qqbot/adapter.py

@@ -260,6 +260,9 @@ def __init__(self, config: PlatformConfig):
         # box; callers can override with set_interaction_callback(None) or
         # register a custom handler.
         self._interaction_callback = self._default_interaction_dispatch
+        # Hermes approval_id → session key. Button payloads carry only the
+        # stable identity so a delayed click cannot consume another FIFO item.
+        self._approval_state: Dict[str, str] = {}
 
     # ------------------------------------------------------------------
     # Properties
@@ -1118,7 +1121,7 @@ async def _default_interaction_dispatch(
     ) -> None:
         """Route ``INTERACTION_CREATE`` button clicks to the right subsystem.
 
-        - ``approve:<session_key>:<decision>`` →
+        - ``approve:<approval_id>:<decision>`` →
           :func:`tools.approval.resolve_gateway_approval`
           (unblocks the agent thread waiting on a dangerous-command approval).
         - ``update_prompt:<answer>`` →
@@ -1137,7 +1140,14 @@ async def _default_interaction_dispatch(
 
         approval = parse_approval_button_data(button_data)
         if approval is not None:
-            session_key, decision = approval
+            approval_id, decision = approval
+            session_key = self._approval_state.get(approval_id)
+            if not session_key:
+                logger.info(
+                    "[%s] Approval %s is already resolved, expired, or unknown",
+                    self._log_tag, approval_id,
+                )
+                return
             choice = self._APPROVAL_BUTTON_TO_CHOICE.get(decision)
             if choice is None:
                 logger.warning(
@@ -1156,11 +1166,16 @@ async def _default_interaction_dispatch(
                 # Import lazily to keep the adapter importable in tests that
                 # don't exercise the approval subsystem.
                 from tools.approval import resolve_gateway_approval
-                count = resolve_gateway_approval(session_key, choice)
+                count = resolve_gateway_approval(
+                    session_key,
+                    choice,
+                    approval_id=approval_id,
+                )
+                self._approval_state.pop(approval_id, None)
                 logger.info(
                     "[%s] Button resolved %d approval(s) for session %s "
-                    "(choice=%s, operator=%s)",
-                    self._log_tag, count, session_key, choice,
+                    "(approval_id=%s, choice=%s, operator=%s)",
+                    self._log_tag, count, session_key, approval_id, choice,
                     event.operator_openid,
                 )
             except Exception as exc:
@@ -2648,7 +2663,7 @@ async def send_approval_request(
         return await self.send_with_keyboard(
             chat_id,
             build_approval_text(req),
-            build_approval_keyboard(req.session_key),
+            build_approval_keyboard(req.approval_id or req.session_key),
             reply_to=reply_to,
         )
 
@@ -2676,7 +2691,11 @@ async def send_exec_approval(
         :func:`tools.approval.resolve_gateway_approval` — dispatched by the
         adapter's interaction callback (:meth:`_default_interaction_dispatch`).
         """
-        del metadata  # QQ doesn't have thread_id / DM targeting overrides.
+        from agent.redact import redact_sensitive_text
+
+        approval_id = str((metadata or {}).get("approval_id") or uuid.uuid4().hex)
+        command = redact_sensitive_text(command, force=True)
+        description = redact_sensitive_text(description, force=True)
 
         # Use the reply-to message for passive-message context when we have one.
         # QQ requires a msg_id on outbound messages to a user we've never
@@ -2686,13 +2705,17 @@ async def send_exec_approval(
         req = ApprovalRequest(
             session_key=session_key,
             title="Execute this command?",
+            approval_id=approval_id,
             description=description,
             command_preview=command,
             timeout_sec=self._APPROVAL_TIMEOUT_SECONDS,
         )
-        return await self.send_approval_request(
+        result = await self.send_approval_request(
             chat_id, req, reply_to=msg_id,
         )
+        if result.success:
+            self._approval_state[approval_id] = session_key
+        return result
 
     _APPROVAL_TIMEOUT_SECONDS = 300  # matches gateway's default gateway_timeout