spec

tit.jobs.spec

Job data model: JobSpec/JobStatus and the small value types they're built from.

Shapes mirror contracts/openapi.yaml (JobKind, JobState, JobProgress, JobError, Artifact, WaitingOn, JobStatus, JobSpec, JobDetail, LockConflict). A few fields exist only on the Python side (persisted to spec.json/status.json for the scheduler's own bookkeeping — locks, cost, pid, create_time, budget_wait) and are dropped by :meth:JobStatus.to_api / :meth:JobSpec.to_api before a response leaves the server, so the wire shape stays exactly what the contract describes plus one harmless additive field (budget_wait, not additionalProperties: false in the schema).

JobKind's frozen v1 contract enum (CONTRACT_JOB_KINDS) includes tools alongside project_init (contracts/CHANGES.md, 2026-08-27 entry, item 1); project_init maps to -m tit.project_init in :mod:tit.jobs.kinds (ra_14 finding #2). report left the enum on 2026-09-28 with the combined preprocessing report it ran (contracts/CHANGES.md).

Cost dataclass

Cost(cpus: float, mem_gb: float)

Estimated resource cost of one job, for scheduler admission (§2.4).

JobSpec dataclass

JobSpec(id: str, kind: str, config: dict[str, Any], subject_ids: list[str], after: list[str] = list(), tags: list[str] = list(), env: dict[str, str] = dict(), locks: list[str] = list(), cost: Cost = (lambda: Cost(cpus=1.0, mem_gb=1.0))(), created_by: str = 'api', created_at: str = utcnow_iso(), group_id: str | None = None, overwrite: bool = False)

The full, persisted job record (jobs/<id>/spec.json).

kind/config/subject_ids/after/tags/overwrite are the fields a caller may set (the wire JobSpec in the contract is exactly this subset, minus id which the server assigns); locks/cost/env/created_by/created_at/group_id are computed or defaulted by :mod:tit.jobs.manager at submission time and never come from the client directly. A group_cap key in an older spec.json (the retired "Subjects in parallel" setting) is ignored: :func:tit.jobs.scheduler.evaluate runs one job per product.

to_api

to_api() -> dict[str, Any]

The contract-shaped JobSpec (request/response body — no server internals).

Source code in tit/jobs/spec.py

def to_api(self) -> dict[str, Any]:
    """The contract-shaped ``JobSpec`` (request/response body — no server internals)."""
    out: dict[str, Any] = {
        "kind": self.kind,
        "config": self.config,
        "subject_ids": list(self.subject_ids),
    }
    if self.after:
        out["after"] = list(self.after)
    if self.tags:
        out["tags"] = list(self.tags)
    if self.overwrite:
        out["overwrite"] = self.overwrite
    return out

PlannedJob dataclass

PlannedJob(label: str, kind: str, config: dict[str, Any], subject_ids: list[str], after_labels: list[str] = list(), tags: list[str] = list(), overwrite: bool = False)

One node of a not-yet-submitted job DAG (tit.jobs.plans.plan_preprocessing).

label is a plan-scoped identifier (e.g. "charm", "dicom") used only to express edges before real job ids exist; :meth:tit.jobs.manager.JobManager.submit_plan resolves after_labels to real job ids in topological order and assigns each a real id.

JobStatus dataclass

JobStatus(id: str, kind: str, state: str, subject_ids: list[str], created_at: str, group_id: str | None = None, started_at: str | None = None, finished_at: str | None = None, progress: JobProgress | None = None, liveness: str | None = None, waiting_on: list[WaitingOn] = list(), exit_code: int | None = None, error: JobError | None = None, artifacts: list[Artifact] = list(), cpu_percent: float | None = None, rss: int | None = None, cpu_percent_peak: float | None = None, cpu_percent_avg: float | None = None, rss_peak: int | None = None, rss_avg: int | None = None, pid: int | None = None, create_time: float | None = None, budget_wait: str | None = None)

to_dict

to_dict() -> dict[str, Any]

Full persisted shape (status.json), internal fields included.

Source code in tit/jobs/spec.py

def to_dict(self) -> dict[str, Any]:
    """Full persisted shape (``status.json``), internal fields included."""
    d = self.to_api()
    d["pid"] = self.pid
    d["create_time"] = self.create_time
    d["budget_wait"] = self.budget_wait
    return d

to_api

to_api(project_dir: str | None = None) -> dict[str, Any]

Contract-shaped JobStatus for API responses.

log_path (ra_13 finding #6) is derived from project_dir on the fly via the same convention :mod:tit.jobs.registry uses for the file itself, rather than persisted -- it's the same deterministic string every time for a given (project_dir, id) pair, so there is nothing to store. None when project_dir isn't supplied (e.g. a bare JobStatus built in a test with no registry behind it).

Source code in tit/jobs/spec.py

def to_api(self, project_dir: str | None = None) -> dict[str, Any]:
    """Contract-shaped ``JobStatus`` for API responses.

    ``log_path`` (ra_13 finding #6) is derived from *project_dir* on the fly via the same
    convention :mod:`tit.jobs.registry` uses for the file itself, rather than persisted --
    it's the same deterministic string every time for a given ``(project_dir, id)`` pair, so
    there is nothing to store. ``None`` when *project_dir* isn't supplied (e.g. a bare
    ``JobStatus`` built in a test with no registry behind it).
    """
    from tit.jobs.registry import stdout_path  # local: registry imports this module

    return {
        "id": self.id,
        "kind": self.kind,
        "state": self.state,
        "subject_ids": list(self.subject_ids),
        "group_id": self.group_id,
        "created_at": self.created_at,
        "started_at": self.started_at,
        "finished_at": self.finished_at,
        "progress": self.progress.to_dict() if self.progress else None,
        "liveness": self.liveness,
        "waiting_on": [w.to_dict() for w in self.waiting_on],
        "exit_code": self.exit_code,
        "error": self.error.to_dict() if self.error else None,
        "artifacts": [a.to_dict() for a in self.artifacts],
        "cpu_percent": self.cpu_percent,
        "rss": self.rss,
        "cpu_percent_peak": self.cpu_percent_peak,
        "cpu_percent_avg": self.cpu_percent_avg,
        "rss_peak": self.rss_peak,
        "rss_avg": self.rss_avg,
        "log_path": stdout_path(project_dir, self.id) if project_dir else None,
    }

new_job_id

new_job_id() -> str

A short, URL-safe, collision-resistant job id.

Source code in tit/jobs/spec.py

def new_job_id() -> str:
    """A short, URL-safe, collision-resistant job id."""
    return uuid.uuid4().hex[:16]

utcnow_iso

utcnow_iso() -> str

Current UTC time as an ISO-8601 string (JobStatus.created_at etc.).

Source code in tit/jobs/spec.py

def utcnow_iso() -> str:
    """Current UTC time as an ISO-8601 string (``JobStatus.created_at`` etc.)."""
    return datetime.now(timezone.utc).isoformat()