fix: little more tool call hardening

application/agents/tools/tool_action_parser.py

@@ -68,9 +68,22 @@ class ToolActionParser:
             # along with it. Only genuinely malformed arguments fail here.
             call_args = self._decode_arguments(call.arguments)
             if call_args is _MALFORMED_ARGUMENTS:
+                # Quote a bounded prefix of the payload: the shape is the only
+                # thing that tells "we concatenated two complete payloads"
+                # (``{}{}`` — the per-index merge in llm/handlers/base.py
+                # appends, so a provider that restates complete arguments on a
+                # second frame produces invalid JSON) apart from "the model
+                # emitted garbage". Both look identical without it, which cost
+                # a day of triage on 2026-08-19. Truncated because tool
+                # arguments carry user content.
+                raw = call.arguments
+                preview = raw[:120] if isinstance(raw, str) else repr(raw)[:120]
                 logger.error(
-                    "Error parsing OpenAI LLM call: arguments are not decodable JSON (%s)",
+                    "Error parsing OpenAI LLM call: arguments are not decodable "
+                    "JSON (%s); length=%s prefix=%r",
                     getattr(call, "name", "<unknown>"),
+                    len(raw) if isinstance(raw, (str, bytes)) else "n/a",
+                    preview,
                 )
                 return None, None, None
 

application/api/answer/routes/stream.py

@@ -8,6 +8,9 @@ from application.api import api
 
 from application.api.answer.routes.base import answer_ns, BaseAnswerResource
 
+from application.api.answer.services.continuation_service import (
+    ResumeInProgressError,
+)
 from application.api.answer.services.persistence_policy import resolve_persistence
 from application.api.answer.services.stream_processor import StreamProcessor
 from application.streaming.sse_keepalive import with_sse_keepalive
@@ -175,6 +178,24 @@ class StreamResource(Resource, BaseAnswerResource):
                 ),
                 mimetype="text/event-stream",
             )
+        except ResumeInProgressError as e:
+            # Another request already owns this conversation's continuation
+            # claim — an expected concurrency outcome, not a bad request. It
+            # must be caught ahead of ``ValueError`` (its base class), or it
+            # lands in the handler below and is reported to the user as a
+            # malformed body while filling the ERROR channel with tracebacks.
+            # ``/v1/chat/completions`` has answered 409 for this since
+            # 2026-07-13; this is the same contract on the primary route.
+            logger.warning(
+                "/stream - resume already in progress for conversation %s",
+                data.get("conversation_id"),
+                extra={"error": str(e)},
+            )
+            return Response(
+                self.error_stream_generate(str(e)),
+                status=409,
+                mimetype="text/event-stream",
+            )
         except ValueError as e:
             message = "Malformed request body"
             logger.error(

application/api/user/reconciliation.py

@@ -60,6 +60,28 @@ def run_reconciliation() -> Dict[str, Any]:
     # ``(user_id, event_type, payload, scope)``.
     events: list[tuple] = []
 
+    # Each sweep below commits in its own transaction, but the queued
+    # user-facing events are fanned out once, at the end. Without the
+    # ``finally``, a transient DB error in a later sweep discards events whose
+    # writes already landed — and the loss is permanent, because every sweep
+    # filters out the rows it just made terminal (``find_and_lock_stalled_
+    # ingests`` looks only at ``status='active'``, ``find_and_lock_stuck_
+    # messages`` only at pending/streaming), so no later tick re-finds them to
+    # re-emit. The user is left with an upload toast spinning on "Training..."
+    # forever. Seen live on 2026-08-21 when an Azure DNS blip took Postgres
+    # unreachable mid-tick.
+    try:
+        return _run_sweeps(engine, summary, events)
+    finally:
+        _publish_events(events)
+
+
+def _run_sweeps(engine, summary: Dict[str, Any], events: list[tuple]) -> Dict[str, Any]:
+    """Run every reconciler sweep, appending user-facing events to ``events``.
+
+    Split out of :func:`run_reconciliation` so the caller can guarantee the
+    queued events are published even when a sweep raises partway through.
+    """
     with engine.begin() as conn:
         repo = ReconciliationRepository(conn)
         pt_repo = PendingToolStateRepository(conn)
@@ -121,8 +143,11 @@ def run_reconciliation() -> Dict[str, Any]:
             repo.mark_tool_call_failed(
                 row["call_id"],
                 error=(
-                    "reconciler: stuck in 'proposed' for >5 min; "
-                    "side effect status unknown"
+                    "reconciler: never advanced past 'proposed' for >5 min. "
+                    "The row is written immediately before the tool runs, so "
+                    "either the tool is still executing or the stream was "
+                    "abandoned before it reported back; in the abandoned case "
+                    "no side effect was committed."
                 ),
             )
             summary["tool_calls_failed"] += 1
@@ -142,8 +167,12 @@ def run_reconciliation() -> Dict[str, Any]:
             repo.mark_tool_call_failed(
                 row["call_id"],
                 error=(
-                    "reconciler: executed-not-confirmed; side effect "
-                    "assumed committed, manual cleanup required"
+                    "reconciler: executed but not confirmed within 15 min. "
+                    "The tool SUCCEEDED and its side effect is committed; "
+                    "confirmation only lands when the whole turn finalizes, "
+                    "so a long-running turn reaches this state while healthy. "
+                    "Check the parent message before assuming cleanup is "
+                    "needed."
                 ),
             )
             summary["tool_calls_failed"] += 1
@@ -328,7 +357,6 @@ def run_reconciliation() -> Dict[str, Any]:
                 },
             )
 
-    _publish_events(events)
     return summary
 
 

application/api/user/tasks.py

@@ -28,12 +28,24 @@ logger = logging.getLogger(__name__)
 # durability story is grep-able next to the body. Combined with
 # ``autoretry_for=(Exception,)`` and a bounded ``max_retries`` so a poison
 # message can't loop forever.
+# ``retry_backoff`` is the factor, NOT a boolean toggle: celery's
+# ``add_autoretry_behaviour`` OVERWRITES ``retry_kwargs["countdown"]`` whenever
+# it is truthy (celery/app/autoretry.py). With the old ``retry_backoff=True``
+# the factor was ``int(max(1.0, True)) == 1``, so the declared 60 s countdown
+# never reached ``task.retry`` and the three waits were jittered 0-1 s, 0-2 s
+# and 0-4 s — a whole retry envelope of at most 7 seconds. A 3.5-minute network
+# blip (2026-08-21) therefore exhausted every attempt and published a terminal
+# ``source.ingest.failed`` / ``attachment.failed`` to the user.
+# Factor 30 gives 30 s / 60 s / 120 s (full-jittered, capped by celery's
+# ``retry_backoff_max`` default of 600 s), which actually spans an outage of
+# that length. ``countdown`` is deliberately absent: leaving it in only
+# documents a value the library discards.
 DURABLE_TASK = dict(
     bind=True,
     acks_late=True,
     autoretry_for=(Exception,),
-    retry_kwargs={"max_retries": 3, "countdown": 60},
-    retry_backoff=True,
+    retry_kwargs={"max_retries": 3},
+    retry_backoff=30,
 )
 
 
@@ -681,10 +693,24 @@ def cleanup_idempotency_dedup(self):
 
 @celery.task(bind=True, acks_late=False)
 def reconciliation_task(self):
-    """Sweep stuck durability rows and escalate them to terminal status + alert."""
+    """Sweep stuck durability rows and escalate them to terminal status + alert.
+
+    Never retries: the task is on a 30 s beat, so the next tick IS the retry,
+    and ``acks_late=False`` keeps a failing run from being redelivered. It only
+    guards itself so a transient connectivity error emits one WARNING instead
+    of a full traceback on the same ERROR channel ``_emit_alert`` uses for real
+    reconciler findings — at 30 s cadence a multi-minute outage would otherwise
+    bury genuine alerts under dozens of identical stack traces.
+    """
     from application.api.user.reconciliation import run_reconciliation
 
-    return run_reconciliation()
+    try:
+        return run_reconciliation()
+    except Exception as exc:  # noqa: BLE001 - housekeeping must never crash the beat loop
+        logging.getLogger(__name__).warning(
+            "reconciliation_task failed; the next beat retries: %s", exc,
+        )
+        return {"messages_failed": 0, "tool_calls_failed": 0, "error": True}
 
 
 @celery.task(bind=True, acks_late=False)

application/storage/db/repositories/conversations.py

@@ -1165,9 +1165,18 @@ class ConversationsRepository:
 
         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.
+        terminal, which leaves an audit trail of what was superseded.
+
+        That stamp does NOT survive: the UPDATE and the DELETE below run the
+        same ``position > :pos`` predicate in one transaction, so every stamped
+        row is removed microseconds later and the owning stream's late
+        ``finalize_message`` correctly reports ``NOT_FOUND``, not
+        ``ALREADY_FAILED``. Do not rely on the stamp for anything but the
+        ``RETURNING id`` log line. A stream superseded less than
+        ``STREAM_HEARTBEAT_INTERVAL`` before it finishes is never cancelled
+        (the ticker in ``api/answer/routes/base.py`` is the only thing that
+        sets the cancel flag), so it runs to completion and raises
+        ``answer_persist_failed`` with an answer nothing can save.
 
         Args:
             conversation_id: Conversation whose tail is being trimmed.

scripts/e2e/bake_template.sh

@@ -9,13 +9,13 @@
 
 set -euo pipefail
 
-PG_BIN="/Users/Shared/DBngin/postgresql/16.2/bin"
+PG_BIN="${PG_BIN:-/Users/Shared/DBngin/postgresql/16.2/bin}"
 PSQL="${PG_BIN}/psql"
 PG_ISREADY="${PG_BIN}/pg_isready"
 
 PG_HOST="127.0.0.1"
 PG_PORT="5432"
-PG_SUPERUSER="postgres"
+PG_SUPERUSER="${PG_SUPERUSER:-postgres}"
 
 TEMPLATE_DB="docsgpt_e2e_template"
 E2E_DB="docsgpt_e2e"

scripts/e2e/mock_llm.py

@@ -72,6 +72,24 @@ STREAM_CHUNK_COUNT = 5
 # module docstring for why hash fixtures cannot serve this purpose.
 EMIT_DIRECTIVE = re.compile(r"\[\[MOCK_LLM_EMIT:([A-Za-z0-9_=\-]+)\]\]")
 
+# In-band directive: ``[[MOCK_LLM_TOOLCALL:<action>:<mode>]]`` makes the stub
+# answer with a tool call instead of content, and controls how the call's
+# ``arguments`` are split across SSE frames. Modes:
+#   ``once``   — one frame carrying the complete arguments (a well-behaved
+#                provider).
+#   ``repeat`` — TWO frames for the same ``index``, each carrying the COMPLETE
+#                arguments. Some OpenAI-compatible gateways restate a short
+#                argument payload on the finish frame rather than sending a
+#                delta; DocsGPT's per-index merge appends, so the two combine
+#                into invalid JSON (``{}`` + ``{}`` -> ``{}{}``).
+#   ``delta``  — arguments split into genuine partial deltas, which is what the
+#                merge's ``+=`` exists to reassemble. The control case.
+# An optional 4th field is a base64url JSON object to send as the arguments;
+# it defaults to ``{}`` (a zero-parameter action such as ``note_view``).
+TOOLCALL_DIRECTIVE = re.compile(
+    r"\[\[MOCK_LLM_TOOLCALL:([A-Za-z0-9_\-]+):(once|repeat|delta)(?::([A-Za-z0-9_=\-]+))?\]\]"
+)
+
 app = Flask(__name__)
 
 
@@ -233,6 +251,49 @@ def _directive_content(messages: list[dict[str, Any]] | None) -> str | None:
     return found
 
 
+def _directive_toolcall(
+    messages: list[dict[str, Any]] | None,
+) -> tuple[str, str, str] | None:
+    """Decoded ``[[MOCK_LLM_TOOLCALL:...]]`` directive, or None.
+
+    Returns:
+        ``(action_name, frame_mode, arguments_json)`` for the last directive
+        found, or ``None`` when the conversation carries none.
+    """
+
+    found: tuple[str, str, str] | None = None
+    for match in TOOLCALL_DIRECTIVE.finditer(_messages_text(messages) or ""):
+        action, mode, raw_args = match.group(1), match.group(2), match.group(3)
+        arguments = "{}"
+        if raw_args:
+            try:
+                padded = raw_args + "=" * (-len(raw_args) % 4)
+                arguments = base64.urlsafe_b64decode(padded.encode("ascii")).decode(
+                    "utf-8"
+                )
+            except (binascii.Error, UnicodeDecodeError, ValueError) as exc:
+                sys.stderr.write(
+                    f"[mock-llm] bad MOCK_LLM_TOOLCALL args {raw_args!r}: {exc}\n"
+                )
+                sys.stderr.flush()
+        found = (action, mode, arguments)
+    return found
+
+
+def _toolcall_arg_frames(arguments: str, mode: str) -> list[str]:
+    """Split ``arguments`` into the per-frame payloads for ``mode``."""
+
+    if mode == "repeat":
+        # The incident shape: the complete payload arrives twice for one index.
+        return [arguments, arguments]
+    if mode == "delta":
+        if len(arguments) < 2:
+            return [arguments]
+        midpoint = len(arguments) // 2
+        return [arguments[:midpoint], arguments[midpoint:]]
+    return [arguments]
+
+
 def _resolve_chat_response(
     payload: dict[str, Any], digest: str
 ) -> tuple[str, list[dict[str, Any]] | None, str, dict[str, int]]:
@@ -243,6 +304,32 @@ def _resolve_chat_response(
     returned and the miss is logged so the dev can convert it.
     """
 
+    # A real provider can only answer with a tool call when the request
+    # actually offered tools. DocsGPT's finalize round deliberately sends
+    # ``tools=None`` to force a text answer, so honouring that here is what
+    # makes the loop terminate the way it does in production.
+    toolcall = _directive_toolcall(payload.get("messages"))
+    if toolcall is not None and payload.get("tools"):
+        action, _mode, arguments = toolcall
+        prompt_tokens = _estimate_tokens(_messages_text(payload.get("messages")))
+        return (
+            "",
+            [
+                {
+                    "index": 0,
+                    "id": f"call_e2e_{digest[:12]}",
+                    "type": "function",
+                    "function": {"name": action, "arguments": arguments},
+                }
+            ],
+            "tool_calls",
+            {
+                "prompt_tokens": prompt_tokens,
+                "completion_tokens": 8,
+                "total_tokens": prompt_tokens + 8,
+            },
+        )
+
     directive = _directive_content(payload.get("messages"))
     if directive is not None:
         prompt_tokens = _estimate_tokens(_messages_text(payload.get("messages")))
@@ -344,6 +431,7 @@ def _stream_chat_response(
     tool_calls: list[dict[str, Any]] | None,
     finish_reason: str,
     chunk_delay_ms: int = 0,
+    toolcall_arg_mode: str | None = None,
 ):
     """Generator yielding SSE frames that match the OpenAI streaming protocol.
 
@@ -378,7 +466,42 @@ def _stream_chat_response(
     # Opening role delta — matches OpenAI's real behavior.
     yield _sse(_base_chunk({"role": "assistant", "content": ""}))
 
-    if tool_calls:
+    if tool_calls and toolcall_arg_mode:
+        # Frame-split mode: the call's ``arguments`` are spread over several
+        # deltas that all share one ``index``, which is what the client-side
+        # merge in application/llm/handlers/base.py reassembles.
+        call = tool_calls[0]
+        frames = _toolcall_arg_frames(call["function"]["arguments"], toolcall_arg_mode)
+        for position, piece in enumerate(frames):
+            _maybe_sleep()
+            if position == 0:
+                delta = {
+                    "tool_calls": [
+                        {
+                            "index": call.get("index", 0),
+                            "id": call.get("id"),
+                            "type": "function",
+                            "function": {
+                                "name": call["function"]["name"],
+                                "arguments": piece,
+                            },
+                        }
+                    ]
+                }
+            else:
+                # Continuation frames carry neither id nor name — only the
+                # index ties them to the call, exactly as OpenAI streams them.
+                delta = {
+                    "tool_calls": [
+                        {
+                            "index": call.get("index", 0),
+                            "function": {"arguments": piece},
+                        }
+                    ]
+                }
+            yield _sse(_base_chunk(delta))
+        yield _sse(_base_chunk({}, final=True))
+    elif tool_calls:
         # Emit tool calls in one delta; content streaming is skipped when
         # tool_calls are present, matching what RAG code paths expect.
         _maybe_sleep()
@@ -454,6 +577,11 @@ def chat_completions() -> Response:
             tool_calls=tool_calls,
             finish_reason=finish_reason,
             chunk_delay_ms=chunk_delay_ms,
+            toolcall_arg_mode=(
+                (_directive_toolcall(payload.get("messages")) or (None, None, None))[1]
+                if payload.get("tools")
+                else None
+            ),
         )
         response = Response(
             stream_with_context(generator),

scripts/e2e/reset_db.sh

@@ -6,13 +6,13 @@
 
 set -euo pipefail
 
-PG_BIN="/Users/Shared/DBngin/postgresql/16.2/bin"
+PG_BIN="${PG_BIN:-/Users/Shared/DBngin/postgresql/16.2/bin}"
 PSQL="${PG_BIN}/psql"
 PG_ISREADY="${PG_BIN}/pg_isready"
 
 PG_HOST="127.0.0.1"
 PG_PORT="5432"
-PG_SUPERUSER="postgres"
+PG_SUPERUSER="${PG_SUPERUSER:-postgres}"
 
 TEMPLATE_DB="docsgpt_e2e_template"
 E2E_DB="docsgpt_e2e"

scripts/e2e/up.sh

@@ -25,7 +25,7 @@ set -euo pipefail
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
 
-PG_BIN="/Users/Shared/DBngin/postgresql/16.2/bin"
+PG_BIN="${PG_BIN:-/Users/Shared/DBngin/postgresql/16.2/bin}"
 DBNGIN_REDIS_BIN="/Users/Shared/DBngin/redis/7.0.0/bin"
 
 # Resolve redis-cli — PATH first, then DBngin's bundled copy.

tests/api/user/test_reconciliation.py

@@ -443,7 +443,10 @@ class TestStuckProposedToolCalls:
             {"id": "cp-1"},
         ).fetchone()
         assert row[0] == "failed"
-        assert "side effect status unknown" in (row[1] or "")
+        # The row is written immediately before the tool runs, so an abandoned
+        # stream leaves no side effect at all — the message must not imply one.
+        assert "never advanced past 'proposed'" in (row[1] or "")
+        assert "no side effect was committed" in (row[1] or "")
 
         assert any(
             getattr(rec, "alert", None) == "reconciler_tool_call_failed_proposed"
@@ -498,8 +501,11 @@ class TestStuckExecutedToolCalls:
             {"id": "ce-1"},
         ).fetchone()
         assert row[0] == "failed"
-        assert "executed-not-confirmed" in (row[1] or "")
-        assert "manual cleanup" in (row[1] or "")
+        # The tool SUCCEEDED here; confirmation only lands when the whole turn
+        # finalizes, so a healthy long turn reaches this state and the message
+        # must not assert that cleanup is required.
+        assert "executed but not confirmed" in (row[1] or "")
+        assert "The tool SUCCEEDED" in (row[1] or "")
         assert any(
             getattr(rec, "alert", None) == "reconciler_tool_call_failed_executed"
             for rec in caplog.records
@@ -950,3 +956,70 @@ class TestPostgresUriMissing:
             "tool_calls_failed": 0,
             "skipped": "POSTGRES_URI not set",
         }
+
+
+# ---------------------------------------------------------------------------
+# Publish-after-commit durability — a sweep that raises must not swallow the
+# events whose DB writes already landed.
+# ---------------------------------------------------------------------------
+
+
+class TestEventsSurviveAFailedSweep:
+    """Regression: 2026-08-21, an Azure DNS blip took Postgres unreachable
+    mid-tick. ``run_reconciliation`` commits each sweep in its own transaction
+    but fans out queued user-facing events once, at the end — so an exception
+    in a later sweep used to discard events for rows that were already
+    terminal. The loss is permanent, because each sweep filters out the rows it
+    just escalated, so no later tick re-finds them.
+    """
+
+    def test_queued_events_are_published_when_a_later_sweep_raises(self, pg_conn):
+        from application.api.user.reconciliation import run_reconciliation
+
+        # A stalled ingest: escalated (and its event queued) by the sweep that
+        # runs before the one we blow up.
+        source_id = "1a000000-0000-0000-0000-0000000000e1"
+        _seed_source(pg_conn, source_id=source_id, user_id="u-blip")
+        _seed_ingest_progress(pg_conn, source_id=source_id, embedded=2, total=50)
+
+        published: list = []
+
+        def _capture(user_id, event_type, payload, *, scope=None):
+            published.append((user_id, event_type, payload, scope))
+            return "1-0"
+
+        calls = {"n": 0}
+
+        @contextmanager
+        def _fake_begin():
+            calls["n"] += 1
+            # Fail a sweep AFTER the ingest sweep has committed its row.
+            if calls["n"] == 6:
+                raise RuntimeError("connection is bad: Network is unreachable")
+            yield pg_conn
+
+        fake_engine = MagicMock()
+        fake_engine.begin = _fake_begin
+
+        with patch(
+            "application.api.user.reconciliation.get_engine",
+            return_value=fake_engine,
+        ), patch(
+            "application.events.publisher.publish_user_event", _capture
+        ):
+            with pytest.raises(RuntimeError):
+                run_reconciliation()
+
+        # The DB write landed...
+        status = pg_conn.execute(
+            text(
+                "SELECT status FROM ingest_chunk_progress "
+                "WHERE source_id = CAST(:sid AS uuid)"
+            ),
+            {"sid": source_id},
+        ).fetchone()[0]
+        assert status == "stalled"
+
+        # ...and so did its event. Without the ``finally`` the user's upload
+        # toast would spin on "Training..." forever.
+        assert [p[1] for p in published] == ["source.ingest.failed"]

tests/e2e/specs/tier-a/failed-turn-replay.spec.ts

@@ -0,0 +1,181 @@
+/**
+ * Failed turns are replayed into the next attempt — why an over-budget
+ * conversation cannot recover.
+ *
+ * Production incident (2026-08-20): a conversation crossed the model's context
+ * window and five further sends over 27 minutes each failed identically. The
+ * compression threshold checker counted 561,822 → 957,474 → 1,353,246 →
+ * 1,748,855 tokens across those attempts — **+395.7k every time, constant to
+ * within 200 tokens** — because each attempt that produced no answer was still
+ * persisted, and then replayed by the next one.
+ *
+ * The mechanism is a single missing predicate.
+ * `ConversationsRepository.get_messages`
+ * (application/storage/db/repositories/conversations.py:633-642) is a bare
+ * `SELECT * FROM conversation_messages ... ORDER BY position` with no `status`
+ * filter, and it is the ONE query behind both consumers: the agent's history
+ * builder and the compression token counter. A `failed` row qualifies for
+ * replay in `BaseAgent._build_messages` because the gate is
+ * `has_completed_turn = "prompt" in i and "response" in i` (base.py:987) — and
+ * the failed row does have a response: the 62-byte
+ * `TERMINATED_RESPONSE_PLACEHOLDER`.
+ *
+ * // Silent-break covered: nothing in the API response reveals this. The proof
+ *    has to come from what the provider actually received, so these specs read
+ *    the mock LLM's own request log rather than the SSE body.
+ *
+ * Note the production rows were five NEW sends, not five retry-button presses:
+ * the retry path passes `index` and calls `truncate_after(keep_up_to=index-1)`,
+ * which deletes the failed row and reuses its position. Five accumulated rows
+ * at five distinct positions can only be fresh appends — which is what the
+ * generic "Please try again later" error trains a user to do.
+ */
+
+import * as playwright from '@playwright/test';
+const { expect, test } = playwright;
+
+import { open, stat } from 'node:fs/promises';
+
+import { authedRequest } from '../../helpers/api.js';
+import { pg } from '../../helpers/db.js';
+import { resetDb } from '../../helpers/reset.js';
+import { newUserContext } from '../../helpers/auth.js';
+import { streamOnce } from '../../helpers/streaming.js';
+
+/** application/api/answer/services/conversation_service.py:32-34 */
+const TERMINATED_PLACEHOLDER =
+  'Response was terminated prior to completion, try regenerating.';
+
+const MOCK_LLM_LOG = process.env.E2E_MOCK_LLM_LOG ?? '/tmp/docsgpt-e2e/mock-llm.log';
+
+async function logSize(path: string): Promise<number> {
+  try {
+    return (await stat(path)).size;
+  } catch {
+    return 0;
+  }
+}
+
+async function logSince(path: string, offset: number): Promise<string> {
+  const handle = await open(path, 'r');
+  try {
+    const size = (await handle.stat()).size;
+    if (size <= offset) return '';
+    const buffer = Buffer.alloc(size - offset);
+    await handle.read(buffer, 0, buffer.length, offset);
+    return buffer.toString('utf8');
+  } finally {
+    await handle.close();
+  }
+}
+
+/**
+ * Append a turn that produced no answer, exactly as the stream's failure path
+ * leaves it: the user's whole question retained in `prompt`, the placeholder
+ * in `response`, `status='failed'`.
+ */
+async function appendFailedTurn(
+  conversationId: string,
+  userId: string,
+  question: string,
+): Promise<void> {
+  await pg.query(
+    `INSERT INTO conversation_messages (
+        conversation_id, position, prompt, response, status, user_id, timestamp
+     )
+     SELECT CAST($1 AS uuid),
+            COALESCE(MAX(position), -1) + 1,
+            $3, $4, 'failed', $2, clock_timestamp()
+       FROM conversation_messages
+      WHERE conversation_id = CAST($1 AS uuid)`,
+    [conversationId, userId, question, TERMINATED_PLACEHOLDER],
+  );
+}
+
+test.describe('failed turns in conversation history', () => {
+  test.beforeEach(async () => {
+    await resetDb();
+  });
+
+  test('a turn that produced no answer is replayed to the model on the next send', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    // Distinctive enough to find verbatim in the provider request log.
+    const deadQuestion = 'DEAD-TURN-MARKER-8f21 summarise the attached dump';
+
+    try {
+      const conversationId = await streamOnce(api, {
+        question: 'first question',
+        save_conversation: true,
+      });
+      await appendFailedTurn(conversationId, sub, deadQuestion);
+
+      const before = await logSize(MOCK_LLM_LOG);
+      await api.post('/stream', {
+        data: {
+          question: 'second question',
+          conversation_id: conversationId,
+          save_conversation: true,
+        },
+      });
+      const sent = await logSince(MOCK_LLM_LOG, before);
+
+      // The dead turn was shipped to the provider — both halves of it: the
+      // user's full question, and the placeholder standing in for an answer
+      // that never existed. It contributes nothing and costs its full length
+      // in every subsequent turn.
+      expect(sent).toContain('DEAD-TURN-MARKER-8f21');
+      expect(sent).toContain('Response was terminated prior to completion');
+    } finally {
+      await api.dispose();
+    }
+  });
+
+  test('each additional failed turn adds its full question to the next request', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    // ~40 KB per dead turn: small enough to keep the spec fast, large enough
+    // that the growth is unambiguous. Production's were ~965 KB each.
+    const filler = 'x'.repeat(40_000);
+
+    try {
+      const conversationId = await streamOnce(api, {
+        question: 'first question',
+        save_conversation: true,
+      });
+
+      const sizes: number[] = [];
+      for (let round = 0; round < 3; round += 1) {
+        await appendFailedTurn(conversationId, sub, `DEAD-${round} ${filler}`);
+        const before = await logSize(MOCK_LLM_LOG);
+        await api.post('/stream', {
+          data: {
+            question: `probe ${round}`,
+            conversation_id: conversationId,
+            save_conversation: true,
+          },
+        });
+        sizes.push((await logSince(MOCK_LLM_LOG, before)).length);
+      }
+
+      // Strictly increasing, by roughly one dead question each round: the
+      // conversation gets bigger every time an attempt fails, so a user who
+      // re-sends after a context-window error is making it worse. This is what
+      // eventually put compression's OWN input past the compressor's window
+      // (application/api/answer/services/compression/orchestrator.py:174-176
+      // compresses every query with no size check), killing the only path that
+      // could have rescued the conversation.
+      expect(sizes[1]).toBeGreaterThan(sizes[0]);
+      expect(sizes[2]).toBeGreaterThan(sizes[1]);
+      expect(sizes[2] - sizes[0]).toBeGreaterThan(60_000);
+    } finally {
+      await api.dispose();
+    }
+  });
+});

tests/e2e/specs/tier-a/resume-claim-lockout.spec.ts

@@ -0,0 +1,181 @@
+/**
+ * Duplicate `tool_actions` resume — the abandoned-claim lockout.
+ *
+ * Production incident (2026-08-19, four occurrences for one brand-new user at
+ * 15:01:17, 15:02:40, 15:13:46 and 15:23:47 UTC — note the ~11 and ~10 minute
+ * spacing). Each attempt to resume a paused tool turn came back as:
+ *
+ *     ERROR /stream - error: Malformed request body - specific error: Resume
+ *     already in progress for this conversation. - traceback: ...
+ *
+ * Three defects compound here:
+ *
+ * 1. `ResumeInProgressError` subclasses `ValueError`
+ *    (application/api/answer/services/continuation_service.py:34), so on
+ *    `/stream` it lands in the generic `except ValueError` handler and is
+ *    reported to the client as **400 "Malformed request body"** and logged at
+ *    ERROR with a full traceback. The correct handling already exists one route
+ *    over — `/v1/chat/completions` catches it explicitly and answers **409**
+ *    `{"type":"conflict_error","code":"resume_in_progress"}`
+ *    (application/api/v1/routes.py:422-433). It was simply never mirrored;
+ *    `git log -S ResumeInProgressError -- .../routes/stream.py` is empty.
+ *
+ * 2. Nothing releases the claim when the client goes away.
+ *    `PendingToolStateRepository.claim_state` flips the row `pending →
+ *    resuming` atomically, and the only recovery is
+ *    `revert_stale_resuming(grace_seconds=600)` from a beat task — so an
+ *    abandoned claim locks the conversation for up to **ten minutes**, which is
+ *    exactly the spacing in the production timestamps.
+ *
+ * 3. The client abandons claims by design: `submitToolActions`
+ *    (frontend/src/conversation/conversationSlice.ts) opens with
+ *    `abortController.abort()` and re-POSTs, and mints a *fresh*
+ *    Idempotency-Key each time — which `/stream` does not read anyway.
+ *
+ * // Silent-break covered: a user who abandons one resume is locked out of
+ *    their own conversation for ten minutes and is told their request was
+ *    malformed. The seeded `status='resuming'` row below is precisely the state
+ *    the app leaves behind when a resume POST is aborted mid-flight.
+ */
+
+import * as playwright from '@playwright/test';
+const { expect, test } = playwright;
+
+import { open, stat } from 'node:fs/promises';
+
+import { authedRequest } from '../../helpers/api.js';
+import { pg } from '../../helpers/db.js';
+import { resetDb } from '../../helpers/reset.js';
+import { newUserContext } from '../../helpers/auth.js';
+import { streamOnce } from '../../helpers/streaming.js';
+
+/**
+ * Insert the row the app itself writes when a turn pauses for a client-side
+ * tool, already claimed — i.e. a resume that was started and then abandoned.
+ */
+async function seedAbandonedClaim(conversationId: string, userId: string): Promise<void> {
+  await pg.query(
+    `INSERT INTO pending_tool_state (
+        conversation_id, user_id, messages, pending_tool_calls,
+        tools_dict, tool_schemas, agent_config, expires_at,
+        status, resumed_at
+     ) VALUES (
+        CAST($1 AS uuid), $2, '[]'::jsonb, '[]'::jsonb,
+        '{}'::jsonb, '[]'::jsonb, '{}'::jsonb,
+        clock_timestamp() + interval '30 minutes',
+        'resuming', clock_timestamp()
+     )`,
+    [conversationId, userId],
+  );
+}
+
+/** Byte length of the Flask log, or 0 when it is not present. */
+async function readLogSize(path: string): Promise<number> {
+  try {
+    return (await stat(path)).size;
+  } catch {
+    return 0;
+  }
+}
+
+/** Everything appended to the Flask log since `offset`. */
+async function readLogSince(path: string, offset: number): Promise<string> {
+  const handle = await open(path, 'r');
+  try {
+    const size = (await handle.stat()).size;
+    if (size <= offset) return '';
+    const buffer = Buffer.alloc(size - offset);
+    await handle.read(buffer, 0, buffer.length, offset);
+    return buffer.toString('utf8');
+  } finally {
+    await handle.close();
+  }
+}
+
+test.describe('duplicate tool_actions resume', () => {
+  test.beforeEach(async () => {
+    await resetDb();
+  });
+
+  test('a resume against an already-claimed conversation is a 409 conflict, not a bad request', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    try {
+      // A real conversation to resume against.
+      const conversationId = await streamOnce(api, {
+        question: 'hello',
+        save_conversation: true,
+      });
+      await seedAbandonedClaim(conversationId, sub);
+
+      const res = await api.post('/stream', {
+        data: {
+          question: '',
+          conversation_id: conversationId,
+          tool_actions: [{ call_id: 'call_abandoned', decision: 'approved' }],
+        },
+      });
+
+      // A conflict, matching what /v1/chat/completions has answered since
+      // 2026-07-13 — not the "Malformed request body" 400 this used to give.
+      expect(res.status()).toBe(409);
+      const text = await res.text();
+      expect(text).not.toContain('Malformed request body');
+      expect(text).toContain('Resume already in progress');
+
+      // The claim is untouched, so every retry inside the 600s grace window
+      // gets the same answer — the ten-minute lockout.
+      const { rows } = await pg.query<{ status: string }>(
+        `SELECT status FROM pending_tool_state WHERE conversation_id = CAST($1 AS uuid)`,
+        [conversationId],
+      );
+      expect(rows).toHaveLength(1);
+      expect(rows[0].status).toBe('resuming');
+    } finally {
+      await api.dispose();
+    }
+  });
+
+  test('the conflict is logged as a warning, without a traceback', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+    const logPath = process.env.E2E_FLASK_LOG ?? '/tmp/docsgpt-e2e/flask.log';
+    const before = await readLogSize(logPath);
+
+    try {
+      const conversationId = await streamOnce(api, {
+        question: 'hello',
+        save_conversation: true,
+      });
+      await seedAbandonedClaim(conversationId, sub);
+
+      await api.post('/stream', {
+        data: {
+          question: '',
+          conversation_id: conversationId,
+          tool_actions: [{ call_id: 'call_abandoned', decision: 'approved' }],
+        },
+      });
+
+      const written = await readLogSince(logPath, before);
+
+      // Racing a resume is an expected concurrency outcome, not a server
+      // fault. Logging it as ERROR-with-traceback is what buried it among real
+      // errors for a full day in production, so the shape of the log line is
+      // part of the contract.
+      expect(written).toContain('resume already in progress for conversation');
+      expect(written).not.toContain(
+        '/stream - error: Malformed request body - specific error: ' +
+          'Resume already in progress',
+      );
+      expect(written).not.toContain('resume_from_tool_actions');
+    } finally {
+      await api.dispose();
+    }
+  });
+});

tests/e2e/specs/tier-a/tool-arg-merge.spec.ts

@@ -0,0 +1,217 @@
+/**
+ * Streaming tool-call argument merge — the "silent tool" failure.
+ *
+ * Production incident (2026-08-19, a brand-new user's first session): every
+ * call to one tool failed to parse for six minutes, the model kept re-emitting
+ * it, the tool never ran once — and every stream still finished `ok`, so the
+ * stream error rate never moved. 22 `tool_call_attempts` rows landed with
+ * `tool_name='unknown'`; the parser logged `Extra data: line 1 column 3
+ * (char 2)`, which is `json.loads` reporting that it consumed a COMPLETE
+ * two-character JSON value (`{}`) and then found more input.
+ *
+ * Mechanism. `LLMHandler.handle_streaming` merges streamed tool-call deltas
+ * keyed by `call.index`, and when both sides are `str` it APPENDS
+ * (application/llm/handlers/base.py:1419-1428). Its `else` branch is commented
+ * "Complete (non-delta) payloads: latest wins" but only fires on a type
+ * mismatch. So a provider that restates a COMPLETE `arguments` payload on a
+ * second frame for the same index gets them concatenated into invalid JSON.
+ * Only OpenAI-compatible providers reach this branch — Anthropic and Google
+ * emit index-less ToolCalls that take the separate "complete call" path at
+ * base.py:1408-1412.
+ *
+ * // Silent-break covered: the tool never runs, yet the turn reports success.
+ *    A regression here is invisible to every error-rate metric we have — the
+ *    only durable evidence is `tool_call_attempts.tool_name = 'unknown'`.
+ *    These specs assert on that row, not on the SSE text.
+ *
+ * The `once` and `delta` cases are the boundary: they prove the frames
+ * themselves are fine and that legitimate partial-delta accumulation — which
+ * is the entire reason the `+=` exists — must keep working. Any fix that
+ * makes `repeat` pass MUST leave those two green.
+ */
+
+import * as playwright from '@playwright/test';
+const { expect, test } = playwright;
+
+import { authedRequest } from '../../helpers/api.js';
+import { pg } from '../../helpers/db.js';
+import { resetDb } from '../../helpers/reset.js';
+import { newUserContext } from '../../helpers/auth.js';
+
+/**
+ * `memory` ships in DEFAULT_CHAT_TOOLS, so `memory_view` is a registered,
+ * runnable action for every user — the point being that this tool COULD have
+ * run. Production's victim was `note_view`, the same shape.
+ */
+const ACTION = 'memory_view';
+
+/** Directive understood by scripts/e2e/mock_llm.py. */
+function toolCallDirective(
+  action: string,
+  mode: 'once' | 'repeat' | 'delta',
+  args?: Record<string, unknown>,
+): string {
+  const base = `[[MOCK_LLM_TOOLCALL:${action}:${mode}`;
+  if (args === undefined) return `${base}]]`;
+  const encoded = Buffer.from(JSON.stringify(args), 'utf8').toString('base64url');
+  return `${base}:${encoded}]]`;
+}
+
+interface AttemptRow {
+  tool_name: string;
+  action_name: string;
+  status: string;
+  error: string | null;
+}
+
+async function attemptsFor(userId: string): Promise<AttemptRow[]> {
+  const { rows } = await pg.query<AttemptRow>(
+    `SELECT tool_name, action_name, status, error
+       FROM tool_call_attempts
+      WHERE user_id = $1
+      ORDER BY attempted_at ASC`,
+    [userId],
+  );
+  return rows;
+}
+
+async function messageStatuses(userId: string): Promise<string[]> {
+  const { rows } = await pg.query<{ status: string }>(
+    `SELECT status FROM conversation_messages WHERE user_id = $1 ORDER BY position ASC`,
+    [userId],
+  );
+  return rows.map((r) => r.status);
+}
+
+test.describe('streaming tool-call argument merge', () => {
+  test.beforeEach(async () => {
+    await resetDb();
+  });
+
+  test('a provider that restates complete arguments makes the tool unrunnable, and the turn still reports success', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    try {
+      const res = await api.post('/stream', {
+        data: {
+          question: `check my memory ${toolCallDirective(ACTION, 'repeat')}`,
+          save_conversation: true,
+        },
+        timeout: 180_000,
+      });
+      expect(res.ok()).toBeTruthy();
+      const body = await res.text();
+
+      // ---- the failure is silent by every signal a user or a dashboard sees
+      expect(body).toContain('"type": "end"');
+      expect(await messageStatuses(sub)).toEqual(['complete']);
+
+      // ---- ...and total by the only signal that records it
+      const attempts = await attemptsFor(sub);
+      expect(attempts.length).toBeGreaterThan(0);
+      expect(attempts.every((a) => a.status === 'failed')).toBe(true);
+      // `tool_name` is 'unknown' because parsing dies before the name is
+      // resolved — the production fingerprint.
+      expect(attempts.every((a) => a.tool_name === 'unknown')).toBe(true);
+      expect(attempts.every((a) => a.action_name === ACTION)).toBe(true);
+      // The tool itself never executed: not one attempt reached the registry.
+      expect(attempts.some((a) => a.tool_name === 'memory')).toBe(false);
+    } finally {
+      await api.dispose();
+    }
+  });
+
+  test('the same call succeeds when the provider sends its arguments once', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    try {
+      const res = await api.post('/stream', {
+        data: {
+          question: `check my memory ${toolCallDirective(ACTION, 'once', { path: '/' })}`,
+          save_conversation: true,
+        },
+        timeout: 180_000,
+      });
+      expect(res.ok()).toBeTruthy();
+
+      const attempts = await attemptsFor(sub);
+      expect(attempts.length).toBeGreaterThan(0);
+      // Resolved to the real tool — so the frames are not the problem.
+      expect(attempts.some((a) => a.tool_name === 'unknown')).toBe(false);
+      expect(attempts[0].tool_name).toBe('memory');
+      expect(attempts[0].action_name).toBe(ACTION);
+    } finally {
+      await api.dispose();
+    }
+  });
+
+  test('legitimate partial-delta arguments still reassemble', async ({ browser }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    try {
+      const res = await api.post('/stream', {
+        data: {
+          question: `check my memory ${toolCallDirective(ACTION, 'delta', { path: '/' })}`,
+          save_conversation: true,
+        },
+        timeout: 180_000,
+      });
+      expect(res.ok()).toBeTruthy();
+
+      const attempts = await attemptsFor(sub);
+      expect(attempts.length).toBeGreaterThan(0);
+      // This is what the `+=` in the merge exists for. A fix for `repeat`
+      // that breaks this case has traded one bug for a worse one.
+      expect(attempts.some((a) => a.tool_name === 'unknown')).toBe(false);
+      expect(attempts[0].tool_name).toBe('memory');
+    } finally {
+      await api.dispose();
+    }
+  });
+
+  test('a registered tool with unparseable arguments is re-asked every round, up to the iteration cap', async ({
+    browser,
+  }) => {
+    const { sub, token } = await newUserContext(browser);
+    const api = await authedRequest(playwright, token);
+
+    try {
+      const res = await api.post('/stream', {
+        data: {
+          question: `check my memory ${toolCallDirective(ACTION, 'repeat')}`,
+          save_conversation: true,
+        },
+        timeout: 180_000,
+      });
+      expect(res.ok()).toBeTruthy();
+      const body = await res.text();
+
+      // The repeat-failure guard (`UNRESOLVABLE_CALL_LIMIT`,
+      // application/agents/tool_executor.py) deliberately does NOT apply here:
+      // it is gated on `llm_name not in self._name_to_tool`, i.e. invented tool
+      // names only, because "a registered tool whose arguments merely failed to
+      // decode is recoverable, and refusing it would strand a working tool for
+      // the rest of the turn". `memory_view` is registered, so every round gets
+      // the plain error and another chance.
+      expect(body).toContain('its arguments were not a valid JSON object');
+      expect(body).not.toContain('has already failed');
+
+      // That reasoning holds when the model can fix its own output. It cannot
+      // here — the arguments are mangled downstream of the model by the merge,
+      // so the retry is unwinnable and the only thing that ends the turn is
+      // MAX_TOOL_ITERATIONS (application/llm/handlers/base.py:17). Every one of
+      // these 25 rounds is a billed provider call that could never succeed.
+      const attempts = await attemptsFor(sub);
+      expect(attempts.length).toBe(25);
+    } finally {
+      await api.dispose();
+    }
+  });
+});

tests/test_celery.py

@@ -40,3 +40,65 @@ def test_celeryconfig_durability_defaults():
     # Project-scoped queue prevents a sibling worker on the same broker
     # from grabbing DocsGPT tasks.
     assert celeryconfig.task_default_queue == "docsgpt"
+
+
+@pytest.mark.unit
+def test_durable_task_retry_envelope_spans_a_multi_minute_outage():
+    """``retry_backoff`` is a FACTOR, not a boolean toggle.
+
+    Celery's ``add_autoretry_behaviour`` overwrites
+    ``retry_kwargs["countdown"]`` whenever ``retry_backoff`` is truthy
+    (celery/app/autoretry.py). With ``retry_backoff=True`` the factor is
+    ``int(max(1.0, True)) == 1``, so the declared 60 s countdown never reached
+    ``task.retry`` and the three waits were jittered 0-1 s, 0-2 s and 0-4 s —
+    a whole envelope of at most 7 seconds, which a 3.5-minute network blip
+    (2026-08-21) exhausted immediately. On exhaustion these tasks publish a
+    terminal ``source.ingest.failed`` / ``attachment.failed``, so the user sees
+    a permanent failure caused by a transient one.
+    """
+    from unittest.mock import MagicMock
+
+    from celery.app.autoretry import add_autoretry_behaviour
+
+    from application.api.user.tasks import DURABLE_TASK
+
+    assert DURABLE_TASK["retry_backoff"] == 30
+    # Leaving a ``countdown`` in would only document a value celery discards.
+    assert "countdown" not in DURABLE_TASK["retry_kwargs"]
+
+    seen: list[dict] = []
+
+    class _Task:
+        max_retries = DURABLE_TASK["retry_kwargs"]["max_retries"]
+
+        def __init__(self) -> None:
+            self.request = MagicMock()
+
+        def run(self):
+            raise ConnectionError("Network is unreachable")
+
+        def retry(self, **kwargs):
+            seen.append(dict(kwargs))
+            return RuntimeError("retry-sentinel")
+
+    task = _Task()
+    add_autoretry_behaviour(
+        task,
+        autoretry_for=DURABLE_TASK["autoretry_for"],
+        retry_kwargs=dict(DURABLE_TASK["retry_kwargs"]),
+        retry_backoff=DURABLE_TASK["retry_backoff"],
+    )
+
+    ceilings = []
+    for attempt in range(DURABLE_TASK["retry_kwargs"]["max_retries"]):
+        task.request.retries = attempt
+        with pytest.raises(RuntimeError, match="retry-sentinel"):
+            task.run()
+        ceilings.append(30 * 2**attempt)
+        # full jitter: the wait is drawn from [0, factor * 2**retries].
+        assert 0 <= seen[-1]["countdown"] <= ceilings[-1]
+
+    # 30 / 60 / 120 — an envelope that can actually outlast a multi-minute
+    # blip, versus the 1 / 2 / 4 the old settings produced.
+    assert ceilings == [30, 60, 120]
+    assert sum(ceilings) >= 200