fix: more test fixes and additions, fix loss on resume

application/alembic/env.py

@@ -28,7 +28,13 @@ if settings.POSTGRES_URI:
     config.set_main_option("sqlalchemy.url", settings.POSTGRES_URI)
 
 if config.config_file_name is not None:
-    fileConfig(config.config_file_name)
+    # ``disable_existing_loggers`` defaults to True, which would switch off
+    # every ``application.*`` logger already imported by the time migrations
+    # run. ``app.py`` calls ``setup_logging()`` (line 16) BEFORE
+    # ``ensure_database_ready()`` (line 58), so the web tier never
+    # re-enables them and loses all application logging for the life of the
+    # process — on exactly the boot where a schema upgrade happened.
+    fileConfig(config.config_file_name, disable_existing_loggers=False)
 
 
 def run_migrations_offline() -> None:

application/api/answer/routes/answer.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
 
@@ -164,6 +167,20 @@ class AnswerResource(Resource, BaseAnswerResource):
             extra_info = stream_result.get("extra")
             if extra_info:
                 result.update(extra_info)
+        except ResumeInProgressError as e:
+            # Another request already owns this conversation's continuation
+            # claim. Same contract as ``/stream`` and ``/v1/chat/completions``:
+            # an expected concurrency outcome the caller can retry, not a
+            # server fault, so it must not land in the 500 handler below and
+            # fill the ERROR channel with tracebacks.
+            logger.warning(
+                "/api/answer - resume already in progress for conversation %s",
+                data.get("conversation_id"),
+                extra={"error": str(e)},
+            )
+            return make_response(
+                {"error": str(e), "code": "resume_in_progress"}, 409,
+            )
         except Exception as e:
             logger.error(
                 f"/api/answer - error: {str(e)} - traceback: {traceback.format_exc()}",

application/api/answer/routes/base.py

@@ -1054,9 +1054,27 @@ class BaseAnswerResource:
                         # that sets it lives in the superseding request's own
                         # process. That is routine and must not page anyone.
                         # Anything else is a genuinely orphaned answer.
-                        superseded = self.conversation_service.was_superseded(
-                            reserved_message_id
-                        )
+                        # Guarded: this opens its own DB connection, and an
+                        # unguarded raise here would escape to the generic
+                        # ``except`` below — which yields a terminal ``error``
+                        # frame, re-finalizes, and releases the resume claim,
+                        # all for a row that is already gone. It would also
+                        # swallow the very alert this branch exists to raise.
+                        # A missing ``superseded_messages`` table (code
+                        # deployed ahead of ``alembic upgrade head``) lands
+                        # here too, so the fallback is the pre-tombstone
+                        # behaviour rather than a user-visible failure.
+                        try:
+                            superseded = self.conversation_service.was_superseded(
+                                reserved_message_id
+                            )
+                        except Exception:
+                            logger.exception(
+                                "was_superseded lookup failed for %s; "
+                                "treating the row as orphaned",
+                                reserved_message_id,
+                            )
+                            superseded = False
                         if superseded:
                             logger.info(
                                 "stream superseded: message row %s was replaced "
@@ -1415,9 +1433,10 @@ class BaseAnswerResource:
             # invisible, and every retry inside the window gets another 409.
             # Not on the ``StreamSuperseded`` path above — there the turn was
             # deliberately replaced and the row is legitimately gone.
+            claim_released = False
             if _continuation and conversation_id:
                 try:
-                    ContinuationService().release_claim(
+                    claim_released = ContinuationService().release_claim(
                         str(conversation_id),
                         decoded_token.get("sub", "local"),
                     )
@@ -1428,6 +1447,21 @@ class BaseAnswerResource:
                         exc_info=True,
                     )
             if reserved_message_id is not None:
+                # Releasing the claim above says "this turn is retryable"; the
+                # row must not simultaneously be stamped terminally failed in a
+                # way the retry cannot overwrite. A resume reuses the SAME
+                # ``reserved_message_id`` (stream_processor stores it in the
+                # persisted ``agent_config``), so the retry's own
+                # ``finalize_message(status="complete")`` is gated by
+                # ``only_if_non_terminal`` and lands ``ALREADY_FAILED`` — the
+                # user watches a correct answer stream in and finds
+                # "Response was terminated prior to completion" on reload.
+                # Mark the failure retryable so the reclaim hole in
+                # ``update_message_by_id`` lets that second answer through,
+                # exactly as it already does for the reconciler's own marker.
+                failure_metadata = dict(query_metadata or {})
+                if claim_released:
+                    failure_metadata["resume_retryable"] = True
                 try:
                     self.conversation_service.finalize_message(
                         reserved_message_id,
@@ -1436,7 +1470,7 @@ class BaseAnswerResource:
                         sources=source_log_docs,
                         tool_calls=tool_calls,
                         model_id=model_id or self.default_model_id,
-                        metadata=query_metadata if query_metadata else None,
+                        metadata=failure_metadata or None,
                         status="failed",
                         error=e,
                     )

application/api/user/reconciliation.py

@@ -30,6 +30,21 @@ logger = logging.getLogger(__name__)
 MAX_MESSAGE_RECONCILE_ATTEMPTS = 3
 
 
+def zero_summary() -> Dict[str, int]:
+    """The counter keys every tick reports, all at zero.
+
+    Single source of truth so the early-return and the caller's error
+    fallback cannot drift from the sweeps that actually populate them.
+    """
+    return {
+        "messages_failed": 0,
+        "tool_calls_failed": 0,
+        "ingests_stalled": 0,
+        "idempotency_pending_failed": 0,
+        "schedule_runs_failed": 0,
+    }
+
+
 def run_reconciliation() -> Dict[str, Any]:
     """Single tick of the reconciler. Five sweeps, FOR UPDATE SKIP LOCKED.
 
@@ -42,20 +57,10 @@ def run_reconciliation() -> Dict[str, Any]:
     sitting in ``pending`` until 24 h TTL.
     """
     if not settings.POSTGRES_URI:
-        return {
-            "messages_failed": 0,
-            "tool_calls_failed": 0,
-            "skipped": "POSTGRES_URI not set",
-        }
+        return {**zero_summary(), "skipped": "POSTGRES_URI not set"}
 
     engine = get_engine()
-    summary = {
-        "messages_failed": 0,
-        "tool_calls_failed": 0,
-        "ingests_stalled": 0,
-        "idempotency_pending_failed": 0,
-        "schedule_runs_failed": 0,
-    }
+    summary: Dict[str, Any] = zero_summary()
     # User-facing events to fan out once their DB writes have committed
     # (publish-after-commit). Each item is
     # ``(user_id, event_type, payload, scope)``.

application/api/user/tasks.py

@@ -60,15 +60,40 @@ logger = logging.getLogger(__name__)
 # ``retry_kwargs["countdown"]`` into it on every retry, so one shared dict
 # spread across all the decorators below would let concurrent retries race
 # and would permanently mutate this module constant.
+#
+# ``dont_autoretry_for`` is the necessary counterweight to that widened
+# envelope. ``autoretry_for=(Exception,)`` retries EVERYTHING, but a
+# ``DocumentParseError`` (unparseable, empty, or image-only file) fails
+# identically on every attempt — with factor 60 it now burns a median 3.5 min
+# and up to ~7 before reaching a terminal state, during which anything polling
+# ``/api/task_status`` (the wiki-convert and GraphRAG-enable modals) just keeps
+# reporting "pending". It is the DEFAULT here so a new durable task cannot
+# forget it; tasks needing a wider set pass their own via ``durable_task()``.
 DURABLE_TASK = dict(
     bind=True,
     acks_late=True,
     autoretry_for=(Exception,),
+    dont_autoretry_for=(DocumentParseError,),
     max_retries=3,
     retry_backoff=60,
 )
 
 
+def durable_task(**overrides) -> Dict:
+    """Return ``DURABLE_TASK`` with per-task overrides applied.
+
+    Needed because ``@celery.task(**DURABLE_TASK, dont_autoretry_for=...)``
+    would be a duplicate keyword argument.
+
+    Args:
+        **overrides: Task options replacing the shared defaults.
+
+    Returns:
+        The merged options dict, for ``@celery.task(**durable_task(...))``.
+    """
+    return {**DURABLE_TASK, **overrides}
+
+
 # operation tag for the poison-path source.ingest.failed event, per task.
 _INGEST_POISON_OPERATION = {
     "ingest": "upload",
@@ -103,10 +128,11 @@ def _emit_ingest_poison_event(task_name, bound):
     )
 
 
-# ``dont_autoretry_for``: a file that cannot be converted to text fails
-# identically on every attempt, so retrying only multiplies the log noise
-# before the same failure — go straight to the poison/failure path.
-@celery.task(**DURABLE_TASK, dont_autoretry_for=(DocumentParseError,))
+# ``dont_autoretry_for=(DocumentParseError,)`` now comes from ``DURABLE_TASK``:
+# a file that cannot be converted to text fails identically on every attempt,
+# so retrying only multiplies the log noise before the same failure — go
+# straight to the poison/failure path.
+@celery.task(**DURABLE_TASK)
 @with_idempotency(task_name="ingest", on_poison=_emit_ingest_poison_event)
 def ingest(
     self,
@@ -285,8 +311,9 @@ def _emit_attachment_poison_event(task_name, bound):
 # deterministic — retrying re-fails identically and multiplies log noise, so it
 # goes straight to the failure path.
 @celery.task(
-    **DURABLE_TASK,
-    dont_autoretry_for=(DataError, AttachmentRejectedError, DocumentParseError),
+    **durable_task(
+        dont_autoretry_for=(DataError, AttachmentRejectedError, DocumentParseError),
+    )
 )
 @with_idempotency(
     task_name="store_attachment", on_poison=_emit_attachment_poison_event,
@@ -722,7 +749,7 @@ def reconciliation_task(self):
     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
+    from application.api.user.reconciliation import run_reconciliation, zero_summary
 
     try:
         return run_reconciliation()
@@ -733,14 +760,9 @@ def reconciliation_task(self):
         # SUCCESS because the exception is swallowed here, so this log is the
         # only channel that carries the stack.
         logger.exception("reconciliation_task failed; the next beat retries")
-        return {
-            "messages_failed": 0,
-            "tool_calls_failed": 0,
-            "ingests_stalled": 0,
-            "attachments_stalled": 0,
-            "schedule_runs_failed": 0,
-            "error": True,
-        }
+        # Keys come from ``zero_summary`` so a failed tick reports the same
+        # shape as a successful one; hand-writing them drifted immediately.
+        return {**zero_summary(), "error": True}
 
 
 @celery.task(bind=True, acks_late=False)

application/parser/embedding_pipeline.py

@@ -5,6 +5,7 @@ from retry import retry
 from tqdm import tqdm
 from application.core.settings import settings
 from application.events.publisher import publish_user_event
+from application.parser.file.base_parser import DocumentParseError
 from application.storage.db.repositories.ingest_chunk_progress import (
     IngestChunkProgressRepository,
 )
@@ -204,7 +205,12 @@ def embed_and_store_documents(
         if str(getattr(d, "text", getattr(d, "page_content", d)) or "").strip()
     ]
     if not docs:
-        raise ValueError(
+        # ``DocumentParseError``, not ``ValueError``: an empty or image-only
+        # file yields nothing on every attempt, so this must reach the Celery
+        # tasks' ``dont_autoretry_for`` and fail once. As a bare ``ValueError``
+        # it was swept up by ``autoretry_for=(Exception,)`` and re-failed
+        # identically across the whole backoff envelope.
+        raise DocumentParseError(
             "No text could be extracted from this file. It may be empty, "
             "image-only, or in an unsupported format."
         )

application/storage/db/repositories/conversations.py

@@ -896,16 +896,23 @@ 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.
+        ``reclaim_reconciler_failed`` punches two narrow holes in that gate,
+        both for rows whose ``failed`` stamp a later successful answer
+        disproves:
+
+        1. A row the **reconciler** failed for staleness. 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". Matched on the reconciler's own error string, and
+           skipping rows whose awaiting-approval prompt it already revoked.
+        2. A row marked ``resume_retryable`` by the stream handler, which sets
+           it only when it also released the continuation claim — i.e. it
+           explicitly handed the turn back for a retry that reuses this same
+           row id.
+
+        Everything else stays terminal: genuine failures (client disconnect,
+        provider errors, the pending-tool cleanup task) are unaffected.
 
         Reclaiming also strips the reconciler's bookkeeping keys, which
         otherwise survive the metadata merge and leave a ``complete`` row
@@ -922,7 +929,7 @@ class ConversationsRepository:
             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.
+                for staleness, and rows a failed resume marked retryable.
 
         Returns:
             MessageUpdateOutcome: What happened to the row.
@@ -946,6 +953,7 @@ class ConversationsRepository:
         # does this fire" query.
         reclaim_strip = (
             " - 'error' - 'reconcile_attempts' - 'last_reconcile_seen_heartbeat'"
+            " - 'resume_retryable'"
             if reclaim_reconciler_failed
             else ""
         )
@@ -1002,6 +1010,18 @@ class ConversationsRepository:
                     "AND COALESCE("
                     "(message_metadata->>'reconciler_cleared_approval')::boolean,"
                     " false) = false"
+                    ") OR ("
+                    # Second hole, same principle: the stream handler that
+                    # failed a resume also handed the turn back by releasing
+                    # the continuation claim, so it stamped the row
+                    # ``resume_retryable``. The retry reuses this very row
+                    # (the reserved id lives in the persisted agent_config);
+                    # without this it would land ALREADY_FAILED and the
+                    # user's second, successful answer would be discarded.
+                    "status = 'failed' "
+                    "AND COALESCE("
+                    "(message_metadata->>'resume_retryable')::boolean,"
+                    " false) = true"
                     "))"
                 )
             update_where.append(terminal_gate)