merge: integrate existing remote master history

README.md

@@ -43,6 +43,62 @@ npm run dev        # 开发模式（默认代理到本机 8000）
 
 前端通过 `VITE_API_BASE_URL` 指定后端地址（默认同源 `/ai-api/resume-agent`）。
 
+### OfferPai 账号入口
+
+后端在 `.env` 保持 `OFFERPAI_AUTH_REQUIRED=true`，前端必须通过以下带 Token 的
+地址进入：
+
+```text
+http://localhost:5173/?token=<OfferPai Token>
+```
+
+前端会立即从地址栏移除 `token`，将它只保存在当前页面内存中，并通过
+`Authorization: Bearer` 附加到后续每个 API 请求。后端先调用 OfferPai 的
+`checkLogin` 和用户信息接口，并校验当前账号拥有对应 session；鉴权成功后才创建、恢复
+或修改会话。前端工作台也只会在取得受认证的 session 后挂载；缺少 Token 时返回
+`401 external_auth_required`，鉴权失败时不会进入服务。
+同一 OfferPai 用户再次从带 Token 的入口进入时会恢复其最近会话。
+
+后端的账号资料只保存外部用户 ID、昵称和默认手机号，不保存原始 Token；另外会保存不含
+Token 的 C 端同步元数据（远端简历 ID、同步状态、revision、内容 hash 和错误码）。默认
+手机号会在手机号选择卡中自动选中，用户仍可改填其他手机号。页面每次完整重新加载都必须
+重新携带 `?token=`。
+
+### OfferPai C 端简历镜像
+
+后端通过以下配置调用 OfferPai C 端简历接口：
+
+```dotenv
+OFFERPAI_RESUME_API_BASE_URL=https://test.offerpai.com.cn/api
+OFFERPAI_RESUME_TIMEOUT_SECONDS=8
+```
+
+手动创建流程进入 `MINIMUM_READY`（完成姓名、邮箱、手机号、求职类型和目标岗位）时，
+系统会自动创建本地工作文档与 OfferPai C 端简历，不再等待用户额外点击“创建简历”。
+之后已确认的主表信息及教育、工作、实习、项目、竞赛五类经历会继续同步到 C 端；导入
+简历确认后也会执行相同同步。候选优化稿、待确认个人总结等 Agent 中间结果不会作为正式
+简历同步。local DB 仍保留会话 FSM、对话与组件生命周期、导入和优化任务状态，以及
+简历 revision 缓存，用于并发校验、恢复会话和失败重试。
+
+简历同步是双向的：时间线读取会先用 `/resume/list` 的 `updateTime` 检查远端是否变化，
+变化后再读取主表和五个经历分区，并把 OfferPai 支持的字段拉回当前简历；教育、工作、
+实习、项目、竞赛条目的段落 ID 用于保留本地条目 ID，不支持的本地分区继续留在本地。
+本地编辑则使用“上次共同版本 B / 当前本地投影 L / 当前远端快照 R”判断：只远端变化时
+拉取，只本地变化时推送，两边都变化时返回冲突（409），不会静默覆盖另一侧。前端工作台
+会每 20 秒轮询时间线，并在窗口重新聚焦或恢复可见时立即检查。
+
+暂时不能用 C 端接口完全替代 local DB：远程 HTTP 写入无法与本地会话事务原子提交，
+且 Agent 仍依赖稳定的 section/entry/bullet ID、乐观 revision、pending proposal、撤销版本
+和优化运行状态。只有在 C 端接口支持幂等写入、条件版本更新、稳定子项 ID，并完成远程
+写入与本地状态的补偿/对账机制后，才适合进一步缩减本地简历缓存。
+
+写操作会在同一 session 锁内完成远端预检查、本地修改和条件推送；远程 HTTP 仍无法与
+本地数据库事务组成真正的跨系统原子提交，因此 C 端超时或拒绝时，本地 revision 可能
+已经更新，失败状态会记录在 `profile.external_resume`，后续进入或读取会话时会按基线
+继续补偿。远端被删除时读取不会自动重建或覆盖未知内容；需要先处理冲突/删除状态后再
+重新开始。“重新开始”会先删除已绑定的 C 端镜像，再删除本地 session；远端已不存在按
+幂等成功处理。
+
 ## 测试
 
 ```bash
@@ -55,5 +111,6 @@ npm run typecheck && npm run build
 ## 部署与安全基线
 
 见 [docs/DEPLOY.md](docs/DEPLOY.md)。要点：试点期单进程 + 单 Postgres 即可，无需容器编排；
-服务无内置认证，必须放在内网或网关之后；不要在环境变量中设置
+所有 session、简历编辑、SSE 和导入接口都会逐请求校验 OfferPai Token 及 session 所属用户；
+仍建议部署在 HTTPS 网关之后。不要在环境变量中设置
 `RESUME_AGENT_DEFAULT_TIER=vip`（会把全量会话提权）。

backend/.env.example

@@ -1,9 +1,9 @@
 # ===== LLM — production uses Volcengine Ark over the OpenAI-compatible protocol =====
 RESUME_AGENT_LLM_PROVIDER=volcengine
-VOLCENGINE_API_KEY=
+VOLCENGINE_API_KEY=replace-with-volcengine-api-key
 VOLCENGINE_BASE_URL=https://ark.cn-beijing.volces.com/api/v3
 # A plain model name (as below) or an Ark inference endpoint ID (ep-xxxxxxxx) both work.
-VOLCENGINE_MODEL=deepseek-v4-flash-260425
+VOLCENGINE_MODEL=doubao-seed-2-0-lite-260215
 
 # Alternative: any OpenAI-compatible gateway. With provider=auto the LLM is used
 # only when OPENAI_API_KEY is non-empty; otherwise requests fall back to rules.
@@ -27,9 +27,16 @@ RESUME_AGENT_LLM_FALLBACK_TO_RULES=false
 RESUME_AGENT_INTENT_ROUTER_MODE=on
 # RESUME_AGENT_INTENT_MODEL=  # dedicated intent-classifier model; defaults to the main model
 
+# ===== OfferPai account integration =====
+OFFERPAI_AUTH_BASE_URL=https://test.offerpai.com.cn
+OFFERPAI_AUTH_TIMEOUT_SECONDS=8
+OFFERPAI_AUTH_REQUIRED=true
+OFFERPAI_RESUME_API_BASE_URL=https://test.offerpai.com.cn/api
+OFFERPAI_RESUME_TIMEOUT_SECONDS=8
+
 # ===== Database — PostgreSQL is required at runtime (create_app fails without it) =====
-DATABASE_URL=postgresql+psycopg://resume_agent:change-me@127.0.0.1:5435/resume_agent
-RESUME_AGENT_TEST_DATABASE_URL=postgresql+psycopg://resume_agent:change-me@127.0.0.1:5435/resume_agent_test
+DATABASE_URL=postgresql+psycopg://postgres:replace-with-password@127.0.0.1:5432/postgres
+RESUME_AGENT_TEST_DATABASE_URL=postgresql+psycopg://postgres:replace-with-password@127.0.0.1:5432/resume_agent_test
 # Schema the tables live in; defaults to resume_agent. Set per environment when
 # several deployments share one database.
 # RESUME_AGENT_DATABASE_SCHEMA=resume_agent

backend/README.md

@@ -23,6 +23,14 @@ the tables, and `python scripts/migrate_sqlite_to_postgres.py` to move existing
 SQLite is used only when `database_path` is passed explicitly (tests, local pilot).
 CORS defaults to `http://localhost:5173`; set a comma-separated `RESUME_AGENT_CORS_ORIGINS`.
 
+Set `OFFERPAI_AUTH_BASE_URL` to enable the OfferPai landing-token bridge. The frontend
+passes the in-memory landing token on every session request via `Authorization: Bearer`;
+the backend validates it with `GET /api/public/checkLogin`, loads
+`GET /api/user/manage/info` using the upstream `Token` cookie, and persists the account
+identity/default phone without persisting the token itself. Resume content uses a
+three-way OfferPai reconciliation baseline: remote-only changes are pulled, local-only
+changes are pushed, and concurrent changes return a conflict instead of overwriting.
+
 The health check at `GET /health` is always open.
 
 ## Tests

backend/app/agent.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
 from copy import deepcopy
+from dataclasses import dataclass
+from functools import wraps
 import logging
-from typing import Any
+from threading import Lock
+from typing import Any, Callable
 from uuid import uuid4
 
 from .database import Database
@@ -41,9 +44,43 @@ from .services import EntryExpander, ExperienceExtractor, ResumeRewriter
 from .profile_summary import ProfileSummaryGenerator, RuleBasedProfileSummaryGenerator
 from .skill_suggester import SkillSuggester
 from .resume_skill_advisor import recommend_skill_candidates
+from .offerpai_auth import (
+    OfferPaiAuthError,
+    OfferPaiIdentity,
+    OfferPaiIdentityProvider,
+)
+from .offerpai_resume import (
+    SUPPORTED_SECTION_KINDS,
+    OfferPaiResumeError,
+    OfferPaiResumeProvider,
+    build_offerpai_resume_payload,
+    merge_offerpai_resume_snapshot,
+    offerpai_payload_hash,
+    offerpai_update_marker,
+)
 from . import builder_conversation
 
 
+@dataclass(frozen=True)
+class OfferPaiReconcileResult:
+    remote_id: str | None
+    action: str
+
+
+def _serialize_offerpai_session(method: Any) -> Any:
+    """Serialize OfferPai writes for one session inside the running process."""
+
+    @wraps(method)
+    def wrapped(
+        self: "ResumeAgent", session_id: str, *args: Any, **kwargs: Any
+    ) -> Any:
+        locks = self._offerpai_session_locks
+        with locks[hash(session_id) % len(locks)]:
+            return method(self, session_id, *args, **kwargs)
+
+    return wrapped
+
+
 class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
     def __init__(
         self,
@@ -55,6 +92,8 @@ class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
         experience_optimizer: ExperienceOptimizer | None = None,
         target_position_suggester: TargetPositionSuggester | None = None,
         profile_summary_generator: ProfileSummaryGenerator | None = None,
+        offerpai_identity_provider: OfferPaiIdentityProvider | None = None,
+        offerpai_resume_provider: OfferPaiResumeProvider | None = None,
     ) -> None:
         self.database = database
         self.extractor = extractor
@@ -64,6 +103,12 @@ class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
         self.experience_optimizer = experience_optimizer or RuleStructuredExperienceOptimizer()
         self.target_position_suggester = target_position_suggester
         self.profile_summary_generator = profile_summary_generator or RuleBasedProfileSummaryGenerator()
+        self.offerpai_identity_provider = offerpai_identity_provider
+        self.offerpai_resume_provider = offerpai_resume_provider
+        # The documented pilot deployment is a single process.  Striped locks
+        # prevent duplicate creates and stale-over-new writes from concurrent
+        # requests without retaining one lock object per session forever.
+        self._offerpai_session_locks = tuple(Lock() for _ in range(64))
 
     def recommend_skills(self, session_id: str, question: str) -> list[dict[str, Any]]:
         with self.database.transaction() as connection:
@@ -76,14 +121,743 @@ class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
             if str(skill).strip()
         ]
         return recommend_skill_candidates(session["profile"], existing, question, self.skill_suggester)
-    def create_session(self, request: CreateSessionRequest) -> TimelineResponse:
+
+    def authenticate_external_token(self, external_token: str) -> OfferPaiIdentity:
+        if self.offerpai_identity_provider is None:
+            raise FSMError(
+                "external_auth_unavailable",
+                "OfferPai 账号服务尚未配置。",
+                status_code=503,
+            )
+        try:
+            return self.offerpai_identity_provider.authenticate(external_token)
+        except OfferPaiAuthError as exc:
+            raise FSMError(
+                exc.code,
+                exc.public_message,
+                status_code=exc.status_code,
+            ) from exc
+
+    def authorize_session(
+        self, session_id: str, external_token: str
+    ) -> OfferPaiIdentity:
+        identity = self.authenticate_external_token(external_token)
+        session = self.database.get_session(session_id)
+        if session is None:
+            raise FSMError("session_not_found", "Session not found", status_code=404)
+        external_account = session["profile"].get("external_account")
+        if not isinstance(external_account, dict) or (
+            external_account.get("provider") != "offerpai"
+            or str(external_account.get("user_id") or "").strip()
+            != identity.user_id
+        ):
+            raise FSMError(
+                "external_auth_forbidden",
+                "当前 OfferPai 账号无权访问该简历会话。",
+                status_code=403,
+            )
+        return identity
+
+    def sync_resume_to_offerpai(
+        self,
+        session_id: str,
+        external_token: str,
+        *,
+        raise_on_error: bool = True,
+    ) -> str | None:
+        result = self.reconcile_resume_with_offerpai(
+            session_id,
+            external_token,
+            raise_on_error=raise_on_error,
+        )
+        return result.remote_id
+
+    @_serialize_offerpai_session
+    def reconcile_resume_with_offerpai(
+        self,
+        session_id: str,
+        external_token: str,
+        *,
+        raise_on_error: bool = True,
+    ) -> OfferPaiReconcileResult:
+        return self._reconcile_resume_with_offerpai_unlocked(
+            session_id,
+            external_token,
+            raise_on_error=raise_on_error,
+        )
+
+    @_serialize_offerpai_session
+    def mutate_with_offerpai_sync(
+        self,
+        session_id: str,
+        external_token: str,
+        mutation: Callable[[], Any],
+        *,
+        allow_pulled_resume: bool = False,
+    ) -> Any:
+        """Hold the per-session sync lock across preflight, mutation and push."""
+
+        result = self._reconcile_resume_with_offerpai_unlocked(
+            session_id,
+            external_token,
+            raise_on_error=True,
+        )
+        if result.action == "pulled" and not allow_pulled_resume:
+            raise FSMError(
+                "offerpai_resume_changed",
+                "The resume changed in OfferPai. Refresh before continuing.",
+                status_code=409,
+            )
+        response = mutation()
+        post_result = self._reconcile_resume_with_offerpai_unlocked(
+            session_id,
+            external_token,
+            raise_on_error=True,
+        )
+        if post_result.action == "pulled":
+            raise FSMError(
+                "offerpai_resume_changed",
+                "The resume changed in OfferPai. Refresh before continuing.",
+                status_code=409,
+            )
+        return response
+
+    def _reconcile_resume_with_offerpai_unlocked(
+        self,
+        session_id: str,
+        external_token: str,
+        *,
+        raise_on_error: bool,
+    ) -> OfferPaiReconcileResult:
+        provider = self.offerpai_resume_provider
+        if provider is None:
+            if raise_on_error:
+                raise FSMError(
+                    "offerpai_resume_unavailable",
+                    "OfferPai resume storage is not configured.",
+                    status_code=503,
+                )
+            return OfferPaiReconcileResult(None, "error")
+
+        with self.database.transaction() as connection:
+            session = self.database.fetch_session(connection, session_id)
+            if session is None:
+                raise FSMError("session_not_found", "Session not found", status_code=404)
+            resume = self.database.fetch_resume(connection, session_id)
+        if resume is None:
+            return OfferPaiReconcileResult(None, "none")
+
+        profile = deepcopy(session["profile"])
+        sync_state = (
+            deepcopy(profile.get("external_resume"))
+            if isinstance(profile.get("external_resume"), dict)
+            else {}
+        )
+        resume_name = str(sync_state.get("resume_name") or "").strip() or (
+            self._offerpai_resume_name(session_id, profile)
+        )
+        remote_id = str(sync_state.get("id") or "").strip() or None
+        main_payload, section_payloads, unsupported = build_offerpai_resume_payload(
+            resume["content"],
+            profile,
+            resume_name=resume_name,
+            resume_id=remote_id,
+        )
+        local_hash = offerpai_payload_hash(main_payload, section_payloads)
+
+        try:
+            listed_resumes = provider.list_resumes(external_token)
+            listed_remote: dict[str, Any] | None = None
+            if remote_id is not None:
+                listed_remote = next(
+                    (
+                        item
+                        for item in listed_resumes
+                        if str(item.get("id") or "").strip() == remote_id
+                    ),
+                    None,
+                )
+                if listed_remote is None:
+                    raise OfferPaiResumeError(
+                        "offerpai_resume_not_found",
+                        "The OfferPai resume was deleted or is no longer accessible.",
+                        status_code=404,
+                    )
+            else:
+                if sync_state.get("status") == "remote_deleted" or (
+                    sync_state.get("last_error_code") == "offerpai_resume_not_found"
+                    and sync_state.get("resume_name")
+                ):
+                    raise OfferPaiResumeError(
+                        "offerpai_resume_not_found",
+                        "The OfferPai resume was deleted or is no longer accessible.",
+                        status_code=404,
+                    )
+                matches = [
+                    item
+                    for item in listed_resumes
+                    if str(item.get("resumeName") or "").strip() == resume_name
+                    and str(item.get("id") or "").strip()
+                ]
+                if len(matches) > 1:
+                    raise OfferPaiResumeError(
+                        "offerpai_resume_conflict",
+                        "Multiple OfferPai resumes have the same generated name.",
+                        status_code=409,
+                    )
+                if matches:
+                    listed_remote = matches[0]
+                    remote_id = str(listed_remote["id"])
+                    sync_state = {
+                        **sync_state,
+                        "provider": "offerpai",
+                        "id": remote_id,
+                        "resume_name": resume_name,
+                    }
+                else:
+                    return self._push_resume_to_offerpai_unlocked(
+                        session_id=session_id,
+                        external_token=external_token,
+                        resume=resume,
+                        sync_state=sync_state,
+                        resume_name=resume_name,
+                        remote_id=None,
+                        main_payload=main_payload,
+                        section_payloads=section_payloads,
+                        unsupported=unsupported,
+                        payload_hash=local_hash,
+                    )
+
+            marker = offerpai_update_marker(listed_remote or {})
+            baseline_hash = (
+                str(sync_state.get("payload_hash") or "").strip() or None
+                if sync_state.get("payload_hash_version") == 2
+                else None
+            )
+            if (
+                baseline_hash is not None
+                and marker is not None
+                and marker == sync_state.get("update_marker")
+                and sync_state.get("status") in {"synced", "partial"}
+            ):
+                if local_hash != baseline_hash:
+                    return self._push_resume_to_offerpai_unlocked(
+                        session_id=session_id,
+                        external_token=external_token,
+                        resume=resume,
+                        sync_state=sync_state,
+                        resume_name=resume_name,
+                        remote_id=remote_id,
+                        main_payload=main_payload,
+                        section_payloads=section_payloads,
+                        unsupported=unsupported,
+                        payload_hash=local_hash,
+                    )
+                refreshed_state = {
+                    **sync_state,
+                    "provider": "offerpai",
+                    "id": remote_id,
+                    "resume_name": resume_name,
+                    "status": "partial" if unsupported else "synced",
+                    "synced_revision": resume["revision"],
+                    "payload_hash": baseline_hash,
+                    "payload_hash_version": 2,
+                    "update_marker": marker,
+                    "unsupported_sections": list(unsupported),
+                    "last_error_code": None,
+                }
+                if refreshed_state != sync_state:
+                    self._update_external_resume_state(session_id, refreshed_state)
+                return OfferPaiReconcileResult(remote_id, "unchanged")
+
+            remote_main, remote_sections, marker = (
+                self._read_stable_offerpai_snapshot_unlocked(
+                    external_token,
+                    remote_id,
+                    initial_marker=marker,
+                )
+            )
+            remote_hash = offerpai_payload_hash(remote_main, remote_sections)
+            remote_resume_name = (
+                str(remote_main.get("resumeName") or "").strip() or resume_name
+            )
+
+            if local_hash == remote_hash:
+                converged_state = {
+                    **sync_state,
+                    "provider": "offerpai",
+                    "id": remote_id,
+                    "resume_name": remote_resume_name,
+                    "status": "partial" if unsupported else "synced",
+                    "synced_revision": resume["revision"],
+                    "payload_hash": remote_hash,
+                    "payload_hash_version": 2,
+                    "update_marker": marker,
+                    "unsupported_sections": list(unsupported),
+                    "last_error_code": None,
+                }
+                self._update_external_resume_state(session_id, converged_state)
+                return OfferPaiReconcileResult(remote_id, "converged")
+
+            if baseline_hash is None:
+                legacy_local_unchanged = (
+                    sync_state.get("status") in {"synced", "partial"}
+                    and sync_state.get("synced_revision") == resume["revision"]
+                )
+                if legacy_local_unchanged:
+                    return self._pull_offerpai_resume_unlocked(
+                        session_id=session_id,
+                        source_session=session,
+                        source_resume=resume,
+                        sync_state=sync_state,
+                        remote_id=remote_id,
+                        remote_main=remote_main,
+                        remote_sections=remote_sections,
+                        remote_hash=remote_hash,
+                        update_marker=marker,
+                    )
+                recoverable_uncertain_create = (
+                    sync_state.get("status") in {"failed", "syncing"}
+                    and sync_state.get("last_error_code")
+                    in {
+                        None,
+                        "offerpai_resume_timeout",
+                        "offerpai_resume_unavailable",
+                        "offerpai_resume_invalid_response",
+                    }
+                )
+                if recoverable_uncertain_create:
+                    return self._push_resume_to_offerpai_unlocked(
+                        session_id=session_id,
+                        external_token=external_token,
+                        resume=resume,
+                        sync_state=sync_state,
+                        resume_name=resume_name,
+                        remote_id=remote_id,
+                        main_payload=main_payload,
+                        section_payloads=section_payloads,
+                        unsupported=unsupported,
+                        payload_hash=local_hash,
+                    )
+                raise OfferPaiResumeError(
+                    "offerpai_resume_conflict",
+                    "Local and OfferPai resumes differ and have no shared sync baseline.",
+                    status_code=409,
+                )
+
+            if local_hash == baseline_hash and remote_hash != baseline_hash:
+                return self._pull_offerpai_resume_unlocked(
+                    session_id=session_id,
+                    source_session=session,
+                    source_resume=resume,
+                    sync_state=sync_state,
+                    remote_id=remote_id,
+                    remote_main=remote_main,
+                    remote_sections=remote_sections,
+                    remote_hash=remote_hash,
+                    update_marker=marker,
+                )
+
+            if remote_hash == baseline_hash and local_hash != baseline_hash:
+                return self._push_resume_to_offerpai_unlocked(
+                    session_id=session_id,
+                    external_token=external_token,
+                    resume=resume,
+                    sync_state=sync_state,
+                    resume_name=resume_name,
+                    remote_id=remote_id,
+                    main_payload=main_payload,
+                    section_payloads=section_payloads,
+                    unsupported=unsupported,
+                    payload_hash=local_hash,
+                )
+
+            raise OfferPaiResumeError(
+                "offerpai_resume_conflict",
+                "The resume changed both locally and in OfferPai. Refresh and resolve the conflict.",
+                status_code=409,
+            )
+        except OfferPaiResumeError as exc:
+            self._record_offerpai_failure(
+                session_id,
+                sync_state=sync_state,
+                resume_name=resume_name,
+                remote_id=remote_id,
+                error=exc,
+            )
+            if raise_on_error:
+                raise FSMError(
+                    exc.code,
+                    exc.public_message,
+                    status_code=exc.status_code,
+                ) from exc
+            return OfferPaiReconcileResult(
+                remote_id,
+                "conflict" if exc.code == "offerpai_resume_conflict" else "error",
+            )
+
+    def _push_resume_to_offerpai_unlocked(
+        self,
+        *,
+        session_id: str,
+        external_token: str,
+        resume: dict[str, Any],
+        sync_state: dict[str, Any],
+        resume_name: str,
+        remote_id: str | None,
+        main_payload: dict[str, Any],
+        section_payloads: dict[str, dict[str, Any]],
+        unsupported: tuple[str, ...],
+        payload_hash: str,
+    ) -> OfferPaiReconcileResult:
+        provider = self.offerpai_resume_provider
+        assert provider is not None
+        created_now = remote_id is None
+        if created_now:
+            if not provider.can_create(external_token):
+                raise OfferPaiResumeError(
+                    "offerpai_resume_limit_reached",
+                    "The OfferPai resume limit has been reached.",
+                    status_code=409,
+                )
+            remote_id = provider.save_main(external_token, main_payload)
+            sync_state = {
+                **sync_state,
+                "provider": "offerpai",
+                "id": remote_id,
+                "resume_name": resume_name,
+                "status": "syncing",
+                "last_error_code": None,
+            }
+            self._update_external_resume_state(session_id, sync_state)
+        else:
+            main_payload = {**main_payload, "resumeId": remote_id}
+            provider.save_main(external_token, main_payload)
+
+        for kind in SUPPORTED_SECTION_KINDS:
+            provider.replace_section(
+                external_token,
+                kind,
+                resume_id=remote_id,
+                items=section_payloads[kind].get("items") or [],
+            )
+
+        listed = provider.list_resumes(external_token)
+        remote_listing = next(
+            (
+                item
+                for item in listed
+                if str(item.get("id") or "").strip() == remote_id
+            ),
+            None,
+        )
+        if remote_listing is None:
+            raise OfferPaiResumeError(
+                "offerpai_resume_not_found",
+                "The OfferPai resume disappeared while it was being saved.",
+                status_code=404,
+            )
+        update_marker = offerpai_update_marker(remote_listing)
+        verified_main, verified_sections, update_marker = (
+            self._read_stable_offerpai_snapshot_unlocked(
+                external_token,
+                remote_id,
+                initial_marker=update_marker,
+            )
+        )
+        verified_hash = offerpai_payload_hash(verified_main, verified_sections)
+        if verified_hash != payload_hash:
+            raise OfferPaiResumeError(
+                "offerpai_resume_conflict",
+                "The OfferPai resume changed while it was being saved; please retry.",
+                status_code=409,
+            )
+
+        final_state = {
+            **sync_state,
+            "provider": "offerpai",
+            "id": remote_id,
+            "resume_name": resume_name,
+            "status": "partial" if unsupported else "synced",
+            "synced_revision": resume["revision"],
+            "payload_hash": payload_hash,
+            "payload_hash_version": 2,
+            "update_marker": update_marker,
+            "unsupported_sections": list(unsupported),
+            "last_error_code": None,
+        }
+        self._update_external_resume_state(session_id, final_state)
+        return OfferPaiReconcileResult(
+            remote_id,
+            "created" if created_now else "pushed",
+        )
+
+    def _read_stable_offerpai_snapshot_unlocked(
+        self,
+        external_token: str,
+        remote_id: str,
+        *,
+        initial_marker: str | None,
+    ) -> tuple[
+        dict[str, Any],
+        dict[str, list[dict[str, Any]]],
+        str | None,
+    ]:
+        """Read the six remote resources and reject a torn multi-request snapshot.
+
+        OfferPai does not expose a conditional version for these endpoints.  The
+        list endpoint's ``updateTime`` is therefore used as a best-effort
+        read-version: when it changes during the six reads, retry once and then
+        surface a conflict instead of merging fields from different versions.
+        """
+
+        provider = self.offerpai_resume_provider
+        assert provider is not None
+        marker_before = initial_marker
+        for attempt in range(2):
+            remote_main = provider.get_main(external_token, remote_id)
+            remote_sections = {
+                kind: provider.list_section(
+                    external_token,
+                    kind,
+                    resume_id=remote_id,
+                )
+                for kind in SUPPORTED_SECTION_KINDS
+            }
+            listed = provider.list_resumes(external_token)
+            remote_listing = next(
+                (
+                    item
+                    for item in listed
+                    if str(item.get("id") or "").strip() == remote_id
+                ),
+                None,
+            )
+            if remote_listing is None:
+                raise OfferPaiResumeError(
+                    "offerpai_resume_not_found",
+                    "The OfferPai resume was deleted or is no longer accessible.",
+                    status_code=404,
+                )
+            marker_after = offerpai_update_marker(remote_listing)
+            if marker_before is not None and marker_after != marker_before:
+                marker_before = marker_after
+                continue
+            if marker_before is None and marker_after is not None and attempt == 0:
+                # There was no trusted marker before the read.  Take one more
+                # snapshot so two equal post-read markers establish stability.
+                marker_before = marker_after
+                continue
+            return remote_main, remote_sections, marker_after
+
+        raise OfferPaiResumeError(
+            "offerpai_resume_conflict",
+            "The OfferPai resume changed while it was being read; please retry.",
+            status_code=409,
+        )
+
+    def _pull_offerpai_resume_unlocked(
+        self,
+        *,
+        session_id: str,
+        source_session: dict[str, Any],
+        source_resume: dict[str, Any],
+        sync_state: dict[str, Any],
+        remote_id: str,
+        remote_main: dict[str, Any],
+        remote_sections: dict[str, list[dict[str, Any]]],
+        remote_hash: str,
+        update_marker: str | None,
+    ) -> OfferPaiReconcileResult:
+        merged_content, merged_profile, unsupported = merge_offerpai_resume_snapshot(
+            source_resume["content"],
+            source_session["profile"],
+            remote_main,
+            remote_sections,
+        )
+        self._invalidate_external_pull_state(merged_profile)
+        remote_resume_name = (
+            str(remote_main.get("resumeName") or "").strip()
+            or str(sync_state.get("resume_name") or "").strip()
+            or self._offerpai_resume_name(session_id, merged_profile)
+        )
+
+        with self.database.transaction(immediate=True) as connection:
+            current_session = self.database.fetch_session(connection, session_id)
+            current_resume = self.database.fetch_resume(connection, session_id)
+            if current_session is None:
+                raise FSMError("session_not_found", "Session not found", status_code=404)
+            if current_resume is None:
+                raise FSMError("resume_not_created", "Create the resume before editing it")
+            if (
+                current_resume["revision"] != source_resume["revision"]
+                or current_session["revision"] != source_session["revision"]
+            ):
+                raise FSMError(
+                    "offerpai_sync_race",
+                    "The local resume changed while OfferPai was being read. Retry.",
+                    status_code=409,
+                )
+
+            updated_resume = self.database.update_resume(
+                connection,
+                session_id,
+                merged_content,
+            )
+            for run in self.database.list_active_optimization_runs(
+                connection, session_id
+            ):
+                state = deepcopy(run["state"])
+                state["stale_reason"] = "offerpai_external_change"
+                self.database.update_optimization_run(
+                    connection,
+                    session_id=session_id,
+                    run_id=run["id"],
+                    status="stale",
+                    state=state,
+                    proposal=run.get("proposal"),
+                )
+            self.database.supersede_active_components(connection, session_id)
+
+            merged_profile["external_resume"] = {
+                **sync_state,
+                "provider": "offerpai",
+                "id": remote_id,
+                "resume_name": remote_resume_name,
+                "status": "partial" if unsupported else "synced",
+                "synced_revision": updated_resume["revision"],
+                "payload_hash": remote_hash,
+                "payload_hash_version": 2,
+                "update_marker": update_marker,
+                "unsupported_sections": list(unsupported),
+                "last_error_code": None,
+            }
+            self.database.update_session(
+                connection,
+                session_id,
+                stage=current_session["stage"],
+                profile=merged_profile,
+            )
+        return OfferPaiReconcileResult(remote_id, "pulled")
+
+    @staticmethod
+    def _invalidate_external_pull_state(profile: dict[str, Any]) -> None:
+        profile.pop("anchor_proposal", None)
+        profile.pop("pending_experience", None)
+        builder = profile.get("builder")
+        if not isinstance(builder, dict):
+            return
+        builder["identity_draft"] = {}
+        builder["pending_entry"] = None
+        builder["editing_entry_id"] = None
+        builder["editing_base_entry"] = None
+        builder["selection_candidates"] = []
+        builder["revision_mode"] = False
+        builder["last_confirmed_entry"] = None
+        builder["pending_skill_candidates"] = []
+        builder.pop("pending_summary_proposal", None)
+
+    def _record_offerpai_failure(
+        self,
+        session_id: str,
+        *,
+        sync_state: dict[str, Any],
+        resume_name: str,
+        remote_id: str | None,
+        error: OfferPaiResumeError,
+    ) -> None:
+        current = self.database.get_session(session_id)
+        current_state = (
+            deepcopy(current["profile"].get("external_resume"))
+            if current is not None
+            and isinstance(current["profile"].get("external_resume"), dict)
+            else {}
+        )
+        failure_state = {
+            **sync_state,
+            **current_state,
+            "provider": "offerpai",
+            "resume_name": resume_name,
+            "status": (
+                "remote_deleted"
+                if error.code == "offerpai_resume_not_found"
+                and (remote_id or current_state.get("id"))
+                else "conflict"
+                if error.code == "offerpai_resume_conflict"
+                else "failed"
+            ),
+            "last_error_code": error.code,
+        }
+        if remote_id:
+            failure_state["id"] = remote_id
+        self._update_external_resume_state(session_id, failure_state)
+
+    def _update_external_resume_state(
+        self, session_id: str, external_resume: dict[str, Any]
+    ) -> None:
+        with self.database.transaction(immediate=True) as connection:
+            session = self.database.fetch_session(connection, session_id)
+            if session is None:
+                raise FSMError("session_not_found", "Session not found", status_code=404)
+            profile = deepcopy(session["profile"])
+            profile["external_resume"] = external_resume
+            self.database.update_session(
+                connection,
+                session_id,
+                stage=session["stage"],
+                profile=profile,
+                increment_revision=False,
+            )
+
+    @staticmethod
+    def _offerpai_resume_name(session_id: str, profile: dict[str, Any]) -> str:
+        name = str(profile.get("name") or "我的").strip()[:16] or "我的"
+        suffix = session_id.rsplit("_", 1)[-1][-8:]
+        return f"AI简历-{name}-{suffix}"
+
+    def create_session(
+        self, request: CreateSessionRequest, *, external_token: str | None = None
+    ) -> TimelineResponse:
+        account_phone = request.account_phone
+        external_account: dict[str, Any] | None = None
+        if external_token:
+            identity = self.authenticate_external_token(external_token)
+            account_phone = identity.mobile_number
+            external_account = identity.profile_value()
+            existing = self.database.find_latest_session_by_external_user_id(
+                identity.user_id
+            )
+            if existing is not None:
+                with self.database.transaction(immediate=True) as connection:
+                    current = self.database.fetch_session(connection, existing["id"])
+                    if current is None:
+                        raise FSMError(
+                            "session_not_found",
+                            "Session not found",
+                            status_code=404,
+                        )
+                    profile = deepcopy(current["profile"])
+                    profile["external_account"] = external_account
+                    profile["account_phone"] = account_phone
+                    if profile.get("phone_source") == "account":
+                        profile["phone"] = account_phone
+                    self.database.update_session(
+                        connection,
+                        current["id"],
+                        stage=current["stage"],
+                        profile=profile,
+                        increment_revision=False,
+                    )
+                return self.timeline(existing["id"])
         session_id = f"session_{uuid4().hex}"
         profile: dict[str, Any] = {
-            "account_phone": request.account_phone,
+            "account_phone": account_phone,
             "metadata": request.metadata,
             "anchor": {},
             "experiences": [],
         }
+        if external_account is not None:
+            profile["external_account"] = external_account
         self.database.create_session(
             session_id,
             Stage.PRIVACY_CONSENT,
@@ -557,7 +1331,76 @@ class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
                 ),
             )
 
-    def delete_session(self, session_id: str) -> None:
+    @_serialize_offerpai_session
+    def delete_session(
+        self, session_id: str, *, external_token: str | None = None
+    ) -> None:
+        session = self.database.get_session(session_id)
+        if session is None:
+            raise FSMError("session_not_found", "Session not found", status_code=404)
+        external_resume = session["profile"].get("external_resume")
+        remote_id = (
+            str(external_resume.get("id") or "").strip()
+            if isinstance(external_resume, dict)
+            else ""
+        )
+        resume_name = (
+            str(external_resume.get("resume_name") or "").strip()
+            if isinstance(external_resume, dict)
+            else ""
+        )
+        if external_token and not remote_id and resume_name:
+            if self.offerpai_resume_provider is None:
+                raise FSMError(
+                    "offerpai_resume_unavailable",
+                    "OfferPai 简历保存服务尚未配置。",
+                    status_code=503,
+                )
+            try:
+                matches = [
+                    item
+                    for item in self.offerpai_resume_provider.list_resumes(
+                        external_token
+                    )
+                    if str(item.get("resumeName") or "").strip() == resume_name
+                    and str(item.get("id") or "").strip()
+                ]
+            except OfferPaiResumeError as exc:
+                raise FSMError(
+                    exc.code,
+                    exc.public_message,
+                    status_code=exc.status_code,
+                ) from exc
+            if len(matches) > 1:
+                raise FSMError(
+                    "offerpai_resume_conflict",
+                    "检测到多份同名 OfferPai 简历，无法安全自动删除。",
+                    status_code=409,
+                )
+            if matches:
+                remote_id = str(matches[0]["id"])
+        if remote_id and external_token:
+            if self.offerpai_resume_provider is None:
+                raise FSMError(
+                    "offerpai_resume_unavailable",
+                    "OfferPai 简历保存服务尚未配置。",
+                    status_code=503,
+                )
+            try:
+                self.offerpai_resume_provider.delete_resume(
+                    external_token, remote_id
+                )
+            except OfferPaiResumeError as exc:
+                if exc.code == "offerpai_resume_not_found":
+                    # Delete is idempotent: a missing remote resume already
+                    # satisfies the requested final state.
+                    pass
+                else:
+                    raise FSMError(
+                        exc.code,
+                        exc.public_message,
+                        status_code=exc.status_code,
+                    ) from exc
         if not self.database.delete_session(session_id):
             raise FSMError("session_not_found", "Session not found", status_code=404)
 
@@ -639,4 +1482,3 @@ class ResumeAgent(ResumeEditingMixin, OptimizationFlowMixin):
     @staticmethod
     def _trace_id() -> str:
         return f"trace_{uuid4().hex}"
-

backend/app/database.py

@@ -186,6 +186,35 @@ class Database:
         with self.transaction() as connection:
             return self.fetch_session(connection, session_id)
 
+    def find_latest_session_by_external_user_id(
+        self, external_user_id: str
+    ) -> dict[str, Any] | None:
+        normalized_user_id = str(external_user_id or "").strip()
+        if not normalized_user_id:
+            return None
+
+        with self.transaction() as connection:
+            row = connection.execute(
+                """SELECT id
+                   FROM sessions
+                   WHERE json_extract(
+                             profile_json, '$.external_account.provider'
+                         ) = 'offerpai'
+                     AND CAST(
+                             json_extract(
+                                 profile_json, '$.external_account.user_id'
+                             ) AS TEXT
+                         ) = ?
+                   ORDER BY updated_at DESC, created_at DESC, id DESC
+                   LIMIT 1""",
+                (normalized_user_id,),
+            ).fetchone()
+            return (
+                self.fetch_session(connection, row["id"])
+                if row is not None
+                else None
+            )
+
     def update_session(
         self,
         connection: sqlite3.Connection,
@@ -610,4 +639,3 @@ class Database:
         with self.transaction(immediate=True) as connection:
             cursor = connection.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
             return cursor.rowcount > 0
-

backend/app/fsm.py

@@ -309,6 +309,9 @@ def process_component_event(
                             "ResumePhoneSelector",
                             has_account_phone=bool(updated.get("account_phone")),
                             masked_phone=mask_phone(updated.get("account_phone")),
+                            default_value=(
+                                "account" if updated.get("account_phone") else None
+                            ),
                         )
                     ],
                 ),

backend/app/main.py

@@ -15,6 +15,8 @@ from .experience_optimizer import ExperienceOptimizer, build_experience_optimize
 from .target_position_suggester import TargetPositionSuggester, build_target_position_suggester
 from .resume_expansion import build_expander
 from .profile_summary import ProfileSummaryGenerator, build_profile_summary_generator
+from .offerpai_auth import OfferPaiAuthClient, OfferPaiIdentityProvider
+from .offerpai_resume import OfferPaiResumeClient, OfferPaiResumeProvider
 from .database import Database
 from .postgres_database import PostgresDatabase
 from .fsm import FSMError
@@ -28,6 +30,7 @@ from .models import (
     CreateSessionRequest,
     ErrorDetail,
     MessageRequest,
+    Stage,
     TimelineResponse,
 )
 from .resume_routes import register_resume_routes
@@ -48,6 +51,20 @@ from .skill_suggester import SkillSuggester, build_skill_suggester
 API_PREFIX = "/ai-api/resume-agent"
 
 
+def _bearer_token(authorization: str | None) -> str | None:
+    if authorization is None:
+        return None
+    scheme, separator, value = authorization.partition(" ")
+    token = value.strip()
+    if separator != " " or scheme.lower() != "bearer" or not token:
+        raise FSMError(
+            "external_auth_header_invalid",
+            "登录凭证格式无效，请重新从 OfferPai 进入。",
+            status_code=401,
+        )
+    return token
+
+
 def create_app(
     *,
     database_path: str | Path | None = None,
@@ -62,6 +79,8 @@ def create_app(
     openai_client: Any | None = None,
     resume_import_service: ResumeImportService | None = None,
     profile_summary_generator: ProfileSummaryGenerator | None = None,
+    offerpai_identity_provider: OfferPaiIdentityProvider | None = None,
+    offerpai_resume_provider: OfferPaiResumeProvider | None = None,
 ) -> FastAPI:
     resolved_settings = settings or load_settings()
     if database_path is not None:
@@ -100,6 +119,14 @@ def create_app(
     profile_summary_generator = profile_summary_generator or build_profile_summary_generator(
         resolved_settings, openai_client
     )
+    offerpai_identity_provider = offerpai_identity_provider or OfferPaiAuthClient(
+        resolved_settings.offerpai_auth_base_url,
+        timeout_seconds=resolved_settings.offerpai_auth_timeout_seconds,
+    )
+    offerpai_resume_provider = offerpai_resume_provider or OfferPaiResumeClient(
+        resolved_settings.offerpai_resume_api_base_url,
+        timeout_seconds=resolved_settings.offerpai_resume_timeout_seconds,
+    )
     agent = ResumeAgent(
         database=database,
         extractor=extractor,
@@ -109,6 +136,8 @@ def create_app(
         experience_optimizer=experience_optimizer,
         target_position_suggester=target_position_suggester,
         profile_summary_generator=profile_summary_generator,
+        offerpai_identity_provider=offerpai_identity_provider,
+        offerpai_resume_provider=offerpai_resume_provider,
     )
     application = FastAPI(
         title="Resume Agent MVP",
@@ -125,6 +154,8 @@ def create_app(
     )
     application.state.database = database
     application.state.resume_agent = agent
+    application.state.offerpai_identity_provider = offerpai_identity_provider
+    application.state.offerpai_resume_provider = offerpai_resume_provider
     if resume_import_service is None:
         import_fallback = RuleBasedResumeImportParser()
         import_parser = import_fallback
@@ -143,6 +174,29 @@ def create_app(
         window_seconds=resolved_settings.light_opt_rate_window_seconds,
     )
 
+    def authorize_session_request(
+        session_id: str, authorization: str | None
+    ) -> str | None:
+        external_token = _bearer_token(authorization)
+        requires_external_auth = resolved_settings.offerpai_auth_required
+        if not requires_external_auth and external_token is None:
+            existing_session = agent.database.get_session(session_id)
+            external_account = (
+                existing_session["profile"].get("external_account")
+                if existing_session is not None
+                else None
+            )
+            requires_external_auth = isinstance(external_account, dict)
+        if requires_external_auth and external_token is None:
+            raise FSMError(
+                "external_auth_required",
+                "缺少 OfferPai 登录凭证，请从 OfferPai 重新进入。",
+                status_code=401,
+            )
+        if external_token is not None:
+            agent.authorize_session(session_id, external_token)
+        return external_token
+
     @application.exception_handler(FSMError)
     async def handle_fsm_error(_request: Any, exc: FSMError) -> JSONResponse:
         trace_id = f"trace_{uuid4().hex}"
@@ -167,15 +221,53 @@ def create_app(
         status_code=status.HTTP_201_CREATED,
         tags=["resume-agent"],
     )
-    def create_session(request: CreateSessionRequest | None = None) -> TimelineResponse:
-        return agent.create_session(request or CreateSessionRequest())
+    def create_session(
+        request: CreateSessionRequest | None = None,
+        authorization: str | None = Header(default=None),
+    ) -> TimelineResponse:
+        external_token = _bearer_token(authorization)
+        if resolved_settings.offerpai_auth_required and external_token is None:
+            raise FSMError(
+                "external_auth_required",
+                "缺少 OfferPai 登录凭证，请从 OfferPai 重新进入。",
+                status_code=401,
+            )
+        response = agent.create_session(
+            request or CreateSessionRequest(),
+            external_token=external_token,
+        )
+        if (
+            external_token is not None
+            and response.stage == Stage.MINIMUM_READY
+            and response.resume_id is None
+        ):
+            agent.mutate_with_offerpai_sync(
+                response.session_id,
+                external_token,
+                lambda: agent.create_resume(
+                    response.session_id, CreateResumeRequest()
+                ),
+            )
+            response = agent.timeline(response.session_id)
+        if external_token is not None:
+            agent.reconcile_resume_with_offerpai(
+                response.session_id, external_token
+            )
+            response = agent.timeline(response.session_id)
+        return response
 
     @application.get(
         f"{API_PREFIX}/sessions/{{session_id}}/timeline",
         response_model=TimelineResponse,
         tags=["resume-agent"],
     )
-    def get_timeline(session_id: str) -> TimelineResponse:
+    def get_timeline(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> TimelineResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        if external_token is not None:
+            agent.reconcile_resume_with_offerpai(session_id, external_token)
         return agent.timeline(session_id)
 
     @application.post(
@@ -184,37 +276,102 @@ def create_app(
         tags=["resume-agent"],
     )
     def post_component_event(
-        session_id: str, request: ComponentEventRequest
+        session_id: str,
+        request: ComponentEventRequest,
+        authorization: str | None = Header(default=None),
     ) -> ActionResponse:
-        return agent.component_event(session_id, request)
+        external_token = authorize_session_request(session_id, authorization)
+
+        def mutate() -> ActionResponse:
+            response = agent.component_event(session_id, request)
+            if (
+                external_token is not None
+                and response.stage == Stage.MINIMUM_READY
+                and response.resume_id is None
+            ):
+                response = agent.create_resume(session_id, CreateResumeRequest())
+            return response
+
+        if external_token is None:
+            return mutate()
+        return agent.mutate_with_offerpai_sync(
+            session_id,
+            external_token,
+            mutate,
+        )
 
     @application.post(
         f"{API_PREFIX}/sessions/{{session_id}}/messages",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def post_message(session_id: str, request: MessageRequest) -> ActionResponse:
-        return agent.add_message(session_id, request)
+    def post_message(
+        session_id: str,
+        request: MessageRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        if external_token is None:
+            return agent.add_message(session_id, request)
+        return agent.mutate_with_offerpai_sync(
+            session_id,
+            external_token,
+            lambda: agent.add_message(session_id, request),
+        )
 
     @application.post(
         f"{API_PREFIX}/sessions/{{session_id}}/messages/stream",
         tags=["resume-agent"],
     )
-    def post_message_stream(session_id: str, request: MessageRequest):
-        return stream_builder_message(lambda: agent.add_message(session_id, request))
+    def post_message_stream(
+        session_id: str,
+        request: MessageRequest,
+        authorization: str | None = Header(default=None),
+    ):
+        external_token = authorize_session_request(session_id, authorization)
+
+        def add_and_sync() -> ActionResponse:
+            if external_token is None:
+                return agent.add_message(session_id, request)
+            return agent.mutate_with_offerpai_sync(
+                session_id,
+                external_token,
+                lambda: agent.add_message(session_id, request),
+            )
+
+        return stream_builder_message(add_and_sync)
     @application.post(
         f"{API_PREFIX}/sessions/{{session_id}}/create",
         response_model=CreateResumeResponse,
         tags=["resume-agent"],
     )
     def create_resume(
-        session_id: str, request: CreateResumeRequest | None = None
+        session_id: str,
+        request: CreateResumeRequest | None = None,
+        authorization: str | None = Header(default=None),
     ) -> CreateResumeResponse:
-        return agent.create_resume(session_id, request or CreateResumeRequest())
+        external_token = authorize_session_request(session_id, authorization)
+        create_request = request or CreateResumeRequest()
+        if external_token is None:
+            return agent.create_resume(session_id, create_request)
+        return agent.mutate_with_offerpai_sync(
+            session_id,
+            external_token,
+            lambda: agent.create_resume(session_id, create_request),
+        )
 
-    register_resume_routes(application, agent, API_PREFIX)
+    register_resume_routes(
+        application,
+        agent,
+        API_PREFIX,
+        authorize_session_request=authorize_session_request,
+    )
     register_resume_import_routes(
-        application, agent, application.state.resume_import_service, API_PREFIX
+        application,
+        agent,
+        application.state.resume_import_service,
+        API_PREFIX,
+        authorize_session_request=authorize_session_request,
     )
 
     @application.delete(
@@ -222,8 +379,12 @@ def create_app(
         status_code=status.HTTP_204_NO_CONTENT,
         tags=["resume-agent"],
     )
-    def delete_session(session_id: str) -> Response:
-        agent.delete_session(session_id)
+    def delete_session(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> Response:
+        external_token = authorize_session_request(session_id, authorization)
+        agent.delete_session(session_id, external_token=external_token)
         return Response(status_code=status.HTTP_204_NO_CONTENT)
 
     return application

backend/app/offerpai_auth.py

@@ -0,0 +1,163 @@
+"""OfferPai account authentication and profile lookup.
+
+The landing token is deliberately used only at the HTTP boundary.  It is never
+returned to callers or persisted in the resume-agent database.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from typing import Any, Protocol
+
+import httpx
+
+from .validators import strict_phone
+
+
+_TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9._~-]{16,4096}$")
+
+
+@dataclass(frozen=True, slots=True)
+class OfferPaiIdentity:
+    user_id: str
+    mobile_number: str
+    nick: str | None = None
+    invite_code: str | None = None
+    create_time: int | None = None
+
+    def profile_value(self) -> dict[str, Any]:
+        return {
+            "provider": "offerpai",
+            "user_id": self.user_id,
+            "mobile_number": self.mobile_number,
+            "nick": self.nick,
+            "invite_code": self.invite_code,
+            "create_time": self.create_time,
+        }
+
+
+class OfferPaiIdentityProvider(Protocol):
+    def authenticate(self, token: str) -> OfferPaiIdentity: ...
+
+
+class OfferPaiAuthError(RuntimeError):
+    def __init__(self, code: str, public_message: str, *, status_code: int) -> None:
+        super().__init__(public_message)
+        self.code = code
+        self.public_message = public_message
+        self.status_code = status_code
+
+
+class OfferPaiAuthClient:
+    """Validate an OfferPai token and load the associated account profile."""
+
+    def __init__(
+        self,
+        base_url: str,
+        *,
+        timeout_seconds: float = 8.0,
+        client: Any | None = None,
+    ) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.timeout_seconds = timeout_seconds
+        self._client = client
+
+    def authenticate(self, token: str) -> OfferPaiIdentity:
+        normalized = token.strip()
+        if not _TOKEN_PATTERN.fullmatch(normalized):
+            raise OfferPaiAuthError(
+                "external_auth_invalid",
+                "登录凭证无效或已过期，请重新从 OfferPai 进入。",
+                status_code=401,
+            )
+
+        owned_client = self._client is None
+        client = self._client or httpx.Client(
+            base_url=self.base_url,
+            timeout=self.timeout_seconds,
+            follow_redirects=False,
+            headers={"Accept": "application/json"},
+        )
+        try:
+            login = self._get_json(client, "/api/public/checkLogin", normalized)
+            if str(login.get("code")) != "0" or login.get("data") is not True:
+                raise OfferPaiAuthError(
+                    "external_auth_invalid",
+                    "登录凭证无效或已过期，请重新从 OfferPai 进入。",
+                    status_code=401,
+                )
+
+            profile = self._get_json(client, "/api/user/manage/info", normalized)
+            data = profile.get("data")
+            if str(profile.get("code")) != "0" or not isinstance(data, dict):
+                raise OfferPaiAuthError(
+                    "external_profile_unavailable",
+                    "暂时无法读取 OfferPai 账号信息，请稍后重试。",
+                    status_code=502,
+                )
+        finally:
+            if owned_client:
+                client.close()
+
+        user_id = str(data.get("id") or "").strip()
+        mobile_number = str(data.get("mobileNumber") or "").strip()
+        if not user_id:
+            raise OfferPaiAuthError(
+                "external_profile_invalid",
+                "OfferPai 账号缺少用户标识，请联系管理员。",
+                status_code=502,
+            )
+        if not strict_phone(mobile_number):
+            raise OfferPaiAuthError(
+                "external_mobile_unavailable",
+                "OfferPai 账号未配置有效手机号，请先完善账号手机号。",
+                status_code=422,
+            )
+
+        create_time = data.get("createTime")
+        return OfferPaiIdentity(
+            user_id=user_id,
+            mobile_number=mobile_number,
+            nick=str(data.get("nick") or "").strip() or None,
+            invite_code=str(data.get("inviteCode") or "").strip() or None,
+            create_time=create_time if isinstance(create_time, int) else None,
+        )
+
+    @staticmethod
+    def _get_json(client: Any, path: str, token: str) -> dict[str, Any]:
+        try:
+            response = client.get(path, headers={"Cookie": f"Token={token}"})
+            response.raise_for_status()
+            payload = response.json()
+        except httpx.TimeoutException as exc:
+            raise OfferPaiAuthError(
+                "external_auth_timeout",
+                "OfferPai 账号服务响应超时，请稍后重试。",
+                status_code=504,
+            ) from exc
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code in {401, 403}:
+                raise OfferPaiAuthError(
+                    "external_auth_invalid",
+                    "登录凭证无效或已过期，请重新从 OfferPai 进入。",
+                    status_code=401,
+                ) from exc
+            raise OfferPaiAuthError(
+                "external_auth_unavailable",
+                "OfferPai 账号服务暂时不可用，请稍后重试。",
+                status_code=502,
+            ) from exc
+        except (httpx.HTTPError, ValueError, TypeError) as exc:
+            raise OfferPaiAuthError(
+                "external_auth_unavailable",
+                "OfferPai 账号服务暂时不可用，请稍后重试。",
+                status_code=502,
+            ) from exc
+        if not isinstance(payload, dict):
+            raise OfferPaiAuthError(
+                "external_auth_unavailable",
+                "OfferPai 账号服务返回了无效数据，请稍后重试。",
+                status_code=502,
+            )
+        return payload

backend/app/postgres_database.py

@@ -66,6 +66,40 @@ class PostgresDatabase:
         with self.transaction() as connection:
             return self.fetch_session(connection, session_id)
 
+    def find_latest_session_by_external_user_id(
+        self, external_user_id: str
+    ) -> dict[str, Any] | None:
+        normalized_user_id = str(external_user_id or "").strip()
+        if not normalized_user_id:
+            return None
+
+        sessions = self.tables["sessions"]
+        with self.transaction() as connection:
+            session_id = connection.execute(
+                select(sessions.c.id)
+                .where(
+                    func.jsonb_extract_path_text(
+                        sessions.c.profile, "external_account", "provider"
+                    )
+                    == "offerpai",
+                    func.jsonb_extract_path_text(
+                        sessions.c.profile, "external_account", "user_id"
+                    )
+                    == normalized_user_id,
+                )
+                .order_by(
+                    sessions.c.updated_at.desc(),
+                    sessions.c.created_at.desc(),
+                    sessions.c.id.desc(),
+                )
+                .limit(1)
+            ).scalar_one_or_none()
+            return (
+                self.fetch_session(connection, session_id)
+                if session_id is not None
+                else None
+            )
+
     def update_session(
         self, connection: Connection, session_id: str, *, stage: str,
         profile: dict[str, Any], draft_id: str | None = None,

backend/app/resume_document_core.py

@@ -11,7 +11,16 @@ from uuid import uuid4
 from .skill_classifier import classify_skills
 
 SCHEMA_VERSION = 3
-META_KEYS = {"pending_proposal", "previous_version", "gap_report"}
+META_KEYS = {
+    "pending_proposal",
+    "previous_version",
+    "gap_report",
+    # OfferPai row/paragraph identifiers are synchronization metadata, not
+    # user-visible resume content and must not make proposals stale by
+    # themselves.
+    "offerpai_record_id",
+    "offerpai_description_ids",
+}
 ITEM_KEY_FIELDS: dict[str, tuple[str, ...]] = {
     "education": ("school", "start_date"),
     "work_experience": ("company", "position", "start_date"),

backend/app/resume_import_routes.py

@@ -3,10 +3,10 @@
 from __future__ import annotations
 
 from copy import deepcopy
-from typing import Any
+from typing import Any, Callable
 from uuid import uuid4
 
-from fastapi import FastAPI, File, UploadFile, status
+from fastapi import FastAPI, File, Header, UploadFile, status
 
 from .fsm import FSMError
 from .models import ActionResponse, Stage
@@ -18,15 +18,28 @@ from . import builder_conversation
 
 
 def register_resume_import_routes(
-    application: FastAPI, agent: Any, service: ResumeImportService, prefix: str
+    application: FastAPI,
+    agent: Any,
+    service: ResumeImportService,
+    prefix: str,
+    *,
+    authorize_session_request: Callable[[str, str | None], str | None] | None = None,
 ) -> None:
+    authorize_session_request = authorize_session_request or (
+        lambda _session_id, _authorization: None
+    )
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume-imports",
         response_model=ResumeImportView,
         status_code=status.HTTP_201_CREATED,
         tags=["resume-agent"],
     )
-    async def create_resume_import(session_id: str, file: UploadFile = File(...)) -> ResumeImportView:
+    async def create_resume_import(
+        session_id: str,
+        file: UploadFile = File(...),
+        authorization: str | None = Header(default=None),
+    ) -> ResumeImportView:
+        authorize_session_request(session_id, authorization)
         _require_import_path(agent, session_id)
         with agent.database.transaction() as connection:
             if agent.database.fetch_resume(connection, session_id) is not None:
@@ -73,7 +86,12 @@ def register_resume_import_routes(
         response_model=ResumeImportView,
         tags=["resume-agent"],
     )
-    def get_resume_import(session_id: str, import_id: str) -> ResumeImportView:
+    def get_resume_import(
+        session_id: str,
+        import_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> ResumeImportView:
+        authorize_session_request(session_id, authorization)
         with agent.database.transaction() as connection:
             record = agent.database.fetch_resume_import(connection, session_id, import_id)
         if record is None:
@@ -86,68 +104,87 @@ def register_resume_import_routes(
         tags=["resume-agent"],
     )
     def apply_resume_import(
-        session_id: str, import_id: str, request: ApplyResumeImportRequest
+        session_id: str,
+        import_id: str,
+        request: ApplyResumeImportRequest,
+        authorization: str | None = Header(default=None),
     ) -> ActionResponse:
-        with agent.database.transaction(immediate=True) as connection:
-            session = agent.database.fetch_session(connection, session_id)
-            if session is None:
-                raise FSMError("session_not_found", "Session not found", status_code=404)
-            _require_import_profile(session["profile"])
-            record = agent.database.fetch_resume_import(connection, session_id, import_id)
-            if record is None:
-                raise FSMError("resume_import_not_found", "Resume import not found", status_code=404)
-            if record["status"] != "awaiting_review":
-                raise FSMError("resume_import_not_applicable", "Resume import is not awaiting review")
-            if agent.database.fetch_resume(connection, session_id) is not None:
-                raise FSMError(
-                    "resume_import_not_allowed",
-                    "当前简历预览已有内容，重新开始后才能导入新的简历。",
+        external_token = authorize_session_request(session_id, authorization)
+
+        def mutate() -> ActionResponse:
+            with agent.database.transaction(immediate=True) as connection:
+                session = agent.database.fetch_session(connection, session_id)
+                if session is None:
+                    raise FSMError("session_not_found", "Session not found", status_code=404)
+                _require_import_profile(session["profile"])
+                record = agent.database.fetch_resume_import(connection, session_id, import_id)
+                if record is None:
+                    raise FSMError("resume_import_not_found", "Resume import not found", status_code=404)
+                if record["status"] != "awaiting_review":
+                    raise FSMError("resume_import_not_applicable", "Resume import is not awaiting review")
+                if agent.database.fetch_resume(connection, session_id) is not None:
+                    raise FSMError(
+                        "resume_import_not_allowed",
+                        "当前简历预览已有内容，重新开始后才能导入新的简历。",
+                    )
+                if request.expected_revision != 0:
+                    raise FSMError("revision_conflict", "Resume was modified; refresh before importing", status_code=409)
+                resume_id = f"resume_{uuid4().hex}"
+                imported_document = record["document"]
+                persisted_document = deepcopy(imported_document)
+                persisted_basics = persisted_document.get("basics")
+                if isinstance(persisted_basics, dict):
+                    raw_phone = persisted_basics.pop("phone", None)
+                    masked_phone = mask_phone(raw_phone)
+                    if masked_phone:
+                        persisted_basics["masked_phone"] = masked_phone
+                agent.database.insert_resume(
+                    connection,
+                    resume_id=resume_id,
+                    session_id=session_id,
+                    idempotency_key=None,
+                    content=merge_ids(None, persisted_document),
                 )
-            if request.expected_revision != 0:
-                raise FSMError("revision_conflict", "Resume was modified; refresh before importing", status_code=409)
-            resume_id = f"resume_{uuid4().hex}"
-            imported_document = record["document"]
-            persisted_document = deepcopy(imported_document)
-            persisted_basics = persisted_document.get("basics")
-            if isinstance(persisted_basics, dict):
-                raw_phone = persisted_basics.pop("phone", None)
-                masked_phone = mask_phone(raw_phone)
-                if masked_phone:
-                    persisted_basics["masked_phone"] = masked_phone
-            agent.database.insert_resume(
-                connection,
-                resume_id=resume_id,
-                session_id=session_id,
-                idempotency_key=None,
-                content=merge_ids(None, persisted_document),
-            )
-            profile, welcome_turn = builder_conversation.welcome_turn(
-                _profile_for_imported_resume(session["profile"], imported_document),
-                resume_id,
-                imported=True,
-            )
-            session = agent.database.update_session(
-                connection,
-                session_id,
-                stage=Stage.BUILDER_CONVERSATION,
-                profile=profile,
-                resume_id=resume_id,
-            )
-            agent.database.supersede_active_components(connection, session_id)
-            turn_id = agent.database.insert_turn(
-                connection,
-                session_id=session_id,
-                **welcome_turn,
-            )
-            agent.database.update_resume_import_status(connection, session_id, import_id, "applied")
-        return agent._action_response(session, agent.database.get_turn(turn_id))
+                profile, welcome_turn = builder_conversation.welcome_turn(
+                    _profile_for_imported_resume(session["profile"], imported_document),
+                    resume_id,
+                    imported=True,
+                )
+                session = agent.database.update_session(
+                    connection,
+                    session_id,
+                    stage=Stage.BUILDER_CONVERSATION,
+                    profile=profile,
+                    resume_id=resume_id,
+                )
+                agent.database.supersede_active_components(connection, session_id)
+                turn_id = agent.database.insert_turn(
+                    connection,
+                    session_id=session_id,
+                    **welcome_turn,
+                )
+                agent.database.update_resume_import_status(connection, session_id, import_id, "applied")
+            return agent._action_response(session, agent.database.get_turn(turn_id))
+
+        if external_token is None:
+            return mutate()
+        return agent.mutate_with_offerpai_sync(
+            session_id,
+            external_token,
+            mutate,
+        )
 
     @application.delete(
         f"{prefix}/sessions/{{session_id}}/resume-imports/{{import_id}}",
         response_model=ResumeImportView,
         tags=["resume-agent"],
     )
-    def cancel_resume_import(session_id: str, import_id: str) -> ResumeImportView:
+    def cancel_resume_import(
+        session_id: str,
+        import_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> ResumeImportView:
+        authorize_session_request(session_id, authorization)
         with agent.database.transaction(immediate=True) as connection:
             record = agent.database.fetch_resume_import(connection, session_id, import_id)
             if record is None:

backend/app/resume_routes.py

@@ -1,6 +1,8 @@
 """FastAPI route registration for resume editing and optimization."""
 
-from fastapi import FastAPI
+from typing import Any, Callable
+
+from fastapi import FastAPI, Header
 from fastapi.responses import StreamingResponse
 
 from .agent import ResumeAgent
@@ -14,38 +16,98 @@ from .optimization_models import (
 from .resume_api_models import SkillRecommendationRequest, SkillRecommendationResponse
 
 
-def register_resume_routes(application: FastAPI, agent: ResumeAgent, prefix: str) -> None:
+def register_resume_routes(
+    application: FastAPI,
+    agent: ResumeAgent,
+    prefix: str,
+    *,
+    authorize_session_request: Callable[[str, str | None], str | None] | None = None,
+) -> None:
+    authorize_session_request = authorize_session_request or (
+        lambda _session_id, _authorization: None
+    )
+
+    def mutate_with_sync(
+        session_id: str,
+        external_token: str | None,
+        operation: Callable[[], Any],
+        *,
+        allow_pulled_resume: bool = False,
+    ) -> Any:
+        if external_token is None:
+            return operation()
+        return agent.mutate_with_offerpai_sync(
+            session_id,
+            external_token,
+            operation,
+            allow_pulled_resume=allow_pulled_resume,
+        )
+
     @application.patch(
         f"{prefix}/sessions/{{session_id}}/resume",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def patch_resume(session_id: str, request: ResumePatchRequest) -> ActionResponse:
-        return agent.patch_resume(session_id, request)
+    def patch_resume(
+        session_id: str,
+        request: ResumePatchRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.patch_resume(session_id, request),
+            allow_pulled_resume=True,
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/profile-summary/generate",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def generate_profile_summary(session_id: str) -> ActionResponse:
-        return agent.generate_profile_summary(session_id)
+    def generate_profile_summary(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.generate_profile_summary(session_id),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/profile-summary/confirm",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def confirm_profile_summary(session_id: str) -> ActionResponse:
-        return agent.confirm_profile_summary(session_id)
+    def confirm_profile_summary(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.confirm_profile_summary(session_id),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/profile-summary/reject",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def reject_profile_summary(session_id: str) -> ActionResponse:
-        return agent.reject_profile_summary(session_id)
+    def reject_profile_summary(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.reject_profile_summary(session_id),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/skills/recommend",
@@ -53,8 +115,13 @@ def register_resume_routes(application: FastAPI, agent: ResumeAgent, prefix: str
         tags=["resume-agent"],
     )
     def recommend_skills(
-        session_id: str, request: SkillRecommendationRequest
+        session_id: str,
+        request: SkillRecommendationRequest,
+        authorization: str | None = Header(default=None),
     ) -> SkillRecommendationResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        if external_token is not None:
+            agent.reconcile_resume_with_offerpai(session_id, external_token)
         return SkillRecommendationResponse(
             candidates=agent.recommend_skills(session_id, request.question)
         )
@@ -64,63 +131,125 @@ def register_resume_routes(application: FastAPI, agent: ResumeAgent, prefix: str
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def optimize_entry(session_id: str, request: OptimizeRequest) -> ActionResponse:
-        return agent.optimize_entry(session_id, request)
+    def optimize_entry(
+        session_id: str,
+        request: OptimizeRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.optimize_entry(session_id, request),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/confirm",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def confirm_optimize(session_id: str, request: OptimizeEntryRequest) -> ActionResponse:
-        return agent.confirm_optimize(session_id, request)
+    def confirm_optimize(
+        session_id: str,
+        request: OptimizeEntryRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.confirm_optimize(session_id, request),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/reject",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def reject_optimize(session_id: str, request: OptimizeEntryRequest) -> ActionResponse:
-        return agent.reject_optimize(session_id, request)
+    def reject_optimize(
+        session_id: str,
+        request: OptimizeEntryRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.reject_optimize(session_id, request),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/undo",
         response_model=ActionResponse,
         tags=["resume-agent"],
     )
-    def undo_optimize(session_id: str, request: OptimizeEntryRequest) -> ActionResponse:
-        return agent.undo_optimize(session_id, request)
+    def undo_optimize(
+        session_id: str,
+        request: OptimizeEntryRequest,
+        authorization: str | None = Header(default=None),
+    ) -> ActionResponse:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.undo_optimize(session_id, request),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/target-position",
         tags=["resume-agent"],
     )
     def set_session_target_position(
-        session_id: str, request: TargetPositionRequest
+        session_id: str,
+        request: TargetPositionRequest,
+        authorization: str | None = Header(default=None),
     ) -> dict[str, object]:
-        return agent.set_target_position(session_id, request.target_position)
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.set_target_position(session_id, request.target_position),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/light",
         response_model=OptimizationRunView,
         tags=["resume-agent"],
     )
-    def optimize_light(session_id: str, request: OptimizationStartRequest) -> OptimizationRunView:
-        limiter = getattr(application.state, "light_opt_limiter", None)
-        if limiter is not None and not limiter.allow(session_id):
-            raise FSMError(
-                "rate_limited",
-                "操作过于频繁，请稍后再试（轻度优化每小时最多 20 次）。",
-                status_code=429,
-            )
-        return agent.optimize_light(session_id, request)
+    def optimize_light(
+        session_id: str,
+        request: OptimizationStartRequest,
+        authorization: str | None = Header(default=None),
+    ) -> OptimizationRunView:
+        external_token = authorize_session_request(session_id, authorization)
+
+        def mutate() -> OptimizationRunView:
+            limiter = getattr(application.state, "light_opt_limiter", None)
+            if limiter is not None and not limiter.allow(session_id):
+                raise FSMError(
+                    "rate_limited",
+                    "操作过于频繁，请稍后再试（轻度优化每小时最多 20 次）。",
+                    status_code=429,
+                )
+            return agent.optimize_light(session_id, request)
+
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            mutate,
+        )
 
     @application.get(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/runs/active",
         response_model=list[OptimizationRunView],
         tags=["resume-agent"],
     )
-    def list_active_optimization_runs(session_id: str) -> list[OptimizationRunView]:
+    def list_active_optimization_runs(
+        session_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> list[OptimizationRunView]:
+        external_token = authorize_session_request(session_id, authorization)
+        if external_token is not None:
+            agent.reconcile_resume_with_offerpai(session_id, external_token)
         return agent.list_active_optimization_runs(session_id)
 
     @application.post(
@@ -128,13 +257,31 @@ def register_resume_routes(application: FastAPI, agent: ResumeAgent, prefix: str
         response_model=OptimizationRunView,
         tags=["resume-agent"],
     )
-    def confirm_deep_optimization(session_id: str, run_id: str) -> OptimizationRunView:
-        return agent.confirm_deep_optimization(session_id, run_id)
+    def confirm_deep_optimization(
+        session_id: str,
+        run_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> OptimizationRunView:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.confirm_deep_optimization(session_id, run_id),
+        )
 
     @application.post(
         f"{prefix}/sessions/{{session_id}}/resume/optimize/runs/{{run_id}}/reject",
         response_model=OptimizationRunView,
         tags=["resume-agent"],
     )
-    def reject_optimization(session_id: str, run_id: str) -> OptimizationRunView:
-        return agent.reject_optimization(session_id, run_id)
+    def reject_optimization(
+        session_id: str,
+        run_id: str,
+        authorization: str | None = Header(default=None),
+    ) -> OptimizationRunView:
+        external_token = authorize_session_request(session_id, authorization)
+        return mutate_with_sync(
+            session_id,
+            external_token,
+            lambda: agent.reject_optimization(session_id, run_id),
+        )

backend/app/settings.py

@@ -65,6 +65,11 @@ class Settings:
     intent_model: str | None = None
     knowledge_admin_token: str | None = field(default=None, repr=False)
     database_url: str | None = field(default=None, repr=False)
+    offerpai_auth_base_url: str = "https://test.offerpai.com.cn"
+    offerpai_auth_timeout_seconds: float = 8.0
+    offerpai_auth_required: bool = True
+    offerpai_resume_api_base_url: str = "https://test.offerpai.com.cn/api"
+    offerpai_resume_timeout_seconds: float = 8.0
     deep_max_questions: int = 6
     deep_min_questions: int = 2
     deep_gap_threshold: float = 5.0
@@ -158,6 +163,25 @@ def load_settings(env_file: str | Path | None = None) -> Settings:
         intent_model=os.getenv("RESUME_AGENT_INTENT_MODEL", "").strip() or None,
         knowledge_admin_token=os.getenv("KNOWLEDGE_ADMIN_TOKEN") or None,
         database_url=os.getenv("DATABASE_URL") or None,
+        offerpai_auth_base_url=os.getenv(
+            "OFFERPAI_AUTH_BASE_URL", "https://test.offerpai.com.cn"
+        ).rstrip("/"),
+        offerpai_auth_timeout_seconds=_as_float(
+            "OFFERPAI_AUTH_TIMEOUT_SECONDS",
+            os.getenv("OFFERPAI_AUTH_TIMEOUT_SECONDS"),
+            8.0,
+        ),
+        offerpai_auth_required=_as_bool(
+            os.getenv("OFFERPAI_AUTH_REQUIRED"), True
+        ),
+        offerpai_resume_api_base_url=os.getenv(
+            "OFFERPAI_RESUME_API_BASE_URL", "https://test.offerpai.com.cn/api"
+        ).rstrip("/"),
+        offerpai_resume_timeout_seconds=_as_float(
+            "OFFERPAI_RESUME_TIMEOUT_SECONDS",
+            os.getenv("OFFERPAI_RESUME_TIMEOUT_SECONDS"),
+            8.0,
+        ),
         deep_max_questions=_as_int(
             "RESUME_AGENT_DEEP_MAX_QUESTIONS",
             os.getenv("RESUME_AGENT_DEEP_MAX_QUESTIONS"),
@@ -184,6 +208,12 @@ def load_settings(env_file: str | Path | None = None) -> Settings:
         raise ValueError("EMBEDDING_DIMENSIONS must be positive")
     if settings.embedding_batch_size < 1:
         raise ValueError("EMBEDDING_BATCH_SIZE must be positive")
+    if not settings.offerpai_auth_base_url.startswith(("http://", "https://")):
+        raise ValueError("OFFERPAI_AUTH_BASE_URL must start with http:// or https://")
+    if not settings.offerpai_resume_api_base_url.startswith(("http://", "https://")):
+        raise ValueError(
+            "OFFERPAI_RESUME_API_BASE_URL must start with http:// or https://"
+        )
     if settings.deep_max_questions < settings.deep_min_questions:
         raise ValueError(
             "RESUME_AGENT_DEEP_MAX_QUESTIONS must be at least "

backend/pyproject.toml

@@ -13,6 +13,7 @@ dependencies = [
   "openai>=1.60,<3",
   "python-dotenv>=1.0,<2",
   "uvicorn[standard]>=0.30,<1",
+  "httpx>=0.27,<1",
   "langgraph>=1.0,<2",
   "sqlalchemy>=2,<3",
   "alembic>=1.13,<2",

backend/tests/conftest.py

@@ -39,7 +39,7 @@ def client(tmp_path: Path) -> TestClient:
         extractor=RuleBasedExperienceExtractor(),
         rewriter=RuleBasedResumeRewriter(),
         expander=RuleBasedEntryExpander(),
-        settings=Settings(llm_provider="rule"),
+        settings=Settings(llm_provider="rule", offerpai_auth_required=False),
     )
     with TestClient(application) as test_client:
         yield test_client

backend/tests/test_api.py

@@ -24,11 +24,13 @@ def event(
     body: dict[str, Any],
     event_name: str,
     payload: dict[str, Any] | None = None,
+    headers: dict[str, str] | None = None,
 ):
     block = active_component(body)
     return client.post(
         f"{BASE}/sessions/{session_id}/component-events",
         json={"component_id": block["id"], "event": event_name, "payload": payload or {}},
+        headers=headers,
     )
 
 

backend/tests/test_offerpai_auth.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import httpx
+from fastapi.testclient import TestClient
+
+from app.main import create_app
+from app.offerpai_auth import (
+    OfferPaiAuthClient,
+    OfferPaiAuthError,
+    OfferPaiIdentity,
+)
+from app.settings import Settings
+from test_api import BASE, active_component, event
+
+
+TOKEN = "header.payload.signature-value"
+
+
+def test_offerpai_client_uses_cookie_for_both_get_requests() -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        assert request.headers.get("cookie") == f"Token={TOKEN}"
+        if request.url.path == "/api/public/checkLogin":
+            return httpx.Response(200, json={"code": "0", "data": True})
+        if request.url.path == "/api/user/manage/info":
+            return httpx.Response(
+                200,
+                json={
+                    "code": "0",
+                    "data": {
+                        "id": "2081575100391407617",
+                        "mobileNumber": "13421012384",
+                        "nick": "用户2384",
+                        "inviteCode": "2GOTI0H0X7",
+                        "createTime": 1785121152000,
+                    },
+                },
+            )
+        return httpx.Response(404)
+
+    with httpx.Client(
+        base_url="https://test.offerpai.com.cn",
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        identity = OfferPaiAuthClient(
+            "https://test.offerpai.com.cn", client=http_client
+        ).authenticate(TOKEN)
+
+    assert [request.method for request in requests] == ["GET", "GET"]
+    assert [request.url.path for request in requests] == [
+        "/api/public/checkLogin",
+        "/api/user/manage/info",
+    ]
+    assert identity.user_id == "2081575100391407617"
+    assert identity.mobile_number == "13421012384"
+    assert identity.nick == "用户2384"
+
+
+def test_offerpai_client_rejects_failed_login_without_profile_request() -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        return httpx.Response(200, json={"code": "0", "data": False})
+
+    with httpx.Client(
+        base_url="https://test.offerpai.com.cn",
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        provider = OfferPaiAuthClient("https://test.offerpai.com.cn", client=http_client)
+        try:
+            provider.authenticate(TOKEN)
+        except OfferPaiAuthError as exc:
+            assert exc.code == "external_auth_invalid"
+            assert exc.status_code == 401
+        else:
+            raise AssertionError("Expected invalid external authentication")
+
+    assert len(requests) == 1
+
+
+class FakeIdentityProvider:
+    def __init__(self, error: OfferPaiAuthError | None = None) -> None:
+        self.error = error
+        self.tokens: list[str] = []
+
+    def authenticate(self, token: str) -> OfferPaiIdentity:
+        self.tokens.append(token)
+        if self.error is not None:
+            raise self.error
+        return OfferPaiIdentity(
+            user_id="2081575100391407617",
+            mobile_number="13421012384",
+            nick="用户2384",
+            invite_code="2GOTI0H0X7",
+            create_time=1785121152000,
+        )
+
+
+def auth_client(
+    tmp_path: Path, provider: Any, *, auth_required: bool = True
+) -> tuple[Any, TestClient]:
+    application = create_app(
+        database_path=tmp_path / "offerpai-auth.db",
+        cors_origins=["http://localhost:5173"],
+        settings=Settings(
+            llm_provider="rule", offerpai_auth_required=auth_required
+        ),
+        offerpai_identity_provider=provider,
+    )
+    return application, TestClient(application)
+
+
+def test_session_creation_authenticates_and_defaults_account_phone(tmp_path: Path) -> None:
+    provider = FakeIdentityProvider()
+    application, client = auth_client(tmp_path, provider)
+    with client:
+        response = client.post(
+            f"{BASE}/sessions",
+            json={},
+            headers={"Authorization": f"Bearer {TOKEN}"},
+        )
+        assert response.status_code == 201, response.text
+        created = response.json()
+        session_id = created["session_id"]
+
+        with application.state.database.transaction() as connection:
+            session = application.state.database.fetch_session(connection, session_id)
+        assert session is not None
+        profile = session["profile"]
+        assert profile["account_phone"] == "13421012384"
+        assert profile["external_account"] == {
+            "provider": "offerpai",
+            "user_id": "2081575100391407617",
+            "mobile_number": "13421012384",
+            "nick": "用户2384",
+            "invite_code": "2GOTI0H0X7",
+            "create_time": 1785121152000,
+        }
+        assert TOKEN not in json.dumps(profile, ensure_ascii=False)
+
+        auth_headers = {"Authorization": f"Bearer {TOKEN}"}
+        source = event(
+            client,
+            session_id,
+            created,
+            "accept",
+            {"accepted": True},
+            headers=auth_headers,
+        ).json()
+        phone_selector = event(
+            client,
+            session_id,
+            source,
+            "select",
+            {"value": "manual"},
+            headers=auth_headers,
+        ).json()
+        data = active_component(phone_selector)["data"]
+        assert data["has_account_phone"] is True
+        assert data["masked_phone"] == "134****2384"
+        assert data["default_value"] == "account"
+
+        personal = event(
+            client,
+            session_id,
+            phone_selector,
+            "select",
+            {"source": "account"},
+            headers=auth_headers,
+        )
+        assert personal.status_code == 200, personal.text
+        with application.state.database.transaction() as connection:
+            updated = application.state.database.fetch_session(connection, session_id)
+        assert updated is not None
+        assert updated["profile"]["phone"] == "13421012384"
+        assert updated["profile"]["phone_source"] == "account"
+
+    assert provider.tokens == [TOKEN, TOKEN, TOKEN, TOKEN]
+
+
+def test_invalid_external_token_does_not_create_session(tmp_path: Path) -> None:
+    provider = FakeIdentityProvider(
+        OfferPaiAuthError(
+            "external_auth_invalid",
+            "登录凭证无效或已过期，请重新从 OfferPai 进入。",
+            status_code=401,
+        )
+    )
+    application, client = auth_client(tmp_path, provider)
+    with client:
+        response = client.post(
+            f"{BASE}/sessions",
+            json={},
+            headers={"Authorization": f"Bearer {TOKEN}"},
+        )
+        assert response.status_code == 401
+        assert response.json()["error"]["code"] == "external_auth_invalid"
+        with application.state.database.transaction() as connection:
+            count = connection.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
+        assert count == 0
+
+
+def test_optional_anonymous_mode_does_not_call_external_provider(tmp_path: Path) -> None:
+    provider = FakeIdentityProvider()
+    _application, client = auth_client(tmp_path, provider, auth_required=False)
+    with client:
+        response = client.post(f"{BASE}/sessions", json={})
+        assert response.status_code == 201
+    assert provider.tokens == []
+
+
+def test_optional_mode_still_protects_sessions_bound_to_offerpai(
+    tmp_path: Path,
+) -> None:
+    provider = FakeIdentityProvider()
+    _application, client = auth_client(tmp_path, provider, auth_required=False)
+    with client:
+        created = client.post(
+            f"{BASE}/sessions",
+            json={},
+            headers={"Authorization": f"Bearer {TOKEN}"},
+        )
+        assert created.status_code == 201, created.text
+        session_id = created.json()["session_id"]
+
+        missing = client.get(f"{BASE}/sessions/{session_id}/timeline")
+        assert missing.status_code == 401
+        assert missing.json()["error"]["code"] == "external_auth_required"
+
+
+def test_authentication_is_required_when_enabled(tmp_path: Path) -> None:
+    provider = FakeIdentityProvider()
+    _application, client = auth_client(tmp_path, provider)
+    with client:
+        response = client.post(f"{BASE}/sessions", json={})
+        assert response.status_code == 401
+        assert response.json()["error"]["code"] == "external_auth_required"
+    assert provider.tokens == []
+
+
+def test_authenticated_start_restores_latest_user_session(tmp_path: Path) -> None:
+    provider = FakeIdentityProvider()
+    _application, client = auth_client(tmp_path, provider, auth_required=True)
+    with client:
+        first = client.post(
+            f"{BASE}/sessions",
+            json={},
+            headers={"Authorization": f"Bearer {TOKEN}"},
+        )
+        second = client.post(
+            f"{BASE}/sessions",
+            json={},
+            headers={"Authorization": f"Bearer {TOKEN}"},
+        )
+        assert first.status_code == 201
+        assert second.status_code == 201
+        assert second.json()["session_id"] == first.json()["session_id"]
+    assert provider.tokens == [TOKEN, TOKEN]

backend/tests/test_offerpai_resume.py

@@ -0,0 +1,922 @@
+from __future__ import annotations
+
+from copy import deepcopy
+import json
+from typing import Any
+
+import httpx
+import pytest
+
+from app.offerpai_resume import (
+    OfferPaiResumeClient,
+    OfferPaiResumeError,
+    OfferPaiResumeProvider,
+    build_offerpai_resume_payload,
+    map_v3_resume_to_offerpai,
+    merge_offerpai_resume_snapshot,
+    offerpai_payload_hash,
+    offerpai_update_marker,
+)
+
+
+TOKEN = "header.payload.signature-value"
+BASE_URL = "https://test.offerpai.com.cn/api/"
+
+
+def test_client_uses_api_base_path_cookie_and_normalizes_ids() -> None:
+    requests: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        assert request.headers.get("cookie") == f"Token={TOKEN}"
+        if request.url.path == "/api/resume/canCreate":
+            return httpx.Response(200, json=True)
+        if request.url.path == "/api/resume/list":
+            return httpx.Response(
+                200,
+                json={
+                    "code": "0",
+                    "data": [
+                        {"id": 2081575100391407617, "resumeName": "First"},
+                        {"resumeId": "resume-local", "resumeName": "Second"},
+                    ],
+                },
+            )
+        if request.url.path == "/api/resume":
+            return httpx.Response(
+                200,
+                json={
+                    "code": "0",
+                    "data": {"resumeId": 2081575100391407618},
+                },
+            )
+        if request.url.path == "/api/resume/work":
+            return httpx.Response(200, json={"id": "2081575100391407618"})
+        if request.url.path == "/api/resume/delete":
+            return httpx.Response(200, json={"code": "0", "data": True})
+        return httpx.Response(404)
+
+    main_payload = {
+        "resumeId": "2081575100391407617",
+        "resumeName": "Backend Resume",
+    }
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        provider: OfferPaiResumeProvider = OfferPaiResumeClient(
+            BASE_URL, client=http_client
+        )
+        assert provider.can_create(TOKEN) is True
+        assert provider.list_resumes(TOKEN) == [
+            {"id": "2081575100391407617", "resumeName": "First"},
+            {
+                "resumeId": "resume-local",
+                "resumeName": "Second",
+                "id": "resume-local",
+            },
+        ]
+        assert provider.save_main(TOKEN, main_payload) == "2081575100391407618"
+        assert (
+            provider.replace_section(
+                TOKEN,
+                "work",
+                resume_id="2081575100391407618",
+                items=[{"companyName": "OfferPai"}],
+            )
+            == "2081575100391407618"
+        )
+        assert provider.delete_resume(TOKEN, "2081575100391407618") is None
+
+    assert main_payload["resumeId"] == "2081575100391407617"
+    assert [request.url.path for request in requests] == [
+        "/api/resume/canCreate",
+        "/api/resume/list",
+        "/api/resume",
+        "/api/resume/work",
+        "/api/resume/delete",
+    ]
+    assert json.loads(requests[2].content) == {
+        "resumeId": 2081575100391407617,
+        "resumeName": "Backend Resume",
+    }
+    assert json.loads(requests[3].content) == {
+        "resumeId": 2081575100391407618,
+        "items": [{"companyName": "OfferPai"}],
+    }
+    assert requests[4].url.params["resumeId"] == "2081575100391407618"
+
+
+def test_client_accepts_raw_list_and_enveloped_boolean() -> None:
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path.endswith("/canCreate"):
+            return httpx.Response(200, json={"code": 0, "data": False})
+        return httpx.Response(200, json=[{"id": 7}])
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        client = OfferPaiResumeClient(BASE_URL, client=http_client)
+        assert client.can_create(TOKEN) is False
+        assert client.list_resumes(TOKEN) == [{"id": "7"}]
+
+
+@pytest.mark.parametrize(
+    ("status", "code", "public_status"),
+    [
+        (401, "external_auth_invalid", 401),
+        (403, "external_auth_invalid", 401),
+        (404, "offerpai_resume_not_found", 404),
+        (409, "offerpai_resume_conflict", 409),
+        (422, "offerpai_resume_rejected", 422),
+        (500, "offerpai_resume_unavailable", 502),
+    ],
+)
+def test_http_errors_are_stable_and_do_not_leak_token(
+    status: int, code: str, public_status: int
+) -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(status, text=f"upstream accidentally echoed {TOKEN}")
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        with pytest.raises(OfferPaiResumeError) as captured:
+            OfferPaiResumeClient(BASE_URL, client=http_client).can_create(TOKEN)
+
+    error = captured.value
+    assert error.code == code
+    assert error.status_code == public_status
+    assert TOKEN not in str(error)
+    assert TOKEN not in error.public_message
+    assert TOKEN not in repr(error)
+
+
+def test_business_error_and_invalid_json_are_token_free() -> None:
+    responses = iter(
+        [
+            httpx.Response(
+                200,
+                json={"code": "RESUME_REJECTED", "msg": TOKEN, "data": None},
+            ),
+            httpx.Response(200, text=f"not-json-{TOKEN}"),
+        ]
+    )
+
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return next(responses)
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        client = OfferPaiResumeClient(BASE_URL, client=http_client)
+        with pytest.raises(OfferPaiResumeError) as business_error:
+            client.can_create(TOKEN)
+        with pytest.raises(OfferPaiResumeError) as invalid_json_error:
+            client.can_create(TOKEN)
+
+    assert business_error.value.code == "offerpai_resume_rejected"
+    assert business_error.value.upstream_code == "RESUME_REJECTED"
+    assert invalid_json_error.value.code == "offerpai_resume_invalid_response"
+    assert TOKEN not in str(business_error.value)
+    assert TOKEN not in str(invalid_json_error.value)
+
+
+def test_timeout_and_bad_local_token_have_stable_errors() -> None:
+    def timeout_handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ReadTimeout(f"timeout with {TOKEN}", request=request)
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(timeout_handler),
+    ) as http_client:
+        with pytest.raises(OfferPaiResumeError) as timeout_error:
+            OfferPaiResumeClient(BASE_URL, client=http_client).list_resumes(TOKEN)
+
+    assert timeout_error.value.code == "offerpai_resume_timeout"
+    assert timeout_error.value.status_code == 504
+    assert TOKEN not in str(timeout_error.value)
+
+    with pytest.raises(OfferPaiResumeError) as token_error:
+        OfferPaiResumeClient(BASE_URL).can_create("short token")
+    assert token_error.value.code == "external_auth_invalid"
+    assert token_error.value.status_code == 401
+    assert "short token" not in str(token_error.value)
+
+
+def test_build_payload_maps_v3_content_without_mutating_inputs() -> None:
+    content: dict[str, Any] = {
+        "basics": {
+            "name": "Ada Lovelace",
+            "email": "ada@example.com",
+            "phone": "134****2384",
+            "city": "Shenzhen",
+            "wechat_number": "ada-wechat",
+            "portfolio_url": "https://example.com/ada",
+        },
+        "target": {"position": "Backend Engineer"},
+        "profile_summary": {"content": "Builds reliable systems."},
+        "skills": ["python"],
+        "skill_groups": [
+            {"category": "Languages", "skills": ["Python", "SQL"]}
+        ],
+        "certificates": ["CET-6"],
+        "sections": [
+            {
+                "kind": "education",
+                "items": [
+                    {
+                        "id": "edu-1",
+                        "school": "Example University",
+                        "major": "Computer Science",
+                        "degree": "Bachelor",
+                        "study_type": "Full-time",
+                        "start_date": "2020-09",
+                        "end_date_or_present": "2024-06",
+                        "description": [
+                            {"id": "paragraph-kept", "text": "Top 10%."}
+                        ],
+                    }
+                ],
+            },
+            {
+                "kind": "work_experience",
+                "items": [
+                    {
+                        "company": "OfferPai",
+                        "position": "Engineer",
+                        "start_date": "2024-07",
+                        "end_date_or_present": "present",
+                        "resume_bullets": "Built APIs.\nReduced latency.",
+                    }
+                ],
+            },
+            {
+                "kind": "internship_experience",
+                "items": [
+                    {
+                        "id": "intern-1",
+                        "company": "Example Labs",
+                        "role": "Intern",
+                        "start_date": "2023-01",
+                        "end_date_or_present": "\u81f3\u4eca",
+                        "description": ["Shipped a service."],
+                    }
+                ],
+            },
+            {
+                "kind": "project_experience",
+                "items": [
+                    {
+                        "id": "project-1",
+                        "project_name": "Resume Agent",
+                        "project_role": "Lead",
+                        "organization": "Personal",
+                        "start_date": "2025-01",
+                        "end_date": "now",
+                        "description": {
+                            "id": "project-paragraph",
+                            "text": "Designed the workflow.",
+                        },
+                    }
+                ],
+            },
+            {
+                "kind": "competition",
+                "items": [
+                    {
+                        "id": "competition-1",
+                        "name": "Hackathon",
+                        "award": "Gold",
+                        "date": "2024-05",
+                        "highlights": ["Won first place."],
+                    }
+                ],
+            },
+            {"kind": "skills", "items": [{"name": "Go"}, "SQL"]},
+            {"kind": "certificates", "items": [{"value": "AWS"}]},
+            {"kind": "campus_experience", "items": []},
+            {"kind": "additional_experience", "items": []},
+        ],
+    }
+    profile: dict[str, Any] = {
+        "phone": "134****2384",
+        "account_phone": "13421012384",
+        "target_position": "Platform Engineer",
+        "external_account": {"user_id": "2081575100391407617"},
+        "tags": {
+            "skills": ["SQL", "Docker"],
+            "certificates": ["CET-6", "PMP"],
+        },
+    }
+    original_content = deepcopy(content)
+    original_profile = deepcopy(profile)
+
+    main, sections, unsupported = build_offerpai_resume_payload(
+        content,
+        profile,
+        resume_name="Candidate Resume",
+        resume_id="2081575100391407617",
+    )
+    second = map_v3_resume_to_offerpai(
+        content,
+        profile,
+        resume_name="Candidate Resume",
+        resume_id="2081575100391407617",
+    )
+
+    assert content == original_content
+    assert profile == original_profile
+    assert main == {
+        "resumeName": "Candidate Resume",
+        "targetPosition": "Platform Engineer",
+        "avatarUrl": "",
+        "name": "Ada Lovelace",
+        "email": "ada@example.com",
+        "mobileNumber": "13421012384",
+        "city": "Shenzhen",
+        "wechatNumber": "ada-wechat",
+        "portfolioUrl": "https://example.com/ada",
+        "skills": ["Python", "SQL", "Docker", "Go"],
+        "certificates": ["CET-6", "PMP", "AWS"],
+        "summary": "Builds reliable systems.",
+        "resumeId": "2081575100391407617",
+    }
+    assert tuple(sections) == (
+        "education",
+        "work",
+        "internship",
+        "project",
+        "competition",
+    )
+    assert all(
+        payload["resumeId"] == "2081575100391407617"
+        for payload in sections.values()
+    )
+    assert sections["education"]["items"] == [
+        {
+            "school": "Example University",
+            "major": "Computer Science",
+            "degree": "Bachelor",
+            "studyType": "Full-time",
+            "startDate": "2020.09",
+            "endDate": "2024.06",
+            "description": [{"id": "paragraph-kept", "text": "Top 10%."}],
+        }
+    ]
+    work = sections["work"]["items"][0]
+    assert work["companyName"] == "OfferPai"
+    assert work["position"] == "Engineer"
+    assert work["startDate"] == "2024.07"
+    assert work["endDate"] == ""
+    assert [paragraph["text"] for paragraph in work["description"]] == [
+        "Built APIs.",
+        "Reduced latency.",
+    ]
+    assert work["description"] == second.sections["work"]["items"][0]["description"]
+    assert all(
+        paragraph["id"].startswith("desc_") for paragraph in work["description"]
+    )
+    assert sections["internship"]["items"][0]["endDate"] == ""
+    assert sections["project"]["items"][0] == {
+        "companyName": "Personal",
+        "projectName": "Resume Agent",
+        "role": "Lead",
+        "startDate": "2025.01",
+        "endDate": "",
+        "description": [
+            {"id": "project-paragraph", "text": "Designed the workflow."}
+        ],
+    }
+    assert sections["competition"]["items"][0]["awardDate"] == "2024.05"
+    assert unsupported == ("campus_experience", "additional_experience")
+
+
+def test_payload_never_sends_a_masked_phone() -> None:
+    main, _sections, _unsupported = build_offerpai_resume_payload(
+        {"basics": {"phone": "13421012384"}},
+        {"phone": "134****2384"},
+        resume_name="Resume",
+    )
+
+    assert main["mobileNumber"] == ""
+
+
+def test_payload_accepts_raw_phone_only_from_session_profile() -> None:
+    main, _sections, _unsupported = build_offerpai_resume_payload(
+        {"basics": {"phone": "134****2384"}},
+        {
+            "phone": "134****2384",
+            "external_account": {"mobileNumber": "13421012384"},
+        },
+        resume_name="Resume",
+    )
+
+    assert main["mobileNumber"] == "13421012384"
+
+
+def test_client_gets_main_and_all_sections_with_cookie_query_and_normalized_ids() -> None:
+    requests: list[httpx.Request] = []
+    section_names = ("education", "work", "internship", "project", "competition")
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        assert request.method == "GET"
+        assert request.headers.get("cookie") == f"Token={TOKEN}"
+        assert request.url.params["resumeId"] == "2081575100391407617"
+        if request.url.path == "/api/resume":
+            return httpx.Response(
+                200,
+                json={
+                    "code": "0",
+                    "data": {
+                        "id": 2081575100391407617,
+                        "resumeId": 2081575100391407617,
+                        "resumeName": "Remote resume",
+                    },
+                },
+            )
+        section = request.url.path.rsplit("/", 1)[-1]
+        assert section in section_names
+        data = [
+            {
+                "id": 2081575100391407700 + section_names.index(section),
+                "resumeId": 2081575100391407617,
+                "description": [
+                    {"id": 700 + section_names.index(section), "text": section}
+                ],
+            }
+        ]
+        # Exercise both documented envelopes and direct/raw response bodies.
+        if section_names.index(section) % 2:
+            return httpx.Response(200, json={"code": 0, "data": data})
+        return httpx.Response(200, json=data)
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        client = OfferPaiResumeClient(BASE_URL, client=http_client)
+        main = client.get_main(TOKEN, "2081575100391407617")
+        section_results = {
+            section: client.list_section(
+                TOKEN,
+                section,  # type: ignore[arg-type]
+                resume_id="2081575100391407617",
+            )
+            for section in section_names
+        }
+
+    assert main["id"] == "2081575100391407617"
+    assert main["resumeId"] == "2081575100391407617"
+    for index, section in enumerate(section_names):
+        item = section_results[section][0]
+        assert item["id"] == str(2081575100391407700 + index)
+        assert item["resumeId"] == "2081575100391407617"
+        assert item["description"] == [{"id": str(700 + index), "text": section}]
+    assert [request.url.path for request in requests] == [
+        "/api/resume",
+        "/api/resume/education",
+        "/api/resume/work",
+        "/api/resume/internship",
+        "/api/resume/project",
+        "/api/resume/competition",
+    ]
+
+
+def test_client_get_main_accepts_a_raw_record() -> None:
+    def handler(_request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200,
+            json={"id": 2081575100391407617, "resumeName": "Raw resume"},
+        )
+
+    with httpx.Client(
+        base_url=BASE_URL,
+        transport=httpx.MockTransport(handler),
+    ) as http_client:
+        result = OfferPaiResumeClient(BASE_URL, client=http_client).get_main(
+            TOKEN, 2081575100391407617
+        )
+
+    assert result == {
+        "id": "2081575100391407617",
+        "resumeName": "Raw resume",
+    }
+
+
+def test_update_marker_supports_instant_and_scalar_values() -> None:
+    assert offerpai_update_marker(
+        {"updateTime": {"seconds": 1785912615, "nanos": 42}}
+    ) == "1785912615:42"
+    assert offerpai_update_marker({"updateTime": "2026-08-05T12:00:00Z"}) == (
+        "2026-08-05T12:00:00Z"
+    )
+    assert offerpai_update_marker({"updateTime": None}) is None
+
+
+def test_external_description_edit_keeps_local_entry_id_by_paragraph_id() -> None:
+    existing_content = {
+        "schema_version": 3,
+        "basics": {},
+        "target": {},
+        "skill_groups": [],
+        "sections": [
+            {
+                "id": "section-work",
+                "kind": "work_experience",
+                "heading": "Work",
+                "items": [
+                    {
+                        "id": "local-alpha",
+                        "company": "Alpha",
+                        "position": "Engineer",
+                        "start_date": "2024-01",
+                        "end_date_or_present": "present",
+                        "description": "Original text",
+                        "offerpai_record_id": "old-alpha-row",
+                        "offerpai_description_ids": ["stable-alpha-paragraph"],
+                        "pending_proposal": {"content": "stale proposal"},
+                        "previous_version": {"description": "older text"},
+                        "gap_report": {"missing": ["metric"]},
+                    },
+                    {
+                        "id": "local-beta",
+                        "company": "Beta",
+                        "position": "Engineer",
+                        "start_date": "2023-01",
+                        "end_date_or_present": "2023-12",
+                        "description": "Beta text",
+                        "offerpai_record_id": "new-alpha-row",
+                        "offerpai_description_ids": ["stable-beta-paragraph"],
+                    },
+                ],
+            }
+        ],
+    }
+    remote_sections = {
+        "education": [],
+        "work": [
+            {
+                # The replacement row ID collides with Beta's old row ID.  The
+                # stable paragraph ID must still associate this record to Alpha.
+                "id": "new-alpha-row",
+                "companyName": "Alpha",
+                "position": "Engineer",
+                "startDate": "2024.01",
+                "endDate": "",
+                "description": [
+                    {
+                        "id": "stable-alpha-paragraph",
+                        "text": "Externally edited text",
+                    }
+                ],
+            }
+        ],
+        "internship": [],
+        "project": [],
+        "competition": [],
+    }
+
+    merged, _profile, _unsupported = merge_offerpai_resume_snapshot(
+        existing_content,
+        {},
+        {"skills": [], "certificates": []},
+        remote_sections,
+    )
+
+    work_section = next(
+        section
+        for section in merged["sections"]
+        if section["kind"] == "work_experience"
+    )
+    assert len(work_section["items"]) == 1
+    item = work_section["items"][0]
+    assert item["id"] == "local-alpha"
+    assert item["offerpai_record_id"] == "new-alpha-row"
+    assert item["offerpai_description_ids"] == ["stable-alpha-paragraph"]
+    assert item["description"] == "Externally edited text"
+    assert item["provenance"] == "external_synced"
+    assert "pending_proposal" not in item
+    assert "previous_version" not in item
+    assert item["gap_report"] == {"missing": ["metric"]}
+
+
+def test_local_remote_pull_roundtrip_preserves_dates_paragraphs_and_unsupported_sections() -> None:
+    content: dict[str, Any] = {
+        "schema_version": 3,
+        "resume_name": "Candidate Resume",
+        "basics": {
+            "name": "Ada Lovelace",
+            "email": "ada@example.com",
+            "city": "Shenzhen",
+            "avatar_url": "https://example.com/avatar.png",
+            "wechat_number": "ada-wechat",
+            "portfolio_url": "https://example.com/ada",
+        },
+        "target": {"position": "Platform Engineer"},
+        "profile_summary": {"content": "Builds reliable systems."},
+        # Deliberately cross category order; hash comparison is semantic.
+        "skill_groups": [
+            {"category": "Tools", "skills": ["Docker", "Python"]}
+        ],
+        "sections": [
+            {
+                "id": "section-education",
+                "kind": "education",
+                "heading": "Education",
+                "items": [
+                    {
+                        "id": "local-education",
+                        "school": "Example University",
+                        "major": "Computer Science",
+                        "degree": "Bachelor",
+                        "study_type": "Full-time",
+                        "start_date": "2020-09",
+                        "end_date_or_present": "2024-06",
+                        "description": [
+                            {"id": "education-paragraph", "text": "Top 10%."}
+                        ],
+                    }
+                ],
+            },
+            {
+                "id": "section-work",
+                "kind": "work_experience",
+                "heading": "Work",
+                "items": [
+                    {
+                        "id": "local-work",
+                        "company": "OfferPai",
+                        "position": "Engineer",
+                        "start_date": "2024-07",
+                        "end_date_or_present": "present",
+                        "description": [
+                            {"id": "work-paragraph", "text": "Built APIs."}
+                        ],
+                    }
+                ],
+            },
+            {
+                "id": "section-internship",
+                "kind": "internship_experience",
+                "heading": "Internship",
+                "items": [
+                    {
+                        "id": "local-internship",
+                        "company": "Example Labs",
+                        "position": "Intern",
+                        "start_date": "2023-01",
+                        "end_date_or_present": "2023-06",
+                        "description": [
+                            {
+                                "id": "internship-paragraph",
+                                "text": "Shipped a service.",
+                            }
+                        ],
+                    }
+                ],
+            },
+            {
+                "id": "section-project",
+                "kind": "project_experience",
+                "heading": "Project",
+                "items": [
+                    {
+                        "id": "local-project",
+                        "company": "Personal",
+                        "project_name": "Resume Agent",
+                        "project_role": "Lead",
+                        "start_date": "2025-01",
+                        "end_date_or_present": "present",
+                        "description": [
+                            {
+                                "id": "project-paragraph",
+                                "text": "Designed the workflow.",
+                            }
+                        ],
+                    }
+                ],
+            },
+            {
+                "id": "section-competition",
+                "kind": "competition",
+                "heading": "Competition",
+                "items": [
+                    {
+                        "id": "local-competition",
+                        "name": "Hackathon",
+                        "award": "Gold",
+                        "date": "2024-05",
+                        "description": [
+                            {
+                                "id": "competition-paragraph",
+                                "text": "Won first place.",
+                            }
+                        ],
+                    }
+                ],
+            },
+            {
+                "id": "section-certificates",
+                "kind": "certificates",
+                "heading": "Certificates",
+                "items": [{"id": "local-certificate", "value": "CET-6"}],
+            },
+            {
+                "id": "section-campus",
+                "kind": "campus_experience",
+                "heading": "Campus",
+                "items": [
+                    {
+                        "id": "local-campus",
+                        "organization": "Student Union",
+                        "role": "Member",
+                    }
+                ],
+            },
+        ],
+    }
+    profile: dict[str, Any] = {
+        "phone": "13421012384",
+        "phone_source": "account",
+        "account_phone": "13421012384",
+        "tags": {"skills": [], "certificates": []},
+    }
+    outbound = map_v3_resume_to_offerpai(
+        content,
+        profile,
+        resume_name="Candidate Resume",
+        resume_id="2081575100391407617",
+    )
+    remote_main = deepcopy(outbound.main)
+    remote_main.pop("resumeId")
+    remote_main["id"] = 2081575100391407617
+    remote_sections: dict[str, list[dict[str, Any]]] = {}
+    for section_index, (kind, payload) in enumerate(outbound.sections.items()):
+        items = deepcopy(payload["items"])
+        for item_index, item in enumerate(items):
+            item["id"] = 2081575100391407700 + section_index * 10 + item_index
+        remote_sections[kind] = items
+
+    merged_content, merged_profile, unsupported = merge_offerpai_resume_snapshot(
+        content,
+        profile,
+        remote_main,
+        remote_sections,  # type: ignore[arg-type]
+    )
+    roundtrip = map_v3_resume_to_offerpai(
+        merged_content,
+        merged_profile,
+        resume_name="Candidate Resume",
+        resume_id="2081575100391407617",
+    )
+
+    assert offerpai_payload_hash(remote_main, remote_sections) == (
+        offerpai_payload_hash(roundtrip.main, roundtrip.sections)
+    )
+    assert unsupported == ("campus_experience",)
+    campus = next(
+        section
+        for section in merged_content["sections"]
+        if section["kind"] == "campus_experience"
+    )
+    assert campus == content["sections"][-1]
+
+    by_kind = {section["kind"]: section for section in merged_content["sections"]}
+    assert by_kind["education"]["items"][0]["id"] == "local-education"
+    assert by_kind["education"]["items"][0]["start_date"] == "2020-09"
+    assert by_kind["education"]["items"][0]["end_date_or_present"] == "2024-06"
+    assert by_kind["work_experience"]["items"][0]["end_date_or_present"] == (
+        "present"
+    )
+    assert by_kind["project_experience"]["items"][0]["project_role"] == "Lead"
+    assert by_kind["competition"]["items"][0]["date"] == "2024-05"
+
+    for kind, payload in outbound.sections.items():
+        assert [item["description"] for item in roundtrip.sections[kind]["items"]] == [
+            item["description"] for item in payload["items"]
+        ]
+    assert all(
+        isinstance(
+            by_kind[local_kind]["items"][0]["offerpai_record_id"], str
+        )
+        for local_kind in (
+            "education",
+            "work_experience",
+            "internship_experience",
+            "project_experience",
+            "competition",
+        )
+    )
+
+
+def test_remote_main_clears_converge_without_reusing_account_identity_phone() -> None:
+    existing_content = {
+        "schema_version": 3,
+        "resume_name": "Old resume",
+        "basics": {
+            "name": "Old name",
+            "email": "old@example.com",
+            "city": "Old city",
+            "avatar_url": "old-avatar",
+            "wechat_number": "old-wechat",
+            "portfolio_url": "old-portfolio",
+            "masked_phone": "134****2384",
+        },
+        "target": {"position": "Old target"},
+        "profile_summary": {"content": "Old summary"},
+        "summary": "Old legacy summary",
+        "skills": ["Old root skill"],
+        "certificates": ["Old root certificate"],
+        "skill_groups": [{"category": "Old", "skills": ["Old grouped skill"]}],
+        "sections": [
+            {
+                "id": "old-certificates",
+                "kind": "certificates",
+                "heading": "Certificates",
+                "items": [{"id": "old-certificate", "value": "Old certificate"}],
+            }
+        ],
+    }
+    profile = {
+        "name": "Old name",
+        "email": "old@example.com",
+        "city": "Old city",
+        "avatar_url": "old-avatar",
+        "wechat_number": "old-wechat",
+        "portfolio_url": "old-portfolio",
+        "phone": "13421012384",
+        "phone_source": "account",
+        "account_phone": "13421012384",
+        "external_account": {"mobile_number": "13421012384"},
+        "target_position": "Old target",
+        "resume_name": "Old resume",
+        "summary": "Old summary",
+        "tags": {
+            "skills": ["Old tagged skill"],
+            "certificates": ["Old tagged certificate"],
+        },
+    }
+    remote_main = {
+        "id": "2081575100391407617",
+        "resumeName": "",
+        "targetPosition": "",
+        "avatarUrl": "",
+        "name": "",
+        "email": "",
+        "mobileNumber": "",
+        "city": "",
+        "wechatNumber": "",
+        "portfolioUrl": "",
+        "skills": [],
+        "certificates": [],
+        "summary": "",
+    }
+    remote_sections = {
+        "education": [],
+        "work": [],
+        "internship": [],
+        "project": [],
+        "competition": [],
+    }
+
+    merged_content, merged_profile, _unsupported = merge_offerpai_resume_snapshot(
+        existing_content,
+        profile,
+        remote_main,
+        remote_sections,
+    )
+    outbound = map_v3_resume_to_offerpai(
+        merged_content,
+        merged_profile,
+        resume_id="2081575100391407617",
+    )
+
+    assert merged_profile["account_phone"] == "13421012384"
+    assert merged_profile["external_account"] == {
+        "mobile_number": "13421012384"
+    }
+    assert merged_profile["phone"] == ""
+    assert merged_profile["phone_source"] == "offerpai_resume"
+    assert outbound.main == {
+        "resumeName": "",
+        "targetPosition": "",
+        "avatarUrl": "",
+        "name": "",
+        "email": "",
+        "mobileNumber": "",
+        "city": "",
+        "wechatNumber": "",
+        "portfolioUrl": "",
+        "skills": [],
+        "certificates": [],
+        "summary": "",
+        "resumeId": "2081575100391407617",
+    }
+    assert offerpai_payload_hash(remote_main, remote_sections) == (
+        offerpai_payload_hash(outbound.main, outbound.sections)
+    )

backend/tests/test_postgres_runtime.py

@@ -7,11 +7,21 @@ from fastapi.testclient import TestClient
 from sqlalchemy import create_engine, text
 
 from app.main import create_app
+from app.offerpai_auth import OfferPaiIdentity
 from app.postgres_database import PostgresDatabase
 from app.services import RuleBasedEntryExpander, RuleBasedExperienceExtractor, RuleBasedResumeRewriter
 from app.settings import Settings
 
 
+class StaticIdentityProvider:
+    def authenticate(self, _token: str) -> OfferPaiIdentity:
+        return OfferPaiIdentity(
+            user_id="2081575100391407617",
+            mobile_number="13421012384",
+            nick="用户2384",
+        )
+
+
 def test_create_app_uses_postgres_when_database_path_is_not_supplied(monkeypatch) -> None:
     schema = f"test_runtime_{uuid4().hex}"
     database_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]
@@ -21,7 +31,11 @@ def test_create_app_uses_postgres_when_database_path_is_not_supplied(monkeypatch
         extractor=RuleBasedExperienceExtractor(),
         rewriter=RuleBasedResumeRewriter(),
         expander=RuleBasedEntryExpander(),
-        settings=Settings(llm_provider="rule", database_url=database_url),
+        settings=Settings(
+            llm_provider="rule",
+            database_url=database_url,
+            offerpai_auth_required=False,
+        ),
     )
     try:
         assert isinstance(application.state.database, PostgresDatabase)
@@ -38,3 +52,36 @@ def test_create_app_uses_postgres_when_database_path_is_not_supplied(monkeypatch
                 connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
         finally:
             engine.dispose()
+
+
+def test_postgres_restores_session_by_external_user_id(monkeypatch) -> None:
+    schema = f"test_external_identity_{uuid4().hex}"
+    database_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]
+    monkeypatch.setenv("RESUME_AGENT_DATABASE_SCHEMA", schema)
+    application = create_app(
+        extractor=RuleBasedExperienceExtractor(),
+        rewriter=RuleBasedResumeRewriter(),
+        expander=RuleBasedEntryExpander(),
+        settings=Settings(
+            llm_provider="rule",
+            database_url=database_url,
+            offerpai_auth_required=True,
+        ),
+        offerpai_identity_provider=StaticIdentityProvider(),
+    )
+    headers = {"Authorization": "Bearer header.payload.signature-value"}
+    try:
+        with TestClient(application) as client:
+            first = client.post("/ai-api/resume-agent/sessions", json={}, headers=headers)
+            second = client.post("/ai-api/resume-agent/sessions", json={}, headers=headers)
+            assert first.status_code == 201
+            assert second.status_code == 201
+            assert second.json()["session_id"] == first.json()["session_id"]
+    finally:
+        application.state.database.engine.dispose()
+        engine = create_engine(database_url)
+        try:
+            with engine.begin() as connection:
+                connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+        finally:
+            engine.dispose()

backend/tests/test_profile_summary.py

@@ -35,7 +35,7 @@ def summary_client(tmp_path: Any) -> tuple[TestClient, CountingSummaryGenerator]
         database_path=tmp_path / "summary.db",
         extractor=RuleBasedExperienceExtractor(),
         rewriter=RuleBasedResumeRewriter(),
-        settings=Settings(llm_provider="rule"),
+        settings=Settings(llm_provider="rule", offerpai_auth_required=False),
         profile_summary_generator=generator,
     )
     return TestClient(app), generator

backend/tests/test_resume_import_api.py

@@ -44,7 +44,7 @@ def client_for_import(tmp_path) -> TestClient:
         extractor=RuleBasedExperienceExtractor(),
         rewriter=RuleBasedResumeRewriter(),
         expander=RuleBasedEntryExpander(),
-        settings=Settings(llm_provider="rule"),
+        settings=Settings(llm_provider="rule", offerpai_auth_required=False),
         resume_import_service=ResumeImportService(storage_root=tmp_path / "imports", parser=FakeResumeImportParser()),
     )
     return TestClient(application)

backend/tests/test_session_identity_lookup.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import os
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import create_engine, text, update
+
+from app.database import Database
+from app.postgres_database import PostgresDatabase
+
+
+_INITIAL_TURN = {
+    "role": "assistant",
+    "content": None,
+    "composer_mode": "ui_only",
+    "blocks": [],
+}
+_TARGET_USER_ID = "2081575100391407617"
+
+
+def _profile(
+    user_id: str | None = _TARGET_USER_ID, *, provider: str = "offerpai"
+) -> dict[str, Any]:
+    if user_id is None:
+        return {}
+    return {
+        "external_account": {
+            "provider": provider,
+            "user_id": user_id,
+        }
+    }
+
+
+def _session_cases() -> list[tuple[str, dict[str, Any], datetime, datetime]]:
+    first = datetime(2026, 1, 1, tzinfo=UTC)
+    second = datetime(2026, 1, 2, tzinfo=UTC)
+    latest = datetime(2026, 1, 3, tzinfo=UTC)
+    excluded_latest = datetime(2026, 1, 4, tzinfo=UTC)
+    return [
+        ("session-old", _profile(), first, first),
+        # updated_at wins first; created_at then beats this lexically larger id.
+        ("session-z-created-old", _profile(), first, latest),
+        ("session-a", _profile(), second, latest),
+        # Same timestamps as session-a: id is the final deterministic tie-breaker.
+        ("session-c", _profile(), second, latest),
+        (
+            "session-other-provider",
+            _profile(provider="another-provider"),
+            excluded_latest,
+            excluded_latest,
+        ),
+        (
+            "session-other-user",
+            _profile("another-user"),
+            excluded_latest,
+            excluded_latest,
+        ),
+        ("session-anonymous", _profile(None), excluded_latest, excluded_latest),
+    ]
+
+
+def _create_sessions(database: Any) -> None:
+    for session_id, profile, _created_at, _updated_at in _session_cases():
+        database.create_session(
+            session_id,
+            "PRIVACY_CONSENT",
+            profile,
+            _INITIAL_TURN,
+        )
+
+
+def test_sqlite_finds_latest_offerpai_session_with_deterministic_order(
+    tmp_path: Path,
+) -> None:
+    database = Database(tmp_path / "identity-lookup.db")
+    database.initialize()
+    _create_sessions(database)
+    with database.transaction(immediate=True) as connection:
+        for session_id, _profile_value, created_at, updated_at in _session_cases():
+            connection.execute(
+                "UPDATE sessions SET created_at = ?, updated_at = ? WHERE id = ?",
+                (created_at.isoformat(), updated_at.isoformat(), session_id),
+            )
+
+    found = database.find_latest_session_by_external_user_id(
+        f"  {_TARGET_USER_ID}\t"
+    )
+
+    assert found is not None
+    assert found["id"] == "session-c"
+    assert database.find_latest_session_by_external_user_id("missing-user") is None
+
+
+def test_sqlite_ignores_blank_external_user_id(tmp_path: Path) -> None:
+    database = Database(tmp_path / "identity-lookup-blank.db")
+    database.initialize()
+
+    for external_user_id in ("", " ", "\t\r\n"):
+        assert database.find_latest_session_by_external_user_id(external_user_id) is None
+
+
+@pytest.fixture
+def postgres_database() -> Any:
+    database_url = os.environ["RESUME_AGENT_TEST_DATABASE_URL"]
+    schema = f"test_session_identity_{uuid4().hex}"
+    database = PostgresDatabase(database_url, schema=schema)
+    database.initialize()
+    try:
+        yield database
+    finally:
+        database.engine.dispose()
+        cleanup_engine = create_engine(database_url)
+        try:
+            with cleanup_engine.begin() as connection:
+                connection.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+        finally:
+            cleanup_engine.dispose()
+
+
+def test_postgres_finds_latest_offerpai_session_with_deterministic_order(
+    postgres_database: PostgresDatabase,
+) -> None:
+    _create_sessions(postgres_database)
+    sessions = postgres_database.tables["sessions"]
+    with postgres_database.transaction() as connection:
+        for session_id, _profile_value, created_at, updated_at in _session_cases():
+            connection.execute(
+                update(sessions)
+                .where(sessions.c.id == session_id)
+                .values(created_at=created_at, updated_at=updated_at)
+            )
+
+    found = postgres_database.find_latest_session_by_external_user_id(
+        f"  {_TARGET_USER_ID}\t"
+    )
+
+    assert found is not None
+    assert found["id"] == "session-c"
+    assert (
+        postgres_database.find_latest_session_by_external_user_id("missing-user")
+        is None
+    )
+
+
+def test_postgres_ignores_blank_external_user_id(
+    postgres_database: PostgresDatabase,
+) -> None:
+    for external_user_id in ("", " ", "\t\r\n"):
+        assert (
+            postgres_database.find_latest_session_by_external_user_id(
+                external_user_id
+            )
+            is None
+        )

docs/DEPLOY.md

@@ -23,14 +23,27 @@
 | `RESUME_AGENT_TEST_DATABASE_URL` | 仅测试 | 测试库 |
 | `RESUME_AGENT_CORS_ORIGINS` | 生产必填 | 逗号分隔的前端来源，接入 officeπ 时加其域名 |
 | `RESUME_AGENT_INTENT_ROUTER_MODE` | 建议 `on` | 对话意图路由（off/shadow/on） |
+| `OFFERPAI_AUTH_REQUIRED` | 必须为 `true` | 强制每个 session 请求携带 OfferPai Token 并校验会话归属 |
+| `OFFERPAI_AUTH_BASE_URL` | 必填 | OfferPai 登录鉴权与用户信息服务 Origin |
+| `OFFERPAI_AUTH_TIMEOUT_SECONDS` | 建议 `8` | 登录鉴权与用户信息请求超时 |
+| `OFFERPAI_RESUME_API_BASE_URL` | 必填 | C 端简历 API 根地址，例如 `https://test.offerpai.com.cn/api` |
+| `OFFERPAI_RESUME_TIMEOUT_SECONDS` | 建议 `8` | C 端简历读写请求超时 |
 
 ## 安全红线（试点必须遵守）
 
-1. **服务无内置认证**：所有接口可匿名调用。只能发布在内网/办公网，或置于带鉴权的网关之后；接入 officeπ 前需补用户绑定与归属校验。
+1. **保持 OfferPai 鉴权开启**：`OFFERPAI_AUTH_REQUIRED=true`。前端 Token 只保存在页面内存，所有 session、SSE、上传和简历编辑请求都必须携带 `Authorization: Bearer`；后端会校验 Token 用户与 session 绑定用户一致。
 2. **不要设置** `RESUME_AGENT_DEFAULT_TIER=vip`：该开关会把所有会话默认提权（后门告警会打日志）。
 3. **不要设置** `RESUME_AGENT_API_DOCS=1`：生产暴露 `/docs` 等于公开 API 结构。
 4. 密钥只放 `.env`（已被 gitignore）；仓库内不得出现明文令牌。
 5. 简历文件含用户 PII：`backend/data/`（上传件、SQLite）不得外传、不得提交。
+6. `/?token=` 会被前端立即从地址栏移除，但首次请求仍可能进入 Nginx/CDN 访问日志；生产网关必须关闭 query string 日志或对 `token` 参数脱敏。
+
+## C 端简历同步语义
+
+- 初步资料完成后自动创建本地工作文档与 C 端简历，后续同步主表及教育、工作、实习、项目、竞赛五类经历。
+- local DB 继续保存 FSM、revision、稳定子项 ID、候选稿、撤销和优化状态；C 端接口目前不能完全替代它。
+- 双写不是分布式事务：本地修改先提交，远端失败时会记录同步失败并由后续请求补偿。调用方重试前应先刷新 timeline，不能假设 HTTP 同步错误代表本地未修改。
+- 删除 session（前端“重新开始”）会同步删除已绑定的 C 端镜像；远端返回不存在按幂等成功处理。
 
 ## 健康检查
 

frontend/index.html

@@ -3,6 +3,7 @@
   <head>
     <meta charset="UTF-8" />
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <meta name="referrer" content="no-referrer" />
     <meta name="theme-color" content="#eef8f7" />
     <meta
       name="description"

frontend/src/App.vue

@@ -1,5 +1,5 @@
 <script setup lang="ts">
-import { computed, onMounted, ref, watch } from 'vue'
+import { computed, onBeforeUnmount, onMounted, ref, watch } from 'vue'
 import AgentTimeline from './components/AgentTimeline.vue'
 import AppHeader from './components/AppHeader.vue'
 import ComposerBar from './components/ComposerBar.vue'
@@ -10,6 +10,7 @@ import { useResumeAgent } from './composables/useResumeAgent'
 import { useResumeDocument } from './composables/useResumeDocument'
 
 const {
+  hasLandingToken,
   sessionId,
   revision,
   stage,
@@ -27,6 +28,7 @@ const {
   errorMessage,
   aiStatus,
   streamedAssistantText,
+  isBusy,
   start,
   refreshTimeline,
   submitComponent,
@@ -41,6 +43,54 @@ const mobilePanel = ref<'chat' | 'resume'>('chat')
 const displayedRevision = computed(() => resumeDocument.resume.value?.revision ?? revision.value)
 const chatEnabled = computed(() => composer.value.mode !== 'ui_only')
 const stageCode = computed(() => stage.value.toUpperCase().replaceAll('_', ' / '))
+const REMOTE_REFRESH_INTERVAL_MS = 20_000
+let remoteRefreshTimer: number | undefined
+let remoteRefreshController: AbortController | null = null
+const remoteRefreshBlocked = computed(
+  () =>
+    !hasLandingToken ||
+    !sessionId.value ||
+    isBusy.value ||
+    Boolean(resumeDocument.busyEntryId.value) ||
+    resumeDocument.importBusy.value ||
+    resumeDocument.skillsBusy.value ||
+    resumeDocument.summaryBusy.value,
+)
+
+function cancelRemoteRefresh() {
+  remoteRefreshController?.abort()
+  remoteRefreshController = null
+}
+
+async function refreshExternalChanges() {
+  if (
+    remoteRefreshBlocked.value ||
+    document.visibilityState !== 'visible' ||
+    remoteRefreshController
+  ) return
+
+  const activeController = new AbortController()
+  remoteRefreshController = activeController
+  try {
+    await refreshTimeline(activeController.signal)
+  } catch {
+    // Background synchronization retries on the next interval or focus event.
+  } finally {
+    if (remoteRefreshController === activeController) remoteRefreshController = null
+  }
+}
+
+function handleWindowFocus() {
+  void refreshExternalChanges()
+}
+
+function handleVisibilityChange() {
+  if (document.visibilityState === 'visible') {
+    void refreshExternalChanges()
+  } else {
+    cancelRemoteRefresh()
+  }
+}
 const stageLabels: Record<string, string> = {
   starting: '准备会话',
   PRIVACY_CONSENT: '隐私确认',
@@ -72,6 +122,10 @@ watch(() => resumeDocument.resumeImport.value?.status, (status) => {
   if (status === 'applied') void refreshTimeline()
 })
 
+watch(remoteRefreshBlocked, (blocked) => {
+  if (blocked) cancelRemoteRefresh()
+}, { flush: 'sync' })
+
 async function retryConnection() {
   clearError()
   if (!sessionId.value) {
@@ -89,22 +143,70 @@ async function confirmReset() {
   if (window.confirm('重新开始会清空当前简历共创记录。确定继续吗？')) await resetSession()
 }
 
-onMounted(start)
+onMounted(() => {
+  void start()
+  remoteRefreshTimer = window.setInterval(() => {
+    void refreshExternalChanges()
+  }, REMOTE_REFRESH_INTERVAL_MS)
+  window.addEventListener('focus', handleWindowFocus)
+  document.addEventListener('visibilitychange', handleVisibilityChange)
+})
+
+onBeforeUnmount(() => {
+  if (remoteRefreshTimer !== undefined) window.clearInterval(remoteRefreshTimer)
+  window.removeEventListener('focus', handleWindowFocus)
+  document.removeEventListener('visibilitychange', handleVisibilityChange)
+  cancelRemoteRefresh()
+})
 </script>
 
 <template>
   <div id="top" class="app-shell">
-    <AppHeader
-      :stage-label="stageLabel"
-      :revision="displayedRevision"
-      :session-id="sessionId"
-      :resetting="resetting"
-      @reset="confirmReset"
-    />
+    <main v-if="!sessionId" class="auth-gate" aria-live="polite">
+      <section class="auth-gate__card" :aria-busy="initializing">
+        <div class="auth-gate__brand" aria-label="OfferPai Resume Agent">
+          <span aria-hidden="true">OP</span>
+          <strong>OfferPai Resume Agent</strong>
+        </div>
+        <p class="auth-gate__eyebrow">
+          {{ initializing ? 'VERIFYING ACCESS' : 'ACCESS REQUIRED' }}
+        </p>
+        <h1>
+          {{ initializing ? '正在验证 OfferPai 登录状态' : '需要先完成 OfferPai 鉴权' }}
+        </h1>
+        <p v-if="initializing" class="auth-gate__message">
+          正在校验登录凭证和账号信息，验证通过后会自动进入简历服务。
+        </p>
+        <p v-else class="auth-gate__message" role="alert">
+          {{ errorMessage || '当前登录凭证不可用，请重新从 OfferPai 进入。' }}
+        </p>
+        <div v-if="initializing" class="auth-gate__progress" aria-hidden="true"><i /></div>
+        <button
+          v-else-if="hasLandingToken"
+          type="button"
+          class="auth-gate__retry"
+          @click="retryConnection"
+        >
+          重新验证
+        </button>
+        <p v-else class="auth-gate__hint">
+          请使用 OfferPai 提供的带 <code>?token=</code> 入口重新打开本页。
+        </p>
+      </section>
+    </main>
 
-    <FeatureNavigation current="builder" />
+    <template v-else>
+      <AppHeader
+        :stage-label="stageLabel"
+        :revision="displayedRevision"
+        :session-id="sessionId"
+        :resetting="resetting"
+        @reset="confirmReset"
+      />
 
-    <main class="workspace">
+      <FeatureNavigation current="builder" />
+
+      <main class="workspace">
       <div class="mobile-view-tabs" role="tablist" aria-label="移动端工作区视图">
         <button
           type="button"
@@ -176,18 +278,134 @@ onMounted(start)
           @send="sendMessage"
         />
       </section>
-    </main>
+      </main>
 
-    <footer class="app-footer">
-      <span>OfferPai Resume Agent</span>
-      <span>你的内容会保留在本次简历会话中</span>
-    </footer>
+      <footer class="app-footer">
+        <span>OfferPai Resume Agent</span>
+        <span>你的内容会保留在本次简历会话中</span>
+      </footer>
+    </template>
   </div>
 </template>
 
 <style scoped>
 .app-shell { min-height: 100vh; }
 
+.auth-gate {
+  display: grid;
+  min-height: 100vh;
+  place-items: center;
+  padding: 28px;
+  background:
+    radial-gradient(circle at 18% 12%, rgba(93, 177, 165, .17), transparent 34%),
+    radial-gradient(circle at 82% 78%, rgba(146, 188, 112, .13), transparent 32%),
+    #f5faf8;
+}
+
+.auth-gate__card {
+  width: min(100%, 520px);
+  padding: clamp(30px, 6vw, 54px);
+  border: 1px solid rgba(174, 207, 201, .82);
+  border-radius: 28px;
+  background: rgba(255, 255, 255, .9);
+  box-shadow: 0 24px 70px rgba(31, 82, 75, .12);
+}
+
+.auth-gate__brand {
+  display: flex;
+  align-items: center;
+  gap: 12px;
+  color: var(--ink);
+  font-size: 14px;
+}
+
+.auth-gate__brand span {
+  display: grid;
+  width: 38px;
+  height: 38px;
+  place-items: center;
+  border-radius: 12px;
+  color: #fff;
+  background: var(--brand-dark);
+  font-family: ui-monospace, "SFMono-Regular", Consolas, monospace;
+  font-size: 11px;
+  font-weight: 800;
+  letter-spacing: .08em;
+}
+
+.auth-gate__eyebrow {
+  margin: 48px 0 0;
+  color: var(--brand-dark);
+  font-family: ui-monospace, "SFMono-Regular", Consolas, monospace;
+  font-size: 10px;
+  font-weight: 800;
+  letter-spacing: .14em;
+}
+
+.auth-gate h1 {
+  margin: 12px 0 0;
+  color: var(--ink);
+  font-family: "Aptos Display", "MiSans", "PingFang SC", sans-serif;
+  font-size: clamp(28px, 6vw, 40px);
+  line-height: 1.18;
+}
+
+.auth-gate__message {
+  margin: 18px 0 0;
+  color: var(--ink-muted);
+  font-size: 14px;
+  line-height: 1.8;
+}
+
+.auth-gate__progress {
+  height: 4px;
+  margin-top: 34px;
+  overflow: hidden;
+  border-radius: 999px;
+  background: #e4efec;
+}
+
+.auth-gate__progress i {
+  display: block;
+  width: 42%;
+  height: 100%;
+  border-radius: inherit;
+  background: var(--brand);
+  animation: auth-progress 1.15s ease-in-out infinite alternate;
+}
+
+.auth-gate__retry {
+  min-height: 44px;
+  margin-top: 28px;
+  padding: 0 22px;
+  border: 0;
+  border-radius: 12px;
+  color: #fff;
+  background: var(--brand-dark);
+  font-size: 14px;
+  font-weight: 750;
+  cursor: pointer;
+}
+
+.auth-gate__retry:hover { filter: brightness(.94); }
+.auth-gate__retry:focus-visible { outline: 3px solid rgba(57, 139, 128, .28); outline-offset: 3px; }
+
+.auth-gate__hint {
+  margin: 24px 0 0;
+  padding-top: 20px;
+  border-top: 1px solid var(--line);
+  color: var(--ink-faint);
+  font-size: 12px;
+  line-height: 1.7;
+}
+
+.auth-gate__hint code {
+  padding: 2px 5px;
+  border-radius: 5px;
+  color: var(--brand-dark);
+  background: #eaf4f1;
+}
+
 .workspace {
   display: grid;
   width: min(1280px, calc(100% - 40px));
@@ -287,6 +505,7 @@ onMounted(start)
 }
 
 @keyframes live-pulse { 50% { opacity: .4; } }
+@keyframes auth-progress { from { transform: translateX(-10%); } to { transform: translateX(150%); } }
 
 @media (max-width: 850px) {
   .workspace { width: min(100% - 28px, 760px); grid-template-columns: 1fr; gap: 22px; padding-top: 22px; }

frontend/src/api/resumeAgent.ts

@@ -11,6 +11,11 @@ import type {
 } from '../types/resumeAgent'
 
 const API_ROOT = `${(import.meta.env.VITE_API_BASE_URL || '').replace(/\/$/, '')}/ai-api/resume-agent`
+let landingToken = ''
+
+function attachAuthorization(headers: Headers): void {
+  if (landingToken) headers.set('Authorization', `Bearer ${landingToken}`)
+}
 
 export class ResumeAgentApiError extends Error {
   readonly status: number
@@ -27,6 +32,7 @@ export class ResumeAgentApiError extends Error {
 async function request<T>(path: string, init: RequestInit = {}): Promise<T> {
   const headers = new Headers(init.headers)
   headers.set('Accept', 'application/json')
+  attachAuthorization(headers)
 
   if (init.body && !(init.body instanceof FormData) && !headers.has('Content-Type')) {
     headers.set('Content-Type', 'application/json')
@@ -65,6 +71,7 @@ async function requestBuilderSse(
 ): Promise<ResumeAgentEnvelope> {
   const headers = new Headers(init.headers)
   headers.set('Accept', 'text/event-stream')
+  attachAuthorization(headers)
   if (init.body && !headers.has('Content-Type')) headers.set('Content-Type', 'application/json')
 
   let response: Response
@@ -124,7 +131,12 @@ function sessionPath(sessionId: string, suffix = ''): string {
 }
 
 export const resumeAgentApi = {
-  createSession(signal?: AbortSignal) {
+  createSession(token: string, signal?: AbortSignal) {
+    const normalizedToken = token.trim()
+    if (!normalizedToken) {
+      throw new ResumeAgentApiError('Missing OfferPai login credential.', 401)
+    }
+    landingToken = normalizedToken
     const accountPhone = import.meta.env.VITE_DEMO_ACCOUNT_PHONE?.trim()
     return request<ResumeAgentEnvelope>('/sessions', {
       method: 'POST',
@@ -325,4 +337,3 @@ function optimizeAction(sessionId: string, action: string, entryId: string, sign
 function optimizationRunPath(sessionId: string, runId: string, suffix: string): string {
   return sessionPath(sessionId, `/resume/optimize/runs/${encodeURIComponent(runId)}${suffix}`)
 }
-

frontend/src/composables/useResumeAgent.ts

@@ -13,6 +13,20 @@ import type {
 } from '../types/resumeAgent'
 
 const STORAGE_KEY = 'offerpai.resume-agent.session-id'
+
+function consumeLandingToken(): string {
+  const url = new URL(window.location.href)
+  if (!url.searchParams.has('token')) return ''
+  const token = url.searchParams.get('token')?.trim() || ''
+  url.searchParams.delete('token')
+  window.history.replaceState(
+    window.history.state,
+    '',
+    `${url.pathname}${url.search}${url.hash}`,
+  )
+  return token
+}
+
 const VALID_BLOCK_TYPES = new Set<TimelineBlockType>([
   'text',
   'component',
@@ -209,6 +223,10 @@ function formatError(error: unknown): string {
 }
 
 export function useResumeAgent() {
+  const landingToken = consumeLandingToken()
+  const hasLandingToken = Boolean(landingToken)
+  let createWithLandingToken = Boolean(landingToken)
+  if (createWithLandingToken) localStorage.removeItem(STORAGE_KEY)
   const sessionId = ref('')
   const draftId = ref('')
   const revision = ref(0)
@@ -259,9 +277,11 @@ export function useResumeAgent() {
     if (!keepTimeline) timeline.value = state.timeline
   }
 
-  async function refreshTimeline() {
-    if (!sessionId.value) return
-    const response = await resumeAgentApi.getTimeline(sessionId.value, controller?.signal)
+  async function refreshTimeline(signal: AbortSignal | undefined = controller?.signal) {
+    const requestedSessionId = sessionId.value
+    if (!requestedSessionId) return
+    const response = await resumeAgentApi.getTimeline(requestedSessionId, signal)
+    if (signal?.aborted || sessionId.value !== requestedSessionId) return
     applyState(response)
   }
 
@@ -273,13 +293,18 @@ export function useResumeAgent() {
   }
 
   async function start() {
+    if (!landingToken) {
+      initializing.value = false
+      errorMessage.value = '缺少 OfferPai 登录凭证，请通过带有 ?token= 的入口重新进入。'
+      return
+    }
     controller?.abort()
     const activeController = new AbortController()
     controller = activeController
     initializing.value = true
     errorMessage.value = ''
 
-    const storedSessionId = localStorage.getItem(STORAGE_KEY)
+    const storedSessionId = createWithLandingToken ? null : localStorage.getItem(STORAGE_KEY)
     try {
       if (storedSessionId) {
         sessionId.value = storedSessionId
@@ -293,8 +318,12 @@ export function useResumeAgent() {
         }
       }
 
-      const response = await resumeAgentApi.createSession(activeController.signal)
+      const response = await resumeAgentApi.createSession(
+        landingToken,
+        activeController.signal,
+      )
       applyState(response)
+      createWithLandingToken = false
       if (!timeline.value.length) await refreshTimeline()
     } catch (error) {
       if (activeController.signal.aborted) return
@@ -407,6 +436,7 @@ export function useResumeAgent() {
     resumeId.value = ''
     resumeHook.value = null
     resetting.value = false
+    createWithLandingToken = Boolean(landingToken)
     await start()
   }
 
@@ -417,6 +447,7 @@ export function useResumeAgent() {
   onBeforeUnmount(() => controller?.abort())
 
   return {
+    hasLandingToken,
     sessionId,
     draftId,
     revision,
@@ -447,5 +478,3 @@ export function useResumeAgent() {
     clearError,
   }
 }
-
-