""" Management command to approve a pending manual gate and resume the pipeline. Usage ----- .. code-block:: sh # Approve all pending gates for a pipeline python manage.py approve_gate # Approve a specific gate python manage.py approve_gate --gate 02 # Reject a gate (pipeline stays paused) python manage.py approve_gate --gate 02 --reject --note "Add missing index" # List pending gates python manage.py approve_gate --list Workflow After Gate Pause -------------------------- 1. Pipeline runs up to Agent 02 (Data Architect) 2. Gate fires → pipeline PAUSED, output saved to .docs/generated//02-data-model.md 3. You review the output file, make changes if needed 4. Run: python manage.py approve_gate 5. Pipeline resumes with step 4 (parallel group A) """ from __future__ import annotations from django.core.management.base import BaseCommand, CommandError class Command(BaseCommand): help = __doc__ def add_arguments(self, parser): parser.add_argument( "pipeline_run_id", nargs="?", help="PipelineRun public_id (UUID).", ) parser.add_argument( "--gate", help="Specific agent ID to approve, e.g. '02', '07', '08'. If omitted, approves ALL pending.", ) parser.add_argument( "--reject", action="store_true", help="Reject the gate instead of approving.", ) parser.add_argument( "--note", default="", help="Optional note (required if rejecting).", ) parser.add_argument( "--list", action="store_true", help="List all pipelines with pending manual gates.", ) def handle(self, **options): if options["list"]: self._list_pending() return if not options["pipeline_run_id"]: raise CommandError("Pipeline run ID is required. Usage: manage.py approve_gate ") self._approve_gate( pipeline_run_id=options["pipeline_run_id"], gate=options["gate"], reject=options["reject"], note=options["note"], ) # ------------------------------------------------------------------ # Approve / reject gate # ------------------------------------------------------------------ def _approve_gate(self, pipeline_run_id: str, gate: str | None, reject: bool, note: str) -> None: from simorgh.apps.automation.agent_models import ( ApprovalGate, GateStatus, PipelineRun, PipelineStatus, ) from django.utils import timezone try: pr = PipelineRun.objects.get(public_id=pipeline_run_id) except PipelineRun.DoesNotExist: raise CommandError(f"Pipeline run not found: {pipeline_run_id}") if pr.status not in (PipelineStatus.PAUSED,): raise CommandError( f"Pipeline {pipeline_run_id} is '{pr.status}', not 'paused'. " f"No gates are awaiting review." ) # Find pending gates gates = pr.gates.filter(status=GateStatus.AWAITING_REVIEW) if gate: gates = gates.filter(after_agent_id=gate) if not gates.exists(): self.stdout.write(self.style.WARNING( f"No pending gates found. Gates:\n" + "\n".join(f" {g.after_agent_id}: {g.get_status_display()}" for g in pr.gates.all()) )) return action = "REJECTED" if reject else "APPROVED" new_status = GateStatus.FAILED if reject else GateStatus.PASSED for g in gates: g.status = new_status g.reviewed_by = "cli" g.reviewed_at = timezone.now() if note: g.notes = note elif not reject: g.notes = "Approved via CLI." g.save(update_fields=["status", "reviewed_by", "reviewed_at", "notes"]) self.stdout.write(self.style.SUCCESS( f" ✅ Gate {g.after_agent_id} ({g.gate_type}) → {action}" )) if reject: self.stdout.write(self.style.WARNING( f"\nPipeline {pipeline_run_id} remains PAUSED. " f"Re-run affected agents or address issues, then approve the gate." )) return # Check if all gates are resolved still_pending = pr.gates.filter(status=GateStatus.AWAITING_REVIEW).exists() if still_pending: self.stdout.write(self.style.WARNING( f"\n⚠️ Other gates still pending. Pipeline remains PAUSED." )) else: # Resume pipeline from simorgh.apps.automation.agent_tasks import build_pipeline_workflow pr.status = PipelineStatus.RUNNING pr.save(update_fields=["status"]) workflow = build_pipeline_workflow(pr.module_name, str(pr.public_id)) result = workflow.apply_async() pr.celery_workflow_id = result.id pr.save(update_fields=["celery_workflow_id"]) self.stdout.write(self.style.SUCCESS( f"\n✅ All gates approved! Pipeline RESUMED.\n" f" Workflow ID: {result.id}\n" f" Monitor: manage.py run_agent_pipeline {pr.module_name} --status" )) # Show next steps next_agents = pr.agent_runs.filter(status="pending").order_by("step_index")[:5] if next_agents.exists(): self.stdout.write(f"\n Next agents:") for ar in next_agents: pg = f" [{ar.parallel_group}]" if ar.parallel_group else "" model = f" [{ar.model_name or pr.default_model}]" self.stdout.write(f" → {ar.agent_id} {ar.agent_name}{pg}{model}") # ------------------------------------------------------------------ # List pending # ------------------------------------------------------------------ def _list_pending(self) -> None: from simorgh.apps.automation.agent_models import ApprovalGate, GateStatus, PipelineRun pipelines = PipelineRun.objects.filter(status="paused").order_by("-updated_at") if not pipelines.exists(): self.stdout.write("No paused pipelines with pending gates.") return for pr in pipelines: pending = pr.gates.filter(status=GateStatus.AWAITING_REVIEW) if not pending.exists(): continue self.stdout.write(f"\n{'='*60}") self.stdout.write(f" Pipeline: {pr.module_name}") self.stdout.write(f" Run ID: {pr.public_id}") self.stdout.write(f" Status: {pr.status}") self.stdout.write(f" Paused at step: {pr.current_step}") self.stdout.write(f" {'='*60}") for g in pending: output_path = f".docs/generated/{pr.module_name}/{g.after_agent_id}-*.md" self.stdout.write(f"\n 🛑 Gate: after Agent {g.after_agent_id}") self.stdout.write(f" Type: {g.gate_type}") self.stdout.write(f" Description: {g.description}") self.stdout.write(f" Output: {output_path}") self.stdout.write(f" Command to approve:") self.stdout.write(f" manage.py approve_gate {pr.public_id} --gate {g.after_agent_id}") self.stdout.write(f" Command to reject:") self.stdout.write(f" manage.py approve_gate {pr.public_id} --gate {g.after_agent_id} --reject --note \"reason\"") self.stdout.write("")