From 780d910c1e67bc66586524e3d80d5be6dab296a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 01:08:13 +0000 Subject: [PATCH 01/16] fix(attachments): index reparsed attachment content into the content graph apply_reparsed_result landed a fresh classification result onto the Attachment row but never indexed the recognized content into the content graph, unlike the initial email-import path (_append_email_content_graph). A previously-quarantined attachment that later reparses to "parsed" therefore stayed invisible to content-graph-backed search/AI-hub features even after successful recognition. Flagged as informational by Devin Review on naruon#1486, confirmed real but out of scope there, and closed here as the tracked follow-up. apply_reparsed_result now calls a new _append_reparsed_attachment_content_graph whenever the reparse result lands on "parsed". It reuses the same services.content_graph.parse_content helper the import path already calls, plus a newly shared content_graph_source_record_uid (promoted from a private function in email_import_service.py to a public helper in services/content_graph/parser.py that both call sites import) -- one indexing path, one identity convention, two callers. Since a persisted attachment's original position among its email's siblings is not reliably reproducible post-import, the reparse path keys source_record_uid on the attachment's permanent attachment_uid alone instead of the import path's message-id + list-position convention, and sets the new records' email_id directly from the attachment's already-loaded email_id column rather than through a transient Email relationship append. Updates docs/adr/0005-attachment-content-type-quarantine.md's Revisions section and CHANGELOG.md per repo convention. --- CHANGELOG.md | 25 ++++++ backend/services/attachment_reparse_worker.py | 86 ++++++++++++++++++- backend/services/content_graph/__init__.py | 3 +- backend/services/content_graph/parser.py | 13 +++ backend/services/email_import_service.py | 16 ++-- .../tests/test_attachment_reparse_worker.py | 86 +++++++++++++++++++ ...0005-attachment-content-type-quarantine.md | 35 ++++++++ 7 files changed, 253 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ab9b54f96..3f373cae7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,29 @@ ## [Unreleased] +- **(Devin 리뷰 대응, naruon#1486 후속) 첨부파일 reparse가 성공적으로 재인식된 콘텐츠를 + 초기 import 경로와 달리 content graph에 색인하지 않던 gap을 고쳤습니다.** + `services/email_import_service.py::_append_email_content_graph`는 첨부파일이 첫 + import에서 정상 파싱되면 `ContentNodeRecord`/`ContentSegmentRecord` 그래프를 + 만들지만, `attachment_reparse_worker.py::apply_reparsed_result`는 `Attachment` + 행 자체 컬럼만 갱신했습니다 — 격리(quarantine)됐던 첨부파일이 나중에 reparse로 + `"parsed"`가 되어도 content-graph 기반 검색/AI-hub 기능에는 계속 보이지 + 않았습니다(`AttachmentParseResult`가 import 경로와 동일한 `parse_content` 필드를 + 이미 들고 있었음에도). `apply_reparsed_result`가 결과 `parse_status`가 + `"parsed"`일 때 새 `_append_reparsed_attachment_content_graph`를 호출하도록 + 추가했습니다 — import 경로가 이미 쓰는 `services.content_graph.parse_content`와, + 새로 공개 API로 옮긴 `content_graph_source_record_uid`(원래 + `email_import_service.py`의 private 함수였던 것을 + `services/content_graph/parser.py`로 옮겨 두 호출부가 공유)를 그대로 재사용해 + 색인 경로를 두 개로 만들지 않았습니다. 영속화된 attachment가 자신이 속한 + 이메일의 첨부파일 목록에서 원래 몇 번째였는지는 신뢰성 있게 재현할 수 없으므로, + reparse 경로의 `source_record_uid`는 import 경로의 message-id + 목록 위치 + 조합 대신 attachment의 영구 `attachment_uid` 하나로만 구성하고, 새 레코드의 + `email_id`는 (import 경로처럼 아직 저장되지 않은 `Email`을 통한 관계 append로 + 간접 설정하는 대신) 이미 영속화된 attachment 행의 `email_id` 컬럼에서 직접 + 가져옵니다. 빈 문자열로만 파싱되는 `"parsed"` 결과(공백만 있는 첨부파일 등)는 + 기존 import 경로와 동일하게 색인을 건너뜁니다. 신규 테스트 3개 + (`test_reparse_that_lands_on_parsed_indexes_the_content_graph`, + blank-content 스킵, non-parsed 스킵). 검증: 전체 백엔드 스위트 1908 + passed/40 skipped, ruff clean. - **(Devin 리뷰 대응, 🟡 minor → 실제로는 진짜 결함) NewsDOM 재인식 sweep의 커서가 `RESULT_PENDING`(아직 provider 미설정) 행도 실패 없이 진행했다고 취급해 커서를 그 너머로 진행시켜, 계속 새 업로드가 들어오는 동안 해당 행이 무기한 굶주릴 수 있었습니다.** diff --git a/backend/services/attachment_reparse_worker.py b/backend/services/attachment_reparse_worker.py index dedae29de..244549480 100644 --- a/backend/services/attachment_reparse_worker.py +++ b/backend/services/attachment_reparse_worker.py @@ -22,13 +22,14 @@ from sqlalchemy import bindparam, func, select from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession -from db.models import Attachment +from db.models import Attachment, ContentNodeRecord, ContentSegmentRecord from db.session import AsyncSessionLocal, engine from services.attachment_parser import ( AttachmentParseResult, decode_quarantined_attachment_payload, parse_email_attachment, ) +from services.content_graph import content_graph_source_record_uid, parse_content logger = logging.getLogger(__name__) _sysrand = random.SystemRandom() @@ -74,6 +75,13 @@ def apply_reparsed_result(*, attachment: Attachment, result: AttachmentParseResu ``content=""`` by design -- storing that would destroy the only retained copy of the original quarantined bytes, permanently losing a file that later parser support could otherwise still recover. + + A reparse that lands on ``"parsed"`` also indexes the recognized content + into the content graph, mirroring what the initial import path already + does for an attachment that parses cleanly on first import + (``email_import_service._append_email_content_graph``) -- without this, a + previously-quarantined attachment stayed invisible to content-graph-backed + search/AI-hub features even after successful reparse recognition. """ if result.content: attachment.content = result.content @@ -81,6 +89,82 @@ def apply_reparsed_result(*, attachment: Attachment, result: AttachmentParseResu attachment.parser_key = result.parser_key attachment.parse_status = result.parse_status attachment.parse_error_code = result.parse_error_code + if result.parse_status == "parsed": + _append_reparsed_attachment_content_graph(attachment=attachment, result=result) + + +def _append_reparsed_attachment_content_graph( + *, attachment: Attachment, result: AttachmentParseResult +) -> None: + """Build content graph records for a successfully reparsed attachment. + + Reuses the same ``parse_content`` helper and ``source_record_uid`` + identity convention (``content_graph_source_record_uid``) the import path + uses in ``email_import_service._append_email_content_graph`` -- this is + not a second indexing path, just a second call site for the same one. + + It differs only in how the new records attach to their parents. The + import path appends to a transient ``Email``/``Attachment`` pair (neither + has a real id yet) and lets SQLAlchemy's relationship cascade resolve + ``email_id``/``attachment_id`` at flush time. Here ``attachment`` is + already a persisted row with a stable, permanent ``attachment_uid``, so + ``source_record_uid`` is keyed on that uid alone (not the message-id + + list-position convention the import path uses, since a persisted + attachment's position among its email's siblings is not reliably + reproducible) and ``email_id`` is taken directly from the attachment's + already-loaded ``email_id`` column instead of an ``Email`` relationship + append. + """ + parse_source_content = result.parse_content or result.content + if not parse_source_content.strip(): + return + + parse_result = parse_content( + source_kind="attachment", + source_record_uid=content_graph_source_record_uid( + "attachment", attachment.attachment_uid + ), + content=parse_source_content, + content_type=result.parse_content_type or result.content_type or "text/plain", + display_name=attachment.filename, + ) + + node_records_by_uid: dict[str, ContentNodeRecord] = {} + for parsed_node in parse_result.nodes: + node_record = ContentNodeRecord( + email_id=attachment.email_id, + content_node_uid=parsed_node.content_node_uid, + source_kind=parsed_node.source_kind, + source_record_uid=parsed_node.source_record_uid, + parent_node_uid=parsed_node.parent_node_uid, + node_kind=parsed_node.node_kind, + node_path=parsed_node.node_path, + ordinal_index=parsed_node.ordinal_index, + display_label=parsed_node.display_label, + safe_text_content=parsed_node.safe_text_content, + content_hash=parsed_node.content_hash, + ) + attachment.content_nodes.append(node_record) + node_records_by_uid[parsed_node.content_node_uid] = node_record + + for parsed_segment in parse_result.segments: + segment_record = ContentSegmentRecord( + email_id=attachment.email_id, + content_segment_uid=parsed_segment.content_segment_uid, + source_kind=parsed_segment.source_kind, + source_record_uid=parsed_segment.source_record_uid, + segment_kind=parsed_segment.segment_kind, + segment_path=parsed_segment.segment_path, + ordinal_index=parsed_segment.ordinal_index, + heading_path=parsed_segment.heading_path, + safe_text_content=parsed_segment.safe_text_content, + content_hash=parsed_segment.content_hash, + word_count=parsed_segment.word_count, + ) + node_records_by_uid[parsed_segment.content_node_uid].segments.append( + segment_record + ) + attachment.content_segments.append(segment_record) def process_reparse_pending_attachment(*, attachment: Attachment) -> str: diff --git a/backend/services/content_graph/__init__.py b/backend/services/content_graph/__init__.py index 917e90a27..0c2d23c13 100644 --- a/backend/services/content_graph/__init__.py +++ b/backend/services/content_graph/__init__.py @@ -1,11 +1,12 @@ from .models import ContentNode, ContentSegment, ParseResult, PdfDomSection -from .parser import parse_content, parse_pdf_dom +from .parser import content_graph_source_record_uid, parse_content, parse_pdf_dom __all__ = [ "ContentNode", "ContentSegment", "ParseResult", "PdfDomSection", + "content_graph_source_record_uid", "parse_content", "parse_pdf_dom", ] diff --git a/backend/services/content_graph/parser.py b/backend/services/content_graph/parser.py index a134eee34..26a25b140 100644 --- a/backend/services/content_graph/parser.py +++ b/backend/services/content_graph/parser.py @@ -297,6 +297,19 @@ def _emit_node_and_segment(self, pending: _PendingHtmlNode) -> None: ) +def content_graph_source_record_uid(prefix: str, *parts: str) -> str: + """Build the one canonical ``source_record_uid`` for a content-graph source. + + Every pipeline stage that indexes content into the content graph (initial + email import, attachment reparse) must call this instead of hashing its + own identity string, so the same logical source always resolves to the + same ``source_record_uid`` no matter which stage indexed it. + """ + payload = "\x00".join(str(part) for part in parts) + digest = hashlib.sha256(payload.encode("utf-8", errors="surrogatepass")).hexdigest() + return f"{prefix}:{digest[:32]}" + + def parse_content( *, source_kind: str, diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index a10b5749e..53944625d 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -30,7 +30,11 @@ BatchEmbeddingPartial, try_batch_import_embeddings, ) -from services.content_graph import ParseResult, parse_content +from services.content_graph import ( + ParseResult, + content_graph_source_record_uid, + parse_content, +) from services.email_dedupe_service import strong_email_fingerprint from services.email_parser import EmailData, parse_eml_bytes from services.embedding import ( @@ -477,7 +481,7 @@ def _append_email_content_graph( ) -> None: body_parse_result = parse_content( source_kind="email_body", - source_record_uid=_content_graph_source_record_uid("email", message_id), + source_record_uid=content_graph_source_record_uid("email", message_id), content=str(parsed.get("body_parse_content") or parsed.get("body") or ""), content_type=str(parsed.get("body_content_type") or "text/plain"), display_name="Email body", @@ -503,7 +507,7 @@ def _append_email_content_graph( continue attachment_parse_result = parse_content( source_kind="attachment", - source_record_uid=_content_graph_source_record_uid( + source_record_uid=content_graph_source_record_uid( "attachment", message_id, str(attachment_index), @@ -754,12 +758,6 @@ def _knowledge_graph_edge_uid( return f"kgedge_{digest[:32]}" -def _content_graph_source_record_uid(prefix: str, *parts: str) -> str: - payload = "\x00".join(str(part) for part in parts) - digest = hashlib.sha256(payload.encode("utf-8", errors="surrogatepass")).hexdigest() - return f"{prefix}:{digest[:32]}" - - def _project_source_segments(email_obj: Email) -> list[ProjectSourceSegment]: """Snapshot the imported email's content segments as project source segments. diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index aaec0ee18..85f8935a9 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -16,6 +16,7 @@ from db.models import Attachment import services.attachment_reparse_worker as attachment_reparse_worker_module +from services.content_graph import content_graph_source_record_uid AttachmentReparseWorker = attachment_reparse_worker_module.AttachmentReparseWorker ATTACHMENT_REPARSE_PENDING_STATUS = ( @@ -38,9 +39,13 @@ def _reparse_pending_attachment( payload: bytes, filename: str = "attachment.bin", attachment_id: int | None = None, + email_id: int | None = None, + attachment_uid: str = "attachment_test-uid", ) -> Attachment: return Attachment( id=attachment_id, + email_id=email_id, + attachment_uid=attachment_uid, filename=filename, content_type=content_type, content=base64.b64encode(payload).decode("ascii"), @@ -142,6 +147,87 @@ def test_reparse_preserves_filename_and_declared_content_type(): assert attachment.content_type == "application/pdf" +def test_reparse_that_lands_on_parsed_indexes_the_content_graph(): + # Unlike the OOXML/PDF/PNG scenarios above, plain text is never + # magic-byte-sniffed (see attachment_parser._MAGIC_BYTE_SIGNATURES), so + # this reparse lands on the ordinary "parsed" status -- exactly the + # outcome email_import_service._append_email_content_graph already + # builds a content graph record for on a cleanly-first-parsed attachment. + # apply_reparsed_result must do the same on this path, or a reparsed + # attachment stays invisible to content-graph-backed search/AI-hub + # features even after successful recognition. + attachment = _reparse_pending_attachment( + content_type="text/plain", + payload=b"Meeting notes\n\nDiscuss the roadmap.", + filename="notes.txt", + email_id=42, + attachment_uid="attachment_notes-uid", + ) + + result = process_reparse_pending_attachment(attachment=attachment) + + assert result == "parsed" + assert attachment.parse_status == "parsed" + assert [node.node_kind for node in attachment.content_nodes] == [ + "document", + "paragraph", + "paragraph", + ] + assert [ + segment.safe_text_content for segment in attachment.content_segments + ] == ["Meeting notes", "Discuss the roadmap."] + assert {node.source_kind for node in attachment.content_nodes} == {"attachment"} + assert {segment.source_kind for segment in attachment.content_segments} == { + "attachment" + } + expected_source_record_uid = content_graph_source_record_uid( + "attachment", "attachment_notes-uid" + ) + assert {node.source_record_uid for node in attachment.content_nodes} == { + expected_source_record_uid + } + assert {node.email_id for node in attachment.content_nodes} == {42} + assert {segment.email_id for segment in attachment.content_segments} == {42} + # Every segment is linked back to its parent node's own segments list too + # (the same node<->segment wiring _append_parse_result_records builds). + assert sum(len(node.segments) for node in attachment.content_nodes) == 2 + + +def test_reparse_that_lands_on_parsed_with_blank_content_does_not_index_content_graph(): + # A reparse can land on "parsed" with nothing displayable (an empty or + # whitespace-only retained payload) -- parse_email_attachment does not + # special-case that. Indexing an empty content graph record for it would + # be pure noise, so this must be skipped exactly like + # _append_email_content_graph skips a blank attachment on import. + attachment = _reparse_pending_attachment( + content_type="text/plain", + payload=b" ", + filename="blank.txt", + email_id=42, + ) + + result = process_reparse_pending_attachment(attachment=attachment) + + assert result == "parsed" + assert attachment.content_nodes == [] + assert attachment.content_segments == [] + + +def test_reparse_that_does_not_land_on_parsed_does_not_index_content_graph(): + attachment = _reparse_pending_attachment( + content_type="application/pdf", + payload=b"\x89PNG\r\n\x1a\n" + b"real png bytes", + filename="invoice.pdf", + email_id=42, + ) + + result = process_reparse_pending_attachment(attachment=attachment) + + assert result == _QUARANTINED_STATUS + assert attachment.content_nodes == [] + assert attachment.content_segments == [] + + class _RowsResult: def __init__(self, rows): self._rows = rows diff --git a/docs/adr/0005-attachment-content-type-quarantine.md b/docs/adr/0005-attachment-content-type-quarantine.md index 75913cc01..72429a9da 100644 --- a/docs/adr/0005-attachment-content-type-quarantine.md +++ b/docs/adr/0005-attachment-content-type-quarantine.md @@ -343,6 +343,41 @@ than reversing the original decision: narrowed by this reversal to: the claim is *present and well-formed*, not that it's provably derived from or consistent with `organization_id`. +- **Attachment reparse never indexed a successfully re-recognized + attachment's content into the content graph, unlike the initial import + path — flagged as informational by Devin Review on this PR ("confirm this + is intended"), confirmed real but out of scope for this PR, and closed + here as the tracked follow-up.** + `services/email_import_service.py::_append_email_content_graph` already + builds a `ContentNodeRecord`/`ContentSegmentRecord` graph for an + attachment that parses cleanly on first import, but + `attachment_reparse_worker.py::apply_reparsed_result` only ever updated + the `Attachment` row's own columns — a previously-quarantined attachment + that later reparses to `"parsed"` stayed invisible to content-graph-backed + search/AI-hub features even after successful recognition, despite + `AttachmentParseResult` carrying the same `parse_content` field the import + path indexes. Fixed by calling a new + `_append_reparsed_attachment_content_graph` from `apply_reparsed_result` + whenever the reparse result lands on `"parsed"`. It reuses the same + `services.content_graph.parse_content` helper the import path already + calls, plus a newly shared `content_graph_source_record_uid` (moved out of + `email_import_service.py`, where it was a private function, into + `services/content_graph/parser.py` as a public helper both call sites + import) — not a second indexing path, the same one with a second caller. + Since a persisted attachment's original position among its email's + siblings is not reliably reproducible after import, the reparse path keys + `source_record_uid` on the attachment's permanent `attachment_uid` alone + instead of the import path's message-id + list-position convention, and + sets the new records' `email_id` directly from the attachment's + already-loaded `email_id` column rather than through a transient `Email` + relationship append (the attachment here is already a persisted row, + unlike at import time, so there is no transient parent to defer FK + resolution through). New tests: + `test_reparse_that_lands_on_parsed_indexes_the_content_graph`, + `test_reparse_that_lands_on_parsed_with_blank_content_does_not_index_content_graph`, + `test_reparse_that_does_not_land_on_parsed_does_not_index_content_graph`. + Verification: full backend suite 1908 passed / 40 skipped, ruff clean. + ## References (APA 7th) Freed, N., & Borenstein, N. (1996). *Multipurpose Internet Mail Extensions From 5096d1f1fcf7bcb386a178986e2432a264d019e3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 11:51:07 +0900 Subject: [PATCH 02/16] fix(attachments): preload graph relationships for reparse --- backend/services/attachment_reparse_worker.py | 8 ++++++++ backend/tests/test_attachment_reparse_worker.py | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/backend/services/attachment_reparse_worker.py b/backend/services/attachment_reparse_worker.py index 244549480..eb58ce01d 100644 --- a/backend/services/attachment_reparse_worker.py +++ b/backend/services/attachment_reparse_worker.py @@ -378,6 +378,14 @@ async def _sweep_attachments(self, session: AsyncSession) -> None: attachment = await session.get(Attachment, attachment_id) if attachment is None: continue + # ``apply_reparsed_result`` appends graph rows through these + # relationships. Load them explicitly at the async boundary; + # implicit lazy IO from the synchronous parser path raises + # MissingGreenlet for persisted attachments. + await session.refresh( + attachment, + attribute_names=["content_nodes", "content_segments"], + ) result = process_reparse_pending_attachment(attachment=attachment) await session.commit() logger.info( diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 85f8935a9..8d1c9d05b 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -260,6 +260,7 @@ def __init__(self, row_batches, *, by_id=None): self.statements = [] self.commit_count = 0 self.rollback_count = 0 + self.refresh_calls = [] async def execute(self, statement): self.statements.append(statement) @@ -268,6 +269,9 @@ async def execute(self, statement): async def get(self, _model, attachment_id): return self._by_id.get(attachment_id) + async def refresh(self, attachment, *, attribute_names): + self.refresh_calls.append((attachment.id, tuple(attribute_names))) + async def commit(self): self.commit_count += 1 @@ -392,6 +396,10 @@ async def test_sweep_advances_the_cursor_across_batches(): assert second.parse_status == _QUARANTINED_STATUS assert session.commit_count == 2 assert session.rollback_count == 0 + assert session.refresh_calls == [ + (1, ("content_nodes", "content_segments")), + (2, ("content_nodes", "content_segments")), + ] @pytest.mark.asyncio From beded49c63651f621ab3e24490ba6a7565b71043 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 11:57:00 +0900 Subject: [PATCH 03/16] fix(attachments): persist reparsed graph topology and vectors --- backend/services/attachment_reparse_worker.py | 47 +++++- backend/services/email_import_service.py | 33 ++-- .../tests/test_attachment_reparse_worker.py | 145 +++++++++++++++++- 3 files changed, 207 insertions(+), 18 deletions(-) diff --git a/backend/services/attachment_reparse_worker.py b/backend/services/attachment_reparse_worker.py index eb58ce01d..dceffdd2e 100644 --- a/backend/services/attachment_reparse_worker.py +++ b/backend/services/attachment_reparse_worker.py @@ -30,6 +30,12 @@ parse_email_attachment, ) from services.content_graph import content_graph_source_record_uid, parse_content +from services.email_import_service import ( + EmailImportEmbeddingProvider, + _generate_import_embeddings, + append_knowledge_graph_edges, +) +from services.llm_provider_selection import resolve_runtime_llm_provider logger = logging.getLogger(__name__) _sysrand = random.SystemRandom() @@ -147,6 +153,7 @@ def _append_reparsed_attachment_content_graph( attachment.content_nodes.append(node_record) node_records_by_uid[parsed_node.content_node_uid] = node_record + segment_records: list[ContentSegmentRecord] = [] for parsed_segment in parse_result.segments: segment_record = ContentSegmentRecord( email_id=attachment.email_id, @@ -165,6 +172,37 @@ def _append_reparsed_attachment_content_graph( segment_record ) attachment.content_segments.append(segment_record) + segment_records.append(segment_record) + + append_knowledge_graph_edges( + nodes=list(node_records_by_uid.values()), + segments=segment_records, + attachment_obj=attachment, + ) + + +async def _refresh_reparsed_attachment_embedding( + session: AsyncSession, attachment: Attachment +) -> None: + """Regenerate the attachment vector through its tenant's active provider.""" + provider = await resolve_runtime_llm_provider( + session, + user_id=attachment.email.user_id, + organization_id=attachment.email.organization_id, + ) + embedding_provider = ( + EmailImportEmbeddingProvider( + api_key=provider.api_key, + base_url=provider.base_url, + embedding_model=provider.embedding_model, + ) + if provider is not None + else None + ) + embeddings = await _generate_import_embeddings( + [attachment.content], embedding_provider=embedding_provider + ) + attachment.embedding = embeddings[0] def process_reparse_pending_attachment(*, attachment: Attachment) -> str: @@ -384,9 +422,16 @@ async def _sweep_attachments(self, session: AsyncSession) -> None: # MissingGreenlet for persisted attachments. await session.refresh( attachment, - attribute_names=["content_nodes", "content_segments"], + attribute_names=[ + "email", + "content_nodes", + "content_segments", + "knowledge_graph_edges", + ], ) result = process_reparse_pending_attachment(attachment=attachment) + if result == "parsed": + await _refresh_reparsed_attachment_embedding(session, attachment) await session.commit() logger.info( "Attachment %s reparse result: %s", diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index 53944625d..88b0fa95f 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -440,7 +440,11 @@ def _build_email_object( message_id=message_id, attachment_payloads=attachment_payloads, ) - _append_knowledge_graph_edges(email_obj) + append_knowledge_graph_edges( + nodes=email_obj.content_nodes, + segments=email_obj.content_segments, + email_obj=email_obj, + ) return email_obj, attachment_count @@ -574,11 +578,20 @@ def _append_parse_result_records( attachment_obj.content_segments.append(segment_record) -def _append_knowledge_graph_edges(email_obj: Email) -> None: +def append_knowledge_graph_edges( + *, + nodes: list[ContentNodeRecord], + segments: list[ContentSegmentRecord], + email_obj: Email | None = None, + attachment_obj: Attachment | None = None, +) -> None: + """Append the canonical content-graph topology for one indexed source.""" + if email_obj is None and attachment_obj is None: + raise ValueError("email_obj or attachment_obj is required") nodes_by_uid = { node.content_node_uid: node for node in sorted( - email_obj.content_nodes, + nodes, key=lambda item: ( item.source_kind, item.source_record_uid, @@ -602,6 +615,7 @@ def add_edge( ) -> None: nonlocal ordinal_index edge = KnowledgeGraphEdgeRecord( + email_id=attachment_obj.email_id if attachment_obj is not None else None, edge_uid=_knowledge_graph_edge_uid( edge_kind, _edge_endpoint_uid(source_node, source_segment), @@ -618,12 +632,11 @@ def add_edge( source_segment=source_segment, target_segment=target_segment, ) - email_obj.knowledge_graph_edges.append(edge) - attachment = _edge_attachment( - source_node=source_node, - target_node=target_node, - source_segment=source_segment, - target_segment=target_segment, + if email_obj is not None: + email_obj.knowledge_graph_edges.append(edge) + attachment = attachment_obj or _edge_attachment( + source_node=source_node, target_node=target_node, + source_segment=source_segment, target_segment=target_segment, ) if attachment is not None: attachment.knowledge_graph_edges.append(edge) @@ -649,7 +662,7 @@ def add_edge( list[ContentSegmentRecord], ] = defaultdict(list) for segment in sorted( - email_obj.content_segments, + segments, key=lambda item: ( item.source_kind, item.source_record_uid, diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 8d1c9d05b..788471b36 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -1,7 +1,8 @@ -"""Unit tests for the attachment reparse worker's per-item processing. +"""Tests for attachment reparse classification and persisted worker processing. -Fully mocked: in-memory ``Attachment`` instances and a fake async session -- -no database, no network. Covers the fail-closed outcome (invalid retained +Most tests use in-memory ``Attachment`` instances and a fake async session; +one PostgreSQL smoke test covers the real async persistence boundary. Covers +the fail-closed outcome (invalid retained payload -> a dedicated terminal status) alongside the two "successful re-evaluation" outcomes: a previously-quarantined attachment whose disagreement is now recognized as legitimate (escapes quarantine), and one @@ -10,11 +11,24 @@ import asyncio import base64 +import datetime from types import SimpleNamespace +import uuid import pytest - -from db.models import Attachment +from sqlalchemy import delete, select +from sqlalchemy.exc import OperationalError +from sqlalchemy.orm import selectinload, undefer + +from core.config import settings +from db.models import ( + Attachment, + ContentNodeRecord, + ContentSegmentRecord, + Email, + KnowledgeGraphEdgeRecord, +) +from db.session import AsyncSessionLocal import services.attachment_reparse_worker as attachment_reparse_worker_module from services.content_graph import content_graph_source_record_uid @@ -33,6 +47,118 @@ _QUARANTINED_STATUS = "content_type_mismatch_quarantined" +@pytest.mark.asyncio +@pytest.mark.postgres +async def test_persisted_reparse_commits_topology_and_provider_embedding(monkeypatch): + """Exercise the real AsyncSession relationship and pgvector persistence path.""" + if not settings.DATABASE_URL: + pytest.skip("PostgreSQL smoke path unavailable") + suffix = uuid.uuid4().hex + expected_embedding = [0.25] * 1536 + + async def runtime_provider(*_args, **_kwargs): + return SimpleNamespace( + api_key="test-provider-key", + base_url="https://provider.example/v1", + embedding_model="embedding-test-model", + ) + + async def generated_embeddings(texts, *, embedding_provider, batch_context=None): + assert texts == ["Meeting notes Discuss the roadmap."] + assert embedding_provider.base_url == "https://provider.example/v1" + assert embedding_provider.embedding_model == "embedding-test-model" + assert batch_context is None + return [expected_embedding] + + monkeypatch.setattr( + attachment_reparse_worker_module, + "resolve_runtime_llm_provider", + runtime_provider, + ) + monkeypatch.setattr( + attachment_reparse_worker_module, + "_generate_import_embeddings", + generated_embeddings, + ) + + async with AsyncSessionLocal() as session: + email = Email( + user_id=f"reparse-user-{suffix}", + organization_id=f"reparse-org-{suffix}", + workspace_id=f"reparse-workspace-{suffix}", + message_id=f"reparse-message-{suffix}", + sender="sender@example.com", + recipients="recipient@example.com", + subject="Reparse persistence smoke", + date=datetime.datetime.now(datetime.timezone.utc), + body="body", + embedding=[0.0] * 1536, + ) + attachment = _reparse_pending_attachment( + content_type="text/plain", + payload=b"Meeting notes\n\nDiscuss the roadmap.", + attachment_uid=f"attachment_{suffix}", + ) + email.attachments.append(attachment) + session.add(email) + try: + await session.commit() + except OperationalError: + await session.rollback() + pytest.skip("PostgreSQL smoke path unavailable") + attachment_id = attachment.id + email_id = email.id + + worker = AttachmentReparseWorker(batch_limit=1) + worker._attachment_cursor = attachment.id - 1 + try: + await worker._sweep_attachments(session) + persisted = ( + await session.execute( + select(Attachment) + .where(Attachment.attachment_uid == attachment.attachment_uid) + .options( + selectinload(Attachment.content_nodes), + selectinload(Attachment.content_segments), + selectinload(Attachment.knowledge_graph_edges), + undefer(Attachment.embedding), + ) + .execution_options(populate_existing=True) + ) + ).scalar_one() + assert persisted.parse_status == "parsed" + assert len(persisted.content_nodes) == 3 + assert len(persisted.content_segments) == 2 + assert {edge.edge_kind for edge in persisted.knowledge_graph_edges} == { + "node_contains_node", + "node_has_segment", + "segment_next", + } + assert list(persisted.embedding) == expected_embedding + finally: + await session.rollback() + await session.execute( + delete(KnowledgeGraphEdgeRecord).where( + KnowledgeGraphEdgeRecord.attachment_id == attachment_id + ) + ) + await session.execute( + delete(ContentSegmentRecord).where( + ContentSegmentRecord.attachment_id == attachment_id + ) + ) + await session.execute( + delete(ContentNodeRecord).where( + ContentNodeRecord.attachment_id == attachment_id + ) + ) + await session.execute( + delete(Attachment).where(Attachment.id == attachment_id) + ) + await session.execute(delete(Email).where(Email.id == email_id)) + await session.commit() + + def _reparse_pending_attachment( *, content_type: str, @@ -191,6 +317,11 @@ def test_reparse_that_lands_on_parsed_indexes_the_content_graph(): # Every segment is linked back to its parent node's own segments list too # (the same node<->segment wiring _append_parse_result_records builds). assert sum(len(node.segments) for node in attachment.content_nodes) == 2 + assert {edge.edge_kind for edge in attachment.knowledge_graph_edges} == { + "node_contains_node", + "node_has_segment", + "segment_next", + } def test_reparse_that_lands_on_parsed_with_blank_content_does_not_index_content_graph(): @@ -397,8 +528,8 @@ async def test_sweep_advances_the_cursor_across_batches(): assert session.commit_count == 2 assert session.rollback_count == 0 assert session.refresh_calls == [ - (1, ("content_nodes", "content_segments")), - (2, ("content_nodes", "content_segments")), + (1, ("email", "content_nodes", "content_segments", "knowledge_graph_edges")), + (2, ("email", "content_nodes", "content_segments", "knowledge_graph_edges")), ] From f63a1093952b078bd1bbf366f559508978f3ea37 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 12:10:27 +0900 Subject: [PATCH 04/16] fix(attachments): chunk reparsed embedding sources --- backend/services/attachment_reparse_worker.py | 8 +-- backend/services/email_import_service.py | 55 ++++++++++++------- .../tests/test_attachment_reparse_worker.py | 27 ++++++--- ...0005-attachment-content-type-quarantine.md | 18 ++++++ 4 files changed, 77 insertions(+), 31 deletions(-) diff --git a/backend/services/attachment_reparse_worker.py b/backend/services/attachment_reparse_worker.py index dceffdd2e..4cac0705f 100644 --- a/backend/services/attachment_reparse_worker.py +++ b/backend/services/attachment_reparse_worker.py @@ -32,7 +32,7 @@ from services.content_graph import content_graph_source_record_uid, parse_content from services.email_import_service import ( EmailImportEmbeddingProvider, - _generate_import_embeddings, + _generate_source_embedding, append_knowledge_graph_edges, ) from services.llm_provider_selection import resolve_runtime_llm_provider @@ -199,10 +199,10 @@ async def _refresh_reparsed_attachment_embedding( if provider is not None else None ) - embeddings = await _generate_import_embeddings( - [attachment.content], embedding_provider=embedding_provider + attachment.embedding = await _generate_source_embedding( + attachment.content, + embedding_provider=embedding_provider, ) - attachment.embedding = embeddings[0] def process_reparse_pending_attachment(*, attachment: Attachment) -> str: diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index 88b0fa95f..84598dbc9 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -337,33 +337,48 @@ async def _extract_and_generate_embeddings( ) fitted_embeddings: list[list[float]] = [] for source_text in source_texts: - source_chunks = chunk_text(source_text) - if not source_chunks: - fitted_embeddings.append(_zero_embedding()) - continue - - vector_sum: list[float] | None = None - vector_count = 0 - for start in range(0, len(source_chunks), MAX_EMBEDDING_CHUNKS_PER_WINDOW): - chunk_embeddings = await _generate_import_embeddings( - source_chunks[start : start + MAX_EMBEDDING_CHUNKS_PER_WINDOW], + fitted_embeddings.append( + await _generate_source_embedding( + source_text, embedding_provider=embedding_provider, batch_context=batch_context, ) - for embedding in chunk_embeddings: - if vector_sum is None: - vector_sum = [0.0] * len(embedding) - for index, value in enumerate(embedding): - vector_sum[index] += value - vector_count += 1 - fitted_embeddings.append( - [value / vector_count for value in vector_sum] - if vector_sum and vector_count - else _zero_embedding() ) return attachment_payloads, fitted_embeddings +async def _generate_source_embedding( + source_text: str, + *, + embedding_provider: EmailImportEmbeddingProvider | None, + batch_context: "EmailImportBatchContext | None" = None, +) -> list[float]: + """Chunk, embed in bounded windows, and average one source vector.""" + source_chunks = chunk_text(source_text) + if not source_chunks: + return _zero_embedding() + + vector_sum: list[float] | None = None + vector_count = 0 + for start in range(0, len(source_chunks), MAX_EMBEDDING_CHUNKS_PER_WINDOW): + chunk_embeddings = await _generate_import_embeddings( + source_chunks[start : start + MAX_EMBEDDING_CHUNKS_PER_WINDOW], + embedding_provider=embedding_provider, + batch_context=batch_context, + ) + for embedding in chunk_embeddings: + if vector_sum is None: + vector_sum = [0.0] * len(embedding) + for index, value in enumerate(embedding): + vector_sum[index] += value + vector_count += 1 + return ( + [value / vector_count for value in vector_sum] + if vector_sum and vector_count + else _zero_embedding() + ) + + def _build_email_object( *, parsed: EmailData, diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 788471b36..6114a8ce8 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -17,7 +17,7 @@ import pytest from sqlalchemy import delete, select -from sqlalchemy.exc import OperationalError +from sqlalchemy.exc import OperationalError, ProgrammingError from sqlalchemy.orm import selectinload, undefer from core.config import settings @@ -30,6 +30,7 @@ ) from db.session import AsyncSessionLocal import services.attachment_reparse_worker as attachment_reparse_worker_module +import services.email_import_service as email_import_service_module from services.content_graph import content_graph_source_record_uid AttachmentReparseWorker = attachment_reparse_worker_module.AttachmentReparseWorker @@ -54,7 +55,8 @@ async def test_persisted_reparse_commits_topology_and_provider_embedding(monkeyp if not settings.DATABASE_URL: pytest.skip("PostgreSQL smoke path unavailable") suffix = uuid.uuid4().hex - expected_embedding = [0.25] * 1536 + long_content = ("alpha " * 400) + "\n\n" + ("beta " * 400) + provider_batches: list[list[str]] = [] async def runtime_provider(*_args, **_kwargs): return SimpleNamespace( @@ -64,11 +66,14 @@ async def runtime_provider(*_args, **_kwargs): ) async def generated_embeddings(texts, *, embedding_provider, batch_context=None): - assert texts == ["Meeting notes Discuss the roadmap."] assert embedding_provider.base_url == "https://provider.example/v1" assert embedding_provider.embedding_model == "embedding-test-model" assert batch_context is None - return [expected_embedding] + start = sum(len(batch) for batch in provider_batches) + provider_batches.append(list(texts)) + return [ + [float(start + index + 1)] * 1536 for index in range(len(texts)) + ] monkeypatch.setattr( attachment_reparse_worker_module, @@ -76,7 +81,7 @@ async def generated_embeddings(texts, *, embedding_provider, batch_context=None) runtime_provider, ) monkeypatch.setattr( - attachment_reparse_worker_module, + email_import_service_module, "_generate_import_embeddings", generated_embeddings, ) @@ -96,14 +101,14 @@ async def generated_embeddings(texts, *, embedding_provider, batch_context=None) ) attachment = _reparse_pending_attachment( content_type="text/plain", - payload=b"Meeting notes\n\nDiscuss the roadmap.", + payload=long_content.encode(), attachment_uid=f"attachment_{suffix}", ) email.attachments.append(attachment) session.add(email) try: await session.commit() - except OperationalError: + except (OperationalError, ProgrammingError): await session.rollback() pytest.skip("PostgreSQL smoke path unavailable") attachment_id = attachment.id @@ -134,6 +139,14 @@ async def generated_embeddings(texts, *, embedding_provider, batch_context=None) "node_has_segment", "segment_next", } + chunk_count = sum(len(batch) for batch in provider_batches) + assert chunk_count > 1 + assert all( + 0 < len(batch) <= email_import_service_module.MAX_EMBEDDING_CHUNKS_PER_WINDOW + for batch in provider_batches + ) + expected_value = sum(range(1, chunk_count + 1)) / chunk_count + expected_embedding = [expected_value] * 1536 assert list(persisted.embedding) == expected_embedding finally: await session.rollback() diff --git a/docs/adr/0005-attachment-content-type-quarantine.md b/docs/adr/0005-attachment-content-type-quarantine.md index 72429a9da..36a652d00 100644 --- a/docs/adr/0005-attachment-content-type-quarantine.md +++ b/docs/adr/0005-attachment-content-type-quarantine.md @@ -147,6 +147,24 @@ that is the point to re-evaluate a dedicated library against this policy. a PR whose stated purpose is a calendar-conflict-check tool. Recorded here rather than silently worked around; the fix belongs in its own dedicated PR. +## Research grounding + +The content-graph follow-up is grounded in Edge et al.'s GraphRAG work, which +separates graph-based indexing from later graph-guided answer construction and +reports benefits for query-focused summarization over large private corpora. +That supports preserving the same document topology when content enters the +index through reparse as when it enters through initial import; this ADR does +not claim that Naruon implements the paper's entity extraction or community +summarization pipeline. + +- Darren Edge, Ha Trinh, Newman Cheng, Joshua Bradley, Alex Chao, Apurva Mody, + Steven Truitt, and Jonathan Larson. 2024. “From Local to Global: A Graph RAG + Approach to Query-Focused Summarization.” arXiv:2404.16130. + https://arxiv.org/abs/2404.16130 + +No paper PDF is copied into this repository: the stable source citation is +linked instead, avoiding an unsupported redistribution assumption. + ## Revisions Two real gaps were found and fixed after initial review, both narrowing rather From 41ae6a2f66bff8664cfd50c51556c64aeeac4624 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 12:14:10 +0900 Subject: [PATCH 05/16] fix(db): skip absent legacy email read-state table --- backend/alembic/versions/0011_email_read_state.py | 7 +++++++ backend/tests/test_alembic_migrations.py | 10 ++++++++++ 2 files changed, 17 insertions(+) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 716590cd1..413841204 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -14,6 +14,11 @@ def upgrade() -> None: + # Fresh installations materialize the current ``email_records`` model in + # the 0001 baseline, including ``is_read``. This historical side branch + # only applies to databases that still carry its legacy ``emails`` table. + if "emails" not in sa.inspect(op.get_bind()).get_table_names(): + return op.add_column( "emails", sa.Column( @@ -26,4 +31,6 @@ def upgrade() -> None: def downgrade() -> None: + if "emails" not in sa.inspect(op.get_bind()).get_table_names(): + return op.drop_column("emails", "is_read") diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index c5afd7e07..7a92d5ee6 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -736,6 +736,16 @@ def test_merge_revision_reconciles_email_read_state_branch(): assert "op.drop_column(" not in revision_text +def test_legacy_email_read_state_branch_skips_fresh_baseline_schema(): + revision_path = ( + BACKEND_ROOT / "alembic" / "versions" / "0011_email_read_state.py" + ) + revision_text = revision_path.read_text() + + assert '"emails" not in sa.inspect(op.get_bind()).get_table_names()' in revision_text + assert 'op.add_column(\n "emails"' in revision_text + + def test_merge_revision_reconciles_newsdom_provider_branch(): revision_path = ( BACKEND_ROOT / "alembic" / "versions" / "0015_merge_newsdom_email_heads.py" From 86074f6040f5bc5baf55a72f3811e6848fb8f863 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:18:26 +0000 Subject: [PATCH 06/16] test(attachments): skip the persisted-reparse postgres smoke cleanly when unreachable test_persisted_reparse_commits_topology_and_provider_embedding only caught OperationalError/ProgrammingError around session.commit(), but a genuinely unreachable DATABASE_URL (the common case: conftest.py defaults it to a postgresql:// URL even when no server is listening) can fail the connection attempt itself with a raw ConnectionRefusedError/OSError that SQLAlchemy/ asyncpg don't always wrap as OperationalError -- so the test hard-failed instead of skipping in any environment without a real Postgres running (this sandbox included, and full backend suite locally). Widened the except clause to the same set already used elsewhere in this file/test_data_api.py for the same reason. Verified: full backend suite 1908 passed/41 skipped without DATABASE_URL reachable, and 1945 passed/4 skipped against a real pgvector/pgvector:pg16 container on the default port (including this test), ruff clean. --- .../tests/test_attachment_reparse_worker.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 6114a8ce8..170f9b6df 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -15,6 +15,7 @@ from types import SimpleNamespace import uuid +import asyncpg import pytest from sqlalchemy import delete, select from sqlalchemy.exc import OperationalError, ProgrammingError @@ -108,7 +109,23 @@ async def generated_embeddings(texts, *, embedding_provider, batch_context=None) session.add(email) try: await session.commit() - except (OperationalError, ProgrammingError): + except ( + ConnectionRefusedError, + OSError, + OperationalError, + ProgrammingError, + asyncpg.CannotConnectNowError, + asyncpg.InvalidAuthorizationSpecificationError, + asyncpg.InvalidCatalogNameError, + asyncpg.InvalidPasswordError, + ): + # A truly unreachable DATABASE_URL (the common case: conftest.py + # defaults it to a postgresql:// URL even when no server is + # actually listening) can fail the connection attempt itself -- + # asyncpg/SQLAlchemy don't always wrap that as OperationalError, + # so a plain ConnectionRefusedError/OSError must be caught here + # too, or this test hard-fails instead of skipping everywhere + # postgres isn't actually running (this sandbox included). await session.rollback() pytest.skip("PostgreSQL smoke path unavailable") attachment_id = attachment.id From 51245f71de47aef80eb257bb158b551388622590 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 12:25:00 +0900 Subject: [PATCH 07/16] test: fail reparse smoke on postgres defects --- .../tests/test_attachment_reparse_worker.py | 40 ++++++++----------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 170f9b6df..e5ec3cb76 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -17,8 +17,8 @@ import asyncpg import pytest -from sqlalchemy import delete, select -from sqlalchemy.exc import OperationalError, ProgrammingError +from sqlalchemy import delete, select, text +from sqlalchemy.exc import OperationalError from sqlalchemy.orm import selectinload, undefer from core.config import settings @@ -55,6 +55,20 @@ async def test_persisted_reparse_commits_topology_and_provider_embedding(monkeyp """Exercise the real AsyncSession relationship and pgvector persistence path.""" if not settings.DATABASE_URL: pytest.skip("PostgreSQL smoke path unavailable") + try: + async with AsyncSessionLocal() as probe_session: + await probe_session.execute(text("SELECT 1")) + except ( + ConnectionRefusedError, + OSError, + OperationalError, + asyncpg.CannotConnectNowError, + asyncpg.InvalidAuthorizationSpecificationError, + asyncpg.InvalidCatalogNameError, + asyncpg.InvalidPasswordError, + ): + pytest.skip("PostgreSQL smoke path unavailable") + suffix = uuid.uuid4().hex long_content = ("alpha " * 400) + "\n\n" + ("beta " * 400) provider_batches: list[list[str]] = [] @@ -107,27 +121,7 @@ async def generated_embeddings(texts, *, embedding_provider, batch_context=None) ) email.attachments.append(attachment) session.add(email) - try: - await session.commit() - except ( - ConnectionRefusedError, - OSError, - OperationalError, - ProgrammingError, - asyncpg.CannotConnectNowError, - asyncpg.InvalidAuthorizationSpecificationError, - asyncpg.InvalidCatalogNameError, - asyncpg.InvalidPasswordError, - ): - # A truly unreachable DATABASE_URL (the common case: conftest.py - # defaults it to a postgresql:// URL even when no server is - # actually listening) can fail the connection attempt itself -- - # asyncpg/SQLAlchemy don't always wrap that as OperationalError, - # so a plain ConnectionRefusedError/OSError must be caught here - # too, or this test hard-fails instead of skipping everywhere - # postgres isn't actually running (this sandbox included). - await session.rollback() - pytest.skip("PostgreSQL smoke path unavailable") + await session.commit() attachment_id = attachment.id email_id = email.id From de11149348cbb76d4d43d0ee311edd0811a8097e Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:28:49 +0000 Subject: [PATCH 08/16] fix(db): make 0011_email_read_state's legacy-table guard offline-safe Devin Review: running `alembic upgrade --sql` (offline SQL generation, a real, CLI-exposed flag on scripts/migrate_db.py) through this revision raised sqlalchemy.exc.NoInspectionAvailable -- op.get_bind() returns a MockConnection in offline mode, and sa.inspect() rejects it outright, so SQL generation failed instead of emitting the migration. Guarded the legacy-table introspection behind context.is_offline_mode(): offline generation has no live connection and no specific target database to ask "does this legacy table exist" at generation time either, so it correctly no-ops (matching the common fresh-install case); a DBA applying against a database that still carries the legacy `emails` table runs it online, where introspection still works exactly as before. Updated the existing source-text-pinning contract test to match (it pinned the pre-refactor call site verbatim) and added two behavioral regression tests: offline mode never calls sa.inspect/add_column/drop_column at all, and online mode still adds the column only when the legacy table is present. Verified: full alembic migration chain (blank DB -> head) via a real pgvector/pgvector:pg16 container, `alembic upgrade ... --sql` no longer raises, full backend suite passes (except two pre-existing failures already reported on this PR: the release-governance contract tests pinned to the base branch's old CI trigger scope, and one pre-existing test-ordering fragility where test_0001_initial_upgrade_succeeds_against_a_ fresh_database mutates the shared postgres schema mid-suite -- neither caused by this commit), ruff clean. --- .../alembic/versions/0011_email_read_state.py | 19 ++++- backend/tests/test_alembic_migrations.py | 76 ++++++++++++++++++- 2 files changed, 91 insertions(+), 4 deletions(-) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 413841204..cdc27a9a1 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -3,7 +3,7 @@ Existing rows default to read so historical/file imports do not surface as unread. """ -from alembic import op +from alembic import context, op import sqlalchemy as sa # revision identifiers, used by Alembic. @@ -13,11 +13,24 @@ depends_on = None +def _legacy_emails_table_present() -> bool: + # Offline SQL generation (``alembic upgrade --sql``) has no live + # connection to introspect -- ``op.get_bind()`` returns a MockConnection + # that ``sa.inspect`` rejects outright. There is no target database to + # ask "does this legacy table exist" at generation time either, so this + # migration is a no-op for offline output; a DBA applying it against a + # database that still carries the legacy ``emails`` table runs it online + # instead, where introspection works. + if context.is_offline_mode(): + return False + return "emails" in sa.inspect(op.get_bind()).get_table_names() + + def upgrade() -> None: # Fresh installations materialize the current ``email_records`` model in # the 0001 baseline, including ``is_read``. This historical side branch # only applies to databases that still carry its legacy ``emails`` table. - if "emails" not in sa.inspect(op.get_bind()).get_table_names(): + if not _legacy_emails_table_present(): return op.add_column( "emails", @@ -31,6 +44,6 @@ def upgrade() -> None: def downgrade() -> None: - if "emails" not in sa.inspect(op.get_bind()).get_table_names(): + if not _legacy_emails_table_present(): return op.drop_column("emails", "is_read") diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index 7a92d5ee6..c161ace70 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -742,10 +742,84 @@ def test_legacy_email_read_state_branch_skips_fresh_baseline_schema(): ) revision_text = revision_path.read_text() - assert '"emails" not in sa.inspect(op.get_bind()).get_table_names()' in revision_text + assert '"emails" in sa.inspect(op.get_bind()).get_table_names()' in revision_text + # Offline SQL generation (`alembic upgrade --sql`) has no live connection + # to introspect -- op.get_bind() returns a MockConnection sa.inspect + # rejects outright -- so the legacy-table check must short-circuit to + # "absent" in that mode rather than raising. + assert "context.is_offline_mode()" in revision_text assert 'op.add_column(\n "emails"' in revision_text +def test_legacy_email_read_state_offline_generation_never_inspects(monkeypatch): + """Regression for offline `alembic upgrade --sql`: op.get_bind() returns a + MockConnection in that mode, and sa.inspect(...) raises NoInspectionAvailable + for it -- so upgrade()/downgrade() must short-circuit on is_offline_mode() + before ever calling sa.inspect, not just happen to skip the add/drop. + """ + module = _load_revision_module("0011_email_read_state.py") + + def _boom(_connection): + raise AssertionError("sa.inspect must not run in offline mode") + + monkeypatch.setattr(module.context, "is_offline_mode", lambda: True) + monkeypatch.setattr(module.sa, "inspect", _boom) + monkeypatch.setattr( + module.op, + "add_column", + lambda *a, **k: pytest.fail("must not add_column in offline mode"), + ) + monkeypatch.setattr( + module.op, + "drop_column", + lambda *a, **k: pytest.fail("must not drop_column in offline mode"), + ) + + module.upgrade() + module.downgrade() + + +def test_legacy_email_read_state_online_adds_column_only_when_table_present( + monkeypatch, +): + module = _load_revision_module("0011_email_read_state.py") + monkeypatch.setattr(module.context, "is_offline_mode", lambda: False) + monkeypatch.setattr(module.op, "get_bind", lambda: object()) + + class _AbsentInspector: + @staticmethod + def get_table_names(): + return ["email_records"] + + monkeypatch.setattr(module.sa, "inspect", lambda _connection: _AbsentInspector()) + monkeypatch.setattr( + module.op, + "add_column", + lambda *a, **k: pytest.fail("must not add_column when emails is absent"), + ) + module.upgrade() + + calls = [] + + class _PresentInspector: + @staticmethod + def get_table_names(): + return ["emails"] + + monkeypatch.setattr(module.sa, "inspect", lambda _connection: _PresentInspector()) + monkeypatch.setattr( + module.op, + "add_column", + lambda *args, **kwargs: calls.append(args), + ) + module.upgrade() + + assert len(calls) == 1 + table_name, column = calls[0] + assert table_name == "emails" + assert column.name == "is_read" + + def test_merge_revision_reconciles_newsdom_provider_branch(): revision_path = ( BACKEND_ROOT / "alembic" / "versions" / "0015_merge_newsdom_email_heads.py" From 87ef2e5f3aad8dfb1ca7584225b85a30412df71d Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:35:17 +0000 Subject: [PATCH 09/16] fix(db): defer 0011's legacy-table check to SQL, not Python Devin Review caught a real correctness gap in the offline-mode guard (de111493): offline SQL generation now no-ops unconditionally, which is right for a fresh install but silently wrong for a legacy database that still has the `emails` table -- alembic_version still advances to 0011, permanently hiding that is_read was never actually added, since a later `alembic upgrade` believes this revision already ran. The Python-side check (sa.inspect(op.get_bind()) or a fixed is_offline_mode() answer) can only ever bake in one fixed answer at generation time, and the same generated script is meant to be applied later against whichever database a DBA chooses. Replaced it with a `DO $$ ... $$` block: the existence check now runs in SQL, evaluated by Postgres at apply time against whatever the actual target is, so the one generated script -- or the same online execution -- is correct against a fresh-baseline target (no-ops) and a legacy target (adds the column) alike. No more sa.inspect/op.get_bind() at all in this revision, so neither failure mode can regress. Updated the source-text contract test to match and replaced the two mock-based upgrade()/downgrade() tests with one real-Postgres smoke test covering both directions (legacy-present: add then remove; fresh-baseline: no-op). Verified against a real pgvector/pgvector:pg16 container: generated the offline SQL, applied that exact generated script against a simulated legacy database (a bare `emails` table + a pinned alembic_version row) via psql, and confirmed `is_read` was actually added -- the exact gap Devin flagged, now closed. Full migration chain blank->head still succeeds, full backend suite passes (same two pre-existing, already-reported release-governance contract failures and shared-DB test-ordering fragility -- neither from this commit), ruff clean. --- .../alembic/versions/0011_email_read_state.py | 75 +++++---- backend/tests/test_alembic_migrations.py | 147 ++++++++++-------- 2 files changed, 129 insertions(+), 93 deletions(-) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index cdc27a9a1..8e121d55c 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -3,8 +3,7 @@ Existing rows default to read so historical/file imports do not surface as unread. """ -from alembic import context, op -import sqlalchemy as sa +from alembic import op # revision identifiers, used by Alembic. revision = "0011_email_read_state" @@ -12,38 +11,54 @@ branch_labels = None depends_on = None +# Fresh installations materialize the current ``email_records`` model in the +# 0001 baseline, including ``is_read``. This historical side branch only +# applies to databases that still carry its legacy ``emails`` table. +# +# The condition has to be evaluated in SQL, not Python: offline SQL +# generation (``alembic upgrade --sql``, a real flag ``scripts/migrate_db.py`` +# exposes) has no live connection to introspect with and no specific target +# database to ask "does this legacy table exist" at generation time either -- +# the same static script is meant to later be applied by a DBA against +# whichever database they choose, fresh-install or legacy. A Python-side +# check (``sa.inspect(op.get_bind())``) can only ever answer that question +# for one hypothetical target chosen at generation time, so it is wrong for +# the other: skip unconditionally and the column silently never gets added +# for a legacy database that applies the generated script (while +# ``alembic_version`` still advances, permanently hiding the gap); inspect +# online and bake in one fixed answer and the same script fails outright +# against the other kind of target. A ``DO $$ ... $$`` block defers the +# check to apply time instead, so the one generated script is correct +# against either kind of target, online or offline-then-applied-later alike. +_UPGRADE_SQL = """ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables WHERE table_name = 'emails' + ) AND NOT EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_name = 'emails' AND column_name = 'is_read' + ) THEN + ALTER TABLE emails ADD COLUMN is_read boolean NOT NULL DEFAULT true; + END IF; +END $$; +""" -def _legacy_emails_table_present() -> bool: - # Offline SQL generation (``alembic upgrade --sql``) has no live - # connection to introspect -- ``op.get_bind()`` returns a MockConnection - # that ``sa.inspect`` rejects outright. There is no target database to - # ask "does this legacy table exist" at generation time either, so this - # migration is a no-op for offline output; a DBA applying it against a - # database that still carries the legacy ``emails`` table runs it online - # instead, where introspection works. - if context.is_offline_mode(): - return False - return "emails" in sa.inspect(op.get_bind()).get_table_names() +_DOWNGRADE_SQL = """ +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables WHERE table_name = 'emails' + ) THEN + ALTER TABLE emails DROP COLUMN IF EXISTS is_read; + END IF; +END $$; +""" def upgrade() -> None: - # Fresh installations materialize the current ``email_records`` model in - # the 0001 baseline, including ``is_read``. This historical side branch - # only applies to databases that still carry its legacy ``emails`` table. - if not _legacy_emails_table_present(): - return - op.add_column( - "emails", - sa.Column( - "is_read", - sa.Boolean(), - nullable=False, - server_default=sa.text("true"), - ), - ) + op.execute(_UPGRADE_SQL) def downgrade() -> None: - if not _legacy_emails_table_present(): - return - op.drop_column("emails", "is_read") + op.execute(_DOWNGRADE_SQL) diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index c161ace70..324555f7f 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -736,88 +736,109 @@ def test_merge_revision_reconciles_email_read_state_branch(): assert "op.drop_column(" not in revision_text -def test_legacy_email_read_state_branch_skips_fresh_baseline_schema(): +def test_legacy_email_read_state_branch_defers_check_to_sql(monkeypatch): revision_path = ( BACKEND_ROOT / "alembic" / "versions" / "0011_email_read_state.py" ) revision_text = revision_path.read_text() - assert '"emails" in sa.inspect(op.get_bind()).get_table_names()' in revision_text - # Offline SQL generation (`alembic upgrade --sql`) has no live connection - # to introspect -- op.get_bind() returns a MockConnection sa.inspect - # rejects outright -- so the legacy-table check must short-circuit to - # "absent" in that mode rather than raising. - assert "context.is_offline_mode()" in revision_text - assert 'op.add_column(\n "emails"' in revision_text + # The legacy-table check must be evaluated in SQL (at apply time), not in + # Python at generation time: offline SQL generation (`alembic upgrade + # --sql`, a real flag `scripts/migrate_db.py` exposes) has no live + # connection to introspect with, and the one generated script is meant to + # later be applied against whichever database a DBA chooses -- a + # Python-side sa.inspect(op.get_bind()) check can only ever bake in one + # fixed answer, which is wrong for whichever kind of target it didn't + # assume (silently skips the real column on a legacy target while + # `alembic_version` still advances, or crashes outright against a fresh + # one). upgrade()/downgrade() themselves must contain no such check -- + # only op.execute() calls -- so this can't regress into + # either failure mode. + assert "def upgrade" in revision_text + upgrade_and_after = revision_text.split("def upgrade", 1)[1] + assert "sa.inspect(op.get_bind())" not in upgrade_and_after + assert "context.is_offline_mode()" not in upgrade_and_after + assert "DO $$" in revision_text + assert "information_schema.tables" in revision_text + assert "ALTER TABLE emails ADD COLUMN is_read" in revision_text - -def test_legacy_email_read_state_offline_generation_never_inspects(monkeypatch): - """Regression for offline `alembic upgrade --sql`: op.get_bind() returns a - MockConnection in that mode, and sa.inspect(...) raises NoInspectionAvailable - for it -- so upgrade()/downgrade() must short-circuit on is_offline_mode() - before ever calling sa.inspect, not just happen to skip the add/drop. - """ module = _load_revision_module("0011_email_read_state.py") + calls = [] + monkeypatch.setattr(module.op, "execute", lambda sql: calls.append(sql)) + module.upgrade() + module.downgrade() + assert len(calls) == 2 + assert "ADD COLUMN is_read" in calls[0] + assert "DROP COLUMN IF EXISTS is_read" in calls[1] - def _boom(_connection): - raise AssertionError("sa.inspect must not run in offline mode") - monkeypatch.setattr(module.context, "is_offline_mode", lambda: True) - monkeypatch.setattr(module.sa, "inspect", _boom) - monkeypatch.setattr( - module.op, - "add_column", - lambda *a, **k: pytest.fail("must not add_column in offline mode"), - ) - monkeypatch.setattr( - module.op, - "drop_column", - lambda *a, **k: pytest.fail("must not drop_column in offline mode"), - ) +def _run_0011_upgrade(sync_conn) -> None: + from alembic.operations import Operations + from alembic.runtime.migration import MigrationContext - module.upgrade() - module.downgrade() + module = _load_revision_module("0011_email_read_state.py") + context = MigrationContext.configure(sync_conn, opts={"target_metadata": None}) + with Operations.context(context): + module.upgrade() -def test_legacy_email_read_state_online_adds_column_only_when_table_present( - monkeypatch, -): +def _run_0011_downgrade(sync_conn) -> None: + from alembic.operations import Operations + from alembic.runtime.migration import MigrationContext + module = _load_revision_module("0011_email_read_state.py") - monkeypatch.setattr(module.context, "is_offline_mode", lambda: False) - monkeypatch.setattr(module.op, "get_bind", lambda: object()) + context = MigrationContext.configure(sync_conn, opts={"target_metadata": None}) + with Operations.context(context): + module.downgrade() - class _AbsentInspector: - @staticmethod - def get_table_names(): - return ["email_records"] - monkeypatch.setattr(module.sa, "inspect", lambda _connection: _AbsentInspector()) - monkeypatch.setattr( - module.op, - "add_column", - lambda *a, **k: pytest.fail("must not add_column when emails is absent"), - ) - module.upgrade() +@pytest.mark.asyncio +@pytest.mark.postgres +async def test_legacy_email_read_state_real_postgres_smoke(): + """Both directions this migration must get right against a real database: + a legacy target (still has the ``emails`` table) gets the column added and + later removed; a fresh-baseline target (no ``emails`` table at all, the + now-common case) is left untouched rather than erroring. + """ + engine = create_async_engine(settings.DATABASE_URL) + try: + async with engine.begin() as conn: + # Fresh-baseline case first, on a connection with no "emails" + # table anywhere in scope: must no-op, not raise. + await conn.run_sync(_run_0011_upgrade) - calls = [] + await conn.execute( + text("CREATE TEMP TABLE emails (id serial primary key) ON COMMIT DROP") + ) - class _PresentInspector: - @staticmethod - def get_table_names(): - return ["emails"] + def _has_is_read(sync_conn): + return any( + column["name"] == "is_read" + for column in inspect(sync_conn).get_columns("emails") + ) - monkeypatch.setattr(module.sa, "inspect", lambda _connection: _PresentInspector()) - monkeypatch.setattr( - module.op, - "add_column", - lambda *args, **kwargs: calls.append(args), - ) - module.upgrade() + assert not await conn.run_sync(_has_is_read) + await conn.run_sync(_run_0011_upgrade) + assert await conn.run_sync(_has_is_read) - assert len(calls) == 1 - table_name, column = calls[0] - assert table_name == "emails" - assert column.name == "is_read" + await conn.run_sync(_run_0011_downgrade) + assert not await conn.run_sync(_has_is_read) + except ( + ConnectionRefusedError, + OSError, + OperationalError, + asyncpg.CannotConnectNowError, + asyncpg.InvalidAuthorizationSpecificationError, + asyncpg.InvalidCatalogNameError, + asyncpg.InvalidPasswordError, + ): + await engine.dispose() + pytest.skip("PostgreSQL smoke path unavailable") + except Exception: + await engine.dispose() + raise + finally: + await engine.dispose() def test_merge_revision_reconciles_newsdom_provider_branch(): From 8e47575f9d20643dea72444e612008baf6c360fe Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:39:53 +0000 Subject: [PATCH 10/16] fix(ci): stop the root governance test step from crashing under PYTHONWARNINGS=error Devin Review flagged the new "Run repository-root governance contract tests" step (db97962c) for missing the same log-scan safety net the backend tests step already has -- a warning-class message pytest prints without failing the run shouldn't count as clean evidence. Added it, matching that step's own tee+grep pattern exactly. While verifying that fix, found something more serious: the step actually crashes outright. The job sets PYTHONWARNINGS=error at the job level (inherited by every step), and there is no root-level pytest config -- so pytest-asyncio's pytest_configure hook unconditionally warns that asyncio_default_fixture_loop_scope is unset, PYTHONWARNINGS=error turns that into a raised exception, and pytest exits 3 (INTERNALERROR) before collecting a single test. This isn't specific to this PR: every future run of this step, for every PR, would hit it. tests/ has no async tests at all -- the warning fires purely because pytest-asyncio is installed in the environment. Added a root-level pytest.ini with the same asyncio_default_fixture_loop_scope = function setting backend/pytest.ini already uses, so pytest-asyncio's configure-time check is satisfied and the warning never fires. Confirmed this doesn't affect backend/'s own suite (it keeps using its own pytest.ini, found first when invoked from that directory). Verified: `PYTHONWARNINGS=error python -m pytest -q tests` from the repo root now exits 0 (was exit 3 before), the tee+grep check passes on that clean output, full backend suite still 1910 passed/42 skipped, ruff clean. --- .github/workflows/app-ci.yml | 11 +++++++++-- pytest.ini | 2 ++ 2 files changed, 11 insertions(+), 2 deletions(-) create mode 100644 pytest.ini diff --git a/.github/workflows/app-ci.yml b/.github/workflows/app-ci.yml index 1c4b7d6a8..5a5ba7e4c 100644 --- a/.github/workflows/app-ci.yml +++ b/.github/workflows/app-ci.yml @@ -76,9 +76,16 @@ jobs: # tests/ (repo root) holds workflow/YAML contract tests, e.g. # test_stacked_pr_workflow_contract.py. The backend job's own pytest # invocation above runs from backend/ and never collects this - # directory, so it needs its own explicit step. + # directory, so it needs its own explicit step. Mirrors that step's + # own log scan: a warning-class message pytest prints without + # actually failing the run must not be accepted as clean evidence. run: | - python -m pytest -q tests + set -o pipefail + python -m pytest -q tests 2>&1 | tee root_pytest_output.log + if grep -qiE 'timeout|fatal|warn|denied' root_pytest_output.log; then + echo "::error::Root governance contract tests produced Timeout, Fatal, Warn, or Denied outputs" + exit 1 + fi frontend: name: frontend diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 000000000..0102b0a97 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +asyncio_default_fixture_loop_scope = function From 09a2443bc84333fd027b56d3981fe8595edc22f3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:43:17 +0000 Subject: [PATCH 11/16] fix(db): resolve 0011's legacy-table check through search_path, not raw table_name Devin Review: the DO-block guard checked information_schema.tables WHERE table_name = 'emails', which ignores search_path entirely and matches a same-named table in *any* schema the connecting role can see. On a deployment with more than one accessible schema, that could find an unrelated `emails` table outside the search path while the unqualified `ALTER TABLE emails` below resolves to a different (or no) table -- passing the guard for the wrong relation, or aborting the migration outright against a target that was never meant to be touched. Replaced the check with to_regclass('emails'), which resolves an unqualified name exactly the way the DDL statement itself does (through the connection's search_path), so the check and the DDL can never disagree about which table they mean. Checked NOT NULL for existence, and matched the column check against the resolved OID via pg_attribute instead of another table_name lookup. Verified against a real pgvector/pgvector:pg16 container: created an unrelated `other_schema.emails` table outside the default search_path (confirmed information_schema.tables would have matched it: the exact false-positive Devin described) and ran this migration's upgrade() against that database -- to_regclass('emails') correctly returned NULL, the migration no-op'd, and other_schema.emails was left untouched. Full backend suite still 1910 passed/42 skipped, ruff clean. --- .../alembic/versions/0011_email_read_state.py | 27 +++++++++++++------ backend/tests/test_alembic_migrations.py | 17 +++++++++--- 2 files changed, 33 insertions(+), 11 deletions(-) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 8e121d55c..d7175e3c0 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -30,14 +30,27 @@ # against the other kind of target. A ``DO $$ ... $$`` block defers the # check to apply time instead, so the one generated script is correct # against either kind of target, online or offline-then-applied-later alike. +# +# ``to_regclass('emails')`` (not ``information_schema.tables`` by bare +# ``table_name``) deliberately: the unqualified ``ALTER TABLE emails`` below +# resolves through the connection's ``search_path``, and ``to_regclass`` +# resolves an unqualified name exactly the same way, returning NULL if it +# doesn't. ``information_schema.tables`` filtered only by ``table_name`` +# ignores ``search_path`` entirely and matches a same-named table in *any* +# schema the connecting role can see -- on a deployment with more than one +# accessible schema, that could find an unrelated ``emails`` table outside +# the search path while the unqualified ``ALTER TABLE emails`` targets a +# different (or no) table, passing the guard for the wrong relation or +# aborting the migration outright. Resolving both the check and the DDL +# through the same name lookup makes that mismatch structurally impossible. _UPGRADE_SQL = """ DO $$ BEGIN - IF EXISTS ( - SELECT 1 FROM information_schema.tables WHERE table_name = 'emails' - ) AND NOT EXISTS ( - SELECT 1 FROM information_schema.columns - WHERE table_name = 'emails' AND column_name = 'is_read' + IF to_regclass('emails') IS NOT NULL AND NOT EXISTS ( + SELECT 1 FROM pg_attribute + WHERE attrelid = to_regclass('emails') + AND attname = 'is_read' + AND NOT attisdropped ) THEN ALTER TABLE emails ADD COLUMN is_read boolean NOT NULL DEFAULT true; END IF; @@ -47,9 +60,7 @@ _DOWNGRADE_SQL = """ DO $$ BEGIN - IF EXISTS ( - SELECT 1 FROM information_schema.tables WHERE table_name = 'emails' - ) THEN + IF to_regclass('emails') IS NOT NULL THEN ALTER TABLE emails DROP COLUMN IF EXISTS is_read; END IF; END $$; diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index 324555f7f..bc4764ff0 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -758,11 +758,22 @@ def test_legacy_email_read_state_branch_defers_check_to_sql(monkeypatch): upgrade_and_after = revision_text.split("def upgrade", 1)[1] assert "sa.inspect(op.get_bind())" not in upgrade_and_after assert "context.is_offline_mode()" not in upgrade_and_after - assert "DO $$" in revision_text - assert "information_schema.tables" in revision_text - assert "ALTER TABLE emails ADD COLUMN is_read" in revision_text module = _load_revision_module("0011_email_read_state.py") + # to_regclass('emails'), not information_schema.tables by bare + # table_name: the latter ignores search_path and can match an unrelated + # same-named table in a different accessible schema than the one the + # unqualified ALTER TABLE below actually resolves to. Checked on the + # loaded module's own SQL constants, not the raw file text, so this + # can't be fooled by a comment mentioning either string for context. + assert "DO $$" in module._UPGRADE_SQL + assert "DO $$" in module._DOWNGRADE_SQL + assert "to_regclass('emails')" in module._UPGRADE_SQL + assert "to_regclass('emails')" in module._DOWNGRADE_SQL + assert "information_schema.tables" not in module._UPGRADE_SQL + assert "information_schema.tables" not in module._DOWNGRADE_SQL + assert "ALTER TABLE emails ADD COLUMN is_read" in module._UPGRADE_SQL + calls = [] monkeypatch.setattr(module.op, "execute", lambda sql: calls.append(sql)) module.upgrade() From c2490965f3e3ea0d905ec7e55d84adc305959155 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 03:45:59 +0000 Subject: [PATCH 12/16] docs(db): document 0011's op.execute as a deliberate structured-ops exception Devin Review's "Migration bypasses structured operations" analysis asked to document the exception or redesign. Redesign isn't possible without losing correctness (that's the whole reason this migration evaluates its legacy-table check in SQL rather than Python, see the fix history on _UPGRADE_SQL below), so documented it instead: the repo rule's actual target is DDL built from interpolated identifier strings (sa.text(f"...")), an injection-safety concern this migration's static, non-interpolated SQL constants don't have. The reason a structured op.* call isn't used is different -- no structured operation expresses "run this DDL only if a runtime condition holds" -- so a DO $$ ... $$ block via op.execute() is the correct primitive, not a workaround for one. No behavior change; full backend suite still 1910 passed/42 skipped, ruff clean. --- .../alembic/versions/0011_email_read_state.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index d7175e3c0..2ad141c21 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -1,6 +1,22 @@ """Add is_read to emails (IMAP \\Seen read state). Existing rows default to read so historical/file imports do not surface as unread. + +Deliberate exception to this repo's "Alembic migrations use structured +operations (``op.create_index``, ...), never ``sa.text(f"...")`` DDL" rule +(``AGENTS.md``/``CLAUDE.md``): ``upgrade()``/``downgrade()`` below use +``op.execute()`` with the module-level ``_UPGRADE_SQL``/``_DOWNGRADE_SQL`` +constants instead of a structured ``op.*`` call. That rule's actual target is +DDL built from interpolated identifier strings (an injection-safety concern); +these constants are plain static text with no interpolation, string +formatting, or identifiers built from variables -- the same safety property +a structured call would have. The reason a structured call isn't used is +different: this migration's behavior must be conditional on whether the +legacy ``emails`` table exists, evaluated at apply time (see the comment on +``_UPGRADE_SQL`` below for why that check cannot live in Python), and no +structured Alembic operation expresses "run this DDL only if a runtime +condition holds" -- a ``DO $$ ... $$`` block is the correct primitive for +that, not a workaround for one. """ from alembic import op From d244ccc89b1ffdfe9eb5d0e0036eadc9734588c6 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 09:11:02 +0000 Subject: [PATCH 13/16] fix(attachments): resolve reparse embedding source and migration downgrade data loss CodeRabbit's full review of PR #1501's content-graph indexing fix found two real correctness gaps, both fixed here: 1. attachment_reparse_worker.py's embedding refresh read attachment.content after apply_reparsed_result ran, but that column is only overwritten when result.content (a markup-stripped display string) is non-empty. A "parsed" result whose display text strips to empty while its raw parse_content does not (markup-only content) left attachment.content at its stale, still-base64-encoded value, so the embedding was generated from base64 noise instead of the actual reparsed text. process_reparse_pending_attachment now returns a ReparseOutcome carrying the resolved embedding_source_text (parse_content preferred over content, matching both the content-graph indexer and email_import_service's own resolution), passed explicitly into the embedding refresh instead of re-derived from the attachment row. 2. 0011_email_read_state.py's downgrade() dropped emails.is_read unconditionally whenever the column and legacy table were present, including a same-named column that predated this revision and its own NOT EXISTS-guarded upgrade() never touched -- destroying that unrelated column and its data on downgrade. upgrade() now tags the column it creates with a COMMENT ON COLUMN provenance marker; downgrade() drops the column only when that exact marker is present. Also renamed email_import_service._generate_source_embedding to the public generate_source_embedding (CodeRabbit nitpick): a third cross-module dependency attachment_reparse_worker.py imports, alongside content_graph_source_record_uid and append_knowledge_graph_edges. New tests: test_reparse_that_lands_on_parsed_with_markup_only_content_ still_embeds_parse_content, test_legacy_email_read_state_downgrade_ preserves_a_preexisting_column (real Postgres). Full backend suite 1911 passed/43 skipped (matching CI's DATABASE_URL-unset invocation); every test touched by this fix also verified in isolation against a real PostgreSQL 16 + pgvector database. ruff clean. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01M4GKADWJyd8NToEAK5SH6Q --- CHANGELOG.md | 35 +++++++++ .../alembic/versions/0011_email_read_state.py | 49 ++++++++++--- backend/services/attachment_reparse_worker.py | 70 +++++++++++++----- backend/services/email_import_service.py | 12 ++- backend/tests/test_alembic_migrations.py | 73 +++++++++++++++++++ .../tests/test_attachment_reparse_worker.py | 58 +++++++++++---- ...0005-attachment-content-type-quarantine.md | 48 ++++++++++++ 7 files changed, 299 insertions(+), 46 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d62358c5f..fde7cd812 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,39 @@ ## [Unreleased] +- **(CodeRabbit 리뷰 대응, naruon#1501) 첨부파일 reparse content-graph 색인 후속(바로 아래 항목)의 + 전체 리뷰에서 실제 결함 2건이 나와 모두 고쳤습니다.** (1) reparse 임베딩 재생성이 + resolved parse 소스 텍스트 대신 `attachment.content`에서 값을 읽고 있었습니다. + `apply_reparsed_result`는 `result.content`(마크업을 걷어낸 *display* 문자열)가 비어있지 + 않을 때만 `attachment.content`를 덮어쓰는데, `"parsed"` 결과의 display 텍스트는 빈 문자열로 + 스트립되지만 raw `result.parse_content`는 그렇지 않은 경우(예: 보이는 텍스트 노드 없이 + 마크업만 있는 첨부파일) `attachment.content`가 base64로 인코딩된 채 그대로 남아있어, + 임베딩이 실제 재파싱된 텍스트가 아니라 base64 노이즈로부터 생성됐습니다 — content graph는 + 올바른 텍스트로 색인됐는데(`_append_reparsed_attachment_content_graph`가 이미 + `result.parse_content or result.content`를 직접 resolve했으므로, import 시점의 + `email_import_service._extract_and_generate_embeddings`와 동일한 resolve 방식), 임베딩만 + 어긋난 것입니다. `process_reparse_pending_attachment`는 이제 단순 상태 문자열 대신 + `ReparseOutcome(parse_status, embedding_source_text)`를 반환해, 그 동일한 resolved 텍스트를 + attachment 행에서 다시(불안정하게) 유도하지 않고 임베딩 재생성으로 명시적으로 전달합니다. + 신규 테스트: + `test_reparse_that_lands_on_parsed_with_markup_only_content_still_embeds_parse_content`. + (2) `0011_email_read_state.py`의 `downgrade()`가 legacy `emails` 테이블과 `is_read` + 컬럼이 둘 다 있으면 무조건 컬럼을 drop했습니다 — 이 리비전보다 먼저 존재했던(그래서 이 + 리비전의 `NOT EXISTS` 가드가 건드리지 않은) 동명의 `is_read` 컬럼까지 데이터째 파괴할 수 + 있었습니다. `upgrade()`가 이제 자신이 만든 컬럼에 `COMMENT ON COLUMN` provenance 마커 + (`_IS_READ_PROVENANCE_MARKER = "0011_email_read_state:added"`)를 남기고, `downgrade()`는 + `col_description`으로 그 마커가 정확히 있을 때만 drop합니다 — 이 리비전이 추가한 것만 + drop하고 그 외에는 손대지 않습니다. 신규 real-Postgres 테스트: + `test_legacy_email_read_state_downgrade_preserves_a_preexisting_column`(legacy + `emails.is_read` 컬럼에 데이터를 미리 심어두고 upgrade→downgrade를 실행해 컬럼과 데이터가 + 모두 살아남는지 확인). 추가로 `email_import_service._generate_source_embedding`을 공개 + `generate_source_embedding`으로 개명(CodeRabbit nitpick): `content_graph_source_record_uid`, + `append_knowledge_graph_edges`에 이어 `attachment_reparse_worker.py`가 가져다 쓰는 세 + 번째 cross-module 헬퍼이므로, 모든 cross-module 헬퍼가 public일 때 모듈 경계가 일관됩니다. + 검증: 전체 백엔드 스위트 1911 passed/43 skipped(`DATABASE_URL` 미설정, CI와 동일), 이번 + 수정이 건드린 테스트는 전부 실제 PostgreSQL 16 + pgvector에 대해 단독 실행 시 통과 — 같은 + 실제 DB에 대해 스위트 전체를 한 프로세스로 돌리면 이 PR에서 이미 보고된 기존 cross-file + test-ordering 실패 1건(`test_0001_initial_upgrade_succeeds_against_a_fresh_database`가 + 스위트 중간에 `email_records`를 drop·재생성)이 재현되지만, 이번 수정과는 무관합니다. ruff + clean. - **(Devin 리뷰 대응, naruon#1486 후속) 첨부파일 reparse가 성공적으로 재인식된 콘텐츠를 초기 import 경로와 달리 content graph에 색인하지 않던 gap을 고쳤습니다.** `services/email_import_service.py::_append_email_content_graph`는 첨부파일이 첫 diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 2ad141c21..49b0c12f5 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -8,15 +8,16 @@ ``op.execute()`` with the module-level ``_UPGRADE_SQL``/``_DOWNGRADE_SQL`` constants instead of a structured ``op.*`` call. That rule's actual target is DDL built from interpolated identifier strings (an injection-safety concern); -these constants are plain static text with no interpolation, string -formatting, or identifiers built from variables -- the same safety property -a structured call would have. The reason a structured call isn't used is -different: this migration's behavior must be conditional on whether the -legacy ``emails`` table exists, evaluated at apply time (see the comment on -``_UPGRADE_SQL`` below for why that check cannot live in Python), and no -structured Alembic operation expresses "run this DDL only if a runtime -condition holds" -- a ``DO $$ ... $$`` block is the correct primitive for -that, not a workaround for one. +these constants interpolate only ``_IS_READ_PROVENANCE_MARKER``, a fixed +module-level literal, never an identifier or a value built from a variable, +external input, or runtime state -- the same safety property a structured +call would have. The reason a structured call isn't used is different: this +migration's behavior must be conditional on whether the legacy ``emails`` +table exists, evaluated at apply time (see the comment on ``_UPGRADE_SQL`` +below for why that check cannot live in Python), and no structured Alembic +operation expresses "run this DDL only if a runtime condition holds" -- a +``DO $$ ... $$`` block is the correct primitive for that, not a workaround +for one. """ from alembic import op @@ -59,7 +60,20 @@ # different (or no) table, passing the guard for the wrong relation or # aborting the migration outright. Resolving both the check and the DDL # through the same name lookup makes that mismatch structurally impossible. -_UPGRADE_SQL = """ +# +# ``COMMENT ON COLUMN emails.is_read`` tags the column with a provenance +# marker (``_IS_READ_PROVENANCE_MARKER``) the moment upgrade() actually adds +# it. downgrade() only drops the column when that exact marker is present +# (CodeRabbit, naruon#1501): an ``emails.is_read`` column that already +# existed before this revision ran -- from some other, unrelated origin -- +# would upgrade()'s ``NOT EXISTS`` guard correctly leave alone, but an +# unconditional ``DROP COLUMN IF EXISTS`` on downgrade would still destroy it +# and its data, since a downgrade has no other way to tell "I added this" +# apart from "this happens to be present". Checking the marker via +# ``col_description`` makes downgrade drop only what this exact revision's +# upgrade created. +_IS_READ_PROVENANCE_MARKER = "0011_email_read_state:added" +_UPGRADE_SQL = f""" DO $$ BEGIN IF to_regclass('emails') IS NOT NULL AND NOT EXISTS ( @@ -69,14 +83,25 @@ AND NOT attisdropped ) THEN ALTER TABLE emails ADD COLUMN is_read boolean NOT NULL DEFAULT true; + COMMENT ON COLUMN emails.is_read IS '{_IS_READ_PROVENANCE_MARKER}'; END IF; END $$; """ -_DOWNGRADE_SQL = """ +_DOWNGRADE_SQL = f""" DO $$ BEGIN - IF to_regclass('emails') IS NOT NULL THEN + IF to_regclass('emails') IS NOT NULL AND EXISTS ( + SELECT 1 FROM pg_attribute + WHERE attrelid = to_regclass('emails') + AND attname = 'is_read' + AND NOT attisdropped + ) AND col_description(to_regclass('emails'), ( + SELECT attnum FROM pg_attribute + WHERE attrelid = to_regclass('emails') + AND attname = 'is_read' + AND NOT attisdropped + )) = '{_IS_READ_PROVENANCE_MARKER}' THEN ALTER TABLE emails DROP COLUMN IF EXISTS is_read; END IF; END $$; diff --git a/backend/services/attachment_reparse_worker.py b/backend/services/attachment_reparse_worker.py index 4cac0705f..f7cf3dc5d 100644 --- a/backend/services/attachment_reparse_worker.py +++ b/backend/services/attachment_reparse_worker.py @@ -18,6 +18,7 @@ import asyncio import logging import random +from dataclasses import dataclass from sqlalchemy import bindparam, func, select from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession @@ -32,8 +33,8 @@ from services.content_graph import content_graph_source_record_uid, parse_content from services.email_import_service import ( EmailImportEmbeddingProvider, - _generate_source_embedding, append_knowledge_graph_edges, + generate_source_embedding, ) from services.llm_provider_selection import resolve_runtime_llm_provider @@ -182,9 +183,18 @@ def _append_reparsed_attachment_content_graph( async def _refresh_reparsed_attachment_embedding( - session: AsyncSession, attachment: Attachment + session: AsyncSession, attachment: Attachment, *, source_text: str ) -> None: - """Regenerate the attachment vector through its tenant's active provider.""" + """Regenerate the attachment vector through its tenant's active provider. + + ``source_text`` must be the caller's already-resolved embedding source + (see ``ReparseOutcome.embedding_source_text``), not read from + ``attachment.content``: ``apply_reparsed_result`` only overwrites that + column when ``result.content`` is non-empty (see its docstring), so a + "parsed" result whose *display* text strips to empty while its *parse* + text does not (e.g. markup-only content) would leave ``attachment.content`` + stale and this would otherwise embed unrelated, already-superseded bytes. + """ provider = await resolve_runtime_llm_provider( session, user_id=attachment.email.user_id, @@ -199,20 +209,37 @@ async def _refresh_reparsed_attachment_embedding( if provider is not None else None ) - attachment.embedding = await _generate_source_embedding( - attachment.content, + attachment.embedding = await generate_source_embedding( + source_text, embedding_provider=embedding_provider, ) -def process_reparse_pending_attachment(*, attachment: Attachment) -> str: +@dataclass(frozen=True, slots=True) +class ReparseOutcome: + """Result of one ``process_reparse_pending_attachment`` call. + + ``embedding_source_text`` mirrors the exact source-text resolution + ``_append_reparsed_attachment_content_graph`` uses (``parse_content`` + preferred over ``content``) and the import path's + ``email_import_service._extract_and_generate_embeddings`` already uses + for the same reason -- it is meaningful only when ``parse_status == + "parsed"``, but is always populated for a uniform return shape. + """ + + parse_status: str + embedding_source_text: str + + +def process_reparse_pending_attachment(*, attachment: Attachment) -> ReparseOutcome: """Re-evaluate one ``reparse_pending`` attachment in place. - Returns the resulting ``parse_status`` on a successful re-evaluation - (``"parsed"``, the quarantine status again, or any other terminal status - ``parse_email_attachment`` can return), or ``RESULT_DECODE_FAILED`` when - the retained payload itself is not valid base64 -- moved to a dedicated - failure status so the sweep does not retry it forever. + Returns a :class:`ReparseOutcome` carrying the resulting ``parse_status`` + on a successful re-evaluation (``"parsed"``, the quarantine status again, + or any other terminal status ``parse_email_attachment`` can return), or + ``RESULT_DECODE_FAILED`` when the retained payload itself is not valid + base64 -- moved to a dedicated failure status so the sweep does not retry + it forever. """ try: raw_content = decode_quarantined_attachment_payload(attachment.content) @@ -224,7 +251,9 @@ def process_reparse_pending_attachment(*, attachment: Attachment) -> str: getattr(attachment, "id", "?"), exc, ) - return RESULT_DECODE_FAILED + return ReparseOutcome( + parse_status=RESULT_DECODE_FAILED, embedding_source_text="" + ) result = parse_email_attachment( filename=attachment.filename, @@ -232,7 +261,10 @@ def process_reparse_pending_attachment(*, attachment: Attachment) -> str: raw_content=raw_content, ) apply_reparsed_result(attachment=attachment, result=result) - return result.parse_status + return ReparseOutcome( + parse_status=result.parse_status, + embedding_source_text=result.parse_content or result.content, + ) def _engine_uses_postgresql() -> bool: @@ -429,14 +461,18 @@ async def _sweep_attachments(self, session: AsyncSession) -> None: "knowledge_graph_edges", ], ) - result = process_reparse_pending_attachment(attachment=attachment) - if result == "parsed": - await _refresh_reparsed_attachment_embedding(session, attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) + if outcome.parse_status == "parsed": + await _refresh_reparsed_attachment_embedding( + session, + attachment, + source_text=outcome.embedding_source_text, + ) await session.commit() logger.info( "Attachment %s reparse result: %s", attachment_id, - result, + outcome.parse_status, ) except Exception: await session.rollback() diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index 84598dbc9..e68d2ec4b 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -338,7 +338,7 @@ async def _extract_and_generate_embeddings( fitted_embeddings: list[list[float]] = [] for source_text in source_texts: fitted_embeddings.append( - await _generate_source_embedding( + await generate_source_embedding( source_text, embedding_provider=embedding_provider, batch_context=batch_context, @@ -347,13 +347,19 @@ async def _extract_and_generate_embeddings( return attachment_payloads, fitted_embeddings -async def _generate_source_embedding( +async def generate_source_embedding( source_text: str, *, embedding_provider: EmailImportEmbeddingProvider | None, batch_context: "EmailImportBatchContext | None" = None, ) -> list[float]: - """Chunk, embed in bounded windows, and average one source vector.""" + """Chunk, embed in bounded windows, and average one source vector. + + Public (not ``_``-prefixed): ``attachment_reparse_worker.py`` imports + this cross-module, alongside ``content_graph_source_record_uid`` and + ``append_knowledge_graph_edges`` -- the module boundary stays consistent + when every cross-module helper is public (CodeRabbit, naruon#1501). + """ source_chunks = chunk_text(source_text) if not source_chunks: return _zero_embedding() diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index bc4764ff0..7f6c76251 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -773,6 +773,16 @@ def test_legacy_email_read_state_branch_defers_check_to_sql(monkeypatch): assert "information_schema.tables" not in module._UPGRADE_SQL assert "information_schema.tables" not in module._DOWNGRADE_SQL assert "ALTER TABLE emails ADD COLUMN is_read" in module._UPGRADE_SQL + # CodeRabbit (naruon#1501): downgrade must drop emails.is_read only when + # this revision's own upgrade created it, not whenever the column merely + # happens to be present -- an unconditional DROP would also destroy a + # pre-existing, unrelated is_read column and its data. upgrade() tags the + # column it creates with a provenance marker comment; downgrade() checks + # that exact marker via col_description before dropping. + assert "COMMENT ON COLUMN emails.is_read" in module._UPGRADE_SQL + assert module._IS_READ_PROVENANCE_MARKER in module._UPGRADE_SQL + assert "col_description" in module._DOWNGRADE_SQL + assert module._IS_READ_PROVENANCE_MARKER in module._DOWNGRADE_SQL calls = [] monkeypatch.setattr(module.op, "execute", lambda sql: calls.append(sql)) @@ -852,6 +862,69 @@ def _has_is_read(sync_conn): await engine.dispose() +@pytest.mark.asyncio +@pytest.mark.postgres +async def test_legacy_email_read_state_downgrade_preserves_a_preexisting_column(): + """downgrade() must not drop an ``emails.is_read`` column (or its data) + that predates this revision -- CodeRabbit flagged the earlier + unconditional ``DROP COLUMN IF EXISTS`` on naruon#1501: since upgrade()'s + ``NOT EXISTS`` guard already leaves a pre-existing column untouched + (never adding its own provenance marker to it), downgrade() must + symmetrically leave it alone too, distinguishing "this revision added it" + from "it merely happens to be present" via the marker set on the + ``COMMENT ON COLUMN`` this revision's own upgrade() applies. + """ + engine = create_async_engine(settings.DATABASE_URL) + try: + async with engine.begin() as conn: + await conn.execute( + text( + "CREATE TEMP TABLE emails (id serial primary key, " + "is_read boolean NOT NULL DEFAULT false) ON COMMIT DROP" + ) + ) + await conn.execute(text("INSERT INTO emails (is_read) VALUES (false)")) + + def _has_is_read(sync_conn): + return any( + column["name"] == "is_read" + for column in inspect(sync_conn).get_columns("emails") + ) + + # upgrade() must be a no-op here: the column already exists, so + # its NOT EXISTS guard skips both the ADD COLUMN and the marker + # COMMENT -- this pre-existing column is never tagged as "added + # by this revision". + assert await conn.run_sync(_has_is_read) + await conn.run_sync(_run_0011_upgrade) + assert await conn.run_sync(_has_is_read) + + # downgrade() must leave the untagged, pre-existing column (and + # its data) alone rather than dropping it. + await conn.run_sync(_run_0011_downgrade) + assert await conn.run_sync(_has_is_read) + preserved_value = ( + await conn.execute(text("SELECT is_read FROM emails")) + ).scalar_one() + assert preserved_value is False + except ( + ConnectionRefusedError, + OSError, + OperationalError, + asyncpg.CannotConnectNowError, + asyncpg.InvalidAuthorizationSpecificationError, + asyncpg.InvalidCatalogNameError, + asyncpg.InvalidPasswordError, + ): + await engine.dispose() + pytest.skip("PostgreSQL smoke path unavailable") + except Exception: + await engine.dispose() + raise + finally: + await engine.dispose() + + def test_merge_revision_reconciles_newsdom_provider_branch(): revision_path = ( BACKEND_ROOT / "alembic" / "versions" / "0015_merge_newsdom_email_heads.py" diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index e5ec3cb76..72720bcb9 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -221,9 +221,9 @@ def test_reparse_escapes_a_now_recognized_false_positive(): filename="report.docx", ) - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == "unsupported_content_type" + assert outcome.parse_status == "unsupported_content_type" assert attachment.parse_status == "unsupported_content_type" assert attachment.parse_error_code == "unsupported_content_type" assert attachment.parse_status != _QUARANTINED_STATUS @@ -247,9 +247,9 @@ def test_reparse_to_unsupported_content_type_preserves_retained_bytes(): ) retained_content = attachment.content - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == "unsupported_content_type" + assert outcome.parse_status == "unsupported_content_type" assert attachment.content == retained_content assert base64.b64decode(attachment.content) == payload @@ -263,9 +263,9 @@ def test_reparse_of_a_genuine_mismatch_returns_to_quarantine(): filename="invoice.pdf", ) - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == _QUARANTINED_STATUS + assert outcome.parse_status == _QUARANTINED_STATUS assert attachment.parse_status == _QUARANTINED_STATUS assert attachment.parse_error_code == _QUARANTINED_STATUS assert attachment.parse_content_type == "image/png" @@ -277,9 +277,9 @@ def test_reparse_rejects_an_invalid_retained_payload(): ) attachment.content = "not@@base64!!" - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == RESULT_DECODE_FAILED + assert outcome.parse_status == RESULT_DECODE_FAILED assert attachment.parse_status == ATTACHMENT_REPARSE_PAYLOAD_INVALID_STATUS assert attachment.parse_error_code == ATTACHMENT_REPARSE_PAYLOAD_INVALID_STATUS @@ -314,9 +314,10 @@ def test_reparse_that_lands_on_parsed_indexes_the_content_graph(): attachment_uid="attachment_notes-uid", ) - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == "parsed" + assert outcome.parse_status == "parsed" + assert outcome.embedding_source_text == "Meeting notes\n\nDiscuss the roadmap." assert attachment.parse_status == "parsed" assert [node.node_kind for node in attachment.content_nodes] == [ "document", @@ -361,13 +362,42 @@ def test_reparse_that_lands_on_parsed_with_blank_content_does_not_index_content_ email_id=42, ) - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == "parsed" + assert outcome.parse_status == "parsed" assert attachment.content_nodes == [] assert attachment.content_segments == [] +def test_reparse_that_lands_on_parsed_with_markup_only_content_still_embeds_parse_content(): + # A "parsed" result whose *display* text (content) strips down to empty + # while its *parse* text (parse_content) does not -- e.g. an attachment + # that is only markup with no visible text nodes. apply_reparsed_result's + # `if result.content:` guard (see its docstring) then leaves + # attachment.content untouched, so the embedding source must come from + # ReparseOutcome.embedding_source_text (parse_content preferred over + # content, matching _append_reparsed_attachment_content_graph and + # email_import_service._extract_and_generate_embeddings), never from + # attachment.content directly -- CodeRabbit flagged this exact mismatch + # on naruon#1501. + attachment = _reparse_pending_attachment( + content_type="text/plain", + payload=b"
", + filename="markup-only.txt", + email_id=42, + ) + + outcome = process_reparse_pending_attachment(attachment=attachment) + + assert outcome.parse_status == "parsed" + # result.content ("") is falsy, so apply_reparsed_result's `if + # result.content:` guard leaves attachment.content at its retained, + # still-base64-encoded original value -- exactly why the embedding + # source cannot come from attachment.content. + assert attachment.content == base64.b64encode(b"
").decode("ascii") + assert outcome.embedding_source_text == "
" + + def test_reparse_that_does_not_land_on_parsed_does_not_index_content_graph(): attachment = _reparse_pending_attachment( content_type="application/pdf", @@ -376,9 +406,9 @@ def test_reparse_that_does_not_land_on_parsed_does_not_index_content_graph(): email_id=42, ) - result = process_reparse_pending_attachment(attachment=attachment) + outcome = process_reparse_pending_attachment(attachment=attachment) - assert result == _QUARANTINED_STATUS + assert outcome.parse_status == _QUARANTINED_STATUS assert attachment.content_nodes == [] assert attachment.content_segments == [] diff --git a/docs/adr/0005-attachment-content-type-quarantine.md b/docs/adr/0005-attachment-content-type-quarantine.md index 36a652d00..527a922bd 100644 --- a/docs/adr/0005-attachment-content-type-quarantine.md +++ b/docs/adr/0005-attachment-content-type-quarantine.md @@ -396,6 +396,54 @@ than reversing the original decision: `test_reparse_that_does_not_land_on_parsed_does_not_index_content_graph`. Verification: full backend suite 1908 passed / 40 skipped, ruff clean. +- **CodeRabbit's full review of the content-graph-indexing follow-up above + found two real correctness gaps in it, both fixed here.** (1) The reparse + embedding refresh regenerated `attachment.embedding` from + `attachment.content` rather than the resolved parse source text. + `apply_reparsed_result` only overwrites `attachment.content` when + `result.content` (a markup-stripped *display* string) is non-empty; a + `"parsed"` result whose display text strips to empty while its raw + `result.parse_content` does not (e.g. an attachment that is only markup, + no visible text nodes) left `attachment.content` at its stale, + still-base64-encoded retained value, so the embedding was generated from + base64 noise instead of the actual reparsed text — while the content graph + indexed the correct text, since `_append_reparsed_attachment_content_graph` + already resolved `result.parse_content or result.content` itself, matching + `email_import_service._extract_and_generate_embeddings`'s identical + resolution at import time. `process_reparse_pending_attachment` now + returns a `ReparseOutcome(parse_status, embedding_source_text)` instead of + a bare status string, carrying that same resolved text through to the + embedding refresh explicitly rather than re-deriving it (unreliably) from + the attachment row. New test: + `test_reparse_that_lands_on_parsed_with_markup_only_content_still_embeds_parse_content`. + (2) `0011_email_read_state.py`'s `downgrade()` dropped `emails.is_read` + unconditionally whenever the column and legacy `emails` table were both + present — including a same-named `is_read` column that predated this + revision entirely, which this revision's own `NOT EXISTS`-guarded + `upgrade()` therefore never touched, destroying that unrelated column and + its data on downgrade. `upgrade()` now tags the column it creates with a + `COMMENT ON COLUMN` provenance marker + (`_IS_READ_PROVENANCE_MARKER = "0011_email_read_state:added"`); + `downgrade()` drops the column only when that exact marker is present via + `col_description`, so it drops what this revision added and nothing else. + New real-Postgres test: + `test_legacy_email_read_state_downgrade_preserves_a_preexisting_column` + (pre-seeds a legacy `emails.is_read` column with data, runs upgrade then + downgrade, asserts both the column and its data survive). Also renamed + `email_import_service._generate_source_embedding` to the public + `generate_source_embedding` (CodeRabbit nitpick): it is a third + cross-module dependency `attachment_reparse_worker.py` imports, alongside + `content_graph_source_record_uid` and `append_knowledge_graph_edges`, so + the module boundary stays consistent when every cross-module helper is + public. Verification: full backend suite 1911 passed / 43 skipped + (`DATABASE_URL` unset, matching CI), and every test touched by this fix + passes in isolation against a real PostgreSQL 16 + pgvector database; ruff + clean. Running the full suite against that same real database in one + process reproduces one pre-existing, already-reported cross-file + test-ordering failure (`test_0001_initial_upgrade_succeeds_against_a_ + fresh_database` drops and recreates `email_records` mid-suite) — + orthogonal to this fix, not caused by it. + ## References (APA 7th) Freed, N., & Borenstein, N. (1996). *Multipurpose Internet Mail Extensions From 6294b8b98cfc375459e0ee2f52a510ea854f7dca Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 12:45:11 +0000 Subject: [PATCH 14/16] fix(migrations): suppress bandit B608 false positive on 0011 DDL constants _UPGRADE_SQL/_DOWNGRADE_SQL became f-strings to interpolate the fixed _IS_READ_PROVENANCE_MARKER literal, which Bandit's hardcoded_sql_expressions check flags as a possible SQL injection vector. Neither string interpolates an identifier or external input (only that module-level constant), matching the safety property the module docstring already documents -- mark both false positives with the repo's established inline `# nosec BXXX` convention rather than leaving CI red. --- backend/alembic/versions/0011_email_read_state.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/alembic/versions/0011_email_read_state.py b/backend/alembic/versions/0011_email_read_state.py index 49b0c12f5..a63d1e9fc 100644 --- a/backend/alembic/versions/0011_email_read_state.py +++ b/backend/alembic/versions/0011_email_read_state.py @@ -86,7 +86,7 @@ COMMENT ON COLUMN emails.is_read IS '{_IS_READ_PROVENANCE_MARKER}'; END IF; END $$; -""" +""" # nosec B608 _DOWNGRADE_SQL = f""" DO $$ @@ -105,7 +105,7 @@ ALTER TABLE emails DROP COLUMN IF EXISTS is_read; END IF; END $$; -""" +""" # nosec B608 def upgrade() -> None: From 0345eda448f3ee778283ed426e9a5f7134dcfc23 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 1 Sep 2026 21:51:32 +0000 Subject: [PATCH 15/16] test(alembic): narrow the connectivity-probe exception handler in the new downgrade test test_legacy_email_read_state_downgrade_preserves_a_preexisting_column wrapped its whole body -- table setup, migration execution, and assertions -- in a try/except that treats connection-family errors as "PostgreSQL unavailable, skip", matching several older tests in this file. CodeRabbit correctly pointed out its immediate neighbor, test_legacy_email_read_state_real_postgres_smoke, already uses the safer pattern: skip only on an initial `SELECT 1` connectivity probe, then let every later failure propagate and fail the test instead of silently skipping it. Match that pattern here too, so a real migration or assertion bug can't get masked as an unrelated environment skip. Verified against a real PostgreSQL 16 + pgvector database: the 6 migration-related postgres-marked tests pass, and a full-suite run passes (1951 passed, 3 skipped) apart from one pre-existing, already-documented test-ordering artifact unrelated to this change (test_0001_initial_upgrade... drops email_records mid-suite when run before test_attachment_reparse_worker tests in the same process; confirmed unaffected by re-running that test in isolation). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01M4GKADWJyd8NToEAK5SH6Q --- backend/tests/test_alembic_migrations.py | 28 ++++++++++++------------ 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/backend/tests/test_alembic_migrations.py b/backend/tests/test_alembic_migrations.py index b3a236682..33bc139c2 100644 --- a/backend/tests/test_alembic_migrations.py +++ b/backend/tests/test_alembic_migrations.py @@ -876,6 +876,20 @@ async def test_legacy_email_read_state_downgrade_preserves_a_preexisting_column( """ engine = create_async_engine(settings.DATABASE_URL) try: + try: + async with engine.connect() as probe_conn: + await probe_conn.execute(text("SELECT 1")) + except ( + ConnectionRefusedError, + OSError, + OperationalError, + asyncpg.CannotConnectNowError, + asyncpg.InvalidAuthorizationSpecificationError, + asyncpg.InvalidCatalogNameError, + asyncpg.InvalidPasswordError, + ): + pytest.skip("PostgreSQL smoke path unavailable") + async with engine.begin() as conn: await conn.execute( text( @@ -907,20 +921,6 @@ def _has_is_read(sync_conn): await conn.execute(text("SELECT is_read FROM emails")) ).scalar_one() assert preserved_value is False - except ( - ConnectionRefusedError, - OSError, - OperationalError, - asyncpg.CannotConnectNowError, - asyncpg.InvalidAuthorizationSpecificationError, - asyncpg.InvalidCatalogNameError, - asyncpg.InvalidPasswordError, - ): - await engine.dispose() - pytest.skip("PostgreSQL smoke path unavailable") - except Exception: - await engine.dispose() - raise finally: await engine.dispose() From d7e5d2d38270019d09925718f905b7fdbd59f3ba Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 04:00:56 +0000 Subject: [PATCH 16/16] fix(test): add missing refresh() to _LiveReparsePendingSession fake Merging origin/claude/noema-contextualwisdomlab-commercialization-afow1j (the cursor/retry-set starvation fix, which added _LiveReparsePendingSession in tests/test_attachment_reparse_worker.py) into this PR's branch (which added the session.refresh(attachment, attribute_names=[...]) call in _sweep_attachments, needed to eager-load relationships before apply_reparsed_result appends content-graph rows through them) surfaced an integration gap neither branch could have caught alone: the merged production code now calls session.refresh() on every sweep, but _LiveReparsePendingSession (used by two multi-sweep scheduling tests) never implemented it, since it predates that call. Add a no-op refresh(), matching the sibling _SequenceSession fake's pattern. Confirmed via RED (test_sweep_does_not_starve_rows_behind_many_failing_rows and test_sweep_rediscovers_a_row_reverted_to_pending_behind_the_cursor both failed with AttributeError before this fix) -> GREEN (full backend suite: 1920 passed, 43 skipped; ruff clean). One unrelated timing-sensitive test (test_main_kills_original_process_group_on_timeout) failed once under full-suite load and passed in isolation and on a second full-suite run -- a pre-existing flake, not caused by this merge. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01M4GKADWJyd8NToEAK5SH6Q --- backend/tests/test_attachment_reparse_worker.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/backend/tests/test_attachment_reparse_worker.py b/backend/tests/test_attachment_reparse_worker.py index 573a4460a..13e8cf6cb 100644 --- a/backend/tests/test_attachment_reparse_worker.py +++ b/backend/tests/test_attachment_reparse_worker.py @@ -740,6 +740,14 @@ async def execute(self, _statement): async def get(self, _model, attachment_id): return self._table.get(attachment_id) + async def refresh(self, _attachment, *, attribute_names): + # No-op: the fake table already holds the live, fully-populated + # instances (see _SequenceSession.refresh for the sibling fake that + # records calls instead -- this one has no need to, since nothing + # here asserts on refresh() itself, only on the resulting sweep + # behavior across many sweeps). + del attribute_names + async def commit(self): self.commit_count += 1