registry

tit.jobs.registry

On-disk job state: <project>/code/ti-toolbox/jobs/<id>/{spec.json, status.json, events.jsonl, stdout.log}.

Persisted inside the project (not a temp dir) so notebook users and a restarted server both see the same jobs. Writes that must never leave a half-written file behind (spec.json, status.json) go through a temp-file + :func:os.replace.

JobRegistry

JobRegistry(project_dir: str)

Filesystem operations for the job store of one project.

Source code in tit/jobs/registry.py

def __init__(self, project_dir: str) -> None:
    self.project_dir = project_dir
    root = jobs_root(project_dir)
    first_time = not os.path.isdir(root)
    os.makedirs(root, exist_ok=True)
    if first_time:
        ensure_bidsignore(project_dir)

read_spec

read_spec(job_id: str) -> JobSpec | None

The job's spec.json, or None when it is missing, unreadable or malformed.

Source code in tit/jobs/registry.py

def read_spec(self, job_id: str) -> JobSpec | None:
    """The job's ``spec.json``, or ``None`` when it is missing, unreadable or malformed."""
    return self._read(job_id, spec_path, JobSpec.from_dict)

read_status

read_status(job_id: str) -> JobStatus | None

The job's status.json, or None when it is missing, unreadable or malformed.

Source code in tit/jobs/registry.py

def read_status(self, job_id: str) -> JobStatus | None:
    """The job's ``status.json``, or ``None`` when it is missing, unreadable or malformed."""
    return self._read(job_id, status_path, JobStatus.from_dict)

prune

prune(*, keep_count: int = DEFAULT_RETENTION_COUNT, keep_days: float = DEFAULT_RETENTION_DAYS, now: float | None = None) -> list[str]

Drop old, terminal jobs beyond keep_count / keep_days. Returns removed ids.

Never removes a job that is not in a terminal state (queued/running survive regardless of age — a long-lived job should never disappear out from under it).

Source code in tit/jobs/registry.py

def prune(
    self,
    *,
    keep_count: int = DEFAULT_RETENTION_COUNT,
    keep_days: float = DEFAULT_RETENTION_DAYS,
    now: float | None = None,
) -> list[str]:
    """Drop old, terminal jobs beyond *keep_count* / *keep_days*. Returns removed ids.

    Never removes a job that is not in a terminal state (queued/running survive regardless
    of age — a long-lived job should never disappear out from under it).
    """
    from tit.jobs.spec import (
        TERMINAL_STATES,
    )  # local import: avoid a cycle at module load

    finished: list[tuple[str, str, str | None]] = []
    for job_id in self.list_ids():
        status = self.read_status(job_id)
        if status is not None and status.state in TERMINAL_STATES:
            finished.append((job_id, status.created_at, status.finished_at))
    removed = expired(finished, keep_count=keep_count, keep_days=keep_days, now=now)

    for job_id in removed:
        self.delete(job_id)
    if removed:
        logger.info("job registry: pruned %d old job(s)", len(removed))
    return removed

expired

expired(finished: list[tuple[str, str, str | None]], *, keep_count: int = DEFAULT_RETENTION_COUNT, keep_days: float = DEFAULT_RETENTION_DAYS, now: float | None = None) -> list[str]

The retention rule for finished records: which of finished to drop.

finished is (id, created_at, finished_at) per finished record (ISO-8601 times). A record finished more than keep_days ago goes, and so do the oldest beyond the newest keep_count. Shared by the job registry (:meth:JobRegistry.prune) and the agent proposals (:func:tit.server.proposals.prune), so both keep the same history.

Source code in tit/jobs/registry.py

def expired(
    finished: list[tuple[str, str, str | None]],
    *,
    keep_count: int = DEFAULT_RETENTION_COUNT,
    keep_days: float = DEFAULT_RETENTION_DAYS,
    now: float | None = None,
) -> list[str]:
    """The retention rule for finished records: which of *finished* to drop.

    *finished* is ``(id, created_at, finished_at)`` per finished record (ISO-8601 times). A record
    finished more than *keep_days* ago goes, and so do the oldest beyond the newest *keep_count*.
    Shared by the job registry (:meth:`JobRegistry.prune`) and the agent proposals
    (:func:`tit.server.proposals.prune`), so both keep the same history.
    """
    now = now if now is not None else time.time()
    cutoff = now - keep_days * 86400

    def oldest_first(item: tuple[str, str, str | None]) -> tuple[int, float]:
        # By instant, not by string: "+05:00", "Z" and fractional seconds do not sort as
        # text. A created_at that cannot be read sorts first (oldest).
        created = _instant(item[1])
        return (0, 0.0) if created is None else (1, created)

    ordered = sorted(finished, key=oldest_first)
    removed: list[str] = []
    for record_id, created_at, finished_at in ordered:
        ts = _instant(finished_at or created_at)
        if ts is not None and ts < cutoff:
            removed.append(record_id)
    remaining = [item[0] for item in ordered if item[0] not in removed]
    overflow = len(remaining) - keep_count
    if overflow > 0:
        removed.extend(remaining[:overflow])
    return removed

ensure_bidsignore

ensure_bidsignore(project_dir: str) -> None

Make sure project_dir's .bidsignore lists :data:BIDSIGNORE_LINE.

Idempotent: a no-op once the line is present. Creates .bidsignore if it doesn't exist yet; otherwise appends to whatever is already there without touching existing lines (same "users curate this file by hand" convention as :func:tit.pre.utils.ensure_bidsignore, which does the same for CT files -- kept separate here rather than imported so :mod:tit.jobs.registry never has to import :mod:tit.pre).

Source code in tit/jobs/registry.py

def ensure_bidsignore(project_dir: str) -> None:
    """Make sure *project_dir*'s ``.bidsignore`` lists :data:`BIDSIGNORE_LINE`.

    Idempotent: a no-op once the line is present. Creates ``.bidsignore`` if it doesn't exist
    yet; otherwise appends to whatever is already there without touching existing lines (same
    "users curate this file by hand" convention as :func:`tit.pre.utils.ensure_bidsignore`,
    which does the same for CT files -- kept separate here rather than imported so
    :mod:`tit.jobs.registry` never has to import :mod:`tit.pre`).
    """
    target = _storage_path(project_dir, os.path.join(project_dir, ".bidsignore"))
    try:
        with open(target, encoding="utf-8") as fh:
            existing = fh.read().splitlines()
    except OSError:
        existing = []
    if BIDSIGNORE_LINE in existing:
        return
    lines = [*existing, BIDSIGNORE_LINE] if existing else [BIDSIGNORE_LINE]
    with open(target, "w", encoding="utf-8") as fh:
        fh.write("\n".join(lines) + "\n")

job_file_path

job_file_path(project_dir: str, job_id: str, filename: str) -> str

A metadata file whose existing symlinks stay within the project.

Source code in tit/jobs/registry.py

def job_file_path(project_dir: str, job_id: str, filename: str) -> str:
    """A metadata file whose existing symlinks stay within the project."""
    if not filename or filename in (".", "..") or "/" in filename or "\\" in filename:
        raise PermissionError("Job metadata filename must be a single entry")
    return _storage_path(
        project_dir, os.path.join(job_dir(project_dir, job_id), filename)
    )