fix: turn main's Backend tests green (3 fixes #75 did not carry)

backend/app/ops.py

@@ -2076,7 +2076,13 @@ async def _persist_external_resume_proposal(**kwargs: Any) -> dict[str, Any]:
 async def _review_resume_proposal_items(**kwargs: Any) -> dict[str, Any]:
     from app.services.resume_workspace import review_resume_proposal_items
 
-    return await review_resume_proposal_items(**kwargs)
+    # A confirmed decision-plan node is the user's adoption decision: once
+    # its review leaves no pending change, the proposal is finalized there.
+    # A direct workspace review keeps the proposal open until the user saves
+    # a version (create_resume_version_record finalizes it then).
+    authorization = _OPERATION_AUTHORIZATION.get()
+    finalize = bool(authorization is not None and getattr(authorization, "plan_id", ""))
+    return await review_resume_proposal_items(**kwargs, finalize_when_complete=finalize)
 
 
 async def _propose_resume_decision_plan(**kwargs: Any) -> dict[str, Any]:

backend/app/services/local_data_reset.py

@@ -377,6 +377,13 @@ async def _cancel_task_objects(tasks: list[asyncio.Task[Any]]) -> int:
     current = asyncio.current_task()
     active = [task for task in tasks if task is not current and not task.done()]
     for task in active:
+        # A worker may already have been asked to stop (cancel_career_task
+        # cancels its live task). A second cancel() would land inside the
+        # worker's final awaited materialization -- e.g. mid-commit -- and
+        # orphan an open SQLite write transaction that holds the database
+        # lock for the rest of the reset. Cancel once, then await it.
+        if task.cancelling():
+            continue
         task.cancel()
     if active:
         await asyncio.gather(*active, return_exceptions=True)

backend/app/services/resume_route_operations.py

@@ -38,7 +38,11 @@
 from app.services.application_workspace import auto_write_job_to_total
 from app.services.resume_drafts import save_resume_draft
 from app.services.resume_fact_gates import validate_generated_content
-from app.services.resume_versions import create_version_snapshot, snapshot_resume
+from app.services.resume_versions import (
+    create_version_snapshot,
+    mark_workspace_proposal_accepted,
+    snapshot_resume,
+)
 from app.services.security_redaction import safe_error_message
 from app.runtime_paths import runtime_uploads_dir
 
@@ -796,17 +800,7 @@ async def create_resume_version_record(
             ).scalars().all()
         )
         for proposal in pending_proposals:
-            change_ids = {
-                item.get("change_id")
-                for item in (proposal.diff_json or [])
-                if isinstance(item, dict) and item.get("change_id")
-            }
-            reviewed_ids = set((proposal.item_reviews_json or {}).keys())
-            if change_ids.issubset(reviewed_ids):
-                proposal.status = "accepted"
-                proposal.accepted_resume_id = resume.id
-                proposal.accepted_resume_version_id = version.id
-                proposal.reviewed_at = datetime.now(timezone.utc).replace(tzinfo=None)
+            mark_workspace_proposal_accepted(proposal, resume=resume, version=version)
         from app.services.career_resume import added_resume_evidence
 
         added_evidence = added_resume_evidence(

backend/app/services/resume_versions.py

@@ -75,3 +75,40 @@ async def create_version_snapshot(
             last_exc = exc
             continue
     raise RuntimeError("resume_version race: failed after 3 attempts") from last_exc
+
+
+_OPEN_WORKSPACE_PROPOSAL_STATES = ("ready", "in_review")
+
+
+def proposal_fully_reviewed(proposal) -> bool:
+    """Every diff row of a workspace-bound proposal has a recorded review.
+
+    A proposal without diff rows has nothing left to review.
+    """
+    change_ids = {
+        str(item.get("change_id"))
+        for item in (proposal.diff_json or [])
+        if isinstance(item, dict) and item.get("change_id")
+    }
+    reviews = proposal.item_reviews_json or {}
+    return all(
+        isinstance(reviews.get(change_id), dict) for change_id in change_ids
+    )
+
+
+def mark_workspace_proposal_accepted(proposal, *, resume: Resume, version: ResumeVersion) -> bool:
+    """Finalize a fully reviewed workspace proposal onto ``version``.
+
+    Shared by the explicit "save version" path and the per-item review path,
+    so both record the same terminal state. Idempotent: an already-finalized
+    (or stale/blocked/rejected) proposal is left untouched.
+    """
+    from datetime import datetime, timezone
+
+    if proposal.status not in _OPEN_WORKSPACE_PROPOSAL_STATES or not proposal_fully_reviewed(proposal):
+        return False
+    proposal.status = "accepted"
+    proposal.accepted_resume_id = resume.id
+    proposal.accepted_resume_version_id = version.id
+    proposal.reviewed_at = datetime.now(timezone.utc).replace(tzinfo=None)
+    return True

backend/app/services/resume_workspace.py

@@ -33,7 +33,11 @@
 from app.services.resume_fact_gates import validate_edited_text
 from app.services.resume_builder import _profile_to_contact_json
 from app.services.resume_optimization import _proposal_detail
-from app.services.resume_versions import create_version_snapshot
+from app.services.resume_versions import (
+    create_version_snapshot,
+    mark_workspace_proposal_accepted,
+    proposal_fully_reviewed,
+)
 from app.services.packet_readiness import project_packet_state
 
 
@@ -632,11 +636,54 @@ async def review_resume_proposal_item(
     return await _review_resume_proposal_items(proposal_id, resume_id, [change_id], action, edited_text)
 
 
+async def _finalize_fully_reviewed_proposal(db, proposal, resume: Resume) -> bool:
+    """Close a workspace proposal once its last change has been reviewed.
+
+    With at least one accepted change the workspace now holds the adopted
+    resume: snapshot it as a ResumeVersion and mark the proposal accepted
+    through the same helper the "save version" path uses. When every change
+    was rejected nothing was adopted, so the proposal is closed as rejected.
+    Runs inside the caller's transaction; a no-op once finalized.
+    """
+    if proposal.status not in {"ready", "in_review"} or not proposal_fully_reviewed(proposal):
+        return False
+    reviews = proposal.item_reviews_json or {}
+    change_ids = [
+        str(item.get("change_id"))
+        for item in (proposal.diff_json or [])
+        if isinstance(item, dict) and item.get("change_id")
+    ]
+    if not any(reviews[change_id].get("action") == "accept" for change_id in change_ids):
+        proposal.status = "rejected"
+        proposal.reviewed_at = _now()
+        return True
+    # Sections added by this review were flushed by resume_id only; reload the
+    # collection so the version snapshot holds the adopted content.
+    await db.flush()
+    await db.refresh(resume, attribute_names=["sections"])
+    version = await create_version_snapshot(
+        db,
+        resume,
+        change_summary=f"采纳简历优化提案 {proposal.proposal_id} 的全部审核结果",
+        created_by="resume_workspace_review",
+    )
+    resume.current_version_id = version.id
+    return mark_workspace_proposal_accepted(proposal, resume=resume, version=version)
+
+
 async def review_resume_proposal_items(
     proposal_id: str, resume_id: int, change_ids: list[str], action: str,
+    *, finalize_when_complete: bool = False,
 ) -> dict[str, Any]:
-    """Apply one explicitly reviewed set atomically; never approve unrelated operations."""
-    return await _review_resume_proposal_items(proposal_id, resume_id, change_ids, action)
+    """Apply one explicitly reviewed set atomically; never approve unrelated operations.
+
+    ``finalize_when_complete`` (set for confirmed decision-plan nodes) closes
+    the proposal in the same transaction once no change is left pending.
+    """
+    return await _review_resume_proposal_items(
+        proposal_id, resume_id, change_ids, action,
+        finalize_when_complete=finalize_when_complete,
+    )
 
 
 async def _review_resume_proposal_items(
@@ -645,6 +692,8 @@ async def _review_resume_proposal_items(
     change_ids: list[str],
     action: str,
     edited_text: str = "",
+    *,
+    finalize_when_complete: bool = False,
 ) -> dict[str, Any]:
     clean_proposal_id = _text(proposal_id, "proposal_id", 80)
     clean_resume_id = _positive_id(resume_id, "resume_id")
@@ -683,6 +732,11 @@ async def _review_resume_proposal_items(
             raise ValueError("该条目已有不同审核结果，请重新查看提案")
         pending_ids = [change_id for change_id in clean_change_ids if not isinstance(reviews.get(change_id), dict)]
         if not pending_ids:
+            # A replayed last review still finalizes a proposal whose earlier
+            # attempt recorded every review but was never finalized.
+            if finalize_when_complete and await _finalize_fully_reviewed_proposal(db, proposal, resume):
+                await db.commit()
+                resume = await _load_resume(db, resume.id)
             return {
                 **await _workspace_payload(db, resume, job=job, proposal_id=proposal.proposal_id),
                 "duplicate": True,
@@ -843,6 +897,8 @@ async def _review_resume_proposal_items(
             proposal.status = "in_review"
         if clean_action == "accept":
             proposal.workspace_snapshot_hash = workspace_content_hash(resume)
+        if finalize_when_complete:
+            await _finalize_fully_reviewed_proposal(db, proposal, resume)
         await db.commit()
         resume = await _load_resume(db, resume.id)
         return {

backend/tests/test_career_questions.py

@@ -280,7 +280,9 @@ async def inspect() -> tuple[
                 events = (
                     await db.execute(
                         select(AutomationEvent).where(
-                            AutomationEvent.dedupe_key.like("career-answer-reprepare:%")
+                            AutomationEvent.dedupe_key.like("career-answer-reprepare:%"),
+                            # The suite shares one database; pin to this test's job.
+                            AutomationEvent.target_id == str(outcome["job_id"]),
                         )
                     )
                 ).scalars().all()
@@ -421,7 +423,9 @@ async def inspect() -> AutomationEvent | None:
                 rows = (
                     await db.execute(
                         select(AutomationEvent).where(
-                            AutomationEvent.dedupe_key.like("career-answer-rediscovery:%")
+                            AutomationEvent.dedupe_key.like("career-answer-rediscovery:%"),
+                            # The suite shares one database; pin to this test's profile.
+                            AutomationEvent.target_id == str(outcome["profile_id"]),
                         )
                     )
                 ).scalars().all()