settled-reach/tooling/domains/jobs/service.py

"""Logic for the `jobs` domain. Transport-agnostic (D-263).

Nothing here prints, exits, or imports typer.
"""

from __future__ import annotations

import calendar
import json
import time
from pathlib import Path
from typing import Any

from tooling.core import process
from tooling.core.errors import ReachError
from tooling.domains.jobs.schemas import Job, Status

# How often a follow/wait loop re-checks. Chosen for a caller that is a program
# rather than an eye: fast enough that `wait` does not add noticeable latency to
# a short job, slow enough not to spin a core on a long one.
POLL_SECONDS = 0.25


def list_jobs(limit: int = 20) -> list[Job]:
    """Recent jobs, newest first, each reconciled against process liveness."""
    directory = process.jobs_dir()
    files = sorted(directory.glob("*.json"), reverse=True)
    jobs = [_load(path.stem) for path in files[:limit]]
    return [job for job in jobs if job is not None]


def get(job_id: str) -> Job:
    """One job by id, reconciled. Raises if it does not exist."""
    job = _load(job_id)
    if job is None:
        known = [path.stem for path in sorted(process.jobs_dir().glob("*.json"), reverse=True)]
        recent = ", ".join(known[:5]) if known else "(no jobs recorded yet)"
        raise ReachError(
            f"no such job: {job_id}",
            fix=f"reach jobs list — most recent are: {recent}",
            exit_code=2,
        )
    return job


def read_events(job_id: str, offset: int = 0) -> tuple[list[dict[str, Any]], int]:
    """Events from `offset`, plus the new offset.

    A byte offset into an append-only file is the entire reason no daemon is
    needed: a caller can read, drop off, and come back with the offset it kept,
    and nothing has to have been holding a subscription open on its behalf.
    """
    return read_events_from(process.log_path(job_id), offset)


def read_events_from(path: Path, offset: int = 0) -> tuple[list[dict[str, Any]], int]:
    """The same, given a path — so a test can drive it without a real job."""
    if not path.is_file():
        return [], offset

    with path.open("rb") as handle:
        handle.seek(offset)
        raw = handle.read()
        new_offset = handle.tell()

    events: list[dict[str, Any]] = []
    consumed = offset
    for line in raw.split(b"\n"):
        if not line.strip():
            consumed += len(line) + 1
            continue
        try:
            events.append(json.loads(line))
        except json.JSONDecodeError:
            # A partial trailing line: the writer is mid-append. Leave the
            # offset before it so the next read picks it up whole rather than
            # discarding an event because we looked a millisecond too early.
            return events, consumed
        consumed += len(line) + 1
    return events, new_offset


def wait(job_id: str, timeout: float | None = None) -> Job:
    """Block until the job finishes; return it. Never returns while running."""
    deadline = None if timeout is None else time.monotonic() + timeout
    while True:
        job = get(job_id)
        if job.status.finished:
            return job
        if deadline is not None and time.monotonic() >= deadline:
            raise ReachError(
                f"timed out after {timeout:g}s waiting for job {job_id}",
                fix=f"reach jobs status {job_id} — the job is still running, not lost",
                exit_code=2,
            )
        time.sleep(POLL_SECONDS)


def prune(keep: int) -> list[str]:
    """Delete all but the `keep` most recent jobs; return the ids removed.

    Genuinely running jobs survive regardless of age — see `process.prune`,
    which checks the process table rather than the recorded status so that a
    crashed job is prunable and a live generator is not.
    """
    if keep < 0:
        raise ReachError(
            f"cannot keep {keep} jobs",
            fix="pass --keep 0 to remove everything, or a positive number to keep some",
            exit_code=2,
        )
    return process.prune(keep)


def duration(job: Job) -> str:
    """Human-readable elapsed time, or how long it has been running so far."""
    start = _parse(job.started_at)
    end = _parse(job.ended_at) if job.ended_at else time.time()
    if start is None or end is None:
        return "?"
    seconds = max(0.0, end - start)
    if seconds < 60:
        return f"{seconds:.1f}s"
    minutes, rest = divmod(int(seconds), 60)
    return f"{minutes}m{rest:02d}s"


def _load(job_id: str) -> Job | None:
    meta = process.read_meta(job_id)
    if meta is None:
        return None
    return _reconcile(Job.model_validate(meta))


def _reconcile(job: Job) -> Job:
    """Correct a recorded status against reality.

    A job whose file says running but whose pid is gone did not keep running —
    it died without being able to record anything. Reporting it as running
    would be the exit-0 trap somewhere nobody is watching, and would hang any
    caller polling for it to finish.
    """
    if job.status is Status.RUNNING and not process.is_alive(job.pid):
        return job.model_copy(update={"status": Status.DIED})
    return job


def _parse(stamp: str | None) -> float | None:
    """Parse a recorded timestamp as UTC.

    `calendar.timegm`, NOT `time.mktime`: the stamps are written in UTC, and
    mktime would read them as local time. That silently adds the UTC offset to
    every duration — a job started seconds ago reported as having run for over
    two hours. The same mismatch bit the job id in T-1276; both directions of
    this conversion need saying out loud.
    """
    if not stamp:
        return None
    try:
        return calendar.timegm(time.strptime(stamp, "%Y-%m-%dT%H:%M:%S"))
    except (ValueError, TypeError):
        return None