feat(webhook): hand off completed sessions to messaging platforms

agent/agent_runtime_helpers.py

@@ -799,6 +799,15 @@ def _is_verification_candidate(m: Dict) -> bool:
                 # bytes previously sent for the pre-merge message) — drop it
                 # so replay can't substitute stale bytes.
                 drop_stale_api_content(prev)
+                # Carry the absorbed row's durable identity when the survivor
+                # has none: a webhook delivery marker on the absorbed row must
+                # stay findable, or a provider retry after a crash-left
+                # user;user wedge appends the already-committed input again.
+                for identity_key in ("message_id", "_platform_message_id"):
+                    if prev.get(identity_key) is None and msg.get(
+                        identity_key
+                    ) is not None:
+                        prev[identity_key] = msg[identity_key]
                 repairs += 1
                 continue
         merged.append(msg)

agent/conversation_loop.py

@@ -24,7 +24,7 @@
 import ssl
 import sys
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from agent.codex_responses_adapter import _summarize_user_message_for_log
 from agent.conversation_compression import (
@@ -1831,6 +1831,8 @@ def run_conversation(
     persist_user_display_kind: Optional[str] = None,
     persist_user_display_metadata: Optional[Dict[str, Any]] = None,
     moa_config: Optional[dict[str, Any]] = None,
+    persist_user_message_id: Optional[str] = None,
+    input_persisted_callback: Optional[Callable[[], None]] = None,
 ) -> Dict[str, Any]:
     """
     Run a complete conversation with tool calling until completion.
@@ -1856,6 +1858,11 @@ def run_conversation(
         persist_user_display_metadata: Optional payload for that event
             (e.g. a delegation's task count).
                 or queuing follow-up prefetch work.
+        persist_user_message_id: Optional opaque inbound identity to persist
+            on the current user row for exact retry recovery.
+        input_persisted_callback: Optional synchronous callback invoked only
+            after that user row is durably committed. Exceptions abort before
+            the primary conversation provider call.
 
     Returns:
         Dict: Complete conversation result with final response and message history
@@ -1907,6 +1914,8 @@ def run_conversation(
         persist_user_timestamp,
         persist_user_display_kind=persist_user_display_kind,
         persist_user_display_metadata=persist_user_display_metadata,
+        persist_user_message_id=persist_user_message_id,
+        input_persisted_callback=input_persisted_callback,
         restore_or_build_system_prompt=_restore_or_build_system_prompt,
         install_safe_stdio=_install_safe_stdio,
         sanitize_surrogates=_sanitize_surrogates,

agent/turn_context.py

@@ -29,7 +29,7 @@
 import time
 import uuid
 from dataclasses import dataclass
-from typing import Any, Dict, List, Mapping, Optional
+from typing import Any, Callable, Dict, List, Mapping, Optional
 
 from agent.conversation_compression import (
     IDLE_COMPACTION_STATUS_TEMPLATE,
@@ -466,6 +466,8 @@ def build_turn_context(
     *,
     persist_user_display_kind: Optional[str] = None,
     persist_user_display_metadata: Optional[Dict[str, Any]] = None,
+    persist_user_message_id: Optional[str] = None,
+    input_persisted_callback: Optional[Callable[[], None]] = None,
     restore_or_build_system_prompt,
     install_safe_stdio,
     sanitize_surrogates,
@@ -491,6 +493,55 @@ def build_turn_context(
     if recovered_history is not None:
         conversation_history = recovered_history
 
+    # A durable webhook admission can crash after committing its user row but
+    # before publishing accepted→running.  Its provider retry carries the same
+    # globally unique marker. Reuse that committed row as THIS turn's input
+    # instead of appending the prompt a second time. The row must still be the
+    # transcript tail: callback admission happens before the primary provider
+    # call, so any later conversational row would prove this is not the safe
+    # pre-call recovery shape.
+    replayed_persisted_user_msg: Optional[Dict[str, Any]] = None
+    if persist_user_message_id and conversation_history:
+        replay_idx: Optional[int] = None
+        for idx in range(len(conversation_history) - 1, -1, -1):
+            candidate = conversation_history[idx]
+            if (
+                isinstance(candidate, dict)
+                and candidate.get("role") == "user"
+                and (
+                    candidate.get("_platform_message_id")
+                    or candidate.get("message_id")
+                )
+                == persist_user_message_id
+            ):
+                replay_idx = idx
+                break
+        if replay_idx is not None:
+            if any(
+                isinstance(message, dict)
+                and message.get("role") in {"user", "assistant", "tool"}
+                for message in conversation_history[replay_idx + 1 :]
+            ):
+                raise RuntimeError(
+                    "durable input marker is not the pending transcript tail"
+                )
+            replayed_content = conversation_history[replay_idx].get("content")
+            if replayed_content is None:
+                # Validate before the pop so a raise leaves the caller's
+                # history list unmutated.
+                raise RuntimeError("durable input marker has no user content")
+            replayed_persisted_user_msg = conversation_history.pop(replay_idx)
+            # JSONL/direct callers may still supply the public transcript key.
+            # Normalize it to the transport-stripped internal form before this
+            # row re-enters a provider-bound message list.
+            replayed_persisted_user_msg.pop("message_id", None)
+            replayed_persisted_user_msg[
+                "_platform_message_id"
+            ] = persist_user_message_id
+            replayed_persisted_user_msg["_db_persisted"] = True
+            user_message = replayed_content
+            persist_user_message = replayed_content
+
     # NOTE: the DB session row is created later, AFTER the system prompt is
     # restored/built (see _ensure_db_session() below the system-prompt block).
     # Creating it here — before _cached_system_prompt is populated — inserts a
@@ -663,7 +714,11 @@ def build_turn_context(
     expected_persist_content = (
         persist_user_message if persist_user_message is not None else user_message
     )
-    if (
+    if replayed_persisted_user_msg is not None:
+        user_msg = replayed_persisted_user_msg
+        if isinstance(pending_cli_message, dict):
+            agent._pending_cli_user_message = None
+    elif (
         isinstance(pending_cli_message, dict)
         and pending_cli_message.get("content") == expected_persist_content
     ):
@@ -682,6 +737,8 @@ def build_turn_context(
     # CLI input is stamped when staged. Gateway input may carry the platform
     # event time. Preserve either value and cover any legacy unstamped handoff.
     stamp_message_timestamp(user_msg, timestamp=persist_user_timestamp)
+    if persist_user_message_id:
+        user_msg["_platform_message_id"] = persist_user_message_id
 
     # Hydrate todo store from conversation history.
     if conversation_history and not agent._todo_store.has_items():
@@ -1384,8 +1441,11 @@ def build_turn_context(
     #
     # Skip prefetch on trivial prompts (greetings, acknowledgements) to
     # prevent memory-context injection on turns that carry no semantic signal.
+    # A durable replay discards recomputed context below to resend the exact
+    # committed bytes — skip the prefetch entirely so no work is wasted and
+    # no recall indicator is shown for context the model never receives.
     ext_prefetch_cache = ""
-    if agent._memory_manager:
+    if agent._memory_manager and replayed_persisted_user_msg is None:
         try:
             _query = original_user_message if isinstance(original_user_message, str) else ""
             if not is_trivial_prompt(_query):
@@ -1404,6 +1464,13 @@ def build_turn_context(
             except Exception:
                 pass
 
+    if replayed_persisted_user_msg is not None:
+        # The committed row already carries the exact api_content sidecar from
+        # the pre-crash prologue. Ignore newly recomputed memory/plugin context
+        # so this retry sends the same bytes and preserves the prompt prefix.
+        ext_prefetch_cache = ""
+        plugin_user_context = ""
+
     # ── api_content sidecar: persist what you send ──
     # The prefetch/plugin context above is injected into the API copy of this
     # turn's user message, never into the stored content — so on the next
@@ -1422,7 +1489,8 @@ def build_turn_context(
     # wire either — skip the stamp rather than persist provably wrong "exact
     # sent bytes" (MoA keeps its pre-sidecar cache behavior).
     if (
-        not moa_active
+        replayed_persisted_user_msg is None
+        and not moa_active
         and getattr(agent, "api_mode", None) != "codex_app_server"
         and 0 <= current_turn_user_idx < len(messages)
         and messages[current_turn_user_idx].get("role") == "user"
@@ -1490,6 +1558,20 @@ def _ensure_and_persist() -> None:
         if not isinstance(pending_cli_message, dict) or pending_cli_message.get("_db_persisted"):
             agent._pending_cli_user_message = None
 
+    if input_persisted_callback is not None:
+        persisted_user = (
+            messages[current_turn_user_idx]
+            if 0 <= current_turn_user_idx < len(messages)
+            else None
+        )
+        if not isinstance(persisted_user, dict) or not persisted_user.get(
+            "_db_persisted"
+        ):
+            raise RuntimeError(
+                "durable input admission requires a committed user row"
+            )
+        input_persisted_callback()
+
     # Title the session from this user message, now — the row exists and the
     # turn has not called the model yet. Titling is derived from the user's
     # ask alone, so it runs concurrently with the turn instead of waiting for

cli-config.yaml.example

@@ -1198,6 +1198,14 @@ platform_toolsets:
 #       # Route scripts default to a 30 second timeout. Scripts must live under
 #       # the active profile's scripts directory and receive webhook JSON on stdin.
 #       script_timeout_seconds: 30
+#       # A trusted route may hand its exact completed agent session to a new
+#       # thread in Discord's configured home channel. This is exclusive with
+#       # ordinary `deliver` output and cannot be combined with deliver_only.
+#       # routes:
+#       #   completed-build:
+#       #     secret: "replace-me"
+#       #     prompt: "Investigate completed build {build.id}"
+#       #     handoff_to: discord
 #
 # Discord-specific settings (config.yaml top-level, not under platforms:):
 #

gateway/config.py

@@ -2025,11 +2025,19 @@ def _enable_from_env(platform: Platform) -> PlatformConfig:
 
     discord_home = getenv("DISCORD_HOME_CHANNEL")
     if discord_home and Platform.DISCORD in config.platforms:
-        config.platforms[Platform.DISCORD].home_channel = HomeChannel(
+        discord_config = config.platforms[Platform.DISCORD]
+        existing_home = discord_config.home_channel
+        same_home = (
+            existing_home is not None
+            and existing_home.chat_id == discord_home
+        )
+        discord_config.home_channel = HomeChannel(
             platform=Platform.DISCORD,
             chat_id=discord_home,
             name=getenv("DISCORD_HOME_CHANNEL_NAME", "Home"),
             thread_id=getenv("DISCORD_HOME_CHANNEL_THREAD_ID") or None,
+            user_id=existing_home.user_id if existing_home and same_home else None,
+            scope_id=existing_home.scope_id if existing_home and same_home else None,
         )
 
     # Reply threading mode for Discord (off/first/all)