diff --git a/backend/app/ops.py b/backend/app/ops.py index df1cf5b5..a20192f0 100644 --- a/backend/app/ops.py +++ b/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]: diff --git a/backend/app/services/local_data_reset.py b/backend/app/services/local_data_reset.py index c4b4d4ba..f661f490 100644 --- a/backend/app/services/local_data_reset.py +++ b/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) diff --git a/backend/app/services/resume_route_operations.py b/backend/app/services/resume_route_operations.py index 0a7f80e9..54f0c173 100644 --- a/backend/app/services/resume_route_operations.py +++ b/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( diff --git a/backend/app/services/resume_versions.py b/backend/app/services/resume_versions.py index 61dcb614..ae038e7b 100644 --- a/backend/app/services/resume_versions.py +++ b/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 diff --git a/backend/app/services/resume_workspace.py b/backend/app/services/resume_workspace.py index b1a4a9cf..c7c56069 100644 --- a/backend/app/services/resume_workspace.py +++ b/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 { diff --git a/backend/tests/test_career_questions.py b/backend/tests/test_career_questions.py index af6b1b89..2fb21be6 100644 --- a/backend/tests/test_career_questions.py +++ b/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()