diff --git a/backend/core/config.py b/backend/core/config.py index d630c45f8..ca4ed4801 100644 --- a/backend/core/config.py +++ b/backend/core/config.py @@ -78,6 +78,9 @@ class Settings(BaseSettings): ALLOW_LOCAL_LLM_PROVIDERS: bool = False ALLOWED_CORS_ORIGINS: str = "" ENABLE_PROMETHEUS_METRICS: bool = False + # Best-effort projection of imported-email content segments into the project + # semantic graph. Off by default; failure never affects email import. + PROJECT_GRAPH_EXTRACTION_ENABLED: bool = False DATA_REGION: str = "kr" SECONDARY_DATA_REGION: str = "eu" SECURITY_CONTENT_SECURITY_POLICY: str = ( diff --git a/backend/services/email_import_service.py b/backend/services/email_import_service.py index c699d6182..118165250 100644 --- a/backend/services/email_import_service.py +++ b/backend/services/email_import_service.py @@ -14,6 +14,7 @@ from sqlalchemy import bindparam, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession +from core.config import settings from db.models import ( Attachment, ContentNodeRecord, @@ -31,6 +32,11 @@ generate_embeddings, ) from services.exceptions import ArchiveError, EmailParseError, EmbeddingGenerationError +from services.project_graph import ( + ProjectSourceSegment, + extract_project_semantics, + persist_project_graph_projection, +) from services.threading_service import ( assign_thread_id, generate_email_fingerprint, @@ -625,6 +631,68 @@ def _content_graph_source_record_uid(prefix: str, *parts: str) -> str: 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. + + Called before commit while ``email_obj.content_segments`` is the in-memory + list populated by ``_append_email_content_graph`` (no DB IO), so it is safe + to build in the async session context. + """ + return [ + ProjectSourceSegment( + content_segment_uid=segment.content_segment_uid, + source_kind=segment.source_kind, + source_record_uid=segment.source_record_uid, + safe_text_content=segment.safe_text_content, + heading_path=segment.heading_path, + segment_path=segment.segment_path, + ordinal_index=segment.ordinal_index, + ) + for segment in email_obj.content_segments + ] + + +async def _persist_project_graph_projection( + session: AsyncSession, + source_segments: list[ProjectSourceSegment], + *, + user_id: str, + organization_id: str, +) -> None: + """Best-effort projection of imported content segments into the project graph. + + Runs after the email is already committed. Flag-gated and defensive: any + failure is logged and rolled back so it never fails the email import. The + workspace scope mirrors the convention enforced by the project graph + repository (``workspace-``). + """ + if not source_segments: + return + try: + extraction = extract_project_semantics(source_segments) + if not extraction.objects: + return + workspace_id = ( + f"workspace-{organization_id}" + if organization_id + else f"workspace-{user_id}" + ) + await persist_project_graph_projection( + session, + extraction=extraction, + user_id=user_id, + organization_id=organization_id, + workspace_id=workspace_id, + ) + await session.commit() + except Exception: + await session.rollback() + logger.warning( + "Project graph projection skipped for imported email", + exc_info=True, + ) + + async def _import_single_eml( session: AsyncSession, *, @@ -685,6 +753,12 @@ async def _import_single_eml( fitted_embeddings=fitted_embeddings, ) + project_source_segments = ( + _project_source_segments(email_obj) + if settings.PROJECT_GRAPH_EXTRACTION_ENABLED + else [] + ) + session.add(email_obj) try: await session.commit() @@ -696,6 +770,13 @@ async def _import_single_eml( reason_code="database_commit_failed", ) + await _persist_project_graph_projection( + session, + project_source_segments, + user_id=user_id, + organization_id=organization_id, + ) + return EmailImportItemResult( filename=display_filename, status="imported", diff --git a/backend/tests/test_project_graph_import_wiring.py b/backend/tests/test_project_graph_import_wiring.py new file mode 100644 index 000000000..604739e09 --- /dev/null +++ b/backend/tests/test_project_graph_import_wiring.py @@ -0,0 +1,135 @@ +"""Tests for wiring project-graph extraction into the email import pipeline. + +The projection is flag-gated and best-effort: it must never affect the (already +committed) email import. These tests exercise the real deterministic extractor +and mock only the DB persistence layer. +""" + +import types + +import pytest +from unittest.mock import AsyncMock + +import services.email_import_service as import_service +from services.email_import_service import ( + _persist_project_graph_projection, + _project_source_segments, +) + + +def _segment(uid: str, text: str, ordinal: int = 0): + return types.SimpleNamespace( + content_segment_uid=uid, + source_kind="email_body", + source_record_uid="email:1", + safe_text_content=text, + heading_path=None, + segment_path=f"body/{ordinal}", + ordinal_index=ordinal, + ) + + +def test_project_source_segments_maps_content_segments(): + email_obj = types.SimpleNamespace( + content_segments=[_segment("seg1", "hello", 0), _segment("seg2", "world", 1)] + ) + + result = _project_source_segments(email_obj) + + assert [segment.content_segment_uid for segment in result] == ["seg1", "seg2"] + assert result[0].safe_text_content == "hello" + assert result[0].source_kind == "email_body" + assert result[1].ordinal_index == 1 + + +@pytest.mark.asyncio +async def test_projection_persists_with_workspace_scope_when_objects_found(monkeypatch): + persist_mock = AsyncMock() + monkeypatch.setattr( + import_service, "persist_project_graph_projection", persist_mock + ) + session = AsyncMock() + segments = [ + _segment("seg1", "The system must support export. This is a requirement.", 0) + ] + + await _persist_project_graph_projection( + session, segments, user_id="user1", organization_id="org1" + ) + + persist_mock.assert_awaited_once() + kwargs = persist_mock.await_args.kwargs + assert kwargs["user_id"] == "user1" + assert kwargs["organization_id"] == "org1" + # Mirrors the scope convention enforced by the project graph repository. + assert kwargs["workspace_id"] == "workspace-org1" + assert kwargs["extraction"].objects # real extractor produced candidates + session.commit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_projection_falls_back_to_user_workspace_without_org(monkeypatch): + persist_mock = AsyncMock() + monkeypatch.setattr( + import_service, "persist_project_graph_projection", persist_mock + ) + session = AsyncMock() + segments = [_segment("seg1", "We must deliver the milestone by 2026-01-01.", 0)] + + await _persist_project_graph_projection( + session, segments, user_id="user1", organization_id="" + ) + + kwargs = persist_mock.await_args.kwargs + assert kwargs["workspace_id"] == "workspace-user1" + + +@pytest.mark.asyncio +async def test_projection_noop_when_no_segments(monkeypatch): + persist_mock = AsyncMock() + monkeypatch.setattr( + import_service, "persist_project_graph_projection", persist_mock + ) + session = AsyncMock() + + await _persist_project_graph_projection( + session, [], user_id="u", organization_id="o" + ) + + persist_mock.assert_not_awaited() + session.commit.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_projection_noop_when_no_objects_extracted(monkeypatch): + persist_mock = AsyncMock() + monkeypatch.setattr( + import_service, "persist_project_graph_projection", persist_mock + ) + session = AsyncMock() + # Neutral text with no rule keywords -> extractor yields nothing. + segments = [_segment("seg1", "hello there, nice weather today", 0)] + + await _persist_project_graph_projection( + session, segments, user_id="u", organization_id="o" + ) + + persist_mock.assert_not_awaited() + session.commit.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_projection_swallows_failure_and_rolls_back(monkeypatch): + persist_mock = AsyncMock(side_effect=RuntimeError("boom")) + monkeypatch.setattr( + import_service, "persist_project_graph_projection", persist_mock + ) + session = AsyncMock() + segments = [_segment("seg1", "The system must support export requirement.", 0)] + + # Best-effort: a projection failure must not propagate to the import. + await _persist_project_graph_projection( + session, segments, user_id="u", organization_id="o" + ) + + session.rollback.assert_awaited_once()