from __future__ import annotations from typing import Any from app.models.schemas import ( WorkflowReviewFindingSummary, WorkflowReviewNextActionItem, WorkflowReviewPolicyCheckResult, WorkflowReviewStartRequest, WorkflowReviewStepOutcome, ) from app.modules.core.application.workflow_persistence.builders import build_close_payload from .finding_lifecycle import WorkflowPersistenceFindingLifecycleMixin class WorkflowPersistenceWritePhaseMixin(WorkflowPersistenceFindingLifecycleMixin): """Core transactional write phase for review sessions.""" def _execute_core_review_write_phase( self, *, request: WorkflowReviewStartRequest, review_payload: dict[str, Any], review_code: str, policy_checks: list[WorkflowReviewPolicyCheckResult], ) -> dict[str, Any]: """Chạy pha ghi core trước side effect.""" def operation(connection: Any) -> dict[str, Any]: session = self._postgres_client.create_dossier_review_session( review_payload, connection=connection ) created_snapshot_ids = [ item.get("snapshot_id", "") for item in session.get("snapshots", []) if item.get("snapshot_id") ] finding_summaries: list[WorkflowReviewFindingSummary] = [] next_actions: list[WorkflowReviewNextActionItem] = [] lifecycle_outcomes: list[WorkflowReviewStepOutcome] = [] workflow_employee_actions = list(request.participating_employees) close_ready = True blocking_policy = any(not item.allowed for item in policy_checks) requires_manual_intervention = any( (not item.allowed) or item.requires_manual_review or item.circuit_breaker_triggered for item in policy_checks ) blocked_reasons = self._collect_policy_blocked_reasons(policy_checks) assigned_findings_count = 0 submitted_findings_count = 0 verified_findings_count = 0 escalated_findings_count = 0 manual_follow_up_count = 0 workflow_employee_actions.extend( self._build_policy_employee_actions( review_code=review_code, policy_checks=policy_checks, project_code=request.project_code, lead_agent_code=request.lead_agent_code, lead_reviewer_employee_code=request.lead_reviewer_employee_code, ) ) lifecycle_outcomes.append( WorkflowReviewStepOutcome( step="start", executed=True, succeeded=True, target_type="review", target_id=session["review_id"], target_code=review_code, status=session.get("status"), reason="Đã khởi tạo session workflow dossier review.", metadata={ "policy_check_count": len(policy_checks), "requires_manual_intervention": requires_manual_intervention, }, ) ) for finding in request.findings: self._process_finding_write_phase( request=request, session=session, review_code=review_code, finding=finding, policy_checks=policy_checks, workflow_employee_actions=workflow_employee_actions, finding_summaries=finding_summaries, next_actions=next_actions, lifecycle_outcomes=lifecycle_outcomes, counters={ "assigned_findings_count": assigned_findings_count, "submitted_findings_count": submitted_findings_count, "verified_findings_count": verified_findings_count, "escalated_findings_count": escalated_findings_count, "manual_follow_up_count": manual_follow_up_count, }, blocking_policy=blocking_policy, connection=connection, ) assigned_findings_count = ( lifecycle_outcomes[-1].metadata.get( "assigned_findings_count", assigned_findings_count, ) if lifecycle_outcomes else assigned_findings_count ) submitted_findings_count = ( lifecycle_outcomes[-1].metadata.get( "submitted_findings_count", submitted_findings_count, ) if lifecycle_outcomes else submitted_findings_count ) verified_findings_count = ( lifecycle_outcomes[-1].metadata.get( "verified_findings_count", verified_findings_count, ) if lifecycle_outcomes else verified_findings_count ) escalated_findings_count = ( lifecycle_outcomes[-1].metadata.get( "escalated_findings_count", escalated_findings_count, ) if lifecycle_outcomes else escalated_findings_count ) manual_follow_up_count = ( lifecycle_outcomes[-1].metadata.get( "manual_follow_up_count", manual_follow_up_count, ) if lifecycle_outcomes else manual_follow_up_count ) close_ready = close_ready and not lifecycle_outcomes[-1].metadata.get( "manual_required_for_finding", False ) close_executed = False resumable_from_step = None automation_status = "completed" post_close_actions: list[WorkflowReviewNextActionItem] = [] if ( close_ready and request.findings and not blocking_policy and not requires_manual_intervention ): close_payload = build_close_payload(review_code) self._postgres_client.close_dossier_review_session( session["review_id"], close_payload.model_dump(exclude_none=True), connection=connection, ) close_executed = True lifecycle_outcomes.append( WorkflowReviewStepOutcome( step="close", executed=True, succeeded=True, target_type="review", target_id=session["review_id"], target_code=review_code, status="closed", reason="Đã tự động đóng review sau khi hoàn tất verify.", ) ) post_close_actions = self._build_post_close_actions( request=request, review_id=session["review_id"], review_code=review_code, ) next_actions.extend(post_close_actions) else: automation_status = "manual_follow_up_required" resumable_from_step = "assign" if manual_follow_up_count else "close" lifecycle_outcomes.append( WorkflowReviewStepOutcome( step="close", executed=False, succeeded=False, requires_manual_intervention=True, blocked=True, target_type="review", target_id=session["review_id"], target_code=review_code, status="open", reason="Chưa thể tự động close do còn finding cần follow-up tay hoặc policy guardrail đang chặn.", metadata={ "close_ready": close_ready, "blocking_policy": blocking_policy, "requires_manual_intervention": requires_manual_intervention, }, ) ) return { "session": session, "created_snapshot_ids": created_snapshot_ids, "finding_summaries": finding_summaries, "next_actions": next_actions, "lifecycle_outcomes": lifecycle_outcomes, "workflow_employee_actions": workflow_employee_actions, "blocked_reasons": blocked_reasons, "close_executed": close_executed, "resumable_from_step": resumable_from_step, "automation_status": automation_status, "post_close_actions": post_close_actions, "assigned_findings_count": assigned_findings_count, "submitted_findings_count": submitted_findings_count, "verified_findings_count": verified_findings_count, "escalated_findings_count": escalated_findings_count, "manual_follow_up_count": manual_follow_up_count, "requires_manual_intervention": requires_manual_intervention or bool(manual_follow_up_count), } transaction_runner = getattr(self._postgres_client, "run_in_transaction", None) if callable(transaction_runner): return transaction_runner(operation) return operation(None)