find_overdue_jobs

Instructions

Implementation Reference

• src/silentwatch_mcp/server.py:84-101 (registration)

  Tool 'find_overdue_jobs' is registered as an MCP Tool with an optional 'grace_minutes' input schema (default 5).

  Tool(
      name="find_overdue_jobs",
      description=(
          "Returns jobs whose schedule indicates they should have run but "
          "haven't, beyond a grace window."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "grace_minutes": {
                  "type": "integer",
                  "description": "Tolerance to avoid flagging jobs about to run (default 5)",
                  "default": 5,
              }
          },
          "required": [],
      },
  ),

• src/silentwatch_mcp/server.py:160-164 (handler)

  Handler for 'find_overdue_jobs' — lists all jobs from the backend and filters those with is_overdue=True, returning a JobListResponse.

  if name == "find_overdue_jobs":
      jobs = await backend.list_jobs()
      overdue = [j for j in jobs if j.is_overdue]
      response = JobListResponse(jobs=overdue, total=len(overdue))
      return _serialize(response)

• src/silentwatch_mcp/types.py:56-74 (schema)

  CronJob model (used in JobListResponse) includes 'is_overdue' and 'overdue_by_minutes' fields that drive the overdue detection.

  class CronJob(BaseModel):
      """A scheduled cron job with summary state."""
  
      model_config = ConfigDict(frozen=True)
  
      id: str
      name: str
      schedule: str
      """Cron expression or schedule descriptor (e.g., '0 */1 * * *' or 'every 1h')."""
      last_run_at: datetime | None = None
      last_run_status: RunStatus | None = None
      last_success_at: datetime | None = None
      runs_24h: int = 0
      successes_24h: int = 0
      silent_fail_count_24h: int = 0
      is_overdue: bool = False
      overdue_by_minutes: int | None = None
      next_expected_run_at: datetime | None = None
      metadata: dict[str, str] = Field(default_factory=dict)

• src/silentwatch_mcp/types.py:98-102 (schema)

  JobListResponse schema — the response type returned by find_overdue_jobs, containing a list of CronJob objects and a total count.

  class JobListResponse(BaseModel):
      """Response for `list_jobs` and `find_overdue_jobs`."""
  
      jobs: list[CronJob]
      total: int

• src/silentwatch_mcp/analysis.py:123-168 (helper)

  compute_overdue_state — core logic that determines if a job is overdue by comparing its computed next run time (via croniter) against the current time with a grace window.

  def compute_overdue_state(
      schedule: str,
      last_run_at: datetime | None,
      grace_minutes: int = 5,
      now: datetime | None = None,
  ) -> tuple[bool, int | None, datetime | None]:
      """Compute overdue state from a cron schedule + last-run timestamp.
  
      Returns ``(is_overdue, overdue_by_minutes, next_expected_run_at)``.
  
      - **is_overdue:** True if `next_expected_run_at` is more than `grace_minutes` in the past.
      - **overdue_by_minutes:** integer minutes past expected run time, or None if not overdue.
      - **next_expected_run_at:** the cron-schedule's next firing time after last_run_at
        (or after `now - 1 day` if last_run_at is None — i.e., we don't know when it last ran).
  
      Returns ``(False, None, None)`` if the schedule string is empty, "unknown", or
      fails to parse — no schedule means we can't reason about overdue state.
      """
      if not schedule or schedule.lower() == "unknown":
          return (False, None, None)
  
      if now is None:
          now = datetime.now(UTC)
  
      # Use last_run_at as the base for next-run computation. If we don't know when it
      # last ran, fall back to "yesterday" — better than computing from now (which would
      # always say next-run is in the future, masking missing-run scenarios).
      base = last_run_at if last_run_at is not None else now - timedelta(days=1)
  
      try:
          iterator = croniter(schedule, base)
          next_run = iterator.get_next(datetime)
      except (CroniterBadCronError, ValueError) as exc:
          logger.warning("Invalid cron schedule %r: %s", schedule, exc)
          return (False, None, None)
  
      # croniter returns naive datetime by default — coerce to UTC for comparison
      if next_run.tzinfo is None:
          next_run = next_run.replace(tzinfo=UTC)
  
      grace = timedelta(minutes=grace_minutes)
      if next_run + grace < now:
          overdue_by = int((now - next_run).total_seconds() / 60)
          return (True, overdue_by, next_run)
  
      return (False, None, next_run)