fix(ingest): 修复 SQLite 外键序导致采集入库与整包状态写入 500

采集 upsert_ingest_candidate 与整包 replace_state 在无 ORM relationship 时,SQLAlchemy 同一 commit 内不保证按外键依赖顺序插入,子表(ingest_items / tasks / recruitment_events)可能在父行(candidates / resume_files / offers)存在前被写入,SQLite 开启 PRAGMA foreign_keys 时报 FOREIGN KEY constraint failed。

- ingest: 在添加 ingest_items 前显式 flush 父行。
- state: 在候选人/Offer/任务各循环后显式 flush,强制 岗位→候选人→Offer→任务→事件 顺序;并对任务/事件的外键做兜底:目标行不存在时置空,避免孤儿引用(如迁入 _pool 又被清空的候选人)导致 500。
- 新增开启外键约束的回归测试(测试夹具引擎未开 PRAGMA,此前掩蔽此类问题)。

backend/app/repositories/ingest_repository.py

@@ -181,6 +181,11 @@ def upsert_ingest_candidate(
             created_at=at,
         )
     )
+    # 先把候选人与简历行落库:SQLite 外键在 ingest_items 插入时会校验
+    # candidate_id / resume_file_id 指向的行必须已存在。这里没有配置 ORM
+    # relationship,同一 commit 内 SQLAlchemy 的插入顺序不保证把父行排在
+    # ingest_items 之前,故显式 flush,避免 FOREIGN KEY constraint failed。
+    session.flush()
     payload = {
         "source": source,
         "sourceUrl": raw.get("sourceUrl") or "",

backend/app/repositories/state_repository.py

@@ -235,6 +235,7 @@ def replace_state(session: Session, owner: str, input_state: dict[str, Any], imp
                     updated_at=state["updatedAt"],
                 )
             )
+    session.flush()
 
     for offer in state["offers"]:
         if not offer.get("candidateId"):
@@ -256,11 +257,30 @@ def replace_state(session: Session, owner: str, input_state: dict[str, Any], imp
                 updated_at=state["updatedAt"],
             )
         )
+    session.flush()
+
+    # 整包覆盖后,子表(任务/事件)可能引用已迁出本工作区(如 _pool)或已被清空的
+    # 候选人/岗位/Offer。外键只跨表校验行是否存在(不看 owner),这里按当前库内
+    # 真实存在的 id 集合兜底:目标不存在时把外键置空,避免 PUT /api/state 因
+    # FOREIGN KEY constraint failed 500。存在的跨工作区引用(如 _pool 候选人)
+    # 仍在库内,不会被误置空。
+    job_ids = set(session.scalars(select(Job.id)))
+    candidate_ids = set(session.scalars(select(Candidate.id)))
+    offer_ids = set(session.scalars(select(Offer.id)))
 
     for task in state["tasks"]:
         task_id = task.get("id") or _id("task")
         task = {**task, "id": task_id}
         created_at = task.get("createdAt") or state["updatedAt"]
+        candidate_id = task.get("candidateId") or None
+        if candidate_id and candidate_id not in candidate_ids:
+            candidate_id = None
+        job_id = task.get("jobId") or None
+        if job_id and job_id not in job_ids:
+            job_id = None
+        offer_id = task.get("offerId") or None
+        if offer_id and offer_id not in offer_ids:
+            offer_id = None
         session.add(
             Task(
                 id=task_id,
@@ -273,28 +293,35 @@ def replace_state(session: Session, owner: str, input_state: dict[str, Any], imp
                 status=task.get("status") or "",
                 type=task.get("type") or "",
                 current_node=task.get("currentNode") or "",
-                candidate_id=task.get("candidateId") or None,
-                job_id=task.get("jobId") or None,
-                offer_id=task.get("offerId") or None,
+                candidate_id=candidate_id,
+                job_id=job_id,
+                offer_id=offer_id,
                 data=dump_json(task),
                 created_at=created_at,
                 updated_at=state["updatedAt"],
             )
         )
+    session.flush()
 
     for event in state["eventLog"]:
         event_id = event.get("id") or _id("evt")
         at = event.get("at") or event.get("timestamp") or state["updatedAt"]
         event = {**event, "id": event_id, "at": at}
+        candidate_id = event.get("candidateId") or None
+        if candidate_id and candidate_id not in candidate_ids:
+            candidate_id = None
+        job_id = event.get("jobId") or None
+        if job_id and job_id not in job_ids:
+            job_id = None
         session.add(
             RecruitmentEvent(
                 id=event_id,
                 owner_id=owner,
                 type=event.get("type") or "unknown",
                 at=at,
-                candidate_id=event.get("candidateId") or None,
+                candidate_id=candidate_id,
                 candidate_name=event.get("candidateName") or "",
-                job_id=event.get("jobId") or None,
+                job_id=job_id,
                 job_title=event.get("jobTitle") or "",
                 source=event.get("source") or "",
                 stage=event.get("stage") or "",

backend/tests/test_ingest.py

@@ -99,3 +99,75 @@ def test_ingest_attributes_to_logged_in_user(client):
     assert resp.json()["ok"] is True
     state = client.get("/api/state").json()
     assert any(c["name"] == "周八" for c in state["candidates"])
+
+
+def test_ingest_with_pdf_does_not_violate_foreign_key(tmp_path):
+    """回归:带 pdfUrl 的采集走 ResumeFile 入库路径,SQLite 外键校验 candidate_id /
+    resume_file_id 指向的行必须先于 ingest_items 存在。
+
+    测试夹具的引擎未开启 PRAGMA foreign_keys(见 conftest),该用例自建引擎开启外键,
+    复现并锁定「同一 commit 内父行晚于 ingest_items 插入导致 FOREIGN KEY constraint
+    failed」的 500。saved_file 依赖上游 save_resume_bytes 的下游字段(与生产一致)。
+    """
+    from backend.app.models import (
+        Base,
+        Candidate,
+        IngestItem,
+        RecruitmentEvent,
+        ResumeFile,
+        ResumeProfile,
+        ResumeText,
+    )
+    from backend.app.repositories.ingest_repository import upsert_ingest_candidate
+    from sqlalchemy import create_engine, event
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(f"sqlite:///{(tmp_path / 'ingest.sqlite').as_posix()}")
+
+    @event.listens_for(engine, "connect")
+    def _enable_fk(dbapi_connection, connection_record):  # noqa: ARG001
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(bind=engine)
+    session = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)()
+
+    saved_file = {
+        "originalName": "platform-resume.pdf",
+        "storedName": "platform-resume.pdf",
+        "storageKey": "resumes/platform-resume.pdf",
+        "storagePath": str(tmp_path / "uploads/resumes/platform-resume.pdf"),
+        "fileHash": "a" * 64,
+        "sizeBytes": 1024,
+        "fileType": "pdf",
+    }
+    raw = {
+        "name": "候选人甲",
+        "source": "Boss直聘",
+        "sourceUrl": "https://www.zhipin.com/web/chat/index",
+        "pdfUrl": "https://example.test/resume.pdf",
+    }
+    parsed = {"name": "候选人甲", "jobTitle": "前端工程师", "resumeText": "Node 工程化", "skills": ["React"]}
+
+    try:
+        result = upsert_ingest_candidate(session, "admin", raw, parsed, saved_file)
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+    assert result["ok"] is True
+    assert result["candidateId"]
+    assert result["resumeFileId"]
+
+    db = create_engine(f"sqlite:///{(tmp_path / 'ingest.sqlite').as_posix()}")
+    assert db.connect().execute(Candidate.__table__.select()).fetchall()
+    assert db.connect().execute(ResumeFile.__table__.select()).fetchall()
+    assert db.connect().execute(ResumeText.__table__.select()).fetchall()
+    assert db.connect().execute(ResumeProfile.__table__.select()).fetchall()
+    assert db.connect().execute(RecruitmentEvent.__table__.select()).fetchall()
+    assert db.connect().execute(IngestItem.__table__.select()).fetchall()
+    db.dispose()

backend/tests/test_state.py

@@ -122,3 +122,81 @@ def test_legacy_offer_unique_constraint_is_dropped_on_startup(tmp_path, monkeypa
 
     dbm.engine.dispose()
     config.get_settings.cache_clear()
+
+
+def _fk_enabled_session(tmp_path):
+    """自建一个开启 SQLite 外键约束的会话(conftest 的夹具引擎未开 PRAGMA,
+    否则测不出 replace_state 的外键序问题)。"""
+    from backend.app.models import Base
+    from sqlalchemy import create_engine, event
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(f"sqlite:///{(tmp_path / 'state.sqlite').as_posix()}")
+
+    @event.listens_for(engine, "connect")
+    def _enable_fk(dbapi_connection, connection_record):  # noqa: ARG001
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(bind=engine)
+    session = sessionmaker(bind=engine, autoflush=False, autocommit=False, expire_on_commit=False)()
+    return session
+
+
+def test_replace_state_preserves_valid_fk_links(tmp_path):
+    """整包覆盖时,任务/事件到岗位/候选人/Offer 的有效外键必须在提交后保存,
+    不得因同 commit 内 SQLAlchemy 插入顺序不稳而被置空或 500。"""
+    from backend.app.models import Offer, Task
+    from backend.app.repositories.state_repository import replace_state
+
+    session = _fk_enabled_session(tmp_path)
+    state = {
+        "jobs": [{"id": "job-1", "title": "销售经理", "status": "进行中"}],
+        "candidates": [{"id": "cand-1", "jobId": "job-1", "name": "薛庆霞", "jobTitle": "销售经理"}],
+        "offers": [{"id": "offer-1", "candidateId": "cand-1", "status": "待 HR 确认"}],
+        "tasks": [
+            {
+                "id": "task-1",
+                "title": "安排面试",
+                "candidateId": "cand-1",
+                "jobId": "job-1",
+                "offerId": "offer-1",
+                "type": "面试安排",
+                "currentNode": "面试日程安排",
+            }
+        ],
+        "eventLog": [{"id": "evt-1", "type": "candidate_created", "candidateId": "cand-1", "jobId": "job-1"}],
+    }
+    try:
+        replace_state(session, "test1", state, "api")
+        task = session.query(Task).filter(Task.id == "task-1").one()
+        assert task.candidate_id == "cand-1"
+        assert task.job_id == "job-1"
+        assert task.offer_id == "offer-1"
+        assert session.query(Offer).filter(Offer.id == "offer-1").one().id == "offer-1"
+    finally:
+        session.close()
+
+
+def test_replace_state_null_dangling_fk_links(tmp_path):
+    """整包覆盖遇到指向库外目标的孤儿引用(如候选人已迁入 _pool 又被清空)时,
+    应把外键置空而不是 FOREIGN KEY constraint failed。"""
+    from backend.app.models import RecruitmentEvent
+    from backend.app.repositories.state_repository import replace_state
+
+    session = _fk_enabled_session(tmp_path)
+    state = {
+        "jobs": [{"id": "job-1", "title": "销售经理"}],
+        "candidates": [{"id": "cand-1", "jobId": "job-1", "name": "薛庆霞"}],
+        "offers": [],
+        "tasks": [],
+        "eventLog": [{"id": "evt-1", "type": "candidate_created", "candidateId": "cand-ghost", "jobId": "job-1"}],
+    }
+    try:
+        replace_state(session, "test1", state, "api")
+        event = session.query(RecruitmentEvent).filter(RecruitmentEvent.id == "evt-1").one()
+        assert event.candidate_id is None
+        assert event.job_id == "job-1"
+    finally:
+        session.close()