"""Delivery-ambiguity classification and shared retry loop for native-forwarder event POSTs. The claude-native, codex-native, and antigravity-native forwarders mirror transcript items into AP as ``external_conversation_item`` POSTs. The server persists those with a random primary key and does NOT dedupe them — producers are responsible for not re-posting items they have already sent. That makes a blind retry after a failed POST unsafe: if the server committed the item and published ``session.input.consumed`` but the response was lost, a retry appends a second copy and the web UI renders a duplicate bubble. The native tmux pane is unaffected, which is why the duplicate is web-only. :func:`post_may_have_been_delivered` is the shared classifier all forwarders use to decide whether a failed POST is safe to retry. :func:`post_session_event_with_retry` is the shared retry loop extracted from the codex/antigravity forwarders so a single implementation is maintained. """ from __future__ import annotations import contextlib import json import logging import time from collections.abc import Callable, Coroutine from dataclasses import dataclass from pathlib import Path import httpx from omnigent._native_forwarder_health import ( note_post_success as note_native_post_success, ) from omnigent._native_forwarder_health import ( record_post_failure as record_native_post_failure, ) _logger = logging.getLogger(__name__) # Dead-letter sink for permanently-undeliverable forward payloads (#1120). _DEAD_LETTER_FILE = "dead_letter.jsonl" # At this size the file rotates to a single .1 backup (keep-newest); disk ~2x this. _DEAD_LETTER_MAX_BYTES = 50 * 1024 * 1024 # 50 MB per session _DEAD_LETTER_BACKUP_FILE = _DEAD_LETTER_FILE + ".1" def append_dead_letter( bridge_dir: Path, *, session_id: str, event_type: str, payload: dict[str, object], reason: str, delivered_ambiguous: bool = False, http_status: int | None = None, transport_error: str | None = None, ) -> None: """ Append one undeliverable forward payload to ``{bridge_dir}/dead_letter.jsonl`` (#1120). Write-only recovery artifact so a permanently-failed transcript/usage POST is recoverable on disk instead of silently lost. Conservative startup replay of the *proven-undelivered* records is layered on top (#1579) and reads the structured classification fields below to decide what is safe to re-POST. Best-effort: never raises (a dead-letter failure must not disrupt forwarding). When the file reaches :data:`_DEAD_LETTER_MAX_BYTES` it is rotated to a single ``.1`` backup and a fresh file is started, so the most recent drops are kept (the oldest rotate out); disk stays bounded at ~2x the cap. :param bridge_dir: Native forwarder bridge directory the dead-letter file lives in. :param session_id: Omnigent conversation id the dropped event targeted, e.g. ``"conv_abc123"``. :param event_type: Session event type that was dropped, e.g. ``"external_conversation_item"``. :param payload: The event ``data`` payload that failed to deliver. :param reason: Short human-readable cause, e.g. ``"permanent HTTP failure after retries"``. :param delivered_ambiguous: Whether the failure was ambiguous (request sent, response lost), so the server may have committed the item. Such records are NEVER replayed — a re-POST risks a duplicate (no server-side dedup). :param http_status: Final HTTP status code when the server responded, e.g. ``503`` or ``400``; ``None`` for a transport failure that saw no response. :param transport_error: Transport-error class name when the POST raised without a response, e.g. ``"ConnectError"``; ``None`` when the server responded. :returns: None. """ try: path = bridge_dir / _DEAD_LETTER_FILE bridge_dir.mkdir(mode=0o700, parents=True, exist_ok=True) # Keep-newest: at the cap, rotate to a single .1 backup and start fresh. if path.exists() and path.stat().st_size >= _DEAD_LETTER_MAX_BYTES: path.replace(bridge_dir / _DEAD_LETTER_BACKUP_FILE) # Log session_id, not path (logging a bridge path trips CodeQL's # clear-text-sensitive-data heuristic; a bridge dir is not a secret). _logger.warning( "dead-letter file reached cap (%d bytes); rotated to %s and " "started fresh (oldest dead-lettered forwards dropped): session=%s", _DEAD_LETTER_MAX_BYTES, _DEAD_LETTER_BACKUP_FILE, session_id, ) line = json.dumps( { "ts": time.time(), "session_id": session_id, "event_type": event_type, "reason": reason, "delivered_ambiguous": delivered_ambiguous, "http_status": http_status, "transport_error": transport_error, "payload": payload, } ) with path.open("a", encoding="utf-8") as fh: fh.write(line + "\n") except Exception as exc: # noqa: BLE001 - dead-lettering must never disrupt forwarding. _logger.warning( "failed to dead-letter undeliverable forward: type=%s session=%s error=%r", event_type, session_id, exc, ) # Transport failures proving a POST never reached the server (no bytes # sent) — safe to retry. See :func:`post_may_have_been_delivered`. _DELIVERY_SAFE_RETRY_ERRORS = ( httpx.ConnectError, httpx.ConnectTimeout, httpx.PoolTimeout, ) def post_may_have_been_delivered(exc: httpx.HTTPError) -> bool: """ Return whether a failed AP POST may have been delivered AND committed by the server despite the error — making a blind retry unsafe for non-idempotent events. - ``HTTPStatusError``: the server responded with a status. The events route returns 2xx only after the item is appended and the consume event is published, so any non-2xx means the item was not committed (4xx rejects at parse time; a 5xx fails before/at the append). No duplicate risk → safe to retry, so ``False``. - Connection-establishment / pool-acquire failures (:data:`_DELIVERY_SAFE_RETRY_ERRORS`): no bytes were sent → not delivered → safe to retry, so ``False``. - Any other transport error (read/write timeout, read/write error, remote protocol error): the request was sent and we never saw a response, so the server may have processed it → ambiguous → ``True``. :param exc: HTTP exception raised while posting an AP event. :returns: ``True`` when a retry could duplicate a server-committed item; ``False`` when retrying is safe. """ if isinstance(exc, httpx.HTTPStatusError): return False if isinstance(exc, _DELIVERY_SAFE_RETRY_ERRORS): return False return True async def post_external_session_status( client: httpx.AsyncClient, *, session_id: str, status: str, output: str | None = None, background_task_count: int | None = None, response_id: str | None = None, ) -> None: """Post one ``external_session_status`` event to the Sessions API. The turn-end edge native forwarders use to drive session status and, for sub-agents, the parent-inbox wake. Shared by the claude-native and cursor-native forwarders so the event shape stays in one place. :param client: Omnigent HTTP client. :param session_id: Omnigent session/conversation id. :param status: Session status value, e.g. ``"idle"`` or ``"failed"``. :param output: Optional text attached to ``data``. On a ``"failed"`` edge the server surfaces it as ``last_task_error`` so the UI shows a detail instead of a bare "failed". Ignored when falsy. :param background_task_count: Background tasks (shells) still running at the edge, forwarded so the UI can show "N background tasks still running". ``None`` omits the field (server leaves its sticky tally untouched) — the default for edges that know nothing about background shells. :param response_id: Optional id of the assistant turn this status edge belongs to. When set, the server attaches it to the ``session.status`` SSE event so ap-web can drive the bubble's streaming lifecycle — that's what makes native forwarded tool cards render LIVE (spinner + elapsed timer) rather than as static completed cards. ``None`` (the default) preserves the bare, turn-agnostic status edges (e.g. the sub-agent quiescence badge) that don't map to a turn. :raises httpx.HTTPError: If the Omnigent request fails or is rejected. """ data: dict[str, object] = {"status": status} if output: data["output"] = output if background_task_count is not None: data["background_task_count"] = background_task_count if response_id is not None: data["response_id"] = response_id resp = await client.post( f"/v1/sessions/{session_id}/events", json={"type": "external_session_status", "data": data}, ) resp.raise_for_status() async def post_session_event_with_retry( *, client: httpx.AsyncClient, url: str, payload: dict[str, object], event_type: str, max_attempts: int, retry_status_codes: frozenset[int], sleep: Callable[[float], Coroutine[None, None, None]], retry_delay: Callable[[int], float], logger_name: str, ) -> httpx.Response | None: """ POST a session event payload with bounded transient retries. Shared retry loop used by the antigravity (and optionally other) native forwarders. Conversation items persist with a random primary key and no server-side dedup, so an ambiguous transport failure (request sent, response lost) is NOT retried — a re-post would duplicate the item. Other event types are idempotent/transient and are retried. :param client: HTTP client for Omnigent event posts. :param url: Full request URL, e.g. ``"/v1/sessions/conv_x/events"``. :param payload: JSON payload to POST, e.g. ``{"type": ..., "data": ...}``. :param event_type: Session event type, e.g. ``"external_conversation_item"``. Used in log messages and to decide whether an ambiguous failure is safe to retry. :param max_attempts: Maximum POST attempts, e.g. ``3``. :param retry_status_codes: HTTP status codes to retry, e.g. ``frozenset({429, 500, 503})``. :param sleep: Async sleep coroutine (stubbable in tests). :param retry_delay: Callable ``attempt -> float`` returning the delay before the next attempt (one-based failed attempt number). :param logger_name: Logger name used for warning messages, e.g. ``"omnigent.antigravity_native_reader"``. :returns: Final HTTP response, or ``None`` when all attempts raised transport errors (or after an ambiguous conversation-item failure). """ log = logging.getLogger(logger_name) for attempt in range(1, max_attempts + 1): try: response = await client.post(url, json=payload) except httpx.HTTPError as exc: # Conversation items persist with a random primary key and no # server-side dedup, so an ambiguous failure (request sent, # response lost — the server may have committed it) must not # be retried: a re-post would duplicate the item. # Other event types are idempotent / transient, so retrying # them on the same errors is safe and preserves delivery. if event_type == "external_conversation_item" and post_may_have_been_delivered(exc): log.warning( "skipping session event after an ambiguous transport " "failure (may already be committed); not retrying to avoid " "a duplicate: type=%s error=%r", event_type, exc, ) return None if attempt >= max_attempts: log.warning( "failed to post session event after retries: type=%s attempts=%s error=%r", event_type, max_attempts, exc, ) # Surface this connectivity failure to the harness idle-turn # watchdog so a stall caused by unreachable-server posts is # reported with its real cause, not a generic "wedged LLM" # reason. record_native_post_failure(event_type, exc) return None await sleep(retry_delay(attempt)) continue # Reaching here means the POST got an HTTP response (no transport # error), proving the server is reachable — clear any stale # connectivity-failure record so the watchdog can't later misattribute # it to an unrelated stall. note_native_post_success() if response.status_code < 400: return response if response.status_code not in retry_status_codes: return response if attempt >= max_attempts: return response await sleep(retry_delay(attempt)) return None @dataclass(frozen=True) class RepostResult: """ Outcome of one dead-letter replay re-POST attempt (#1579). Returned by the forwarder-supplied ``repost`` callable so :func:`replay_dead_letters` can decide whether to drop the record or keep it, and refresh its classification when a retained record's safety changed (e.g. a proven-undelivered transport record that now fails *ambiguously* must never be auto-replayed again). :param delivered: ``True`` when the server accepted the re-POST (a sub-400 response). The record is removed on success. :param delivered_ambiguous: ``True`` when the re-POST failed ambiguously (request sent, response lost), so the item may now be committed. The record is kept but reclassified so replay never touches it again. :param http_status: Final HTTP status code when the server responded, or ``None`` for a transport failure that saw no response. Used to refresh the retained record's classification. """ delivered: bool delivered_ambiguous: bool = False http_status: int | None = None def _dead_letter_record_replayable( record: object, *, retryable_status_codes: frozenset[int], ) -> bool: """ Return whether a dead-letter record is safe to re-POST on startup (#1579). Only *proven-undelivered* records are replayable: a transport failure that never reached the server, or a retryable status (e.g. ``503``) exhausted after the forwarder's bounded retries. Ambiguous failures (the server may have committed the item) and permanent rejections (a 4xx the server will just reject again) are never replayed. Records written before classification was added (#1579) lack the ``delivered_ambiguous`` field; they are treated as unsafe (forensic only) so a pre-classification ambiguous drop is never replayed into a duplicate. :param record: One parsed dead-letter record, or any non-dict entry (malformed line) which is never replayable. :param retryable_status_codes: HTTP statuses the forwarder treats as transient/retryable, e.g. ``frozenset({429, 500, 503})``. A recorded status in this set is proven-undelivered-but-recoverable. :returns: ``True`` only for proven-undelivered records with the routing fields (``session_id``, ``event_type``, ``payload``) needed to re-POST. """ if not isinstance(record, dict): return False session_id = record.get("session_id") event_type = record.get("event_type") if not (isinstance(session_id, str) and session_id): return False if not (isinstance(event_type, str) and event_type): return False if not isinstance(record.get("payload"), dict): return False if "delivered_ambiguous" not in record: return False if record.get("delivered_ambiguous"): return False http_status = record.get("http_status") if http_status is None: # Transport failure with no response (ambiguous already excluded above) # — proven undelivered, so safe to re-POST. return True # The server responded: only a retryable status is recoverable; a permanent # 4xx would just be rejected again. return http_status in retryable_status_codes def _read_dead_letter_entries(path: Path) -> list[object] | None: """ Read one dead-letter file into ordered entries, preserving malformed lines. :param path: Dead-letter file path (current or ``.1`` backup). :returns: Ordered entries — parsed ``dict`` records, or the raw ``str`` line for any line that failed to parse (kept verbatim so a rewrite never drops forensic data) — or ``None`` when the file does not exist. """ if not path.exists(): return None entries: list[object] = [] for line in path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped: continue try: entries.append(json.loads(stripped)) except json.JSONDecodeError: entries.append(line) return entries def _rewrite_dead_letter_entries(path: Path, entries: list[object]) -> None: """ Atomically rewrite one dead-letter file with the retained entries (#1579). Writes a sibling ``.tmp`` then ``os.replace``-es it into place so a crash mid-rewrite never leaves a half-written file. An empty entry list removes the file. :param path: Dead-letter file path to rewrite. :param entries: Retained entries in original order (``dict`` records are re-serialized; raw ``str`` lines are written verbatim). :returns: None. """ if not entries: with contextlib.suppress(FileNotFoundError): path.unlink() return lines = [entry if isinstance(entry, str) else json.dumps(entry) for entry in entries] tmp = path.with_name(path.name + ".tmp") tmp.write_text("\n".join(lines) + "\n", encoding="utf-8") tmp.replace(path) async def replay_dead_letters( bridge_dir: Path, *, repost: Callable[[dict[str, object]], Coroutine[None, None, RepostResult]], retryable_status_codes: frozenset[int], logger_name: str | None = None, max_records: int | None = None, deadline_seconds: float | None = None, ) -> int: """ Re-POST proven-undelivered dead-lettered forwards on forwarder startup (#1579). Reads the ``.1`` backup first, then the current file, preserving append order, and re-POSTs only *proven-undelivered* records (see :func:`_dead_letter_record_replayable`) via the supplied ``repost`` callable. A delivered record is removed; a still-failing one is retained, with its classification refreshed from the latest attempt so a record that now fails ambiguously (or is permanently rejected) is never auto-replayed again. Ambiguous and permanent-4xx records are left untouched as a forensic record. Bounded so a large dead-letter file or a slow/hung server cannot stall startup: at most ``max_records`` records are re-POSTed and the whole drain is abandoned once ``deadline_seconds`` elapses. Records left over by either bound are retained unchanged (deferred to a later startup) and logged — never silently dropped. The remaining latency lever, a short per-POST timeout and a single attempt, is the caller's responsibility (via ``repost``). Intended to run once at startup, before live forwarding begins, so no other writer races the dead-letter files for this ``bridge_dir``. :param bridge_dir: Native forwarder bridge directory holding the dead-letter files. :param repost: Async callable that re-POSTs one record's ``(session_id, event_type, payload)`` and returns a :class:`RepostResult`. :param retryable_status_codes: HTTP statuses the forwarder treats as transient/retryable, used to classify which recorded statuses are recoverable. :param logger_name: Optional logger name for the recovery summary line; defaults to this module's logger. :param max_records: Maximum number of records to re-POST this run, e.g. ``500``; ``None`` for no cap. Bounds the number of network POSTs (and thus startup latency) even against a healthy server. :param deadline_seconds: Wall-clock budget for the whole drain, e.g. ``30.0``; ``None`` for no deadline. Once exceeded, the remaining replayable records are deferred to a later startup. :returns: The number of records successfully replayed (and removed). """ log = logging.getLogger(logger_name) if logger_name else _logger sources = [ bridge_dir / _DEAD_LETTER_BACKUP_FILE, bridge_dir / _DEAD_LETTER_FILE, ] loaded: list[tuple[Path, list[object]]] = [] any_replayable = False for path in sources: entries = _read_dead_letter_entries(path) if entries is None: continue loaded.append((path, entries)) if any( _dead_letter_record_replayable(entry, retryable_status_codes=retryable_status_codes) for entry in entries ): any_replayable = True if not any_replayable: # Nothing recoverable — leave the forensic files untouched. return 0 replayed = 0 attempted = 0 deferred = 0 stop = False deadline = time.monotonic() + deadline_seconds if deadline_seconds is not None else None # ``loaded`` preserves the .1-then-current source order, so records replay # in the order they were originally dropped. for path, entries in loaded: retained: list[object] = [] changed = False for entry in entries: if not _dead_letter_record_replayable( entry, retryable_status_codes=retryable_status_codes ): # Forensic record (ambiguous / permanent-4xx / malformed) — keep as is. retained.append(entry) continue if stop: deferred += 1 retained.append(entry) continue over_records = max_records is not None and attempted >= max_records over_deadline = deadline is not None and time.monotonic() >= deadline if over_records or over_deadline: # Out of budget — defer this and every later replayable record # to the next startup rather than stall here. stop = True deferred += 1 retained.append(entry) continue assert isinstance(entry, dict) # narrowed by _dead_letter_record_replayable attempted += 1 result = await repost(entry) if result.delivered: replayed += 1 changed = True continue if ( entry.get("delivered_ambiguous") != result.delivered_ambiguous or entry.get("http_status") != result.http_status ): entry = { **entry, "delivered_ambiguous": result.delivered_ambiguous, "http_status": result.http_status, } changed = True retained.append(entry) if changed: _rewrite_dead_letter_entries(path, retained) if replayed: log.info( "replayed %d proven-undelivered dead-lettered forward(s) on startup", replayed, ) if deferred: log.info( "deferred %d replayable dead-lettered forward(s) to a later startup " "(replay budget reached: max_records=%s deadline_seconds=%s)", deferred, max_records, deadline_seconds, ) return replayed