"""Scheduled jobs — admin API. Read-mostly by design: the admin console lets IT admin see timers, trigger one ad-hoc, pause/resume, and tweak cadence. Creating brand-new jobs is *not* an API operation — new jobs ship via a migration so the code and the row stay in lockstep. """ from __future__ import annotations from typing import Any from uuid import UUID from fastapi import APIRouter, Depends, Request from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( check_sensitive_endpoint_rate_limit, get_real_ip, require_admin, require_admin_read, require_browser_session, ) from app.core.response import paginated_response, success_response from app.scheduled_jobs import chain, hygiene from app.scheduled_jobs import service as sched_service from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun router = APIRouter(prefix="/api/v1/scheduled-jobs", tags=["scheduled-jobs"]) class ScheduledJobResponse(BaseModel): id: UUID name: str display_name: str description: str | None = None handler_key: str trigger_type: str interval_seconds: int | None = None cron_expr: str | None = None timezone: str enabled: bool config: dict | None = None last_run_at: str | None = None last_status: str | None = None last_duration_ms: int | None = None last_error: str | None = None last_result_summary: dict | None = None run_count: int failure_count: int created_at: str updated_at: str #: Ledger-hygiene verdict — why this row may look alive but do #: nothing. One of ``ok`` / ``handler_missing`` / #: ``integration_missing`` / ``integration_disabled`` / ``never_run`` #: / ``idle_skipping``. See ``scheduled_jobs/hygiene.py``. health: str = "ok" #: Execution weight class from the handler registry (``light`` / #: ``heavy``). Drives which concurrency gate the run queues on, so #: it belongs next to the schedule when an admin is reasoning about #: why a job started late. weight: str = "light" #: Predecessor for a ``trigger_type='chained'`` row: this job runs #: when that one finishes, not on a clock. chain_after_job_id: UUID | None = None #: Resolved in one batch query so the list can render "接在 X 之后" #: without a per-row round-trip. chain_after_name: str | None = None class ScheduledJobRunResponse(BaseModel): id: UUID job_id: UUID started_at: str finished_at: str | None = None duration_ms: int | None = None status: str triggered_by: str error_message: str | None = None result_summary: dict | None = None acknowledged_at: str | None = None acknowledged_by_id: str | None = None acknowledged_by_name: str | None = None ack_reason: str | None = None class AcknowledgeRunRequest(BaseModel): reason: str | None = Field(default=None, max_length=512) class ScheduledJobPatchRequest(BaseModel): enabled: bool | None = None interval_seconds: int | None = Field(default=None, ge=30) cron_expr: str | None = None timezone: str | None = None config: dict | None = None display_name: str | None = None description: str | None = None #: Chain this job behind another. Setting it converts the row to #: ``trigger_type='chained'`` and drops its cron/interval — a job #: cannot have both a predecessor and a clock, because that is two #: independent triggers, i.e. two runs per cycle. chain_after_job_id: UUID | None = None #: Detach from the predecessor. A separate flag, not "send null": #: null is also what a PATCH that simply does not mention the chain #: sends, and conflating absence with removal is how a partial PATCH #: silently unhooks a job from its predecessor. clear_chain: bool = False def _serialize_job( job: ScheduledJob, *, health: str = "ok", chain_after_name: str | None = None, ) -> dict: return ScheduledJobResponse( health=health, weight=hygiene.handler_weight(job.handler_key), chain_after_job_id=job.chain_after_job_id, chain_after_name=chain_after_name, id=job.id, name=job.name, display_name=job.display_name, description=job.description, handler_key=job.handler_key, trigger_type=job.trigger_type, interval_seconds=job.interval_seconds, cron_expr=job.cron_expr, timezone=job.timezone, enabled=job.enabled, config=job.config, last_run_at=job.last_run_at.isoformat() if job.last_run_at else None, last_status=job.last_status, last_duration_ms=job.last_duration_ms, last_error=job.last_error, last_result_summary=job.last_result_summary, run_count=job.run_count, failure_count=job.failure_count, created_at=job.created_at.isoformat(), updated_at=job.updated_at.isoformat(), ).model_dump(mode="json") def _serialize_run( run: ScheduledJobRun, *, ack_by_name: str | None = None, ) -> dict: return ScheduledJobRunResponse( id=run.id, job_id=run.job_id, started_at=run.started_at.isoformat(), finished_at=run.finished_at.isoformat() if run.finished_at else None, duration_ms=run.duration_ms, status=run.status, triggered_by=run.triggered_by, error_message=run.error_message, result_summary=run.result_summary, acknowledged_at=( run.acknowledged_at.isoformat() if run.acknowledged_at else None ), acknowledged_by_id=( str(run.acknowledged_by_id) if run.acknowledged_by_id else None ), acknowledged_by_name=ack_by_name, ack_reason=run.ack_reason, ).model_dump(mode="json") @router.get("", summary="List all scheduled jobs") async def list_jobs( request: Request, page: int = 1, page_size: int = 20, keyword: str | None = None, enabled: bool | None = None, last_status: str | None = None, category: str | None = None, health: str | None = None, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): jobs, total = await sched_service.list_jobs( db, page=page, page_size=page_size, keyword=keyword, enabled=enabled, last_status=last_status, category=category, health=health, ) # Two extra queries for the whole page (integration states + a # grouped run aggregate), never one per row — see # ``hygiene.annotate_health``. health_map = await hygiene.annotate_health(db, jobs) # One extra query for the whole page, never one per row. parent_names = await sched_service.resolve_chain_parent_names(db, jobs) return paginated_response( data=[ _serialize_job( j, health=health_map.get(str(j.id), "ok"), chain_after_name=parent_names.get(str(j.id)), ) for j in jobs ], total=total, page=page, page_size=page_size, request=request, ) @router.get("/handlers", summary="Available handlers registered in-process") async def list_available_handlers( request: Request, _: AuthUser = Depends(require_admin_read()), ): return success_response(data=sched_service.available_handlers(), request=request) @router.get( "/health", summary=( "Scheduler-liveness summary — surfaces APScheduler wedge " "before the worker watchdog catches it" ), ) async def scheduler_health( request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Returns ``{stale_count, late_count, stale_jobs, fastest_*}`` derived from existing ``scheduled_jobs.last_run_at`` rows. Why it exists: ``app/worker.py``'s in-process watchdog catches the 30-minute wedge case and triggers a docker restart, but the admin should be able to SEE the wedge happening BEFORE 30 minutes of orphans pile up. This endpoint backs a banner on ``/admin/scheduled-jobs`` that lights up as soon as the fastest job slips past 5× its interval. Computed inline from the existing table — no new state, no new background job, no chicken-and-egg "the alarm relies on the scheduler that's broken" failure mode. """ summary = await sched_service.scheduler_health_summary(db) return success_response(data=summary, request=request) @router.get( "/analytics", summary=( "Scheduled-job analytics — totals, daily run trend, " "by-category rollup, top failing / slowest jobs" ), ) async def get_scheduled_jobs_analytics( request: Request, window_days: int = 7, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Aggregate signal for the ``/admin/scheduled-jobs`` 分析 tab. ``window_days`` clamps to 1..90 so a runaway client can't ask the DB to scan all of ``scheduled_job_runs`` (the index keeps even 90d cheap, but anything past 90d isn't useful operationally and the cap defends the worker DB). """ # ``is None`` check, not falsy short-circuit: an explicit # ``?window_days=0`` should clamp to 1, not silently resolve to # the default 7 via ``0 or 7``. window_days = max(1, min(int(window_days if window_days is not None else 7), 90)) data = await sched_service.analytics_summary(db, window_days=window_days) return success_response(data=data, request=request) @router.get( "/schedule-forecast", summary=( "Forward-looking + recent-past schedule for the timeline tab — " "per-job upcoming fires, recent runs, and hourly load buckets" ), ) async def get_schedule_forecast( request: Request, past_hours: int = 6, future_hours: int = 18, category: str | None = None, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Backs the 排程时间轴 tab on ``/admin/scheduled-jobs``. Computes future fire instants from the live APScheduler trigger (cron) or interval arithmetic — identical maths to the worker, so the plotted times match what actually fires — plus recent past runs and an hourly upcoming-concurrency series. Read-only, admin-gated, no new state (everything is derived from ``scheduled_jobs`` / ``scheduled_job_runs``). ``past_hours`` clamps to 0..72, ``future_hours`` to 1..168 (7 days) so a runaway client can't ask the worker to enumerate an unbounded cron horizon. """ data = await sched_service.schedule_forecast( db, past_hours=past_hours, future_hours=future_hours, category=category, ) return success_response(data=data, request=request) @router.get( "/capacity", summary=( "Execution-capacity model — demand vs concurrency-slot ceiling, " "by weight class, by hour of day, and by job" ), ) async def get_capacity( request: Request, window_days: int = 7, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Backs the 容量 panel on the 分析 tab. Answers "why did this job start 20 minutes late" with the only two facts that can explain it: how much slot-time the fleet demands per day against the ceiling (``SCHEDULED_JOB_MAX_CONCURRENCY`` / ``…_HEAVY_MAX_CONCURRENCY``), and which hour of the day that demand lands in. Derived entirely from ``scheduled_job_runs`` — observed wall-clock, not a cron-expression estimate. ``window_days`` clamps to 1..30: the ledger keeps only the last 100 runs per job, so a longer window silently mixes a full week of a 15-minute job with a whole quarter of a monthly one and the per-day arithmetic stops meaning anything. """ window_days = max(1, min(int(window_days if window_days is not None else 7), 30)) data = await hygiene.capacity_summary(db, window_days=window_days) return success_response(data=data, request=request) @router.get( "/hygiene", summary=( "Ledger-hygiene census — every job row that looks alive but " "cannot do work, grouped by cause" ), ) async def get_hygiene_census( request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Fleet-wide health census backing the 台账体检 panel. Deliberately scans every row rather than the current page — "is my ledger clean" is a different question from "what is on screen", and a census restricted to one page would answer neither. """ data = await hygiene.health_census(db) return success_response(data=data, request=request) @router.get( "/chains", summary=( "Job chains — every ordering dependency, rooted at its " "clock-triggered job, with the measured cost of each hop" ), ) async def get_chains( request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Backs the 任务链 panel. Answers the question an operator actually has about a chain — "how long does this take end to end, and where does it spend the time" — because that is what decides whether the chain still fits in its window. Also lists ``orphans``: chained rows whose predecessor was deleted (the FK sets the edge NULL rather than cascading the delete), which can never fire again and are invisible in the job list. """ data = await chain.chain_graph(db) return success_response(data=data, request=request) @router.get("/categories", summary="Scheduled-job count by category") async def job_category_counts( request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Returns one row per category (``hr``, ``cmdb``, ``notification``, ``provisioning``, ``other``) with its total count. Used by the admin page's Segmented filter to render pill counters without paginating through every category.""" counts = await sched_service.job_category_counts(db) return success_response(data=counts, request=request) @router.get("/runs", summary="List recent runs across every scheduled job") async def list_all_runs( request: Request, status: str | None = None, category: str | None = None, keyword: str | None = None, triggered_by_prefix: str | None = None, since_hours: int = 24, page: int = 1, page_size: int = 50, include_acknowledged: bool = False, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): """Global execution-log view. Backs ``/admin/scheduled-jobs/runs`` so the admin can skim the last N hours of runs across every scheduled job without drilling into each per-job drawer — the user surfaced that friction on 2026-04-24. Query params mirror ``list_jobs`` where they apply (``category``, ``keyword``) plus run-only fields (``status``, ``triggered_by_prefix``, ``since_hours``, ``include_acknowledged``). ``include_acknowledged`` only affects the failed-status filter — when False (default) the red queue hides admin-acknowledged failures so a known issue doesn't permanently colour the dashboard. Admin-only.""" pairs, total = await sched_service.list_all_runs( db, page=page, page_size=page_size, status=status, category=category, keyword=keyword, triggered_by_prefix=triggered_by_prefix, since_hours=since_hours, include_acknowledged=include_acknowledged, ) # Resolve acknowledger person names in one batched query so the # UI can render "确认人: 张三" without a second round-trip. ack_ids = { str(run.acknowledged_by_id) for run, _ in pairs if run.acknowledged_by_id } ack_names: dict[str, str] = {} if ack_ids: from sqlalchemy import select as _select from app.identity.person_models import IdentityPerson rows = ( await db.execute( _select(IdentityPerson.id, IdentityPerson.preferred_name, IdentityPerson.full_name) .where(IdentityPerson.id.in_(ack_ids)) ) ).all() for row in rows: pid, preferred, full = row ack_names[str(pid)] = preferred or full or str(pid) data = [] for run, job in pairs: ack_id = str(run.acknowledged_by_id) if run.acknowledged_by_id else None item = _serialize_run( run, ack_by_name=ack_names.get(ack_id) if ack_id else None, ) item["job_name"] = job.name item["job_display_name"] = job.display_name item["job_handler_key"] = job.handler_key # Include the job's stored config so admin overrides that # need to identify the underlying resource (e.g. drift-sweep # force-override needs ``integration_id``) can render an # action button without a second round-trip. item["job_config"] = dict(job.config or {}) data.append(item) return paginated_response( data=data, total=total, page=page, page_size=page_size, request=request, ) @router.get("/{job_id}", summary="Get scheduled job detail") async def get_job( job_id: UUID, request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ): job = await sched_service.get_job(db, job_id) return success_response(data=_serialize_job(job), request=request) @router.get("/{job_id}/runs", summary="Recent runs for a scheduled job") async def list_runs( job_id: UUID, request: Request, _: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), limit: int = 50, ): runs = await sched_service.list_runs(db, job_id, limit=limit) ack_ids = {str(r.acknowledged_by_id) for r in runs if r.acknowledged_by_id} ack_names: dict[str, str] = {} if ack_ids: from sqlalchemy import select as _select from app.identity.person_models import IdentityPerson rows = ( await db.execute( _select(IdentityPerson.id, IdentityPerson.preferred_name, IdentityPerson.full_name) .where(IdentityPerson.id.in_(ack_ids)) ) ).all() for row in rows: pid, preferred, full = row ack_names[str(pid)] = preferred or full or str(pid) return success_response( data=[ _serialize_run( r, ack_by_name=( ack_names.get(str(r.acknowledged_by_id)) if r.acknowledged_by_id else None ), ) for r in runs ], request=request, ) @router.patch("/{job_id}", summary="Enable/disable or reschedule a job") async def patch_job( job_id: UUID, body: ScheduledJobPatchRequest, request: Request, current_user: AuthUser = Depends(require_admin()), # Pre-launch P1 fix (Round 45, CR-102 + Round 46 ordering fix): # high-privilege admin cron mutation must require a browser # session. A stolen IT-admin PAT was sufficient before; now # PAT-bearer requests are refused so an attacker can't reschedule # / disable / silently retune critical jobs (cmdb.scope_decommission, # hr cron, etc.). Parameter order: ``require_admin`` MUST run # before ``require_browser_session`` because the latter reads # ``request.state.auth_via`` which is set by ``get_current_user`` # transitively via ``require_admin``. Declaring it first made the # gate fail-close on every JWT call (PR #467 CI catch). _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): job = await sched_service.update_job( db, job_id, enabled=body.enabled, interval_seconds=body.interval_seconds, cron_expr=body.cron_expr, timezone=body.timezone, config=body.config, display_name=body.display_name, description=body.description, chain_after_job_id=body.chain_after_job_id, clear_chain=body.clear_chain, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=current_user.id, actor_type="user", action="scheduled_job.update", resource_type="scheduled_job", resource_id=str(job.id), status="success", # ``mode="json"`` is a no-op on today's all-primitive # ``ScheduledJobPatchRequest``, but ``audit_logs.detail`` is a # JSON column with a bare ``json.dumps`` serializer: the first # Decimal / datetime / UUID field this schema grows would 500 # the PATCH at the audit flush, AFTER the job row was already # updated. Enforced by ``test_audit_detail_model_dump_mode``. detail=body.model_dump(exclude_none=True, mode="json"), ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() parent_names = await sched_service.resolve_chain_parent_names(db, [job]) return success_response( data=_serialize_job( job, chain_after_name=parent_names.get(str(job.id)), ), request=request, ) @router.post( "/runs/{run_id}/acknowledge", summary="Acknowledge a failed scheduled-job run", ) async def acknowledge_run( run_id: UUID, body: AcknowledgeRunRequest, request: Request, current_user: AuthUser = Depends(require_admin()), # Pre-launch P1 fix (Round 45, CR-102 + Round 46 ordering fix): # browser-session gate to prevent a stolen PAT from silently # ACK-ing every failure. Declared AFTER ``require_admin`` so # ``request.state.auth_via`` is populated before the gate reads # it — fail-closed default otherwise rejects every JWT call. _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): """Mark a failed run as acknowledged so the default red queue stops showing it. The run row keeps ``status='failed'`` — we never rewrite history (audit, post-mortems, and ``failure_count`` all key off the truth). The UI hides acknowledged rows behind an ``include_acknowledged=true`` toggle. Notifications still fire on the NEXT failure with the same signature because the dedup gate treats an acked prior failure as "admin already saw a similar one and decided to silence it; a fresh occurrence deserves a fresh push". 409 (``SCHEDULED_RUN_NOT_FAILED``) if the run isn't ``failed``. 200 (no-op, returns the existing row unchanged) if already acked — the original ``acknowledged_at`` / ``acknowledged_by_id`` are kept because the FIRST acknowledgement is the audit-relevant one. """ run = await sched_service.acknowledge_run( db, run_id, actor_person_id=current_user.id, reason=body.reason, ) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=current_user.id, actor_type="user", action="scheduled_job_run.acknowledge", resource_type="scheduled_job_run", resource_id=str(run.id), status="success", detail={"reason": body.reason or None, "job_id": str(run.job_id)}, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() return success_response(data=_serialize_run(run), request=request) class TriggerJobBody(BaseModel): config_override: dict[str, Any] | None = None @router.post("/{job_id}/trigger", summary="Run a scheduled job immediately") async def trigger_job( job_id: UUID, request: Request, body: TriggerJobBody = TriggerJobBody(), _rate: None = Depends(check_sensitive_endpoint_rate_limit), current_user: AuthUser = Depends(require_admin()), # Pre-launch P1 fix (Round 45, CR-102 + Round 46 ordering fix): # browser-session required for immediate-trigger. Declared AFTER # ``require_admin`` so ``request.state.auth_via`` is populated # before the gate reads it. _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ): job = await sched_service.get_job(db, job_id) audit = AuditService(db) await audit.log( trace_id=request.state.trace_id, actor_user_id=current_user.id, actor_type="user", action="scheduled_job.trigger", resource_type="scheduled_job", resource_id=str(job.id), status="success", detail={"handler_key": job.handler_key, "config_override": body.config_override}, ip_address=get_real_ip(request), user_agent=request.headers.get("user-agent"), ) await db.commit() # Execute OUTSIDE the request DB session — execute_job opens its own. result = await sched_service.execute_job( job.id, triggered_by=f"manual:{current_user.id}", config_override=body.config_override, ) return success_response(data=result, request=request)