Mongo Journal

MongoJournal stores job records and exposes operational views.

Focus areas in this API page:

• status lifecycle and monotonic transitions
• event timeline and derived JobProgressView
• worker capability registration for local placement
• resubmission/recovery support fields (payloads, node maps, policies)

lingo.storage.mongo.journal

MongoDB-based journal for job state and metadata storage.

JobEvent

Bases: BaseModel

Single event in a job timeline.

Source code in lingo/storage/mongo/journal.py

class JobEvent(BaseModel):
    """Single event in a job timeline."""

    at: str
    event: str
    task_name: Optional[str] = None
    attempt: Optional[int] = None
    message: Optional[str] = None
    data: Optional[dict[str, Any]] = None

JobNotFoundError

Bases: LookupError

Raised when a job record cannot be recovered from memory or persistence.

Source code in lingo/storage/mongo/journal.py

class JobNotFoundError(LookupError):
    """Raised when a job record cannot be recovered from memory or persistence."""

JobProgressView

Bases: BaseModel

User-facing progress view composed from job status + timeline events.

Source code in lingo/storage/mongo/journal.py

class JobProgressView(BaseModel):
    """User-facing progress view composed from job status + timeline events."""

    id: str
    status: JobStatusCode
    created_at: str
    updated_at: str
    attempts: int = 0
    started_at: Optional[str] = None
    completed_at: Optional[str] = None
    retries: int = 0
    events: list[JobEvent] = Field(default_factory=list)
    subjobs: list[SubJobProgress] = Field(default_factory=list)

JobRecord

Bases: BaseModel

A complete job record with status and metadata.

Source code in lingo/storage/mongo/journal.py

class JobRecord(BaseModel):
    """A complete job record with status and metadata."""
    id: str
    status: JobStatusCode
    info: Optional[Any] = None
    details: Optional[JobStatusDetails] = None
    result: Optional[Any] = None
    phrase_payload: Optional[dict[str, Any]] = None
    dag_payload: Optional[dict[str, Any]] = None
    node_task_map: Optional[dict[str, str]] = None
    compiled_node_specs: Optional[dict[str, dict[str, Any]]] = None
    policy_payload: Optional[dict[str, Any]] = None
    created_at: str
    updated_at: str
    attempts: int = 0
    events: list[JobEvent] = Field(default_factory=list)

    @property
    def code(self) -> JobStatusCode:
        """Compatibility alias for status used by API endpoints."""
        return self.status

    def progress(self) -> JobProgressView:
        subjobs: dict[str, SubJobProgress] = {}
        started_at: Optional[str] = None
        completed_at: Optional[str] = None
        retries = 0

        for evt in self.events:
            if evt.event in {"job.started", "job.running"} and started_at is None:
                started_at = evt.at
            if evt.event in {"job.completed", "job.failed", "job.cancelled", "job.timed_out"}:
                completed_at = evt.at
            if evt.event == "job.retry_scheduled":
                retries += 1

            if evt.task_name:
                node = subjobs.get(evt.task_name)
                if node is None:
                    node = SubJobProgress(task_name=evt.task_name)
                    subjobs[evt.task_name] = node

                if evt.event == "subjob.started":
                    node.attempts = max(node.attempts, int(evt.attempt or 1))
                    if node.started_at is None:
                        node.started_at = evt.at
                    node.status = JobStatusCode.RUNNING
                elif evt.event == "subjob.completed":
                    node.attempts = max(node.attempts, int(evt.attempt or node.attempts or 1))
                    node.completed_at = evt.at
                    node.status = JobStatusCode.COMPLETED
                elif evt.event == "subjob.failed":
                    node.attempts = max(node.attempts, int(evt.attempt or node.attempts or 1))
                    node.status = JobStatusCode.FAILED
                elif evt.event == "subjob.retry":
                    node.retries += 1

        if started_at is None and self.status in {
            JobStatusCode.RUNNING,
            JobStatusCode.COMPLETED,
            JobStatusCode.FAILED,
            JobStatusCode.CANCELLED,
            JobStatusCode.TIMED_OUT,
        }:
            started_at = self.created_at
        if completed_at is None and self.status.is_terminal():
            completed_at = self.updated_at

        return JobProgressView(
            id=self.id,
            status=self.status,
            created_at=self.created_at,
            updated_at=self.updated_at,
            attempts=self.attempts,
            started_at=started_at,
            completed_at=completed_at,
            retries=retries,
            events=list(self.events),
            subjobs=list(subjobs.values()),
        )

code: JobStatusCode property

Compatibility alias for status used by API endpoints.

JobSortByOptions

Bases: str, Enum

Sorting options for job searches.

Source code in lingo/storage/mongo/journal.py

class JobSortByOptions(str, Enum):
    """Sorting options for job searches."""
    CREATED_AT = "created_at"
    UPDATED_AT = "updated_at"
    STATUS = "status"

JobStatusCode

Bases: str, Enum

Status codes for job execution.

Source code in lingo/storage/mongo/journal.py

class JobStatusCode(str, Enum):
    """Status codes for job execution."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"
    TIMED_OUT = "timed_out"

    def is_terminal(self) -> bool:
        return self in {
            JobStatusCode.COMPLETED,
            JobStatusCode.FAILED,
            JobStatusCode.CANCELLED,
            JobStatusCode.TIMED_OUT,
        }

    def rank(self) -> int:
        if self == JobStatusCode.PENDING:
            return 0
        if self == JobStatusCode.RUNNING:
            return 1
        return 2

JobStatusDetails

Bases: BaseModel

Details about a job's status.

Source code in lingo/storage/mongo/journal.py

class JobStatusDetails(BaseModel):
    """Details about a job's status."""
    message: Optional[str] = None
    error: Optional[str] = None
    heartbeat_at: Optional[str] = None
    progress: Optional[float] = None
    attempt: int = 0

MongoJournal

Bases: Journal

MongoDB-based implementation of the Journal interface.

Source code in lingo/storage/mongo/journal.py

class MongoJournal(Journal):
    """MongoDB-based implementation of the Journal interface."""

    def __init__(
        self,
        margins_type: Optional[type] = None,
        table: str = "jobs",
        connection_string: Optional[str] = None,
        db_name: str = "lingo",
        heartbeat_timeout_seconds: int = 300,
        store_full_dag: bool = False,
    ):
        self.margins_type = margins_type
        self.table = table
        self.connection_string = connection_string or "mongodb://localhost:27017"
        self.db_name = db_name
        self.heartbeat_timeout_seconds = heartbeat_timeout_seconds
        self.store_full_dag = store_full_dag
        self.job_records: dict[str, JobRecord] = {}
        self._status_providers: dict[str, Callable[[], JobStatusCode]] = {}
        self._result_providers: dict[str, Callable[[], Any]] = {}
        self._current_id: Optional[str] = None
        self._mongo_available = False
        self._mongo_checked = False
        self._mongo_client: Optional[AsyncIOMotorClient] = None
        self._mongo_collection = None
        self._mongo_workers_collection = None
        self._workers: dict[str, WorkerCapabilities] = {}

    def _now(self) -> str:
        return datetime.now(timezone.utc).isoformat()

    def _validate_margins(self, value: Any) -> Any:
        if self.margins_type is None or value is None:
            return value

        if isinstance(value, self.margins_type):
            return value

        validator = getattr(self.margins_type, "model_validate", None)
        if validator is not None and isinstance(value, dict):
            return validator(value)

        return value

    @staticmethod
    def _parse_iso(value: str) -> datetime:
        # Support naive or UTC timestamp strings from mixed producers.
        dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
            return dt.replace(tzinfo=timezone.utc)
        return dt

    @staticmethod
    def _status_rank(code: JobStatusCode) -> int:
        return code.rank()

    @staticmethod
    def _is_terminal(code: JobStatusCode) -> bool:
        return code.is_terminal()

    async def _ensure_mongo(self) -> None:
        if self._mongo_checked:
            return
        self._mongo_checked = True
        try:
            client = AsyncIOMotorClient(self.connection_string, serverSelectionTimeoutMS=500)
            await client.admin.command("ping")
            self._mongo_client = client
            self._mongo_collection = client[self.db_name][self.table]
            self._mongo_workers_collection = client[self.db_name][f"{self.table}_workers"]
            self._mongo_available = True
        except Exception:
            self._mongo_available = False

    def _record_to_doc(self, record: JobRecord) -> dict[str, Any]:
        payload = record.model_dump(mode="python")
        payload["status"] = record.status.value
        payload["info"] = serialize_value(record.info)
        payload["details"] = serialize_value(record.details.model_dump(mode="python") if record.details else None)
        payload["result"] = serialize_value(record.result)
        payload["events"] = [event.model_dump(mode="python") for event in record.events]
        return payload

    def _doc_to_record(self, doc: dict[str, Any]) -> JobRecord:
        details_payload = deserialize_value(doc.get("details"))
        details = JobStatusDetails(**details_payload) if isinstance(details_payload, dict) else None
        return JobRecord(
            id=doc["id"],
            status=JobStatusCode(doc["status"]),
            info=self._validate_margins(deserialize_value(doc.get("info"))),
            details=details,
            result=deserialize_value(doc.get("result")),
            phrase_payload=doc.get("phrase_payload"),
            dag_payload=doc.get("dag_payload"),
            node_task_map=doc.get("node_task_map"),
            compiled_node_specs=doc.get("compiled_node_specs"),
            policy_payload=doc.get("policy_payload"),
            created_at=doc.get("created_at", self._now()),
            updated_at=doc.get("updated_at", self._now()),
            attempts=doc.get("attempts", 0),
            events=[JobEvent(**item) for item in doc.get("events", [])],
        )

    async def _persist_record(self, record: JobRecord) -> None:
        await self._ensure_mongo()
        if not self._mongo_available:
            return
        assert self._mongo_collection is not None
        doc = self._record_to_doc(record)
        await self._mongo_collection.replace_one({"id": record.id}, doc, upsert=True)

    async def _load_record(self, job_id: str) -> Optional[JobRecord]:
        await self._ensure_mongo()
        if not self._mongo_available:
            return None
        assert self._mongo_collection is not None
        doc = await self._mongo_collection.find_one({"id": job_id})
        if not doc:
            return None
        return self._doc_to_record(doc)

    async def _search_records(self, query: dict[str, Any], limit: int) -> list[JobRecord]:
        await self._ensure_mongo()
        if not self._mongo_available:
            return []
        assert self._mongo_collection is not None
        rows = self._mongo_collection.find(query).sort("updated_at", -1).limit(limit)
        out: list[JobRecord] = []
        async for doc in rows:
            out.append(self._doc_to_record(doc))
        return out

    async def _persist_worker(self, worker: WorkerCapabilities) -> None:
        await self._ensure_mongo()
        if not self._mongo_available:
            return
        assert self._mongo_workers_collection is not None
        await self._mongo_workers_collection.replace_one(
            {"worker_id": worker.worker_id},
            worker.model_dump(mode="python"),
            upsert=True,
        )

    async def _load_workers(self) -> list[WorkerCapabilities]:
        await self._ensure_mongo()
        if not self._mongo_available:
            return []
        assert self._mongo_workers_collection is not None
        rows = self._mongo_workers_collection.find({})
        workers: list[WorkerCapabilities] = []
        async for doc in rows:
            workers.append(WorkerCapabilities(**doc))
        return workers

    def _schedule_persist(self, record: JobRecord) -> None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            return
        loop.create_task(self._persist_record(record))

    def _schedule_worker_persist(self, worker: WorkerCapabilities) -> None:
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            return
        loop.create_task(self._persist_worker(worker))

    def _new_record(
        self,
        job_id: str,
        status: JobStatusCode,
        info: Optional[Any] = None,
        details: Optional[JobStatusDetails] = None,
        result: Any = None,
        phrase_payload: Optional[dict[str, Any]] = None,
        dag_payload: Optional[dict[str, Any]] = None,
        node_task_map: Optional[dict[str, str]] = None,
        compiled_node_specs: Optional[dict[str, dict[str, Any]]] = None,
        policy_payload: Optional[dict[str, Any]] = None,
    ) -> JobRecord:
        now = self._now()
        return JobRecord(
            id=job_id,
            status=status,
            info=self._validate_margins(info),
            details=details,
            result=result,
            phrase_payload=phrase_payload,
            dag_payload=dag_payload,
            node_task_map=node_task_map,
            compiled_node_specs=compiled_node_specs,
            policy_payload=policy_payload,
            created_at=now,
            updated_at=now,
            events=[],
        )

    def upsert_job(
        self,
        job_id: str,
        *,
        status: JobStatusCode,
        info: Optional[Any] = None,
        details: Optional[JobStatusDetails] = None,
        result: Any = None,
        phrase_payload: Optional[dict[str, Any]] = None,
        dag_payload: Optional[dict[str, Any]] = None,
        node_task_map: Optional[dict[str, str]] = None,
        compiled_node_specs: Optional[dict[str, dict[str, Any]]] = None,
        policy_payload: Optional[dict[str, Any]] = None,
        bump_attempt: bool = False,
        force: bool = False,
    ) -> JobRecord:
        """Insert or update a job record in the in-memory store."""
        existing = self.job_records.get(job_id)
        if existing is None:
            record = self._new_record(
                job_id,
                status=status,
                info=info,
                details=details,
                result=result,
                phrase_payload=phrase_payload,
                dag_payload=dag_payload,
                node_task_map=node_task_map,
                compiled_node_specs=compiled_node_specs,
                policy_payload=policy_payload,
            )
            if bump_attempt:
                record.attempts += 1
            self.job_records[job_id] = record
            self._schedule_persist(record)
            return record

        # Enforce monotonic transitions by default.
        if not force:
            if self._is_terminal(existing.status) and status != existing.status:
                return existing
            if self._status_rank(status) < self._status_rank(existing.status):
                return existing

        existing.status = status
        existing.updated_at = self._now()
        if info is not None:
            existing.info = self._validate_margins(info)
        if details is not None:
            existing.details = details
        if result is not None:
            existing.result = result
        if phrase_payload is not None:
            existing.phrase_payload = phrase_payload
        if dag_payload is not None:
            existing.dag_payload = dag_payload
        if node_task_map is not None:
            existing.node_task_map = node_task_map
        if compiled_node_specs is not None:
            existing.compiled_node_specs = compiled_node_specs
        if policy_payload is not None:
            existing.policy_payload = policy_payload
        if bump_attempt:
            existing.attempts += 1
        self._schedule_persist(existing)
        return existing

    def register_providers(
        self,
        job_id: str,
        *,
        status_provider: Optional[Callable[[], JobStatusCode]] = None,
        result_provider: Optional[Callable[[], Any]] = None,
    ) -> None:
        """Register external providers for dynamic status/result lookup."""
        if status_provider is not None:
            self._status_providers[job_id] = status_provider
        if result_provider is not None:
            self._result_providers[job_id] = result_provider

    def register_worker(
        self,
        *,
        worker_id: str,
        node_id: str,
        tasks: list[str],
        queues: list[str],
        private_queue: Optional[str] = None,
    ) -> WorkerCapabilities:
        worker = WorkerCapabilities(
            worker_id=worker_id,
            node_id=node_id,
            tasks=sorted(set(tasks)),
            queues=sorted(set(queues)),
            private_queue=private_queue,
            updated_at=self._now(),
        )
        self._workers[worker_id] = worker
        self._schedule_worker_persist(worker)
        return worker

    async def list_workers(self, stale_after_seconds: int = 300) -> list[WorkerCapabilities]:
        for worker in await self._load_workers():
            self._workers[worker.worker_id] = worker

        now = self._parse_iso(self._now())
        out: list[WorkerCapabilities] = []
        for worker in self._workers.values():
            age = now - self._parse_iso(worker.updated_at)
            if age.total_seconds() <= stale_after_seconds:
                out.append(worker)
        return out

    async def find_worker_for_tasks(self, task_names: list[str], stale_after_seconds: int = 300) -> Optional[WorkerCapabilities]:
        required = set(task_names)
        if not required:
            return None

        for worker in await self.list_workers(stale_after_seconds=stale_after_seconds):
            if required.issubset(set(worker.tasks)):
                return worker
        return None

    def append_event(
        self,
        job_id: str,
        *,
        event: str,
        task_name: Optional[str] = None,
        attempt: Optional[int] = None,
        message: Optional[str] = None,
        data: Optional[dict[str, Any]] = None,
    ) -> JobEvent:
        """Append a timeline event to a job record."""
        record = self.job_records.get(job_id)
        if record is None:
            record = self._new_record(job_id, status=JobStatusCode.PENDING)
            self.job_records[job_id] = record

        evt = JobEvent(
            at=self._now(),
            event=event,
            task_name=task_name,
            attempt=attempt,
            message=message,
            data=data,
        )
        record.events.append(evt)
        record.updated_at = self._now()
        self._schedule_persist(record)
        return evt

    async def get_events(self, id: str) -> list[JobEvent]:
        """Return timeline events for a job in recorded order."""
        record = await self.recover(id)
        return list(record.events)

    async def get_progress(self, id: str) -> JobProgressView:
        """Build a progress snapshot including subjob timing and retry count."""
        record = await self.recover(id)
        return record.progress()

    async def search_runs(
        self,
        created_after: Optional[str] = None,
        created_before: Optional[str] = None,
        status: Optional[JobStatusCode] = None,
        info_filter: Optional[Callable] = None,
        limit: int = 50,
        sort_by: JobSortByOptions = JobSortByOptions.CREATED_AT,
    ) -> List[JobRecord]:
        """Search for job runs with filters."""
        if self._mongo_available or not self._mongo_checked:
            mongo_query: dict[str, Any] = {}
            if status:
                mongo_query["status"] = status.value
            if created_after or created_before:
                time_filter: dict[str, Any] = {}
                if created_after:
                    time_filter["$gte"] = created_after
                if created_before:
                    time_filter["$lte"] = created_before
                mongo_query["created_at"] = time_filter

            mongo_results = await self._search_records(mongo_query, limit)
            if mongo_results:
                for record in mongo_results:
                    self.job_records[record.id] = record

        after_dt = datetime.fromisoformat(created_after) if created_after else None
        before_dt = datetime.fromisoformat(created_before) if created_before else None

        results: list[JobRecord] = []
        for record in self.job_records.values():
            record.info = self._validate_margins(record.info)
            if status and record.status != status:
                continue

            created_at_dt = datetime.fromisoformat(record.created_at)
            if after_dt and created_at_dt < after_dt:
                continue
            if before_dt and created_at_dt > before_dt:
                continue

            if info_filter and not info_filter(record.info):
                continue
            results.append(record)

        if sort_by == JobSortByOptions.UPDATED_AT:
            results.sort(key=lambda r: r.updated_at, reverse=True)
        elif sort_by == JobSortByOptions.STATUS:
            results.sort(key=lambda r: r.status.value)
        else:
            results.sort(key=lambda r: r.created_at, reverse=True)

        return results[:limit]

    async def recover(self, id: str) -> JobRecord:
        """Recover a job by ID."""
        in_memory = self.job_records.get(id)
        if in_memory is not None:
            in_memory.info = self._validate_margins(in_memory.info)
            return in_memory

        loaded = await self._load_record(id)
        if loaded is not None:
            self.job_records[id] = loaded
            return loaded

        raise JobNotFoundError(f"Job '{id}' not found")

    async def recover_status(self, id: str) -> JobRecord:
        """Recover the status of a job by ID."""
        record = await self.recover(id)
        # Ensure status reflects provider-backed live state (e.g., Celery AsyncResult).
        return await self.get_status(record)

    async def get_status(self, job: Any) -> JobRecord:
        """Get the current status of a job."""
        job_id = getattr(job, "id", None)
        if not job_id:
            return self._new_record("unknown", JobStatusCode.PENDING)

        provider = self._status_providers.get(job_id)
        if provider is not None:
            try:
                provider_status = provider()
                previous = self.job_records.get(job_id)
                previous_heartbeat = None
                if previous and previous.details:
                    previous_heartbeat = previous.details.heartbeat_at

                heartbeat_at = self._now()
                if provider_status == JobStatusCode.RUNNING and previous_heartbeat:
                    heartbeat_at = previous_heartbeat

                details = JobStatusDetails(
                    message="Status from broker",
                    heartbeat_at=heartbeat_at,
                    progress=None,
                )
                record = self.upsert_job(job_id, status=provider_status, details=details)
                # Handle heartbeat timeout for long-running tasks.
                if record.status == JobStatusCode.RUNNING and record.details and record.details.heartbeat_at:
                    elapsed = self._parse_iso(self._now()) - self._parse_iso(record.details.heartbeat_at)
                    if elapsed.total_seconds() > self.heartbeat_timeout_seconds:
                        record = self.upsert_job(
                            job_id,
                            status=JobStatusCode.TIMED_OUT,
                            details=JobStatusDetails(
                                message="Heartbeat timed out",
                                heartbeat_at=record.details.heartbeat_at,
                                error="Timeout",
                                progress=record.details.progress,
                            ),
                            force=True,
                        )
                return record
            except Exception as exc:
                return self.upsert_job(
                    job_id,
                    status=JobStatusCode.FAILED,
                    details=JobStatusDetails(message=str(exc), error=type(exc).__name__),
                    force=True,
                )

        record = self.job_records.get(job_id)
        if record:
            return record

        # Create a basic record if not found
        raw_code = getattr(job, "code", JobStatusCode.PENDING)
        status = raw_code if isinstance(raw_code, JobStatusCode) else JobStatusCode(raw_code)
        return self.upsert_job(job_id, status=status, info=getattr(job, "info", None))

    async def get_result(self, job: Any) -> Any:
        """Get the result of a completed job."""
        job_id = getattr(job, "id", None)
        if not job_id:
            return None

        provider = self._result_providers.get(job_id)
        if provider is not None:
            try:
                value = provider()
                if value is not None:
                    self.upsert_job(job_id, status=JobStatusCode.COMPLETED, result=value)
                return value
            except Exception:
                # Keep fallback behavior for transient broker reads.
                pass

        loaded = await self._load_record(job_id)
        if loaded is not None:
            self.job_records[job_id] = loaded
            if loaded.result is not None:
                return loaded.result

        record = self.job_records.get(job_id)
        if record is not None:
            return record.result

        return getattr(job, "result", lambda block=False: None)(block=False)

    async def restart_recent(self, limit: int = 50) -> List[JobRecord]:
        """Restart recent failed jobs."""
        failed = [
            r for r in self.job_records.values()
            if r.status == JobStatusCode.FAILED
        ]
        failed.sort(key=lambda r: r.updated_at, reverse=True)
        restarted: list[JobRecord] = []
        for record in failed[:limit]:
            record.status = JobStatusCode.PENDING
            record.updated_at = self._now()
            record.attempts += 1
            self._schedule_persist(record)
            restarted.append(record)
        return restarted

    @contextmanager
    def pre_generate_id(self):
        """
        Context manager for pre-generating job IDs.

        Usage:
            with journal.pre_generate_id() as job_id:
                # Use job_id within this context
                ...
        """
        import uuid
        job_id = str(uuid.uuid4())
        self._current_id = job_id
        try:
            yield job_id
        finally:
            self._current_id = None

append_event(job_id: str, *, event: str, task_name: Optional[str] = None, attempt: Optional[int] = None, message: Optional[str] = None, data: Optional[dict[str, Any]] = None) -> JobEvent

Append a timeline event to a job record.

Source code in lingo/storage/mongo/journal.py

def append_event(
    self,
    job_id: str,
    *,
    event: str,
    task_name: Optional[str] = None,
    attempt: Optional[int] = None,
    message: Optional[str] = None,
    data: Optional[dict[str, Any]] = None,
) -> JobEvent:
    """Append a timeline event to a job record."""
    record = self.job_records.get(job_id)
    if record is None:
        record = self._new_record(job_id, status=JobStatusCode.PENDING)
        self.job_records[job_id] = record

    evt = JobEvent(
        at=self._now(),
        event=event,
        task_name=task_name,
        attempt=attempt,
        message=message,
        data=data,
    )
    record.events.append(evt)
    record.updated_at = self._now()
    self._schedule_persist(record)
    return evt

get_events(id: str) -> list[JobEvent] async

Return timeline events for a job in recorded order.

Source code in lingo/storage/mongo/journal.py

async def get_events(self, id: str) -> list[JobEvent]:
    """Return timeline events for a job in recorded order."""
    record = await self.recover(id)
    return list(record.events)

get_progress(id: str) -> JobProgressView async

Build a progress snapshot including subjob timing and retry count.

Source code in lingo/storage/mongo/journal.py

async def get_progress(self, id: str) -> JobProgressView:
    """Build a progress snapshot including subjob timing and retry count."""
    record = await self.recover(id)
    return record.progress()

get_result(job: Any) -> Any async

Get the result of a completed job.

Source code in lingo/storage/mongo/journal.py

async def get_result(self, job: Any) -> Any:
    """Get the result of a completed job."""
    job_id = getattr(job, "id", None)
    if not job_id:
        return None

    provider = self._result_providers.get(job_id)
    if provider is not None:
        try:
            value = provider()
            if value is not None:
                self.upsert_job(job_id, status=JobStatusCode.COMPLETED, result=value)
            return value
        except Exception:
            # Keep fallback behavior for transient broker reads.
            pass

    loaded = await self._load_record(job_id)
    if loaded is not None:
        self.job_records[job_id] = loaded
        if loaded.result is not None:
            return loaded.result

    record = self.job_records.get(job_id)
    if record is not None:
        return record.result

    return getattr(job, "result", lambda block=False: None)(block=False)

get_status(job: Any) -> JobRecord async

Get the current status of a job.

Source code in lingo/storage/mongo/journal.py

async def get_status(self, job: Any) -> JobRecord:
    """Get the current status of a job."""
    job_id = getattr(job, "id", None)
    if not job_id:
        return self._new_record("unknown", JobStatusCode.PENDING)

    provider = self._status_providers.get(job_id)
    if provider is not None:
        try:
            provider_status = provider()
            previous = self.job_records.get(job_id)
            previous_heartbeat = None
            if previous and previous.details:
                previous_heartbeat = previous.details.heartbeat_at

            heartbeat_at = self._now()
            if provider_status == JobStatusCode.RUNNING and previous_heartbeat:
                heartbeat_at = previous_heartbeat

            details = JobStatusDetails(
                message="Status from broker",
                heartbeat_at=heartbeat_at,
                progress=None,
            )
            record = self.upsert_job(job_id, status=provider_status, details=details)
            # Handle heartbeat timeout for long-running tasks.
            if record.status == JobStatusCode.RUNNING and record.details and record.details.heartbeat_at:
                elapsed = self._parse_iso(self._now()) - self._parse_iso(record.details.heartbeat_at)
                if elapsed.total_seconds() > self.heartbeat_timeout_seconds:
                    record = self.upsert_job(
                        job_id,
                        status=JobStatusCode.TIMED_OUT,
                        details=JobStatusDetails(
                            message="Heartbeat timed out",
                            heartbeat_at=record.details.heartbeat_at,
                            error="Timeout",
                            progress=record.details.progress,
                        ),
                        force=True,
                    )
            return record
        except Exception as exc:
            return self.upsert_job(
                job_id,
                status=JobStatusCode.FAILED,
                details=JobStatusDetails(message=str(exc), error=type(exc).__name__),
                force=True,
            )

    record = self.job_records.get(job_id)
    if record:
        return record

    # Create a basic record if not found
    raw_code = getattr(job, "code", JobStatusCode.PENDING)
    status = raw_code if isinstance(raw_code, JobStatusCode) else JobStatusCode(raw_code)
    return self.upsert_job(job_id, status=status, info=getattr(job, "info", None))

pre_generate_id()

Context manager for pre-generating job IDs.

Usage

with journal.pre_generate_id() as job_id: # Use job_id within this context ...

Source code in lingo/storage/mongo/journal.py

@contextmanager
def pre_generate_id(self):
    """
    Context manager for pre-generating job IDs.

    Usage:
        with journal.pre_generate_id() as job_id:
            # Use job_id within this context
            ...
    """
    import uuid
    job_id = str(uuid.uuid4())
    self._current_id = job_id
    try:
        yield job_id
    finally:
        self._current_id = None

recover(id: str) -> JobRecord async

Recover a job by ID.

Source code in lingo/storage/mongo/journal.py

async def recover(self, id: str) -> JobRecord:
    """Recover a job by ID."""
    in_memory = self.job_records.get(id)
    if in_memory is not None:
        in_memory.info = self._validate_margins(in_memory.info)
        return in_memory

    loaded = await self._load_record(id)
    if loaded is not None:
        self.job_records[id] = loaded
        return loaded

    raise JobNotFoundError(f"Job '{id}' not found")

recover_status(id: str) -> JobRecord async

Recover the status of a job by ID.

Source code in lingo/storage/mongo/journal.py

async def recover_status(self, id: str) -> JobRecord:
    """Recover the status of a job by ID."""
    record = await self.recover(id)
    # Ensure status reflects provider-backed live state (e.g., Celery AsyncResult).
    return await self.get_status(record)

register_providers(job_id: str, *, status_provider: Optional[Callable[[], JobStatusCode]] = None, result_provider: Optional[Callable[[], Any]] = None) -> None

Register external providers for dynamic status/result lookup.

Source code in lingo/storage/mongo/journal.py

def register_providers(
    self,
    job_id: str,
    *,
    status_provider: Optional[Callable[[], JobStatusCode]] = None,
    result_provider: Optional[Callable[[], Any]] = None,
) -> None:
    """Register external providers for dynamic status/result lookup."""
    if status_provider is not None:
        self._status_providers[job_id] = status_provider
    if result_provider is not None:
        self._result_providers[job_id] = result_provider

restart_recent(limit: int = 50) -> List[JobRecord] async

Restart recent failed jobs.

Source code in lingo/storage/mongo/journal.py

async def restart_recent(self, limit: int = 50) -> List[JobRecord]:
    """Restart recent failed jobs."""
    failed = [
        r for r in self.job_records.values()
        if r.status == JobStatusCode.FAILED
    ]
    failed.sort(key=lambda r: r.updated_at, reverse=True)
    restarted: list[JobRecord] = []
    for record in failed[:limit]:
        record.status = JobStatusCode.PENDING
        record.updated_at = self._now()
        record.attempts += 1
        self._schedule_persist(record)
        restarted.append(record)
    return restarted

search_runs(created_after: Optional[str] = None, created_before: Optional[str] = None, status: Optional[JobStatusCode] = None, info_filter: Optional[Callable] = None, limit: int = 50, sort_by: JobSortByOptions = JobSortByOptions.CREATED_AT) -> List[JobRecord] async

Search for job runs with filters.

Source code in lingo/storage/mongo/journal.py

async def search_runs(
    self,
    created_after: Optional[str] = None,
    created_before: Optional[str] = None,
    status: Optional[JobStatusCode] = None,
    info_filter: Optional[Callable] = None,
    limit: int = 50,
    sort_by: JobSortByOptions = JobSortByOptions.CREATED_AT,
) -> List[JobRecord]:
    """Search for job runs with filters."""
    if self._mongo_available or not self._mongo_checked:
        mongo_query: dict[str, Any] = {}
        if status:
            mongo_query["status"] = status.value
        if created_after or created_before:
            time_filter: dict[str, Any] = {}
            if created_after:
                time_filter["$gte"] = created_after
            if created_before:
                time_filter["$lte"] = created_before
            mongo_query["created_at"] = time_filter

        mongo_results = await self._search_records(mongo_query, limit)
        if mongo_results:
            for record in mongo_results:
                self.job_records[record.id] = record

    after_dt = datetime.fromisoformat(created_after) if created_after else None
    before_dt = datetime.fromisoformat(created_before) if created_before else None

    results: list[JobRecord] = []
    for record in self.job_records.values():
        record.info = self._validate_margins(record.info)
        if status and record.status != status:
            continue

        created_at_dt = datetime.fromisoformat(record.created_at)
        if after_dt and created_at_dt < after_dt:
            continue
        if before_dt and created_at_dt > before_dt:
            continue

        if info_filter and not info_filter(record.info):
            continue
        results.append(record)

    if sort_by == JobSortByOptions.UPDATED_AT:
        results.sort(key=lambda r: r.updated_at, reverse=True)
    elif sort_by == JobSortByOptions.STATUS:
        results.sort(key=lambda r: r.status.value)
    else:
        results.sort(key=lambda r: r.created_at, reverse=True)

    return results[:limit]

upsert_job(job_id: str, *, status: JobStatusCode, info: Optional[Any] = None, details: Optional[JobStatusDetails] = None, result: Any = None, phrase_payload: Optional[dict[str, Any]] = None, dag_payload: Optional[dict[str, Any]] = None, node_task_map: Optional[dict[str, str]] = None, compiled_node_specs: Optional[dict[str, dict[str, Any]]] = None, policy_payload: Optional[dict[str, Any]] = None, bump_attempt: bool = False, force: bool = False) -> JobRecord

Insert or update a job record in the in-memory store.

Source code in lingo/storage/mongo/journal.py

def upsert_job(
    self,
    job_id: str,
    *,
    status: JobStatusCode,
    info: Optional[Any] = None,
    details: Optional[JobStatusDetails] = None,
    result: Any = None,
    phrase_payload: Optional[dict[str, Any]] = None,
    dag_payload: Optional[dict[str, Any]] = None,
    node_task_map: Optional[dict[str, str]] = None,
    compiled_node_specs: Optional[dict[str, dict[str, Any]]] = None,
    policy_payload: Optional[dict[str, Any]] = None,
    bump_attempt: bool = False,
    force: bool = False,
) -> JobRecord:
    """Insert or update a job record in the in-memory store."""
    existing = self.job_records.get(job_id)
    if existing is None:
        record = self._new_record(
            job_id,
            status=status,
            info=info,
            details=details,
            result=result,
            phrase_payload=phrase_payload,
            dag_payload=dag_payload,
            node_task_map=node_task_map,
            compiled_node_specs=compiled_node_specs,
            policy_payload=policy_payload,
        )
        if bump_attempt:
            record.attempts += 1
        self.job_records[job_id] = record
        self._schedule_persist(record)
        return record

    # Enforce monotonic transitions by default.
    if not force:
        if self._is_terminal(existing.status) and status != existing.status:
            return existing
        if self._status_rank(status) < self._status_rank(existing.status):
            return existing

    existing.status = status
    existing.updated_at = self._now()
    if info is not None:
        existing.info = self._validate_margins(info)
    if details is not None:
        existing.details = details
    if result is not None:
        existing.result = result
    if phrase_payload is not None:
        existing.phrase_payload = phrase_payload
    if dag_payload is not None:
        existing.dag_payload = dag_payload
    if node_task_map is not None:
        existing.node_task_map = node_task_map
    if compiled_node_specs is not None:
        existing.compiled_node_specs = compiled_node_specs
    if policy_payload is not None:
        existing.policy_payload = policy_payload
    if bump_attempt:
        existing.attempts += 1
    self._schedule_persist(existing)
    return existing

SubJobProgress

Bases: BaseModel

Derived progress summary for a task in the job graph.

Source code in lingo/storage/mongo/journal.py

class SubJobProgress(BaseModel):
    """Derived progress summary for a task in the job graph."""

    task_name: str
    started_at: Optional[str] = None
    completed_at: Optional[str] = None
    status: Optional[JobStatusCode] = None
    attempts: int = 0
    retries: int = 0

WorkerCapabilities

Bases: BaseModel

Worker capability registration used by local-placement routing.

Source code in lingo/storage/mongo/journal.py

class WorkerCapabilities(BaseModel):
    """Worker capability registration used by local-placement routing."""

    worker_id: str
    node_id: str
    tasks: list[str] = Field(default_factory=list)
    queues: list[str] = Field(default_factory=list)
    private_queue: Optional[str] = None
    updated_at: str