From 29f661f3c7a0a102e3d749c94ca3fb6c1a47c29d Mon Sep 17 00:00:00 2001 From: Alex Date: Sat, 22 Aug 2026 12:57:58 +0100 Subject: [PATCH] fix: more test fixes and additions, fix loss on resume --- application/alembic/env.py | 8 +- application/api/answer/routes/answer.py | 17 ++ application/api/answer/routes/base.py | 44 ++- application/api/user/reconciliation.py | 29 +- application/api/user/tasks.py | 52 +++- application/parser/embedding_pipeline.py | 8 +- .../storage/db/repositories/conversations.py | 42 ++- frontend/src/conversation/AnswerFlow.tsx | 6 +- .../src/conversation/ConversationBubble.tsx | 4 + frontend/src/conversation/MarkdownAnswer.tsx | 21 +- .../src/conversation/sandboxLinks.test.ts | 60 ++++ frontend/src/conversation/sandboxLinks.ts | 31 ++- tests/api/answer/routes/test_answer.py | 41 +++ tests/api/answer/test_base_routes.py | 64 +++++ tests/api/user/test_reconciliation.py | 15 +- tests/api/user/test_tasks.py | 35 +++ tests/core/test_logging_config.py | 62 +++++ .../tier-a/resume-retry-persists.spec.ts | 259 ++++++++++++++++++ tests/e2e/specs/tier-b/streaming-ux.spec.ts | 36 ++- .../test_reconciliation_liveness.py | 50 ++++ tests/test_celery.py | 94 +++++++ tests/test_coverage_gaps.py | 3 +- tests/test_remaining_coverage.py | 13 +- .../test_pgvector_schema_and_pool.py | 12 + 24 files changed, 921 insertions(+), 85 deletions(-) create mode 100644 tests/e2e/specs/tier-a/resume-retry-persists.spec.ts 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