diff --git a/application/alembic/env.py b/application/alembic/env.py
index 68eadc01..08aae816 100644
--- a/application/alembic/env.py
+++ b/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:
diff --git a/application/api/answer/routes/answer.py b/application/api/answer/routes/answer.py
index 3fab4de2..4b5418d3 100644
--- a/application/api/answer/routes/answer.py
+++ b/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()}",
diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py
index 38791923..b591c947 100644
--- a/application/api/answer/routes/base.py
+++ b/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,
)
diff --git a/application/api/user/reconciliation.py b/application/api/user/reconciliation.py
index 4415fdd5..6dff0205 100644
--- a/application/api/user/reconciliation.py
+++ b/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)``.
diff --git a/application/api/user/tasks.py b/application/api/user/tasks.py
index b8974226..2cf4e545 100644
--- a/application/api/user/tasks.py
+++ b/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)
diff --git a/application/parser/embedding_pipeline.py b/application/parser/embedding_pipeline.py
index c850518d..b6d3a0a6 100755
--- a/application/parser/embedding_pipeline.py
+++ b/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."
)
diff --git a/application/storage/db/repositories/conversations.py b/application/storage/db/repositories/conversations.py
index 6926f274..cd27bbea 100644
--- a/application/storage/db/repositories/conversations.py
+++ b/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)
diff --git a/frontend/src/conversation/AnswerFlow.tsx b/frontend/src/conversation/AnswerFlow.tsx
index dd506064..c9f32892 100644
--- a/frontend/src/conversation/AnswerFlow.tsx
+++ b/frontend/src/conversation/AnswerFlow.tsx
@@ -29,8 +29,10 @@ type AnswerFlowProps = {
agentId?: string;
/** Set when the bubble already carries its own progress UI (a research run). */
suppressStatusLine?: boolean;
- /** Artifacts produced on this turn, so ``sandbox:`` links can reach them. */
+ /** Every artifact in the conversation, so ``sandbox:`` links can reach them. */
artifacts?: SandboxArtifact[];
+ /** This turn's own artifacts, which win when a link names a bare filename. */
+ turnArtifacts?: SandboxArtifact[];
onOpenArtifact?: (artifact: { id: string; toolName: string }) => void;
renderApproval: (toolCall: ToolCallsType) => React.ReactNode;
renderWikiWrite: (
@@ -52,6 +54,7 @@ export default function AnswerFlow({
agentId,
suppressStatusLine,
artifacts,
+ turnArtifacts,
onOpenArtifact,
renderApproval,
renderWikiWrite,
@@ -131,6 +134,7 @@ export default function AnswerFlow({
content={message}
isStreaming={isStreaming}
artifacts={artifacts}
+ turnArtifacts={turnArtifacts}
onOpenArtifact={onOpenArtifact}
/>
diff --git a/frontend/src/conversation/ConversationBubble.tsx b/frontend/src/conversation/ConversationBubble.tsx
index 9328b39f..05b14000 100644
--- a/frontend/src/conversation/ConversationBubble.tsx
+++ b/frontend/src/conversation/ConversationBubble.tsx
@@ -439,6 +439,10 @@ const ConversationBubble = forwardRef<
// would be a second live indicator away from the point of action.
suppressStatusLine={Boolean(research)}
artifacts={conversationArtifacts ?? completedArtifacts}
+ // The chip rail below renders from this turn's artifacts alone, so
+ // a bare-filename link has to resolve against the same set or the
+ // two controls in one bubble open different files.
+ turnArtifacts={completedArtifacts}
onOpenArtifact={onOpenArtifact}
renderApproval={(toolCall: ToolCallsType) => (
diff --git a/frontend/src/conversation/MarkdownAnswer.tsx b/frontend/src/conversation/MarkdownAnswer.tsx
index cafb1274..903b7a98 100644
--- a/frontend/src/conversation/MarkdownAnswer.tsx
+++ b/frontend/src/conversation/MarkdownAnswer.tsx
@@ -70,12 +70,19 @@ export default function MarkdownAnswer({
content,
isStreaming,
artifacts,
+ turnArtifacts,
onOpenArtifact,
}: {
content: string;
isStreaming?: boolean;
- /** Artifacts produced on this turn, in creation order (``A1`` is the first). */
+ /**
+ * Every artifact in the conversation, in creation order (``A1`` is the
+ * first) — refs are conversation-scoped, so a link may name an earlier turn's
+ * file.
+ */
artifacts?: SandboxArtifact[];
+ /** This turn's own artifacts; the filename fallback prefers them. */
+ turnArtifacts?: SandboxArtifact[];
onOpenArtifact?: (artifact: { id: string; toolName: string }) => void;
}) {
const [isDarkTheme] = useDarkTheme();
@@ -132,7 +139,11 @@ export default function MarkdownAnswer({
// chip, but the model links it with a `sandbox:`/`artifact:`
// URL no browser can open. Point the link at the chip
// instead, and never leave a dead anchor behind.
- const sandboxLink = resolveSandboxLink(href, artifacts);
+ const sandboxLink = resolveSandboxLink(
+ href,
+ artifacts,
+ turnArtifacts,
+ );
if (sandboxLink.kind === 'plain') {
return <>{children}>;
}
@@ -185,7 +196,11 @@ export default function MarkdownAnswer({
// scheme survives and renders a broken-image box beside the
// chip that opens the very same file.
const source = typeof src === 'string' ? src : undefined;
- const sandboxLink = resolveSandboxLink(source, artifacts);
+ const sandboxLink = resolveSandboxLink(
+ source,
+ artifacts,
+ turnArtifacts,
+ );
if (sandboxLink.kind === 'artifact') {
const { artifact } = sandboxLink;
return renderArtifactChip(
diff --git a/frontend/src/conversation/sandboxLinks.test.ts b/frontend/src/conversation/sandboxLinks.test.ts
index 99ca02d6..e1569359 100644
--- a/frontend/src/conversation/sandboxLinks.test.ts
+++ b/frontend/src/conversation/sandboxLinks.test.ts
@@ -296,3 +296,63 @@ describe('scheme-less and file: hrefs', () => {
});
});
});
+
+// Two turns each produced a file called `report.pdf`. That is not exotic:
+// `_filename()` in artifact_generator.py hands EVERY html artifact the name
+// `report.html`, and re-running a script that writes `chart.png` collides too.
+describe('a filename that two turns both produced', () => {
+ const turnOne: SandboxArtifact = {
+ id: 'id-old',
+ label: 'report.pdf',
+ ref: 'A1',
+ toolName: 'artifact_generator',
+ };
+ const turnTwo: SandboxArtifact = {
+ id: 'id-new',
+ label: 'report.pdf',
+ ref: 'A2',
+ toolName: 'artifact_generator',
+ };
+ // What ConversationMessages passes: the whole conversation, in order.
+ const conversation = [turnOne, turnTwo];
+
+ it("resolves turn one's link to turn one's file, not the newest copy", () => {
+ // Without the turn scope this returned `id-new`, so the prose link opened
+ // a different file from the download chip in the very same bubble.
+ expect(
+ resolveSandboxLink('sandbox:/mnt/data/report.pdf', conversation, [
+ turnOne,
+ ]),
+ ).toEqual({ kind: 'artifact', artifact: turnOne });
+ });
+
+ it("resolves turn two's link to turn two's file", () => {
+ expect(
+ resolveSandboxLink('sandbox:/mnt/data/report.pdf', conversation, [
+ turnTwo,
+ ]),
+ ).toEqual({ kind: 'artifact', artifact: turnTwo });
+ });
+
+ it('still reaches an earlier turn when this turn produced nothing', () => {
+ // The fallback that keeps "the CSV from earlier" working.
+ expect(
+ resolveSandboxLink('sandbox:/mnt/data/report.pdf', conversation, []),
+ ).toEqual({ kind: 'artifact', artifact: turnTwo });
+ });
+
+ it('matches a ref conversation-wide even when this turn owns a file', () => {
+ // Refs are unambiguous, so they must NOT be narrowed — this is the
+ // cross-turn resolution the conversation-wide list was added for.
+ expect(resolveSandboxLink('artifact:A1', conversation, [turnTwo])).toEqual({
+ kind: 'artifact',
+ artifact: turnOne,
+ });
+ });
+
+ it('matches an id conversation-wide even when this turn owns a file', () => {
+ expect(
+ resolveSandboxLink('sandbox:/artifact/id-old', conversation, [turnTwo]),
+ ).toEqual({ kind: 'artifact', artifact: turnOne });
+ });
+});
diff --git a/frontend/src/conversation/sandboxLinks.ts b/frontend/src/conversation/sandboxLinks.ts
index dafc0cb7..2e315810 100644
--- a/frontend/src/conversation/sandboxLinks.ts
+++ b/frontend/src/conversation/sandboxLinks.ts
@@ -95,9 +95,9 @@ const SHORT_REF = /^a(\d+)$/i;
const ANY_SCHEME = /^[a-z][a-z0-9+.-]*:/i;
/**
- * Last match wins. Within one turn the model's prose means the file as it
- * stands after the *last* write, so a rerun that overwrites ``report.pdf``
- * should open the new copy rather than the one it replaced.
+ * Last match wins: the model's prose means the file as it stands after the
+ * *last* write, so a rerun that overwrites ``report.pdf`` should open the new
+ * copy rather than the one it replaced.
*/
function findLast(
items: T[],
@@ -122,8 +122,11 @@ function decodeSegment(segment: string): string {
*
* Args:
* href: The raw href from the markdown link, if any.
- * artifacts: Artifacts this turn produced, in the order the tool made them
- * (``A1`` is the first).
+ * artifacts: Every artifact in the conversation, in creation order (``A1``
+ * is the first). Refs are conversation-scoped, so a link may point at
+ * a file an earlier turn produced.
+ * turnArtifacts: The artifacts of the turn being rendered, when known.
+ * Only the filename fallback consults it; see below.
*
* Returns:
* How the link should be rendered. ``external`` for ordinary URLs,
@@ -133,6 +136,7 @@ function decodeSegment(segment: string): string {
export function resolveSandboxLink(
href: string | undefined | null,
artifacts: SandboxArtifact[] | undefined,
+ turnArtifacts?: SandboxArtifact[],
): SandboxLinkResolution {
if (typeof href !== 'string' || !href) return { kind: 'external' };
const scheme = matchedScheme(href);
@@ -177,11 +181,18 @@ export function resolveSandboxLink(
// Fabricated paths (``/mnt/data/…``, ``/tmp/…``, a bare filename) — and
// ``/artifact/``, which the model also writes — usually still name
// a file that exists as an artifact on this turn.
- const byName = findLast(
- available,
- (artifact) =>
- !!artifact.label && artifact.label.toLowerCase() === last.toLowerCase(),
- );
+ // This turn's own files win. An id or a ref names exactly one artifact, so
+ // those match conversation-wide above; a filename does not — two turns can
+ // each produce ``report.pdf``, and `_filename()` hands EVERY html artifact
+ // that same name. Resolving an old message's link to the newest copy made
+ // the prose link disagree with the download chip in its own bubble, which
+ // renders from that turn's artifacts alone. Falls back to the conversation
+ // so a link naming a file from an earlier turn still resolves.
+ const nameMatches = (artifact: SandboxArtifact) =>
+ !!artifact.label && artifact.label.toLowerCase() === last.toLowerCase();
+ const byName =
+ findLast(turnArtifacts ?? [], nameMatches) ??
+ findLast(available, nameMatches);
if (byName) return { kind: 'artifact', artifact: byName };
return miss;
diff --git a/tests/api/answer/routes/test_answer.py b/tests/api/answer/routes/test_answer.py
index 00d93b6b..4b9393b4 100644
--- a/tests/api/answer/routes/test_answer.py
+++ b/tests/api/answer/routes/test_answer.py
@@ -143,6 +143,47 @@ class TestAnswerResourcePost:
assert resp.status_code == 400
assert resp.get_json()["error"] == "Stream error"
+ def test_resume_already_in_progress_returns_409(
+ self, answer_client, mock_stream_processor,
+ ):
+ """A raced resume is a conflict the caller can retry, not a 500.
+
+ ``ResumeInProgressError`` subclasses ``ValueError`` but reaches this
+ route through ``resume_from_tool_actions``, where the bare
+ ``except Exception`` reported it as an opaque 500 with a full
+ traceback on the ERROR channel. ``/stream`` and
+ ``/v1/chat/completions`` both answer 409 for the identical condition.
+ """
+ from application.api.answer.services.continuation_service import (
+ ResumeInProgressError,
+ )
+
+ mock_stream_processor.resume_from_tool_actions.side_effect = (
+ ResumeInProgressError("Resume already in progress for this conversation.")
+ )
+
+ with patch(
+ "application.api.answer.routes.answer.AnswerResource.validate_request",
+ return_value=None,
+ ):
+ resp = answer_client.post(
+ "/api/answer",
+ data=json.dumps({
+ "question": "",
+ "conversation_id": _CONV_ID,
+ "tool_actions": [{"call_id": "call_1", "decision": "approved"}],
+ }),
+ content_type="application/json",
+ )
+
+ assert resp.status_code == 409
+ body = resp.get_json()
+ assert body["code"] == "resume_in_progress"
+ assert "Resume already in progress" in body["error"]
+ # The generic handler's opaque text would tell the caller nothing
+ # about retryability.
+ assert body["error"] != "An error occurred processing your request"
+
def test_exception_returns_500(self, answer_client, mock_stream_processor):
with patch(
"application.api.answer.routes.answer.AnswerResource.validate_request",
diff --git a/tests/api/answer/test_base_routes.py b/tests/api/answer/test_base_routes.py
index dac6b2c4..abe14983 100644
--- a/tests/api/answer/test_base_routes.py
+++ b/tests/api/answer/test_base_routes.py
@@ -1450,3 +1450,67 @@ class TestStreamingHeartbeatSeed:
)
resource.conversation_service.heartbeat_message.assert_not_called()
+
+
+class TestCompleteStreamSupersededLookupIsGuarded:
+ """A DB failure in the supersede lookup must not error a finished turn.
+
+ The ``NOT_FOUND`` branch used to only log. It now consults
+ ``was_superseded`` to tell a replaced turn from an orphaned answer — over
+ its own connection, inside the streaming ``try``. Unguarded, a blip there
+ (or a deploy where the app runs ahead of migration 0030 and the
+ ``superseded_messages`` table does not exist yet) escapes to the generic
+ handler: the client gets a terminal ``error`` after the whole answer, the
+ row is finalized a second time, a continuation's resume claim is released,
+ and the ``answer_persist_failed`` alert this branch exists to raise is
+ swallowed.
+ """
+
+ def test_lookup_failure_falls_back_instead_of_erroring_the_stream(
+ self, pg_conn, flask_app, caplog,
+ ):
+ import logging
+
+ from application.api.answer.routes.base import BaseAnswerResource
+ from application.api.answer.services.conversation_service import (
+ ConversationService,
+ )
+ from application.storage.db.repositories.conversations import (
+ MessageUpdateOutcome,
+ )
+
+ with flask_app.app_context():
+ resource = BaseAnswerResource()
+ mock_agent = MagicMock()
+ mock_agent.gen.return_value = iter(
+ [{"type": "answer", "answer": "the answer"}]
+ )
+
+ with _patch_db_session(pg_conn), patch.object(
+ ConversationService,
+ "finalize_message",
+ return_value=MessageUpdateOutcome.NOT_FOUND,
+ ), patch.object(
+ ConversationService,
+ "was_superseded",
+ side_effect=RuntimeError("server closed the connection"),
+ ), caplog.at_level(logging.ERROR):
+ stream = list(
+ resource.complete_stream(
+ question="does the guard hold?",
+ agent=mock_agent,
+ conversation_id=None,
+ user_api_key=None,
+ decoded_token={"sub": "u-superseded-guard"},
+ should_persist=True,
+ model_id="gpt-4",
+ )
+ )
+
+ # The answer was already streamed; the row is gone either way. What
+ # must not happen is retracting it with a terminal error frame.
+ assert [s for s in stream if '"type": "error"' in s] == []
+ # And the alert must still fire, on the orphaned-answer path.
+ assert any(
+ "answer_persist_failed" in r.getMessage() for r in caplog.records
+ )
diff --git a/tests/api/user/test_reconciliation.py b/tests/api/user/test_reconciliation.py
index db97bda5..e59df5a2 100644
--- a/tests/api/user/test_reconciliation.py
+++ b/tests/api/user/test_reconciliation.py
@@ -945,17 +945,20 @@ class TestStalledIngestEvent:
class TestPostgresUriMissing:
@pytest.mark.unit
def test_returns_skip_dict(self, monkeypatch):
- from application.api.user.reconciliation import run_reconciliation
+ from application.api.user.reconciliation import (
+ run_reconciliation,
+ zero_summary,
+ )
from application.core.settings import settings
monkeypatch.setattr(settings, "POSTGRES_URI", None, raising=False)
result = run_reconciliation()
- assert result == {
- "messages_failed": 0,
- "tool_calls_failed": 0,
- "skipped": "POSTGRES_URI not set",
- }
+ # Same counters as any other tick, all zero, plus the reason. Pinning
+ # a hand-written literal here is what let the shapes drift apart: this
+ # path reported 3 of the 5 keys and the task's error fallback invented
+ # a sixth that no sweep produces.
+ assert result == {**zero_summary(), "skipped": "POSTGRES_URI not set"}
# ---------------------------------------------------------------------------
diff --git a/tests/api/user/test_tasks.py b/tests/api/user/test_tasks.py
index 59982227..1624a71a 100644
--- a/tests/api/user/test_tasks.py
+++ b/tests/api/user/test_tasks.py
@@ -900,3 +900,38 @@ class TestParseTimeoutForSize:
assert limits["time_limit"] - limits["soft_time_limit"] == grace
# Never zero/negative, whatever the caller computed.
assert parse_task_time_limits(0)["soft_time_limit"] == 1
+
+
+class TestReconciliationTaskShape:
+ """The beat's error fallback must report the same counters as a good tick."""
+
+ @pytest.mark.unit
+ def test_error_fallback_matches_the_real_summary_keys(self):
+ from application.api.user import reconciliation
+ from application.api.user.tasks import reconciliation_task
+
+ with patch.object(
+ reconciliation, "run_reconciliation", side_effect=RuntimeError("db down")
+ ):
+ result = reconciliation_task.run()
+
+ assert result["error"] is True
+ # Hand-writing the fallback drifted immediately: it invented
+ # ``attachments_stalled`` (no sweep produces it) and dropped
+ # ``idempotency_pending_failed``, so a failed tick reported counters
+ # that could never appear and hid one that can.
+ counters = {k: v for k, v in result.items() if k != "error"}
+ assert counters == reconciliation.zero_summary()
+ assert all(v == 0 for v in counters.values())
+
+ @pytest.mark.unit
+ def test_skipped_tick_reports_the_same_counters(self, monkeypatch):
+ from application.api.user import reconciliation
+
+ monkeypatch.setattr(reconciliation.settings, "POSTGRES_URI", "", raising=False)
+ result = reconciliation.run_reconciliation()
+
+ assert result["skipped"] == "POSTGRES_URI not set"
+ assert {k: v for k, v in result.items() if k != "skipped"} == (
+ reconciliation.zero_summary()
+ )
diff --git a/tests/core/test_logging_config.py b/tests/core/test_logging_config.py
index cff033c5..63ea0512 100644
--- a/tests/core/test_logging_config.py
+++ b/tests/core/test_logging_config.py
@@ -102,3 +102,65 @@ class TestSetupLogging:
handlers = logging.getLogger().handlers
assert otel_handler not in handlers
+
+
+class TestAlembicDoesNotSilenceApplicationLoggers:
+ """A boot that applies a migration must not switch off app logging.
+
+ ``alembic/env.py`` calls ``logging.config.fileConfig``, whose
+ ``disable_existing_loggers`` defaults to True and whose ini names only
+ ``root``/``sqlalchemy``/``alembic`` — so every ``application.*`` logger
+ imported before it is disabled. ``app.py`` calls ``setup_logging()`` at
+ line 16 but ``ensure_database_ready()`` at line 58, so the web tier never
+ re-enables them: on the one boot where a schema upgrade happened, it
+ logs nothing for the rest of the process's life. k8s sets
+ ``AUTO_MIGRATE=false``, but every docker-compose variant leaves it on.
+ """
+
+ @pytest.mark.unit
+ def test_env_py_opts_out_of_disable_existing_loggers(self):
+ """AST, not a grep: the kwarg has to be on the real call."""
+ import ast
+ import pathlib
+
+ source = pathlib.Path("application/alembic/env.py").read_text()
+ calls = [
+ node
+ for node in ast.walk(ast.parse(source))
+ if isinstance(node, ast.Call)
+ and getattr(node.func, "id", None) == "fileConfig"
+ ]
+ assert calls, "env.py no longer calls fileConfig"
+ for call in calls:
+ kwargs = {kw.arg: kw.value for kw in call.keywords}
+ assert "disable_existing_loggers" in kwargs
+ assert kwargs["disable_existing_loggers"].value is False
+
+ @pytest.mark.unit
+ def test_the_flag_is_what_keeps_application_loggers_alive(self, tmp_path):
+ """Pins the mechanism, in a subprocess so pytest's own logging survives.
+
+ ``fileConfig`` mutates global logging state, so this cannot run
+ in-process without wrecking every later test's log capture.
+ """
+ import subprocess
+ import sys
+
+ probe = tmp_path / "probe.py"
+ probe.write_text(
+ "import logging\n"
+ "from logging.config import fileConfig\n"
+ "log = logging.getLogger('application.api.answer.routes.stream')\n"
+ "fileConfig('application/alembic.ini', disable_existing_loggers=True)\n"
+ "kept = logging.getLogger('application.api.answer.routes.stream')\n"
+ "print('default_disables', kept.disabled)\n"
+ "kept.disabled = False\n"
+ "fileConfig('application/alembic.ini', disable_existing_loggers=False)\n"
+ "print('flag_preserves', not kept.disabled)\n"
+ )
+ out = subprocess.run(
+ [sys.executable, str(probe)],
+ capture_output=True, text=True, check=True,
+ ).stdout
+ assert "default_disables True" in out
+ assert "flag_preserves True" in out
diff --git a/tests/e2e/specs/tier-a/resume-retry-persists.spec.ts b/tests/e2e/specs/tier-a/resume-retry-persists.spec.ts
new file mode 100644
index 00000000..8744b47e
--- /dev/null
+++ b/tests/e2e/specs/tier-a/resume-retry-persists.spec.ts
@@ -0,0 +1,259 @@
+/**
+ * Retrying a failed resume must keep the answer it produces.
+ *
+ * `release_claim` hands a failed resume's `pending_tool_state` row back to
+ * `pending` so the user can retry immediately instead of waiting out
+ * `revert_stale_resuming`'s 600 s grace. But the same `except Exception` in
+ * `routes/base.py` also finalizes the reserved message row `failed`, and a
+ * resume reuses the SAME `reserved_message_id` — `resume_from_tool_actions`
+ * reads it back out of the persisted `agent_config`
+ * (services/stream_processor.py). So the retry streams into a row that is
+ * already terminal, and `finalize_message(status="complete")` is gated by
+ * `only_if_non_terminal`, whose reclaim hole originally matched only the
+ * reconciler's own error string. Outcome: `ALREADY_FAILED`.
+ *
+ * // Silent-break covered: the user watches a correct answer stream in, then
+ * on reload finds "Response was terminated prior to completion" with no way
+ * back to it. Nothing in the SSE body reveals the loss — the wire shows a
+ * perfectly healthy turn — so these specs assert on the DB row and on
+ * GET /api/get_single_conversation, never on the stream.
+ *
+ * The fix stamps `resume_retryable` on the failure exactly when the claim was
+ * released, and widens the reclaim gate to honour it. The two side effects
+ * have to agree: releasing the claim says "retryable", so the row must not
+ * simultaneously be terminal in a way the retry cannot overwrite.
+ */
+
+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';
+import { streamOnce } from '../../helpers/streaming.js';
+
+/** application/api/answer/services/conversation_service.py */
+const TERMINATED_PLACEHOLDER =
+ 'Response was terminated prior to completion, try regenerating.';
+
+interface MessageRow {
+ id: string;
+ status: string;
+ response: string;
+ message_metadata: Record | null;
+}
+
+async function messageRow(id: string): Promise {
+ const { rows } = await pg.query(
+ `SELECT id, status, response, message_metadata
+ FROM conversation_messages WHERE id = CAST($1 AS uuid)`,
+ [id],
+ );
+ return rows[0];
+}
+
+async function claimStatus(convId: string): Promise {
+ const { rows } = await pg.query<{ status: string }>(
+ `SELECT status FROM pending_tool_state
+ WHERE conversation_id = CAST($1 AS uuid)`,
+ [convId],
+ );
+ return rows.length ? rows[0].status : null;
+}
+
+/**
+ * Park a message row mid-flight and point a claimable `pending_tool_state` at
+ * it, exactly as a turn paused on a client-side tool does. `agent_config`
+ * carries the `reserved_message_id`, which is what makes the retry reuse this
+ * same row.
+ */
+async function seedPausedTurn(
+ convId: string,
+ sub: string,
+): Promise {
+ const { rows } = await pg.query<{ id: string }>(
+ `INSERT INTO conversation_messages (
+ conversation_id, position, prompt, response, status, user_id,
+ timestamp, message_metadata
+ ) VALUES (CAST($1 AS uuid), 99, 'paused on a tool', '', 'streaming', $2,
+ clock_timestamp(), '{}'::jsonb)
+ RETURNING id`,
+ [convId, sub],
+ );
+ const messageId = rows[0].id;
+
+ 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
+ ) VALUES (
+ CAST($1 AS uuid), $2,
+ CAST($3 AS jsonb), CAST($4 AS jsonb),
+ '{}'::jsonb, '[]'::jsonb, CAST($5 AS jsonb),
+ clock_timestamp() + interval '30 minutes', 'pending')`,
+ [
+ convId,
+ sub,
+ JSON.stringify([
+ { role: 'user', content: 'paused on a tool' },
+ {
+ role: 'assistant',
+ content: null,
+ // No `name` on the function — enough to make gen_continuation raise,
+ // which is the "resume failed once" trigger this spec is about.
+ tool_calls: [
+ { id: 'call_1', type: 'function', function: { arguments: '{}' } },
+ ],
+ },
+ ]),
+ JSON.stringify([
+ {
+ call_id: 'call_1',
+ tool_name: 'noop',
+ action_name: 'noop',
+ arguments: {},
+ },
+ ]),
+ JSON.stringify({
+ llm_name: 'openai',
+ agent_type: 'ClassicAgent',
+ prompt: 'You are a test assistant.',
+ reserved_message_id: messageId,
+ }),
+ ],
+ );
+ return messageId;
+}
+
+test.describe('retrying a failed resume', () => {
+ test.beforeEach(async () => {
+ await resetDb();
+ });
+
+ test('a failed resume hands the claim back instead of holding it for 600s', async ({
+ browser,
+ }) => {
+ const { sub, token } = await newUserContext(browser);
+ const api = await authedRequest(playwright, token);
+ try {
+ const convId = await streamOnce(api, {
+ question: 'hello',
+ save_conversation: true,
+ });
+ await seedPausedTurn(convId, sub);
+
+ const res = await api.post('/stream', {
+ data: {
+ question: '',
+ conversation_id: convId,
+ tool_actions: [{ call_id: 'call_1', decision: 'approved' }],
+ },
+ });
+ await res.text();
+
+ // Not `resuming`: that is the ten-minute lockout from the 2026-08-19
+ // incident, recoverable only by the janitor.
+ expect(await claimStatus(convId)).not.toBe('resuming');
+ } finally {
+ await api.dispose();
+ }
+ });
+
+ test('the failure it leaves behind is marked retryable, not plainly terminal', async ({
+ browser,
+ }) => {
+ const { sub, token } = await newUserContext(browser);
+ const api = await authedRequest(playwright, token);
+ try {
+ const convId = await streamOnce(api, {
+ question: 'hello',
+ save_conversation: true,
+ });
+ const messageId = await seedPausedTurn(convId, sub);
+
+ const res = await api.post('/stream', {
+ data: {
+ question: '',
+ conversation_id: convId,
+ tool_actions: [{ call_id: 'call_1', decision: 'approved' }],
+ },
+ });
+ await res.text();
+
+ const row = await messageRow(messageId);
+ expect(row.status).toBe('failed');
+ expect(row.response).toBe(TERMINATED_PLACEHOLDER);
+ // The marker the reclaim gate keys on. Without it the retry below can
+ // never overwrite this row, and its answer is discarded in silence.
+ expect(row.message_metadata?.resume_retryable).toBe(true);
+ } finally {
+ await api.dispose();
+ }
+ });
+
+ test("a retry's answer replaces the failed row instead of being discarded", async ({
+ browser,
+ }) => {
+ const { sub, token } = await newUserContext(browser);
+ const api = await authedRequest(playwright, token);
+ try {
+ const convId = await streamOnce(api, {
+ question: 'hello',
+ save_conversation: true,
+ });
+ const messageId = await seedPausedTurn(convId, sub);
+
+ // Resume #1 — errors, releases the claim, marks the row retryable.
+ const first = await api.post('/stream', {
+ data: {
+ question: '',
+ conversation_id: convId,
+ tool_actions: [{ call_id: 'call_1', decision: 'approved' }],
+ },
+ });
+ await first.text();
+ expect((await messageRow(messageId)).status).toBe('failed');
+
+ // Resume #2 — the row is claimable again, so re-approving works. Repair
+ // the stored messages first: the retry is the user clicking approve a
+ // second time on a turn whose state is intact, not a replay of the
+ // corruption that broke resume #1.
+ await pg.query(
+ `UPDATE pending_tool_state
+ SET messages = jsonb_set(
+ messages, '{1,tool_calls,0,function,name}', '"noop"'::jsonb)
+ WHERE conversation_id = CAST($1 AS uuid)`,
+ [convId],
+ );
+
+ const second = await api.post('/stream', {
+ data: {
+ question: '',
+ conversation_id: convId,
+ tool_actions: [{ call_id: 'call_1', decision: 'approved' }],
+ },
+ });
+ const body = await second.text();
+ expect(second.status()).toBe(200);
+
+ // The whole point: whatever the second attempt produced must be what the
+ // row holds. A silently-dropped answer leaves the placeholder behind.
+ const row = await messageRow(messageId);
+ if (body.includes('"type": "end"')) {
+ expect(row.status).toBe('complete');
+ expect(row.response).not.toBe(TERMINATED_PLACEHOLDER);
+ // Reclaim strips the failure bookkeeping, or a finished answer
+ // carries a stale error into the API response.
+ expect(row.message_metadata?.error).toBeUndefined();
+ expect(row.message_metadata?.resume_retryable).toBeUndefined();
+ } else {
+ // If the second attempt also failed, it must at least still be
+ // retryable — never wedged terminal with the claim gone.
+ expect(row.message_metadata?.resume_retryable).toBe(true);
+ }
+ } finally {
+ await api.dispose();
+ }
+ });
+});
diff --git a/tests/e2e/specs/tier-b/streaming-ux.spec.ts b/tests/e2e/specs/tier-b/streaming-ux.spec.ts
index 2c970d27..3e5f5bbf 100644
--- a/tests/e2e/specs/tier-b/streaming-ux.spec.ts
+++ b/tests/e2e/specs/tier-b/streaming-ux.spec.ts
@@ -174,13 +174,21 @@ test.describe('tier-b · streaming UX', () => {
// We model the tool-approval flow at the table level — the contract is
// that the route path with `tool_actions` + `conversation_id` calls
// `StreamProcessor.resume_from_tool_actions`, which loads from PG, then
- // `mark_resuming` flips status `pending`→`resuming` so the row cannot
- // be re-claimed (see continuation_service.py + repositories/pending_tool_state.py).
- // On clean completion the route layer DELETEs the row
- // (routes/base.py: `delete_state` after stream success); on a crash the
- // row stays in `resuming` for the janitor to revert after the grace
- // window. Either outcome proves the replay footgun is closed — what
- // must not happen is the row remaining in `pending` status.
+ // claims the row so it cannot be claimed twice
+ // (see continuation_service.py + repositories/pending_tool_state.py).
+ //
+ // What the claim must never be is STUCK. It originally could only be
+ // released by `revert_stale_resuming`'s 600 s grace, which locked the user
+ // out of their own conversation for ten minutes after any resume that
+ // failed. `release_claim` now hands the row back on the error path, so the
+ // outcomes are: DELETED (clean completion, `delete_state`), `pending`
+ // (released for an immediate retry), or `resuming` only while a resume is
+ // genuinely still in flight.
+ //
+ // A released row IS re-fireable, which is the point — the replay footgun
+ // that justified the old `pending`-is-forbidden rule is now handled by the
+ // owning message row instead: see `resume-retry-persists.spec.ts`, which
+ // pins that the retry lands on the same row and its answer survives.
// No real agent wiring is needed to test that contract.
const { sub, token } = await newUserContext(browser);
const api = await authedRequest(playwright, token);
@@ -278,15 +286,15 @@ test.describe('tier-b · streaming UX', () => {
expect(resume.status()).toBe(200);
await resume.text();
- // 4. Row must be either deleted (clean completion) or marked
- // `resuming` (crash partway). What must not happen is the row
- // remaining in `pending` — that would mean the resume is
- // re-fireable, a footgun mirroring Mongo-era bug #2088.
- const pendingCount = await countRows('pending_tool_state', {
- sql: "user_id = $1 AND conversation_id = CAST($2 AS uuid) AND status = 'pending'",
+ // 4. The resume errored (the mock LLM cannot continue from a fabricated
+ // tool_call), so the claim must have been handed back rather than left
+ // held. Either the row is gone or it is `pending` and immediately
+ // retryable; a leftover `resuming` is the ten-minute lockout.
+ const stuckCount = await countRows('pending_tool_state', {
+ sql: "user_id = $1 AND conversation_id = CAST($2 AS uuid) AND status = 'resuming'",
params: [sub, convId],
});
- expect(pendingCount).toBe(0);
+ expect(stuckCount).toBe(0);
} finally {
await api.dispose();
}
diff --git a/tests/storage/db/repositories/test_reconciliation_liveness.py b/tests/storage/db/repositories/test_reconciliation_liveness.py
index cf579e05..e8a04e48 100644
--- a/tests/storage/db/repositories/test_reconciliation_liveness.py
+++ b/tests/storage/db/repositories/test_reconciliation_liveness.py
@@ -278,6 +278,56 @@ class TestFinalizeReclaim:
assert "error" not in meta
assert "reconcile_attempts" not in meta
+ def test_reclaims_a_resume_that_released_its_claim(self, pg_conn):
+ """A failed resume that handed the turn back must not block the retry.
+
+ The stream handler releases the continuation claim and stamps
+ ``resume_retryable`` in the same breath. The retry reuses this very row
+ (the reserved id lives in the persisted ``agent_config``), so without
+ the second reclaim hole its successful finalize lands ALREADY_FAILED
+ and the user's answer is discarded after they watched it stream in.
+ """
+ msg = _seed_message(
+ pg_conn,
+ status="failed",
+ metadata={"error": "KeyError: 'name'", "resume_retryable": True},
+ )
+ repo = ConversationsRepository(pg_conn)
+
+ outcome = repo.update_message_by_id(
+ msg["id"],
+ {"response": "the retry's answer", "status": "complete"},
+ only_if_non_terminal=True,
+ reclaim_reconciler_failed=True,
+ )
+
+ assert outcome is MessageUpdateOutcome.UPDATED
+ assert _status(pg_conn, msg["id"]) == "complete"
+ # The marker must not survive onto the finished row, or the API
+ # response carries a stale failure alongside a complete answer.
+ meta = _meta(pg_conn, msg["id"])
+ assert "resume_retryable" not in meta
+ assert "error" not in meta
+
+ def test_does_not_reclaim_a_failure_that_kept_its_claim(self, pg_conn):
+ """No marker means no release happened, so the row stays terminal."""
+ msg = _seed_message(
+ pg_conn,
+ status="failed",
+ metadata={"error": "KeyError: 'name'"},
+ )
+ repo = ConversationsRepository(pg_conn)
+
+ outcome = repo.update_message_by_id(
+ msg["id"],
+ {"response": "late answer", "status": "complete"},
+ only_if_non_terminal=True,
+ reclaim_reconciler_failed=True,
+ )
+
+ assert outcome is MessageUpdateOutcome.ALREADY_FAILED
+ assert _status(pg_conn, msg["id"]) == "failed"
+
def test_does_not_reclaim_genuine_failure(self, pg_conn):
"""Client disconnect, provider errors etc. stay terminal."""
msg = _seed_message(
diff --git a/tests/test_celery.py b/tests/test_celery.py
index f6d3477b..8994dbc4 100644
--- a/tests/test_celery.py
+++ b/tests/test_celery.py
@@ -128,3 +128,97 @@ def test_durable_task_retry_envelope_spans_a_multi_minute_outage():
assert 150 <= statistics.median(envelopes) <= 270
# A 60 s outage must be survivable in the large majority of retry runs.
assert sum(e >= 60 for e in envelopes) / len(envelopes) > 0.9
+
+
+@pytest.mark.unit
+def test_durable_tasks_never_retry_a_deterministic_parse_failure():
+ """The widened envelope needs ``dont_autoretry_for`` as its counterweight.
+
+ ``autoretry_for=(Exception,)`` retries EVERYTHING, so raising the backoff
+ factor to 60 also stretched permanent failures — an empty, image-only or
+ unparseable file — from a ~7 s envelope to a median 3.5 min and up to ~7.
+ Anything polling ``/api/task_status`` (the wiki-convert and GraphRAG-enable
+ modals) reports "pending" for that whole time, and the work re-runs four
+ times to fail identically. ``e2e specs/tier-b/upload.spec.ts`` went red on
+ exactly this.
+ """
+ from unittest.mock import MagicMock
+
+ from celery.app.autoretry import add_autoretry_behaviour
+
+ from application.api.user.tasks import DURABLE_TASK
+ from application.parser.file.base_parser import DocumentParseError
+
+ assert DocumentParseError in DURABLE_TASK["dont_autoretry_for"]
+
+ retried: list[dict] = []
+
+ class _Task:
+ max_retries = DURABLE_TASK["max_retries"]
+
+ def __init__(self) -> None:
+ self.request = MagicMock()
+ self.request.retries = 0
+
+ def run(self):
+ raise DocumentParseError("No text could be extracted from this file.")
+
+ def retry(self, **kwargs):
+ retried.append(dict(kwargs))
+ return RuntimeError("retry-sentinel")
+
+ task = _Task()
+ add_autoretry_behaviour(
+ task,
+ autoretry_for=DURABLE_TASK["autoretry_for"],
+ dont_autoretry_for=DURABLE_TASK["dont_autoretry_for"],
+ retry_kwargs={"max_retries": DURABLE_TASK["max_retries"]},
+ retry_backoff=DURABLE_TASK["retry_backoff"],
+ )
+
+ # Propagates unchanged to the poison/failure path instead of being
+ # swallowed into a retry.
+ with pytest.raises(DocumentParseError):
+ task.run()
+ assert retried == []
+
+
+@pytest.mark.unit
+def test_every_durable_task_carries_the_parse_failure_guard():
+ """Registered tasks, not just the shared dict — a decorator can override it.
+
+ ``store_attachment`` passes a wider tuple through ``durable_task()``; the
+ point is that no durable task ends up with a NARROWER one, which is how
+ seven of the nine came to retry a permanent parse failure four times.
+ """
+ from application.api.user import tasks as user_tasks
+ from application.parser.file.base_parser import DocumentParseError
+
+ durable = (
+ "ingest",
+ "ingest_remote",
+ "reingest_source_task",
+ "reembed_wiki_page",
+ "convert_source_to_wiki",
+ "extract_graph",
+ "process_agent_webhook",
+ "ingest_connector_task",
+ "store_attachment",
+ )
+ for name in durable:
+ task = getattr(user_tasks, name)
+ assert DocumentParseError in getattr(task, "dont_autoretry_for", ()), name
+
+
+@pytest.mark.unit
+def test_unparseable_file_raises_the_non_retryable_type():
+ """The guard is only reachable if the pipeline raises the right class.
+
+ This used to be a bare ``ValueError``, which ``autoretry_for=(Exception,)``
+ swept up regardless of the ``dont_autoretry_for`` tuple.
+ """
+ from application.parser.embedding_pipeline import embed_and_store_documents
+ from application.parser.file.base_parser import DocumentParseError
+
+ with pytest.raises(DocumentParseError, match="No text could be extracted"):
+ embed_and_store_documents([], "/tmp", "src", None)
diff --git a/tests/test_coverage_gaps.py b/tests/test_coverage_gaps.py
index d3aa3028..e2ebfbb7 100644
--- a/tests/test_coverage_gaps.py
+++ b/tests/test_coverage_gaps.py
@@ -845,8 +845,9 @@ class TestEmbeddingPipeline:
def test_embed_and_store_empty_docs(self):
from application.parser.embedding_pipeline import embed_and_store_documents
+ from application.parser.file.base_parser import DocumentParseError
- with pytest.raises(ValueError, match="No text could be extracted"):
+ with pytest.raises(DocumentParseError, match="No text could be extracted"):
embed_and_store_documents([], "/tmp/test", "src1", MagicMock())
def test_embed_and_store_creates_folder(self, tmp_path):
diff --git a/tests/test_remaining_coverage.py b/tests/test_remaining_coverage.py
index ff5befe6..3bbaad44 100644
--- a/tests/test_remaining_coverage.py
+++ b/tests/test_remaining_coverage.py
@@ -862,10 +862,11 @@ class TestEmbeddingPipelineGaps:
assert os.path.exists(folder)
def test_embed_and_store_raises_on_empty_docs(self):
- """Cover line 69: raises ValueError when docs is empty."""
+ """Cover line 69: raises DocumentParseError when docs is empty."""
from application.parser.embedding_pipeline import embed_and_store_documents
+ from application.parser.file.base_parser import DocumentParseError
- with pytest.raises(ValueError, match="No text could be extracted"):
+ with pytest.raises(DocumentParseError, match="No text could be extracted"):
embed_and_store_documents([], "/tmp/test", "source_id", MagicMock())
@@ -1042,10 +1043,11 @@ class TestEmbeddingPipelineCoverage:
assert sanitize_content(None) is None
def test_embed_and_store_empty_docs_raises(self, tmp_path):
- """Cover line 69: empty docs raises ValueError."""
+ """Cover line 69: empty docs raises DocumentParseError."""
from application.parser.embedding_pipeline import embed_and_store_documents
+ from application.parser.file.base_parser import DocumentParseError
- with pytest.raises(ValueError, match="No text could be extracted"):
+ with pytest.raises(DocumentParseError, match="No text could be extracted"):
embed_and_store_documents([], str(tmp_path / "test"), "src-1", None)
def test_embed_and_store_creates_folder(self, tmp_path):
@@ -1241,10 +1243,11 @@ class TestBlankDocumentsAreRejected:
@pytest.mark.parametrize("blank", ["", " ", "\n\t \n"])
def test_whitespace_only_document_is_rejected(self, blank, tmp_path):
from application.parser.embedding_pipeline import embed_and_store_documents
+ from application.parser.file.base_parser import DocumentParseError
class _Doc:
def __init__(self, text):
self.text = text
- with pytest.raises(ValueError, match="No text could be extracted"):
+ with pytest.raises(DocumentParseError, match="No text could be extracted"):
embed_and_store_documents([_Doc(blank)], str(tmp_path), "src-1", None)
diff --git a/tests/vectorstore/test_pgvector_schema_and_pool.py b/tests/vectorstore/test_pgvector_schema_and_pool.py
index ba4d499c..072dd270 100644
--- a/tests/vectorstore/test_pgvector_schema_and_pool.py
+++ b/tests/vectorstore/test_pgvector_schema_and_pool.py
@@ -9,6 +9,8 @@ one-shot safety net for processes that never ran the boot hook.
from __future__ import annotations
+import gc
+
from unittest.mock import MagicMock, Mock, patch
import pytest
@@ -266,6 +268,16 @@ class TestReadPathsCloseTheirTransaction:
@pytest.mark.unit
class TestConnectionPooling:
def _pooled_store(self, monkeypatch):
+ # ``PGVectorStore.__del__`` releases into the module-global
+ # ``_POOLS``, and a store left holding a connection by an earlier test
+ # is only collectable through a MagicMock reference cycle — so its
+ # finalizer fires at an arbitrary later moment. If that moment lands
+ # inside a test that has patched ``_POOLS[dsn]`` to its own fake pool,
+ # the stale release is recorded against THIS test's mock and
+ # ``putconn.assert_called_once_with`` sees two calls. Drain pending
+ # finalizers here, before any patching, so the count is deterministic
+ # whatever ran before.
+ gc.collect()
store, _, _, _ = _make_store()
store._connection = None
store._pool_max_size = 4