feat(project-graph): activate semantic extraction on email import (flag-gated)

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 = (

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-<organization_id>``).
+    """
+    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",

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()