feat(project-graph): activate semantic extraction on import + LLM-grounded extractor (Phases 1–2, folds #927)

backend/core/config.py

@@ -4,6 +4,7 @@
 from pydantic import SecretStr, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from core.env_paths import ENV_FILE_PATHS, operator_env_file_paths
 from core.runtime_secrets import (
     validate_auth_session_hmac_secret_value,
 )
@@ -83,6 +84,13 @@ class Settings(BaseSettings):
     ALLOWED_SCOPEWEAVE_HOSTS: str = ""
     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
+    # Which extractor projects segments into the graph: "keyword"
+    # (deterministic baseline) or "llm" (grounded extraction with enforced
+    # segment citations; falls back to keyword on any failure).
+    PROJECT_GRAPH_EXTRACTOR: str = "keyword"
     DATA_REGION: str = "kr"
     SECONDARY_DATA_REGION: str = "eu"
     SECURITY_CONTENT_SECURITY_POLICY: str = (
@@ -113,11 +121,15 @@ class Settings(BaseSettings):
     ALLOWED_OIDC_HOSTS: str = ""
 
     model_config = SettingsConfigDict(
-        env_file=("~/.env", "../.env", ".env"),
+        env_file=ENV_FILE_PATHS,
         env_file_encoding="utf-8",
         extra="ignore",
     )
 
+    def __init__(self, **values: Any) -> None:
+        values.setdefault("_env_file", operator_env_file_paths())
+        super().__init__(**values)
+
     @field_validator("READONLY_DATABASE_URL", mode="before")
     @classmethod
     def normalize_blank_readonly_database_url(cls, value: Any) -> Any:

backend/core/env_paths.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import os
+from collections.abc import Iterable
+from pathlib import Path
+
+ENV_FILE_PATHS = ("~/.env", "../.env", ".env")
+
+
+def operator_home() -> Path:
+    """Return the operator home directory, preferring explicit HOME overrides."""
+    configured_home = os.environ.get("HOME")
+    if configured_home:
+        return Path(configured_home).expanduser()
+    return Path.home()
+
+
+def expand_operator_path(path: str | os.PathLike[str]) -> Path:
+    """Expand leading ``~`` against the operator home directory."""
+    path_text = os.fspath(path)
+    if path_text == "~":
+        return operator_home()
+    if path_text.startswith("~/") or path_text.startswith("~\\"):
+        return operator_home() / path_text[2:]
+    return Path(path_text).expanduser()
+
+
+def operator_env_file_paths(paths: Iterable[str] = ENV_FILE_PATHS) -> tuple[str, ...]:
+    """Return env-file paths with operator-home expansion applied."""
+    return tuple(str(expand_operator_path(path)) for path in paths)

backend/scripts/private_mail_http_smoke.py

@@ -22,6 +22,12 @@
 from urllib.parse import urlsplit
 from zipfile import BadZipFile, ZipFile
 
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+if str(BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(BACKEND_ROOT))
+
+from core.env_paths import operator_home  # noqa: E402
+
 SESSION_COOKIE_NAME = "naruon_session"
 SUPPORTED_SUFFIXES = {".eml", ".emlx", ".mbox", ".zip"}
 MATCH_SEPARATORS = str.maketrans("", "", " \t\r\n-_./\\()[]{}:;·ㆍ")
@@ -279,7 +285,12 @@ def add_raw(raw: bytes) -> None:
         final_dir = Path(
             os.environ.get(
                 "NARUON_PRIVATE_MAIL_CACHE",
-                str(Path.home() / ".cache" / "naruon" / "private-mail-upload-cache"),
+                str(
+                    operator_home()
+                    / ".cache"
+                    / "naruon"
+                    / "private-mail-upload-cache"
+                ),
             )
         )
         final_dir.mkdir(mode=0o700, parents=True, exist_ok=True)

backend/scripts/start_backend.py

@@ -10,6 +10,7 @@
 if str(BACKEND_ROOT) not in sys.path:
     sys.path.insert(0, str(BACKEND_ROOT))
 
+from core.env_paths import ENV_FILE_PATHS, expand_operator_path  # noqa: E402
 from core.runtime_secrets import validate_auth_session_hmac_secret_value  # noqa: E402
 from core.url_validation import (  # noqa: E402
     parse_allowed_hosts,
@@ -19,7 +20,6 @@
 
 DEFAULT_SERVER_HOST = "127.0.0.1"
 DEFAULT_SERVER_PORT = 8000
-ENV_FILE_PATHS = ("~/.env", "../.env", ".env")
 REQUIRED_SETTINGS = ("DATABASE_URL", "AUTH_SESSION_HMAC_SECRET")
 OIDC_SETTINGS = ("OIDC_ISSUER_URL", "OIDC_CLIENT_ID", "OIDC_JWKS_URL")
 
@@ -55,7 +55,7 @@ def _runtime_values() -> tuple[dict[str, str], list[Path]]:
     values: dict[str, str] = {}
     checked_paths: list[Path] = []
     for env_file in ENV_FILE_PATHS:
-        path = Path(env_file).expanduser()
+        path = expand_operator_path(env_file)
         checked_paths.append(path)
         values.update(_read_env_file(path))
     values.update({key: value for key, value in os.environ.items() if value})

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,
@@ -32,6 +33,12 @@
     generate_embeddings,
 )
 from services.exceptions import ArchiveError, EmailParseError, EmbeddingGenerationError
+from services.project_graph import (
+    ProjectSourceSegment,
+    extract_project_semantics,
+    persist_project_graph_projection,
+)
+from services.project_graph.llm_extractor import extract_project_semantics_llm
 from services.threading_service import (
     assign_thread_id,
     generate_email_fingerprint,
@@ -650,6 +657,103 @@ 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 _extract_project_semantics_for_import(
+    source_segments: list[ProjectSourceSegment],
+    *,
+    embedding_provider: EmailImportEmbeddingProvider | None,
+):
+    """Select the configured extractor; LLM failures fall back to keyword.
+
+    The LLM extractor reuses the import's OpenAI-compatible provider
+    credentials and enforces segment citations, so it cannot introduce
+    uncited claims; any provider/parse failure degrades to the deterministic
+    keyword baseline instead of losing the projection entirely.
+    """
+    if (
+        settings.PROJECT_GRAPH_EXTRACTOR == "llm"
+        and embedding_provider is not None
+        and embedding_provider.api_key
+    ):
+        try:
+            return await extract_project_semantics_llm(
+                source_segments,
+                api_key=embedding_provider.api_key,
+                base_url=embedding_provider.base_url,
+                model=settings.OPENAI_MODEL,
+            )
+        except Exception:
+            logger.warning(
+                "LLM project graph extraction failed; falling back to keyword",
+                exc_info=True,
+            )
+    return extract_project_semantics(source_segments)
+
+
+async def _persist_project_graph_projection(
+    session: AsyncSession,
+    source_segments: list[ProjectSourceSegment],
+    *,
+    user_id: str,
+    organization_id: str,
+    embedding_provider: EmailImportEmbeddingProvider | None = None,
+) -> 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 = await _extract_project_semantics_for_import(
+            source_segments, embedding_provider=embedding_provider
+        )
+        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,
     *,
@@ -711,6 +815,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()
@@ -722,6 +832,14 @@ 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,
+        embedding_provider=embedding_provider,
+    )
+
     return EmailImportItemResult(
         filename=display_filename,
         status="imported",
@@ -828,12 +946,12 @@ async def _generate_import_embeddings(
 
 def _read_eml_bytes(eml_path: Path) -> bytes:
     no_follow_flag = getattr(os, "O_NOFOLLOW", None)
-    if no_follow_flag is None:
-        raise EmailParseError(
-            "Email import requires O_NOFOLLOW support (unavailable on this platform)"
-        )
+    if no_follow_flag is None and eml_path.is_symlink():
+        raise EmailParseError("Failed to read email file")
 
-    open_flags = os.O_RDONLY | no_follow_flag
+    open_flags = os.O_RDONLY
+    if no_follow_flag is not None:
+        open_flags |= no_follow_flag
     file_descriptor_transferred = False
     try:
         file_descriptor = os.open(eml_path, open_flags)