From 520d05c3b4374d08718ec17f949306b51af908aa Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 16 Sep 2026 01:50:15 +0900 Subject: [PATCH 1/4] fix(import): preserve attachments on security owner lineage --- backend/import_fixtures.py | 63 +++++++++++++++++++++++++------------- 1 file changed, 41 insertions(+), 22 deletions(-) diff --git a/backend/import_fixtures.py b/backend/import_fixtures.py index 34d41e0d5..db1983eb7 100644 --- a/backend/import_fixtures.py +++ b/backend/import_fixtures.py @@ -34,6 +34,17 @@ async def generate_fixture_embedding(text: str) -> list[float]: return fit_embedding_vector(embeddings[0], EMBEDDING_DIMENSION) +async def _email_already_imported(session, message_id: str) -> bool: + existing = await session.execute( + select(Email).where( + Email.message_id == message_id, + Email.user_id == IMPORT_USER_ID, + Email.organization_id == IMPORT_ORGANIZATION_ID, + ) + ) + return existing.scalar_one_or_none() is not None + + async def import_eml_file(session, eml_file: Path) -> bool: try: parsed = parse_eml(eml_file) @@ -41,16 +52,13 @@ async def import_eml_file(session, eml_file: Path) -> bool: logger.error("Fixture email parsing failed") return False - existing = await session.execute( - select(Email).where( - Email.message_id == parsed["message_id"], - Email.user_id == IMPORT_USER_ID, - Email.organization_id == IMPORT_ORGANIZATION_ID, - ) - ) - if existing.scalar_one_or_none(): + # Reject duplicates before model/provider work, then explicitly end the + # read transaction so external enrichment never runs while holding it open. + if await _email_already_imported(session, parsed["message_id"]): + await session.rollback() logger.info(f"Email {parsed['message_id']} already exists, skipping.") return False + await session.rollback() body_text = parsed["body"] if parsed["body"].strip() else "Empty body" try: @@ -59,6 +67,30 @@ async def import_eml_file(session, eml_file: Path) -> bool: logger.error("Fixture email body embedding failed") return False + prepared_attachments: list[Attachment] = [] + for att in parsed.get("attachments", []): + att_text = att["content"] if att["content"].strip() else "Empty attachment" + try: + att_emb = await generate_fixture_embedding(att_text) + except Exception: + logger.error("Fixture attachment embedding failed") + att_emb = None + prepared_attachments.append( + Attachment( + filename=att["filename"], + content=att["content"], + embedding=att_emb, + ) + ) + + # Re-check after external work. From here through commit, only database + # operations remain, so the transaction is short and the duplicate guard is + # refreshed after the enrichment window. + if await _email_already_imported(session, parsed["message_id"]): + await session.rollback() + logger.info(f"Email {parsed['message_id']} already exists, skipping.") + return False + thread_id = await assign_thread_id( session, parsed, @@ -81,20 +113,7 @@ async def import_eml_file(session, eml_file: Path) -> bool: embedding=body_emb, thread_id=thread_id, ) - - for att in parsed.get("attachments", []): - att_text = att["content"] if att["content"].strip() else "Empty attachment" - try: - att_emb = await generate_fixture_embedding(att_text) - email_obj.attachments.append( - Attachment( - filename=att["filename"], - content=att["content"], - embedding=att_emb, - ) - ) - except Exception: - logger.error("Fixture attachment embedding failed") + email_obj.attachments.extend(prepared_attachments) session.add(email_obj) try: From 8d3cb9eaf79f02adf093913679d70ac69b55dc1a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 16 Sep 2026 01:50:54 +0900 Subject: [PATCH 2/4] test(import): align fixture doubles with transaction release --- backend/tests/test_import_fixtures.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/backend/tests/test_import_fixtures.py b/backend/tests/test_import_fixtures.py index bad57ec13..2a649016b 100644 --- a/backend/tests/test_import_fixtures.py +++ b/backend/tests/test_import_fixtures.py @@ -35,6 +35,9 @@ def add(self, obj): async def commit(self): self.committed = True + async def rollback(self): + pass + eml_file = tmp_path / "reply.eml" eml_file.write_text("Message-ID: \n\nBody") parsed = { @@ -86,6 +89,9 @@ def add(self, obj): async def commit(self): pass + async def rollback(self): + pass + eml_file = tmp_path / "duplicate-scope.eml" eml_file.write_text("Message-ID: \n\nBody") parsed = { @@ -143,6 +149,9 @@ def add(self, obj): async def commit(self): pass + async def rollback(self): + pass + eml_file = tmp_path / "root.eml" eml_file.write_text("Message-ID: \n\nBody") parsed = { @@ -189,6 +198,9 @@ def add(self, obj): async def commit(self): pass + async def rollback(self): + pass + eml_file = tmp_path / "empty-embedding.eml" eml_file.write_text("Message-ID: \n\nBody") parsed = { From 44da32ce20c794f6f3a5bb99cedd7392549c4370 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 16 Sep 2026 01:51:15 +0900 Subject: [PATCH 3/4] test(import): preserve attachment source through PostgreSQL --- ...est_import_fixture_attachment_integrity.py | 229 ++++++++++++++++++ 1 file changed, 229 insertions(+) create mode 100644 backend/tests/test_import_fixture_attachment_integrity.py diff --git a/backend/tests/test_import_fixture_attachment_integrity.py b/backend/tests/test_import_fixture_attachment_integrity.py new file mode 100644 index 000000000..1ba911c4d --- /dev/null +++ b/backend/tests/test_import_fixture_attachment_integrity.py @@ -0,0 +1,229 @@ +"""Regression coverage for fixture-import attachment enrichment failures.""" + +import datetime +import uuid +from unittest.mock import AsyncMock, patch + +import pytest + +import import_fixtures + + +class _QueryResult: + def __init__(self, existing=None): + self._existing = existing + + def scalar_one_or_none(self): + return self._existing + + +class _RecordingSession: + def __init__(self, results=None): + self.database_transaction_active = False + self.added = None + self.committed = False + self.rollback_count = 0 + self._results = iter(results or [None, None]) + + async def execute(self, _query): + self.database_transaction_active = True + return _QueryResult(next(self._results)) + + def add(self, obj): + self.added = obj + + async def commit(self): + self.committed = True + self.database_transaction_active = False + + async def rollback(self): + self.rollback_count += 1 + self.database_transaction_active = False + + +@pytest.mark.asyncio +async def test_attachment_survives_embedding_failure_outside_database_transaction( + tmp_path, +): + eml_file = tmp_path / "attachment-fallback.eml" + eml_file.write_text("Message-ID: \n\nBody") + parsed = { + "message_id": "", + "sender": "sender@example.com", + "recipients": "user@example.com", + "subject": "Attachment fallback", + "date": datetime.datetime.now(datetime.timezone.utc), + "body": "Body", + "attachments": [ + { + "filename": "evidence.txt", + "content": "attachment source content", + } + ], + } + session = _RecordingSession() + + async def generate_embedding(text: str): + assert session.database_transaction_active is False + if text == "attachment source content": + raise RuntimeError("embedding provider unavailable") + return [0.0] * import_fixtures.EMBEDDING_DIMENSION + + with patch.object(import_fixtures, "parse_eml", return_value=parsed), patch.object( + import_fixtures, + "generate_fixture_embedding", + side_effect=generate_embedding, + ), patch.object( + import_fixtures, + "assign_thread_id", + new_callable=AsyncMock, + return_value="attachment-fallback-thread", + ): + imported = await import_fixtures.import_eml_file(session, eml_file) + + assert imported is True + assert session.rollback_count == 1 + assert session.committed is True + assert session.added is not None + assert len(session.added.attachments) == 1 + attachment = session.added.attachments[0] + assert attachment.filename == "evidence.txt" + assert attachment.content == "attachment source content" + assert attachment.embedding is None + + +@pytest.mark.asyncio +async def test_duplicate_fixture_skips_enrichment_and_releases_read_transaction(tmp_path): + eml_file = tmp_path / "duplicate.eml" + eml_file.write_text("Message-ID: \n\nBody") + parsed = { + "message_id": "", + "sender": "sender@example.com", + "recipients": "user@example.com", + "subject": "Duplicate", + "date": datetime.datetime.now(datetime.timezone.utc), + "body": "Body", + "attachments": [], + } + session = _RecordingSession(results=[object()]) + + with patch.object(import_fixtures, "parse_eml", return_value=parsed), patch.object( + import_fixtures, + "generate_fixture_embedding", + new_callable=AsyncMock, + ) as embedding_mock, patch.object( + import_fixtures, + "assign_thread_id", + new_callable=AsyncMock, + ) as thread_mock: + imported = await import_fixtures.import_eml_file(session, eml_file) + + assert imported is False + assert session.rollback_count == 1 + assert session.database_transaction_active is False + assert session.committed is False + embedding_mock.assert_not_awaited() + thread_mock.assert_not_awaited() + + +@pytest.mark.postgres +@pytest.mark.asyncio +async def test_attachment_embedding_failure_persists_source_in_real_postgres(tmp_path): + """Persist nullable derived enrichment and prove Email attachment cascade.""" + from asyncpg.exceptions import ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + ) + from sqlalchemy import select, text + from sqlalchemy.exc import OperationalError + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + from sqlalchemy.orm import selectinload + + from core.config import settings + from db.models import Attachment, Base, Email + + engine = create_async_engine(settings.DATABASE_URL) + session_factory = async_sessionmaker(engine, expire_on_commit=False) + message_id = f"" + attachment_filename = f"evidence-{uuid.uuid4().hex}.txt" + attachment_content = "authoritative attachment source content" + eml_file = tmp_path / "attachment-postgres.eml" + eml_file.write_text(f"Message-ID: {message_id}\n\nBody") + parsed = { + "message_id": message_id, + "sender": "sender@example.com", + "recipients": "user@example.com", + "subject": "Attachment PostgreSQL acceptance", + "date": datetime.datetime.now(datetime.timezone.utc), + "body": "Body", + "attachments": [ + { + "filename": attachment_filename, + "content": attachment_content, + } + ], + } + + async def generate_embedding(text_value: str): + if text_value == attachment_content: + raise RuntimeError("embedding provider unavailable") + return [0.0] * import_fixtures.EMBEDDING_DIMENSION + + try: + async with engine.begin() as connection: + await connection.execute(text("SELECT 1")) + await connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector")) + await connection.run_sync(Base.metadata.create_all) + + async with session_factory() as session: + with patch.object( + import_fixtures, "parse_eml", return_value=parsed + ), patch.object( + import_fixtures, + "generate_fixture_embedding", + side_effect=generate_embedding, + ), patch.object( + import_fixtures, + "assign_thread_id", + new_callable=AsyncMock, + return_value="attachment-postgres-thread", + ): + imported = await import_fixtures.import_eml_file(session, eml_file) + + assert imported is True + + async with session_factory() as session: + persisted_email = ( + await session.execute( + select(Email) + .options(selectinload(Email.attachments)) + .where( + Email.message_id == message_id, + Email.user_id == import_fixtures.IMPORT_USER_ID, + Email.organization_id == import_fixtures.IMPORT_ORGANIZATION_ID, + ) + ) + ).scalar_one() + assert len(persisted_email.attachments) == 1 + persisted_attachment = persisted_email.attachments[0] + email_id = persisted_email.id + attachment_id = persisted_attachment.id + assert persisted_attachment.filename == attachment_filename + assert persisted_attachment.content == attachment_content + assert persisted_attachment.embedding is None + + await session.delete(persisted_email) + await session.commit() + + async with session_factory() as session: + assert await session.get(Email, email_id) is None + assert await session.get(Attachment, attachment_id) is None + except ( + InvalidAuthorizationSpecificationError, + InvalidPasswordError, + OperationalError, + OSError, + ) as exc: + pytest.skip(f"PostgreSQL smoke database unavailable: {exc}") + finally: + await engine.dispose() From 23ab3653543240c9c194e98530bb793d037489bc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 16 Sep 2026 01:51:35 +0900 Subject: [PATCH 4/4] docs(import): record security-owner stack and DB evidence boundary --- .../fixture-import-attachment-enrichment.md | 38 +++++++++++++++++++ 1 file changed, 38 insertions(+) create mode 100644 docs/doctoring/fixture-import-attachment-enrichment.md diff --git a/docs/doctoring/fixture-import-attachment-enrichment.md b/docs/doctoring/fixture-import-attachment-enrichment.md new file mode 100644 index 000000000..b6c66867f --- /dev/null +++ b/docs/doctoring/fixture-import-attachment-enrichment.md @@ -0,0 +1,38 @@ +# Fixture import attachment-enrichment invariant + +Date: 2026-09-16 +Status: Proposed repair for #1697, stacked on #1612 security owner + +## Authority and stack boundary + +This repair is an ordinary successor to #1612 (`3da3ae8e60e1bb049f59ae86bfe82db12b7e3cc7`) because both lanes modify `backend/import_fixtures.py`. #1612 owns exception-redaction behavior on that file; this successor preserves its static, non-secret failure logging while adding the attachment/transaction repair from #1697. The earlier direct-develop #1699 branch is predecessor evidence and must not merge independently because it reintroduced protected-base raw exception interpolation relative to #1612. + +## Problem + +The root fixture importer built an `Attachment` only after embedding generation succeeded. When enrichment failed, the exception path logged the failure and then committed the enclosing `Email` without the attachment. A derived vector was therefore acting as an accidental persistence prerequisite for source content. + +The importer also reused one SQLAlchemy session across fixture files. Its duplicate-check `SELECT` starts an implicit transaction, so remote/model embedding must not run until that read transaction has been explicitly ended. Moving the duplicate check after enrichment would avoid an idle transaction but would also waste provider work for already imported fixtures. + +## Invariant + +Parsed source content is authoritative for fixture persistence. Embedding is derived enrichment. + +- Check owner-scoped `message_id` duplication before enrichment. End that read transaction with `rollback()` before any body or attachment embedding call. +- A duplicate fixture returns without invoking enrichment and without leaving an open transaction for the next fixture. +- A parsed attachment remains attached to the Email aggregate when embedding generation fails; its `embedding` is nullable and may be `None`. +- Body embedding remains required by the current importer contract; if it fails, the email is not imported. +- After enrichment, re-check the duplicate guard because another importer may have written during the external-I/O window. From that re-check through thread lookup and commit, only database work remains. +- A database commit failure rolls the aggregate back and returns failure. +- Failure logs retain #1612's non-secret static messages; the attachment repair must not restore raw exception values, filenames, provider responses, or traceback-derived secrets. + +This is intentionally scoped to `backend/import_fixtures.py`; it does not redefine the separate production import service contract. + +## Verification + +`backend/tests/test_import_fixture_attachment_integrity.py` covers the causal ordering with a recording session: attachment enrichment failure preserves the original filename/content, all enrichment occurs after the initial duplicate-read transaction is released, and duplicate fixtures invoke neither enrichment nor thread lookup. + +The same file contains a `@pytest.mark.postgres` acceptance path. Against the real pgvector-backed SQLAlchemy schema it imports an Email whose attachment embedding deliberately fails, reloads the aggregate with `selectinload`, and requires the persisted attachment to retain its original filename/content with `embedding=None`. It then deletes the Email and verifies that the related `email_attachments` row is removed through the model relationship cascade. Unique message/attachment identifiers isolate the smoke row from concurrent tests. + +Existing fixture-import test doubles implement the rollback operation required by the real session protocol so they continue to exercise the same transaction boundary rather than bypassing it. + +The PostgreSQL test is source-backed acceptance, not a fabricated receipt. Merge still requires that this exact test execute successfully against a real PostgreSQL/pgvector service on the final exact head. Until the stacked-PR/migration CI foundation (#1691 and its prerequisites) supplies that hosted database path, a skipped or non-executed PostgreSQL test is not GREEN evidence. Alembic migration compatibility remains a separate required receipt; `Base.metadata.create_all()` in this smoke proves current-model persistence/cascade behavior, not migration-history correctness.