feat: improve tool call durability on long calls

application/agents/tool_executor.py

@@ -1,9 +1,11 @@
 import logging
 import re
 import uuid
-from collections import Counter
+from collections import Counter, OrderedDict
 from typing import Any, Dict, List, Optional, Tuple
 
+from sqlalchemy.exc import IntegrityError
+
 from application.agents.default_tools import (
     is_headless_excluded_tool,
     resolve_tool_by_id,
@@ -24,6 +26,16 @@ from application.storage.db.session import db_readonly, db_session
 logger = logging.getLogger(__name__)
 
 
+def _is_foreign_key_violation(exc: BaseException) -> bool:
+    """Whether ``exc`` is a Postgres FK violation (SQLSTATE 23503)."""
+    if not isinstance(exc, IntegrityError):
+        return False
+    pgcode = getattr(getattr(exc, "orig", None), "sqlstate", None) or getattr(
+        getattr(exc, "orig", None), "pgcode", None
+    )
+    return pgcode == "23503"
+
+
 # Tightest provider limit on function-call names (OpenAI: ^[a-zA-Z0-9_-]{1,64}$).
 _MAX_LLM_NAME_LEN = 64
 
@@ -122,6 +134,41 @@ def _redact_args_for_log(args: Any) -> Any:
     return redacted
 
 
+# ``message_id``s whose ``conversation_messages`` parent has been found
+# missing. Every tool-call journal write for such a message FK-fails
+# identically, and a long tool loop produces one ERROR *pair* per call — 60
+# ERRORs from a single stream in production, the whole api-tier error volume
+# for that day. Latch the first failure and log the rest at debug.
+# Bounded so a long-lived worker cannot grow this without limit.
+_MISSING_PARENTS: "OrderedDict[str, None]" = OrderedDict()
+_MISSING_PARENTS_MAX = 256
+
+
+def _is_missing_parent(message_id: Optional[str]) -> bool:
+    return bool(message_id) and message_id in _MISSING_PARENTS
+
+
+def _note_missing_parent(message_id: Optional[str], exc: BaseException) -> bool:
+    """Record a FK failure against ``message_id``. True if newly latched.
+
+    Args:
+        message_id: The message the write was scoped to, if any.
+        exc: The exception raised by the write.
+
+    Returns:
+        True when this is the first sighting for the message (so the caller
+        should log loudly), False when already latched or not a FK error.
+    """
+    if not message_id or not _is_foreign_key_violation(exc):
+        return False
+    if message_id in _MISSING_PARENTS:
+        return False
+    _MISSING_PARENTS[message_id] = None
+    while len(_MISSING_PARENTS) > _MISSING_PARENTS_MAX:
+        _MISSING_PARENTS.popitem(last=False)
+    return True
+
+
 def _journal_key(call_id: str, message_id: Optional[str]) -> str:
     """Namespace the durability-journal key by the per-turn ``message_id``.
 
@@ -177,8 +224,26 @@ def _record_proposed(
                 extra={"alert": "tool_call_id_collision", "call_id": call_id},
             )
         return inserted
-    except Exception:
-        logger.exception("tool_call_attempts proposed write failed for %s", call_id)
+    except Exception as exc:
+        if _note_missing_parent(message_id, exc):
+            logger.warning(
+                "tool_call_attempts: parent message row %s is gone "
+                "(deleted mid-stream); suppressing further journal errors "
+                "for this message. First failing call: %s",
+                message_id,
+                call_id,
+            )
+        elif _is_missing_parent(message_id):
+            logger.debug(
+                "tool_call_attempts proposed write skipped for %s "
+                "(parent %s already known missing)",
+                call_id,
+                message_id,
+            )
+        else:
+            logger.exception(
+                "tool_call_attempts proposed write failed for %s", call_id
+            )
         return False
 
 
@@ -229,8 +294,18 @@ def _mark_executed(
                 user_id=user_id,
                 agent_id=(str(agent_id) if agent_id and looks_like_uuid(str(agent_id)) else None),
             )
-    except Exception:
-        logger.exception("tool_call_attempts executed write failed for %s", call_id)
+    except Exception as exc:
+        if _note_missing_parent(message_id, exc) or _is_missing_parent(message_id):
+            logger.debug(
+                "tool_call_attempts executed write skipped for %s "
+                "(parent %s gone)",
+                call_id,
+                message_id,
+            )
+        else:
+            logger.exception(
+                "tool_call_attempts executed write failed for %s", call_id
+            )
 
 
 def _mark_failed(

application/api/answer/routes/base.py

@@ -1,6 +1,7 @@
 import datetime
 import json
 import logging
+import threading
 import time
 import uuid
 from typing import Any, Dict, Generator, List, Optional
@@ -37,6 +38,16 @@ from application.utils import check_required_fields
 
 logger = logging.getLogger(__name__)
 
+# Seconds between liveness stamps on an in-flight message row. 30 s keeps
+# three stamps inside the reconciler's 5-minute staleness window and inside
+# the replay watchdog's 90 s producer-idle window, so a single missed tick is
+# never enough to trip either.
+STREAM_HEARTBEAT_INTERVAL = 30
+# Ceiling on how long the ticker will keep a row alive. Above the realistic
+# worst case for a 25-round tool loop, but finite, so a wedged-but-alive
+# stream still gets swept eventually.
+STREAM_HEARTBEAT_MAX_SECONDS = 3600
+
 
 answer_ns = Namespace("answer", description="Answer related operations", path="/")
 
@@ -318,12 +329,10 @@ class BaseAnswerResource:
         # ``streaming``, yet must still count as live).
         streaming_marked = False
         # Heartbeat goes into ``metadata.last_heartbeat_at`` (not
-        # ``updated_at``, which reconciler-side writes share) and uses
-        # ``time.monotonic`` so a blocked event loop can't fake fresh.
+        # ``updated_at``, which reconciler-side writes share).
         # ``heartbeat_message`` only touches non-terminal rows, so stamping a
         # still-``pending`` row is safe and does NOT change its status.
-        STREAM_HEARTBEAT_INTERVAL = 60
-        last_heartbeat_at = time.monotonic()
+        heartbeat_stop: Optional[threading.Event] = None
 
         def _mark_streaming_once() -> None:
             """Flip the reserved row ``pending → streaming`` exactly once.
@@ -335,7 +344,7 @@ class BaseAnswerResource:
             ``_heartbeat_streaming``), so a thought-only reasoning phase that
             never reaches this point still stays live.
             """
-            nonlocal streaming_marked, last_heartbeat_at
+            nonlocal streaming_marked
             if streaming_marked or not reserved_message_id:
                 return
             try:
@@ -348,7 +357,7 @@ class BaseAnswerResource:
                     reserved_message_id,
                 )
             # Re-stamp last_heartbeat_at on the transition too; harmless given
-            # the seed at generation start and the per-interval pump below.
+            # the seed at generation start and the background ticker below.
             try:
                 self.conversation_service.heartbeat_message(
                     reserved_message_id,
@@ -359,43 +368,68 @@ class BaseAnswerResource:
                     reserved_message_id,
                 )
             streaming_marked = True
-            last_heartbeat_at = time.monotonic()
 
-        def _heartbeat_streaming() -> None:
-            """Pump the liveness heartbeat once per ``STREAM_HEARTBEAT_INTERVAL``.
+        def _start_heartbeat_ticker() -> "Optional[threading.Event]":
+            """Stamp the liveness heartbeat on a timer for the stream's life.
 
-            Deliberately gated on ``reserved_message_id`` only — NOT on
-            ``streaming_marked``. The loop calls this for *every* chunk
-            (including ``thought``/``metadata``), so a reasoning model that
-            streams only ``thought`` chunks while it "thinks" keeps a still-
-            ``pending`` row's ``last_heartbeat_at`` fresh and stays out of the
-            reconciler's staleness sweep. ``heartbeat_message`` only updates
-            non-terminal rows, so this never resurrects or restatuses a
-            terminal row.
+            This replaces a per-chunk pump that could only stamp when a chunk
+            flowed, which made liveness a function of *output* rather than of
+            the stream actually being alive. Four windows are routinely silent
+            — a provider round that emits only tool-call deltas, the body of a
+            tool call, a same-primary retry or cross-provider fallback, and
+            mid-execution compression — and any of them longer than the
+            reconciler's threshold got a healthy stream swept and its answer
+            discarded. The old pump was also interval-gated against the *last
+            stamp*, so bursty output could leave the row 60 s staler than the
+            last chunk suggested.
 
-            Residual: a model that emits NO chunks at all (not even
-            ``thought``) for longer than the reconciler threshold still goes
-            stale, because this pump only ticks when a chunk flows. Covering a
-            fully-silent stream would require a background-thread heartbeat or
-            a higher staleness threshold; both are out of scope here. The
-            realistic reasoning case (``thought`` chunks streaming) is covered.
+            Liveness stays honest because the ticker is an in-process daemon
+            thread: every real death mode we see in production (gunicorn
+            SIGKILL, worker OOM, ``max_requests`` recycle, host freeze) takes
+            the thread with it, so the row goes stale on schedule and the
+            reconciler still does its job. ``STREAM_HEARTBEAT_MAX_SECONDS``
+            bounds the other direction — a hung-but-alive stream cannot keep a
+            row alive forever.
+
+            Returns:
+                The stop event for the ticker, or None when there is no
+                reserved row to stamp (headless/``/v1`` continuation rounds).
             """
-            nonlocal last_heartbeat_at
             if not reserved_message_id:
-                return
-            now_mono = time.monotonic()
-            if now_mono - last_heartbeat_at < STREAM_HEARTBEAT_INTERVAL:
-                return
-            try:
-                self.conversation_service.heartbeat_message(
-                    reserved_message_id,
-                )
-            except Exception:
-                logger.exception(
-                    "stream heartbeat update failed for %s",
-                    reserved_message_id,
-                )
-            last_heartbeat_at = now_mono
+                return None
+            stop = threading.Event()
+            message_id = reserved_message_id
+            service = self.conversation_service
+
+            def _tick() -> None:
+                deadline = time.monotonic() + STREAM_HEARTBEAT_MAX_SECONDS
+                while not stop.wait(STREAM_HEARTBEAT_INTERVAL):
+                    if time.monotonic() > deadline:
+                        logger.warning(
+                            "stream heartbeat ticker hit its %ss ceiling for "
+                            "message_id=%s; stopping",
+                            STREAM_HEARTBEAT_MAX_SECONDS,
+                            message_id,
+                        )
+                        return
+                    try:
+                        # Returns False once the row is terminal, which is the
+                        # natural stop signal for a reclaimed or finished row.
+                        if not service.heartbeat_message(message_id):
+                            return
+                    except Exception:
+                        # Swallowed deliberately: a transient DB blip must not
+                        # kill the stream. The reconciler is the backstop.
+                        logger.exception(
+                            "stream heartbeat update failed for %s", message_id,
+                        )
+
+            threading.Thread(
+                target=_tick,
+                daemon=True,
+                name=f"stream-heartbeat-{message_id[:8]}",
+            ).start()
+            return stop
 
         # Correlates tool_call_attempts rows with this message.
         if reserved_message_id and getattr(agent, "tool_executor", None):
@@ -517,14 +551,12 @@ class BaseAnswerResource:
                         "generation-start heartbeat seed failed for %s",
                         reserved_message_id,
                     )
-                last_heartbeat_at = time.monotonic()
+
+            # The seed above covers t=0; the ticker takes over from the first
+            # interval onwards, independently of whether anything is flowing.
+            heartbeat_stop = _start_heartbeat_ticker()
 
             for line in gen_iter:
-                # Cheap closure check that only hits the DB when the heartbeat
-                # interval has elapsed. Runs for *every* chunk (incl. ``thought``
-                # / ``metadata``), so a still-``pending`` reasoning stream stays
-                # live without waiting for the ``streaming`` status flip.
-                _heartbeat_streaming()
                 if "metadata" in line:
                     query_metadata.update(line["metadata"])
                 elif "answer" in line:
@@ -882,7 +914,7 @@ class BaseAnswerResource:
 
             if should_persist:
                 if reserved_message_id is not None:
-                    self.conversation_service.finalize_message(
+                    finalize_outcome = self.conversation_service.finalize_message(
                         reserved_message_id,
                         response_full,
                         thought=thought,
@@ -902,6 +934,29 @@ class BaseAnswerResource:
                         } if llm is not None else None,
                         async_title_generation=llm is not None,
                     )
+                    # The outcome used to be discarded here, which is how a
+                    # finished answer could vanish silently: if the row was
+                    # deleted mid-stream (retry/edit truncation) the write
+                    # lands nowhere and `activity_finished` still reports
+                    # `ok`, because activity logging never observes the DB.
+                    # Emit a distinct, countable signal instead.
+                    if finalize_outcome is MessageUpdateOutcome.NOT_FOUND:
+                        logger.error(
+                            "answer_persist_failed: message row %s no longer "
+                            "exists; %d chars of answer were produced and "
+                            "could not be saved (conversation=%s)",
+                            reserved_message_id,
+                            len(response_full or ""),
+                            conversation_id,
+                            extra={
+                                "alert": "answer_persist_failed",
+                                "message_id": reserved_message_id,
+                                "conversation_id": (
+                                    str(conversation_id) if conversation_id else None
+                                ),
+                                "answer_length": len(response_full or ""),
+                            },
+                        )
                 else:
                     conversation_id = self.conversation_service.save_conversation(
                         conversation_id,
@@ -1227,6 +1282,14 @@ class BaseAnswerResource:
             if journal_writer is not None:
                 journal_writer.close()
             return
+        finally:
+            # Every exit path — normal, client abort, error — must stop the
+            # ticker, or a leaked thread keeps stamping a row nobody owns.
+            # Harmless if it ever does leak (``heartbeat_message`` no-ops on
+            # terminal rows) but the thread would live until the worker
+            # recycles.
+            if heartbeat_stop is not None:
+                heartbeat_stop.set()
 
     def _finalize_stateless_tool_pause(
         self,

application/api/answer/services/conversation_service.py

@@ -402,11 +402,20 @@ class ConversationService:
         # Atomic message update + tool_call_attempts confirm; the
         # ``only_if_non_terminal`` guard prevents a late stream from
         # retracting a row the reconciler already escalated.
+        #
+        # Exception: a *successful* finalize is allowed to reclaim a row the
+        # reconciler failed for staleness. A stream that reaches this point
+        # with an answer was self-evidently not stuck, so refusing it throws
+        # away real output and leaves the user staring at "Response was
+        # terminated prior to completion". Only on success — a late failure
+        # must never overwrite the reconciler's failure with a different one.
+        reclaim = status == "complete"
         with db_session() as conn:
             repo = ConversationsRepository(conn)
             outcome = repo.update_message_by_id(
                 message_id, update_fields,
                 only_if_non_terminal=True,
+                reclaim_reconciler_failed=reclaim,
             )
             if outcome is not MessageUpdateOutcome.UPDATED:
                 logger.warning(

application/api/user/reconciliation.py

@@ -86,6 +86,13 @@ def run_reconciliation() -> Dict[str, Any]:
                     and conversation_id
                     and pt_repo.delete_state(str(conversation_id), str(user_id))
                 ):
+                    # Mark the row as having had its approval revoked. A late
+                    # finalize is allowed to reclaim a reconciler-failed row
+                    # (see ``finalize_message``), but must not reclaim this
+                    # one: the resumable state is gone and the UI has already
+                    # been told the approval was cleared, so landing an answer
+                    # here would contradict what the user was shown.
+                    repo.mark_message_approval_cleared(msg["id"])
                     events.append(
                         (
                             str(user_id),

application/parser/file/docling_parser.py

@@ -94,6 +94,46 @@ def _exceeds_tabular_gate(file: Path) -> bool:
     return _tabular_content_size(file) > max_bytes
 
 
+def _delegate(
+    parser: BaseParser, file: Path, errors: str
+) -> Union[str, List[str]]:
+    """Run a lightweight fallback parser, normalizing its failures.
+
+    Anything the fallback raises is converted to ``DocumentParseError``,
+    which matters twice over. The Celery tasks that own these paths list
+    ``DocumentParseError`` in ``dont_autoretry_for``, so a deterministic
+    content error now fails once instead of retrying four times (each retry
+    re-downloading the file from S3); and ``SimpleDirectoryReader.load_data``
+    catches only ``DocumentParseError``, so one poison spreadsheet is skipped
+    into ``failed_files`` instead of aborting an entire multi-file ingest.
+
+    This restores the symmetry with ``DoclingParser.parse_file``, whose own
+    failures are already wrapped — the delegation branch bypassed that.
+
+    Args:
+        parser: The lightweight parser to delegate to.
+        file: Path to the file being parsed.
+        errors: Decoding error policy, forwarded to the parser.
+
+    Returns:
+        The parsed text, or list of row strings.
+
+    Raises:
+        DocumentParseError: If the fallback parser fails for any reason.
+    """
+    try:
+        return parser.parse_file(file, errors)
+    except DocumentParseError:
+        raise
+    except Exception as e:
+        logger.error(
+            f"Lightweight parse of {file.name} failed: {e}", exc_info=True
+        )
+        raise DocumentParseError(
+            f"Failed to parse {file.name}: the file could not be read."
+        ) from e
+
+
 def _capped_markup_copy(file: Path) -> Optional[str]:
     """Head-truncate an oversized markup file to a temp copy for docling.
 
@@ -413,7 +453,7 @@ class DoclingXLSXParser(DoclingParser):
             )
             from application.parser.file.tabular_parser import ExcelParser
 
-            return ExcelParser().parse_file(file, errors)
+            return _delegate(ExcelParser(), file, errors)
         return super().parse_file(file, errors)
 
 
@@ -466,7 +506,7 @@ class DoclingCSVParser(DoclingParser):
             )
             from application.parser.file.tabular_parser import CSVParser
 
-            return CSVParser().parse_file(file, errors)
+            return _delegate(CSVParser(), file, errors)
         return super().parse_file(file, errors)
 
 

application/parser/file/tabular_parser.py

@@ -3,12 +3,63 @@
 Contains parsers for tabular data files.
 
 """
+import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Union
 
 from application.parser.file.base_parser import BaseParser
 
 
+def cell_to_text(value: Any) -> str:
+    """Render one spreadsheet/CSV cell as text that is always safe to join.
+
+    ``Series.astype(str)`` cannot be trusted for this. Up to pandas 2.x it
+    rendered missing values as the string ``"nan"``; pandas 3.0 *preserves*
+    them, so ``.tolist()`` hands back Python ``float`` NaNs and ``str.join``
+    raises ``TypeError: sequence item N: expected str instance, float found``.
+    A single blank cell anywhere in a sheet was therefore enough to destroy
+    an entire upload.
+
+    Missing values (blank cells, ``NaT``, Excel error values, and formulas
+    with no cached result) become the empty string. Dates and times are
+    rendered ISO-8601 rather than pandas' repr. Integral floats drop the
+    ``.0`` that pandas adds when a blank upcasts an integer column — an ID
+    column with one gap otherwise turns every id into ``1001.0``. NUL bytes
+    are stripped here because they crash the downstream Postgres text write.
+
+    Args:
+        value: A single cell value, of any type pandas or openpyxl may yield.
+
+    Returns:
+        str: The cell rendered as text; ``""`` for missing values.
+    """
+    if value is None:
+        return ""
+    # NaN is the only float that is not equal to itself. Checked before the
+    # pandas call because it is by far the common case and needs no import.
+    if isinstance(value, float) and value != value:
+        return ""
+    try:
+        import pandas as pd
+
+        # Raises on array-like cells, which are legitimate values, not NA.
+        if pd.isna(value):
+            return ""
+    except (ImportError, TypeError, ValueError):
+        pass
+    if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
+        return value.isoformat()
+    if isinstance(value, float) and value.is_integer():
+        return str(int(value))
+    text = str(value)
+    return text.replace("\x00", "") if "\x00" in text else text
+
+
+def _row_to_texts(row: Any) -> List[str]:
+    """Render a pandas row as a list of join-safe strings."""
+    return [cell_to_text(v) for v in row.tolist()]
+
+
 class CSVParser(BaseParser):
     """CSV parser.
 
@@ -114,22 +165,22 @@ class PandasCSVParser(BaseParser):
             raise ValueError("pandas module is required to read CSV files.")
 
         df = pd.read_csv(file, **self._pandas_config)
-        headers = df.columns.tolist()
+        headers = [cell_to_text(h) for h in df.columns.tolist()]
         header_row = f"{self._header_prefix}{self._col_joiner.join(headers)}"
 
         if not self._concat_rows:
             return df.apply(
-                lambda row: (self._col_joiner).join(row.astype(str).tolist()), axis=1
+                lambda row: (self._col_joiner).join(_row_to_texts(row)), axis=1
             ).tolist()
-        
+
         text_list = []
         if self._header_period != 1:
             text_list.append(header_row)
-        
+
         for i, row in df.iterrows():
             if (self._header_period > 1 and i > 0 and i % self._header_period == 0):
                 text_list.append(header_row)
-            text_list.append(self._col_joiner.join(row.astype(str).tolist()))
+            text_list.append(self._col_joiner.join(_row_to_texts(row)))
             if self._header_period == 1 and i < len(df) - 1:
                 text_list.append(header_row)
 
@@ -200,14 +251,14 @@ class ExcelParser(BaseParser):
             raise ValueError("pandas module is required to read Excel files.")
 
         df = pd.read_excel(file, **self._pandas_config)
-        headers = [str(h) for h in df.columns.tolist()]
+        headers = [cell_to_text(h) for h in df.columns.tolist()]
         header_row = f"{self._header_prefix}{self._col_joiner.join(headers)}"
-        
+
         if not self._concat_rows:
             return df.apply(
-                lambda row: (self._col_joiner).join(row.astype(str).tolist()), axis=1
+                lambda row: (self._col_joiner).join(_row_to_texts(row)), axis=1
             ).tolist()
-        
+
         text_list = []
         if self._header_period != 1:
             text_list.append(header_row)
@@ -215,7 +266,7 @@ class ExcelParser(BaseParser):
         for i, row in df.iterrows():
             if (self._header_period > 1 and i > 0 and i % self._header_period == 0):
                 text_list.append(header_row)
-            text_list.append(self._col_joiner.join(row.astype(str).tolist()))
+            text_list.append(self._col_joiner.join(_row_to_texts(row)))
             if self._header_period == 1 and i < len(df) - 1:
                 text_list.append(header_row)
         return self._row_joiner.join(text_list)

application/storage/db/repositories/conversations.py

@@ -873,6 +873,7 @@ class ConversationsRepository:
     def update_message_by_id(
         self, message_id: str, fields: dict,
         *, only_if_non_terminal: bool = False,
+        reclaim_reconciler_failed: bool = False,
     ) -> MessageUpdateOutcome:
         """Update specific fields on a message identified by its UUID.
 
@@ -882,9 +883,36 @@ class ConversationsRepository:
         True, the update is gated so a late finalize cannot retract a
         reconciler-set ``failed`` (or a prior ``complete``).
 
+        ``reclaim_reconciler_failed`` punches one hole in that gate: a row
+        the **reconciler** failed for staleness may still be finalized by the
+        stream that owns it, because such a stream demonstrably was not dead.
+        Production showed a 20-minute tool loop swept at minute 6 and then
+        completing normally at minute 20, its answer refused and replaced by
+        "Response was terminated prior to completion". The hole is deliberately
+        narrow — it matches only the reconciler's own error string, so genuine
+        terminal failures (client disconnect, provider errors, the pending-tool
+        cleanup task) stay terminal, and it skips rows whose awaiting-approval
+        prompt the reconciler already revoked.
+
+        Reclaiming also strips the reconciler's bookkeeping keys, which
+        otherwise survive the metadata merge and leave a ``complete`` row
+        carrying a stale ``error``.
+
         The return value discriminates "I updated the row" from "the
         row was already at a terminal state" so the abort handler can
-        journal ``end`` when the normal-path finalize already ran.
+        journal ``end`` when the normal-path finalize already ran. A reclaim
+        returns plain ``UPDATED``: callers that branch on the outcome should
+        treat it as an ordinary successful finalize.
+
+        Args:
+            message_id: UUID of the message row.
+            fields: Field/value pairs to write.
+            only_if_non_terminal: Gate the write on a non-terminal status.
+            reclaim_reconciler_failed: Also allow rows the reconciler failed
+                for staleness.
+
+        Returns:
+            MessageUpdateOutcome: What happened to the row.
         """
         if not looks_like_uuid(message_id):
             return MessageUpdateOutcome.INVALID
@@ -899,6 +927,16 @@ class ConversationsRepository:
 
         api_to_col = {"metadata": "message_metadata"}
 
+        # On reclaim the reconciler's bookkeeping must not survive into the
+        # finished row: ``error`` would show a stale failure in the API
+        # response, and leaving the counters behind corrupts any "how often
+        # does this fire" query.
+        reclaim_strip = (
+            " - 'error' - 'reconcile_attempts' - 'last_reconcile_seen_heartbeat'"
+            if reclaim_reconciler_failed
+            else ""
+        )
+
         set_parts = []
         params: dict = {"id": message_id}
         for key, val in filtered.items():
@@ -908,7 +946,7 @@ class ConversationsRepository:
                     set_parts.append(f"{col} = NULL")
                 else:
                     set_parts.append(
-                        f"{col} = COALESCE({col}, '{{}}'::jsonb) "
+                        f"{col} = (COALESCE({col}, '{{}}'::jsonb){reclaim_strip}) "
                         f"|| CAST(:{col} AS jsonb)"
                     )
                     params[col] = (
@@ -932,9 +970,28 @@ class ConversationsRepository:
                 params[col] = val
 
         set_parts.append("updated_at = now()")
+        # Metadata may not be among the written fields (a finalize with no
+        # query metadata passes none), but the strip must still happen.
+        if reclaim_reconciler_failed and "metadata" not in filtered:
+            set_parts.append(
+                f"message_metadata = COALESCE(message_metadata, '{{}}'::jsonb)"
+                f"{reclaim_strip}"
+            )
         update_where = ["id = CAST(:id AS uuid)"]
         if only_if_non_terminal:
-            update_where.append("status NOT IN ('complete', 'failed')")
+            terminal_gate = "status NOT IN ('complete', 'failed')"
+            if reclaim_reconciler_failed:
+                terminal_gate = (
+                    "(" + terminal_gate + " OR ("
+                    "status = 'failed' "
+                    "AND message_metadata->>'error' LIKE "
+                    "'reconciler: stuck in pending/streaming%' "
+                    "AND COALESCE("
+                    "(message_metadata->>'reconciler_cleared_approval')::boolean,"
+                    " false) = false"
+                    "))"
+                )
+            update_where.append(terminal_gate)
         # Single-statement attempt + prior-status probe. Both CTEs see
         # the same MVCC snapshot, so ``prior.status`` reflects the row
         # state before the UPDATE — exactly what we need to tell
@@ -958,6 +1015,15 @@ class ConversationsRepository:
             return MessageUpdateOutcome.NOT_FOUND
         updated, prior_status = row[0], row[1]
         if updated:
+            if reclaim_reconciler_failed and prior_status == "failed":
+                # Distinct, greppable signal: each of these is one user-visible
+                # answer that would previously have been discarded. Should
+                # trend to zero once false sweeps stop.
+                logger.info(
+                    "finalize_message: reclaimed reconciler-failed row "
+                    "message_id=%s",
+                    message_id,
+                )
             return MessageUpdateOutcome.UPDATED
         if prior_status is None:
             return MessageUpdateOutcome.NOT_FOUND
@@ -1041,11 +1107,62 @@ class ConversationsRepository:
 
         Mirrors Mongo's ``$push`` + ``$slice`` that trims queries after an
         index-based update.
+
+        Rows being deleted may still be **owned by a running stream**: this
+        runs on the retry/edit path, and a stream survives client disconnect
+        (the SSE keepalive pump drains the generator in a daemon thread), while
+        a stream paused on a client-side tool can resume many minutes later.
+        Deleting such a row cascades its ``message_events`` away and makes
+        every subsequent journal and ``tool_call_attempts`` write FK-fail — in
+        production one paused stream, deleted 9 minutes before it resumed,
+        produced 342 WARNs plus 60 ERRORs and finalized into nothing.
+
+        Deleting is still correct — the user explicitly asked to replace these
+        turns — so the rows go, but any non-terminal one is first stamped
+        terminal. That makes the owning stream's late ``finalize_message``
+        report ``ALREADY_FAILED`` (a known, quiet outcome) rather than
+        ``NOT_FOUND``, and leaves an audit trail of what was superseded.
+
+        Args:
+            conversation_id: Conversation whose tail is being trimmed.
+            keep_up_to: Highest position to keep.
+
+        Returns:
+            Number of rows deleted.
         """
         # Shape-gate: see ``rename`` — prevents transaction poisoning when
         # a non-UUID id reaches this code path.
         if not looks_like_uuid(conversation_id):
             return 0
+        superseded = self._conn.execute(
+            text(
+                """
+                UPDATE conversation_messages
+                SET status = 'failed',
+                    message_metadata = jsonb_set(
+                        COALESCE(message_metadata, '{}'::jsonb),
+                        '{error}',
+                        to_jsonb(
+                            CAST('superseded by a newer turn at this position'
+                                 AS text)
+                        )
+                    )
+                WHERE conversation_id = CAST(:conv_id AS uuid)
+                  AND position > :pos
+                  AND status NOT IN ('complete', 'failed')
+                RETURNING id
+                """
+            ),
+            {"conv_id": conversation_id, "pos": keep_up_to},
+        ).fetchall()
+        if superseded:
+            logger.info(
+                "truncate_after: superseding %d in-flight message(s) in "
+                "conversation %s: %s",
+                len(superseded),
+                conversation_id,
+                ", ".join(str(r[0]) for r in superseded),
+            )
         result = self._conn.execute(
             text(
                 "DELETE FROM conversation_messages "

application/storage/db/repositories/reconciliation.py

@@ -14,7 +14,8 @@ class ReconciliationRepository:
         self._conn = conn
 
     def find_and_lock_stuck_messages(
-        self, *, age_minutes: int = 5, limit: int = 100,
+        self, *, age_minutes: int = 5, tool_grace_minutes: int = 15,
+        limit: int = 100,
     ) -> list[dict]:
         """Lock stuck pending/streaming messages skipping live resumes.
 
@@ -27,6 +28,33 @@ class ReconciliationRepository:
         waiting for resume) and ``resuming`` (actively executing)
         ``pending_tool_state`` rows so a paused message survives until
         the PT row's own TTL retires it.
+
+        A second exemption covers **server-executed** tools.
+        ``pending_tool_state`` is only written on the pause path (client-side
+        tools, approval gates), so an agent grinding through
+        ``google_search``/``read_webpage``/``code_executor`` rounds has no PT
+        row and was swept as "stuck" while perfectly healthy. A message with a
+        ``proposed``/``executed`` ``tool_call_attempts`` row that transitioned
+        within ``tool_grace_minutes`` is therefore also exempt. This is
+        deliberately redundant with the stream's timed heartbeat: that write
+        shares the app's connection pool, so pool pressure is exactly the
+        moment liveness should not depend on it, whereas these rows were
+        committed earlier by a different code path.
+
+        Note the coupling — the ``proposed`` (5 min) and ``executed`` (15 min)
+        sweeps below both flip rows to ``failed``, which drops them out of this
+        exemption. That is intentional (a genuinely dead stream's exemption
+        expires rather than lasting forever), but it means shortening either
+        sweep silently shortens this grace too.
+
+        Args:
+            age_minutes: Staleness threshold for the message row.
+            tool_grace_minutes: How recently a tool call must have transitioned
+                for its message to count as live.
+            limit: Maximum rows to lock per tick.
+
+        Returns:
+            The locked rows as dicts.
         """
         result = self._conn.execute(
             text(
@@ -54,12 +82,24 @@ class ReconciliationRepository:
                                  > now() - interval '10 minutes')
                         )
                   )
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM tool_call_attempts tca
+                      WHERE tca.message_id = cm.id
+                        AND tca.status IN ('proposed', 'executed')
+                        AND tca.updated_at
+                            > now() - make_interval(mins => :tool_grace)
+                  )
                 ORDER BY cm.timestamp ASC
                 LIMIT :limit
                 FOR UPDATE OF cm SKIP LOCKED
                 """
             ),
-            {"age": age_minutes, "limit": limit},
+            {
+                "age": age_minutes,
+                "tool_grace": tool_grace_minutes,
+                "limit": limit,
+            },
         )
         return [row_to_dict(r) for r in result.fetchall()]
 
@@ -149,21 +189,60 @@ class ReconciliationRepository:
         return result.rowcount > 0
 
     def increment_message_reconcile_attempts(self, message_id: str) -> int:
-        """Bump ``message_metadata.reconcile_attempts`` and return the new count."""
+        """Bump ``reconcile_attempts``, resetting it if the row proved alive.
+
+        The counter means *consecutive* stale ticks, not a lifetime total. It
+        used to be a pure accumulator, so a long tool loop that stalled, then
+        recovered, then stalled again died on its second stall — with a
+        counter whose name lied about what it counted. Each tick records the
+        ``last_heartbeat_at`` it observed; if the heartbeat has advanced since
+        the previous tick, the stream demonstrably produced life in between
+        and the count restarts at 1.
+
+        This also absorbs a race the timed heartbeat introduces: the
+        reconciler can lock a row microseconds before the ticker's stamp
+        commits, and without the reset that phantom attempt would be
+        permanent.
+
+        Safe against masking a dead stream: ``last_heartbeat_at`` is only ever
+        written by ``heartbeat_message``, which runs in the process producing
+        the stream. A dead producer cannot advance it, so the reset condition
+        can never be met and escalation proceeds unchanged.
+
+        Args:
+            message_id: The message row to bump.
+
+        Returns:
+            The new consecutive-stale-tick count.
+        """
         result = self._conn.execute(
             text(
                 """
                 UPDATE conversation_messages
-                SET message_metadata = jsonb_set(
-                    COALESCE(message_metadata, '{}'::jsonb),
-                    '{reconcile_attempts}',
-                    to_jsonb(
+                SET message_metadata =
+                    jsonb_set(
+                        jsonb_set(
+                            COALESCE(message_metadata, '{}'::jsonb),
+                            '{reconcile_attempts}',
+                            to_jsonb(
+                                CASE
+                                    WHEN message_metadata->>'last_heartbeat_at'
+                                         IS DISTINCT FROM
+                                         message_metadata->>'last_reconcile_seen_heartbeat'
+                                    THEN 1
+                                    ELSE COALESCE(
+                                        (message_metadata->>'reconcile_attempts')::int,
+                                        0
+                                    ) + 1
+                                END
+                            )
+                        ),
+                        '{last_reconcile_seen_heartbeat}',
                         COALESCE(
-                            (message_metadata->>'reconcile_attempts')::int,
-                            0
-                        ) + 1
+                            message_metadata->'last_heartbeat_at',
+                            'null'::jsonb
+                        )
                     )
-                )
                 WHERE id = CAST(:message_id AS uuid)
                 RETURNING (message_metadata->>'reconcile_attempts')::int
                          AS new_count
@@ -193,6 +272,35 @@ class ReconciliationRepository:
         )
         return result.rowcount > 0
 
+    def mark_message_approval_cleared(self, message_id: str) -> bool:
+        """Record that this sweep also revoked an awaiting-approval prompt.
+
+        Read by ``finalize_message``'s reclaim path, which refuses to revive a
+        row whose ``pending_tool_state`` the reconciler already deleted and
+        whose ``tool.approval.cleared`` event has already been published.
+
+        Args:
+            message_id: The message row to stamp.
+
+        Returns:
+            True when the row was stamped.
+        """
+        result = self._conn.execute(
+            text(
+                """
+                UPDATE conversation_messages
+                SET message_metadata = jsonb_set(
+                    COALESCE(message_metadata, '{}'::jsonb),
+                    '{reconciler_cleared_approval}',
+                    'true'::jsonb
+                )
+                WHERE id = CAST(:message_id AS uuid)
+                """
+            ),
+            {"message_id": message_id},
+        )
+        return result.rowcount > 0
+
     def mark_tool_call_failed(self, call_id: str, *, error: str) -> bool:
         """Flip a tool_call_attempts row to ``failed`` with ``error``."""
         result = self._conn.execute(

application/streaming/message_journal.py

@@ -233,6 +233,9 @@ class BatchedJournalWriter:
         self._buffer: list[tuple[int, str, dict[str, Any]]] = []
         self._last_flush_mono_ms = time.monotonic() * 1000.0
         self._closed = False
+        # Latched once the parent row is found to be gone. A deleted parent
+        # never returns, so every subsequent write would FK-fail identically.
+        self._parent_missing = False
 
     def record(
         self,
@@ -250,6 +253,10 @@ class BatchedJournalWriter:
                 event_type,
             )
             return False
+        if self._parent_missing:
+            # Parent row deleted mid-stream; nothing can be replayed and no
+            # future write can succeed. Already logged once by ``flush``.
+            return False
         if not event_type:
             logger.warning(
                 "BatchedJournalWriter.record without event_type: "
@@ -317,15 +324,23 @@ class BatchedJournalWriter:
                 )
         except IntegrityError as exc:
             if _is_foreign_key_violation(exc):
-                # The whole batch references a conversation_messages
-                # parent that doesn't exist — every per-row retry would
-                # FK-fail too, so drop the batch once instead of N
-                # pointless re-attempts. The buffer is already cleared.
+                # The parent ``conversation_messages`` row is gone. In
+                # production this means the row was DELETED out from under a
+                # still-running stream — ``truncate_after`` on a retry or an
+                # edited question removes the row a live (or paused) stream
+                # still owns, and ``message_events`` cascades with it.
+                #
+                # It never comes back, so latch: one WARN for the stream, then
+                # stop re-attempting and stop re-logging. Un-latched, a single
+                # 3.5-minute research stream produced 342 identical WARNs and
+                # was the entire WARN volume for the day.
+                self._parent_missing = True
                 logger.warning(
                     "BatchedJournalWriter: no conversation_messages row "
                     "for message_id=%s (foreign-key violation); dropping "
-                    "%d event(s). The parent message row must be committed "
-                    "before its events are journaled.",
+                    "%d event(s) and all further events for this stream. "
+                    "The parent row was deleted while the stream was still "
+                    "running (retry/edit truncation) or never committed.",
                     self._message_id,
                     len(pending),
                 )