feat(jobs): expose the timestamp, progress and queue-position properties on Job/AsyncJob

CHANGELOG.md

@@ -12,6 +12,19 @@ the fuller account of each version, including verification notes.
 
 ### Added
 
+- **`Job` and `AsyncJob` expose the rest of the job they already hold.**
+  `created_at`, `started_at`, `completed_at`, `expires_at`, `progress`,
+  `queue_position`, `metrics` and `urls` join `id` / `status` / `outputs` /
+  `error`; previously the only way to read one was the private model attribute.
+  The timestamps are timezone-aware `datetime`s, so `job.completed_at -
+  job.started_at` is the run's duration, and the nullable ones (`started_at`,
+  `completed_at`, `progress`, `queue_position`, `metrics`) are `None` rather
+  than absent. Like the existing properties, these are views onto the state on
+  the handle — nothing re-fetches.
+- `Progress` gained `current_node_class`, the one field of the contract's
+  progress schema the event decoder was dropping. It is appended to the
+  dataclass rather than placed beside `current_node`, so the existing
+  positional order is unchanged.
 - `RouterRunResult.credits_used` — what Comfy Router reported a run cost, lifted from the
   `X-Comfy-Credits-Used` response header onto what `models.run_detailed()` returns. It is a
   price rather than a settled ledger entry, absent means "not reported" and never "free", and

README.md

@@ -304,6 +304,37 @@ live UI feedback, and `wait()`/`result()`/`run()` for the definitive answer.
 output handles regardless of which node produced them (`job.get_outputs(node_id)`
 filters to one node, as in the quickstart above).
 
+### What else a job handle carries
+
+| | |
+|---|---|
+| `job.created_at` | when the server accepted the job |
+| `job.started_at` | when execution began — `None` while queued |
+| `job.completed_at` | when it reached a terminal state — `None` before then |
+| `job.expires_at` | retention deadline: when the job and its outputs stop being readable |
+| `job.progress` | the latest `Progress` snapshot, or `None` |
+| `job.queue_position` | place in the queue, or `None` |
+| `job.metrics` | server timings in ms (`queue_ms`, `execution_ms`), or `None` |
+| `job.urls` | the follow-up links — `self` / `events` / `cancel` |
+
+The timestamps are timezone-aware `datetime`s, so a duration is a
+subtraction:
+
+```python
+job = client.run(wf)
+print(job.completed_at - job.started_at)    # how long the run took
+```
+
+Every one of these is a view onto the state the handle already holds — the
+same as `status` and `outputs`, and for the same reason: nothing here
+re-fetches, so `refresh()` (or `wait()` / `result()`, which call it) is what
+moves them. `AsyncJob` exposes all of them identically, and none of them is
+awaitable — there is nothing to await in a read of local state.
+
+`job.progress` is whatever snapshot came back on the last poll, and not every
+surface fills that in — `None` there means "nothing on this handle", not "no
+progress". For live progress, use `job.events()`.
+
 ## Getting a job's workflow back
 
 The SDK only holds the workflow it submitted for as long as the originating

src/comfy_sdk/events.py

@@ -19,6 +19,7 @@
 from typing import Any
 
 from comfy_low.models import Output as LowOutput
+from comfy_low.models import Progress as LowProgress
 from comfy_low.sse import RawEvent
 
 from .outputs import AsyncOutput, Output
@@ -33,6 +34,11 @@ class Progress:
     current_node: str | None = None
     step: int | None = None
     steps: int | None = None
+    # Appended rather than slotted in beside `current_node` on purpose: the
+    # fields above are a published positional order, and reordering them
+    # would silently re-bind any `Progress(0.42, "KSampler 21/50", ...)`
+    # built positionally by a caller.
+    current_node_class: str | None = None
 
 
 @dataclass
@@ -77,11 +83,32 @@ def _progress(data: dict[str, Any]) -> Progress:
         nodes_done=data.get("nodes_done"),
         nodes_total=data.get("nodes_total"),
         current_node=data.get("current_node"),
+        current_node_class=data.get("current_node_class"),
         step=data.get("step"),
         steps=data.get("steps"),
     )
 
 
+def progress_from_model(model: LowProgress) -> Progress:
+    """Lift the generated progress model into the SDK's :class:`Progress`.
+
+    The same dataclass the ``progress`` frames of a job's event stream carry,
+    so a snapshot read off a job handle and one received live are the same
+    type — which is what makes ``case Progress()`` match either. Field for
+    field with the model; nothing is dropped.
+    """
+    return Progress(
+        value=model.value,
+        message=model.message,
+        nodes_done=model.nodes_done,
+        nodes_total=model.nodes_total,
+        current_node=model.current_node,
+        current_node_class=model.current_node_class,
+        step=model.step,
+        steps=model.steps,
+    )
+
+
 def _preview(data: dict[str, Any]) -> Preview:
     raw = data.get("data_base64", "")
     try:

src/comfy_sdk/jobs.py

@@ -13,17 +13,19 @@
 import time
 from collections.abc import AsyncIterator, Iterator
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any, Literal
 
 import httpx
 
 from comfy_low.errors import ApiError
 from comfy_low.models import Job as LowJob
+from comfy_low.models import JobUrls
 from comfy_low.models import Output as LowOutput
 from comfy_low.transport import AsyncComfyLow, ComfyLow
 
 from . import _core
-from .events import Event, StatusChange, event_from_raw
+from .events import Event, Progress, StatusChange, event_from_raw, progress_from_model
 from .exceptions import JobFailed, to_sdk_error, translating
 from .outputs import AsyncOutput, Output
 
@@ -72,6 +74,78 @@ def outputs(self) -> list[Output]:
     def error(self) -> Any:
         return self._model.error
 
+    # -- lifecycle --------------------------------------------------------
+    #
+    # Views onto whatever state this handle currently holds — reading one
+    # never re-fetches, exactly like the properties above. Call
+    # :meth:`refresh` (or :meth:`wait` / :meth:`result`) first if you need
+    # the server's latest.
+
+    @property
+    def created_at(self) -> datetime:
+        """When the server accepted the job. Timezone-aware; always set."""
+        return self._model.created_at
+
+    @property
+    def started_at(self) -> datetime | None:
+        """When execution began, or ``None`` while the job is still queued."""
+        return self._model.started_at
+
+    @property
+    def completed_at(self) -> datetime | None:
+        """When the job reached a terminal state, or ``None`` before then.
+
+        With :attr:`started_at`, this is how long a job took to run::
+
+            duration = job.completed_at - job.started_at
+        """
+        return self._model.completed_at
+
+    @property
+    def expires_at(self) -> datetime:
+        """Retention deadline — when the job and its outputs stop being
+        readable. A platform property, not an API constant, so read it rather
+        than assuming a window.
+        """
+        return self._model.expires_at
+
+    @property
+    def progress(self) -> Progress | None:
+        """The latest progress snapshot on this handle, or ``None``.
+
+        Same :class:`~comfy_sdk.events.Progress` shape the ``progress`` frames
+        of :meth:`events` carry. Not every surface fills this in on a poll, so
+        ``None`` means "no snapshot on this handle" and never "no progress" —
+        :meth:`events` is the live source, and the one to use for a UI.
+        """
+        model = self._model.progress
+        return None if model is None else progress_from_model(model)
+
+    @property
+    def queue_position(self) -> int | None:
+        """Place in the queue, or ``None`` when the surface does not report
+        one (and once the job is no longer queued).
+        """
+        return self._model.queue_position
+
+    @property
+    def metrics(self) -> dict[str, int | None] | None:
+        """Server-reported timings in milliseconds (e.g. ``queue_ms``,
+        ``execution_ms``), or ``None``. Individual values are nullable too: a
+        metric that is not available yet is ``None`` rather than absent.
+        """
+        return self._model.metrics
+
+    @property
+    def urls(self) -> JobUrls:
+        """The server's follow-up links (``self`` / ``events`` / ``cancel``).
+
+        Follow these rather than building URLs. A link may be host-relative,
+        and one pointing off the deployment's own origin is never sent the
+        API key.
+        """
+        return self._model.urls
+
     def get_outputs(self, node_id: str) -> list[Output]:
         """The outputs produced by one node, in server order.
 
@@ -205,6 +279,49 @@ def outputs(self) -> list[AsyncOutput]:
     def error(self) -> Any:
         return self._model.error
 
+    # -- lifecycle (mirrors :class:`Job`; reads the handle, never re-fetches)
+
+    @property
+    def created_at(self) -> datetime:
+        """:attr:`Job.created_at`."""
+        return self._model.created_at
+
+    @property
+    def started_at(self) -> datetime | None:
+        """:attr:`Job.started_at`."""
+        return self._model.started_at
+
+    @property
+    def completed_at(self) -> datetime | None:
+        """:attr:`Job.completed_at`."""
+        return self._model.completed_at
+
+    @property
+    def expires_at(self) -> datetime:
+        """:attr:`Job.expires_at`."""
+        return self._model.expires_at
+
+    @property
+    def progress(self) -> Progress | None:
+        """:attr:`Job.progress`."""
+        model = self._model.progress
+        return None if model is None else progress_from_model(model)
+
+    @property
+    def queue_position(self) -> int | None:
+        """:attr:`Job.queue_position`."""
+        return self._model.queue_position
+
+    @property
+    def metrics(self) -> dict[str, int | None] | None:
+        """:attr:`Job.metrics`."""
+        return self._model.metrics
+
+    @property
+    def urls(self) -> JobUrls:
+        """:attr:`Job.urls`."""
+        return self._model.urls
+
     def get_outputs(self, node_id: str) -> list[AsyncOutput]:
         """:meth:`Job.get_outputs`, bound to async outputs. Not a coroutine —
         it reads state already on the handle, so no ``await``.

tests/conftest.py

@@ -92,6 +92,19 @@ class ServerState:
     )
     job_workflow_format: str = "api"
     job_workflow_not_found: bool = False
+    # The nullable lifecycle fields of a served job. The defaults are the
+    # shape a queued job has on the wire (nothing started, nothing finished,
+    # no snapshot); a test that needs the populated shape sets them, so both
+    # halves come from a real response rather than a hand-built model.
+    # `job_metrics` is the one that defaults to populated, since a server
+    # reports queue timings from the start.
+    job_started_at: str | None = None
+    job_completed_at: str | None = None
+    job_progress: dict[str, Any] | None = None
+    job_queue_position: int | None = 0
+    job_metrics: dict[str, int | None] | None = field(
+        default_factory=lambda: {"queue_ms": 9000, "execution_ms": None}
+    )
 
     # --- POST /v2/models/{provider}/{model} (the awaited model run) ---
     # The provider's native payload the run resolves to. Deliberately not a
@@ -359,19 +372,21 @@ def _asset_json(asset_id: str, hash_: str, created_new: bool, size: int) -> dict
     }
 
 
-def _job_json(job_id: str, status: str, outputs: list[dict] | None = None) -> dict:
+def _job_json(
+    state: ServerState, job_id: str, status: str, outputs: list[dict] | None = None
+) -> dict:
     return {
         "id": job_id,
         "status": status,
         "created_at": "2026-07-10T18:20:00Z",
-        "started_at": None,
-        "completed_at": None,
+        "started_at": state.job_started_at,
+        "completed_at": state.job_completed_at,
         "expires_at": "2026-07-11T18:20:00Z",
-        "queue_position": 0,
-        "progress": None,
+        "queue_position": state.job_queue_position,
+        "progress": state.job_progress,
         "outputs": outputs or [],
         "error": None,
-        "metrics": {"queue_ms": 9000, "execution_ms": None},
+        "metrics": state.job_metrics,
         "urls": {
             "self": f"/api/v2/jobs/{job_id}",
             "events": f"/api/v2/jobs/{job_id}/events",
@@ -568,7 +583,7 @@ def _serve_job(self, job_id: str) -> None:
             else:
                 status = "running"
                 outputs = []
-            self._json(200, _job_json(job_id, status, outputs))
+            self._json(200, _job_json(state, job_id, status, outputs))
 
         def _serve_job_workflow(self, job_id: str) -> None:
             if state.job_workflow_not_found:
@@ -662,7 +677,7 @@ def do_POST(self) -> None:
                 return
             m = re.match(r"/api/v2/jobs/([^/]+)/cancel$", self.path)
             if m:
-                self._json(200, _job_json(m.group(1), "canceling"))
+                self._json(200, _job_json(state, m.group(1), "canceling"))
                 return
             self._read_body()
             self._err(404, "not_found")
@@ -1008,7 +1023,7 @@ def _post_jobs(self) -> None:
             job_id = f"job_{state.submit_count:02d}"
             if key:
                 state.idempotency[key] = job_id
-            self._json(201, _job_json(job_id, "queued"))
+            self._json(201, _job_json(state, job_id, "queued"))
 
     return Handler
 

tests/test_event_types.py

@@ -7,7 +7,19 @@
 from __future__ import annotations
 
 from comfy_low.sse import RawEvent
-from comfy_sdk.events import Log, Preview, event_from_raw
+from comfy_sdk.events import Log, Preview, Progress, event_from_raw, progress_from_model
+
+#: A progress payload with every optional field of the schema present.
+_PROGRESS_WIRE = {
+    "value": 0.42,
+    "nodes_done": 11,
+    "nodes_total": 31,
+    "current_node": "12",
+    "current_node_class": "KSampler",
+    "step": 21,
+    "steps": 50,
+    "message": "KSampler 21/50",
+}
 
 
 def _binder(model):  # only OutputReady needs a real binder; unused here
@@ -39,3 +51,32 @@ def test_preview_survives_undecodable_base64():
 
 def test_unknown_event_name_is_skipped():
     assert event_from_raw(RawEvent(event="mystery", data={}), _binder) is None
+
+
+def test_progress_event_carries_every_field_of_the_schema():
+    # `current_node_class` was the one field of the progress schema the
+    # decoder dropped. The stub server's frames do not send it, so this is
+    # where it is pinned.
+    ev = event_from_raw(RawEvent(event="progress", data=dict(_PROGRESS_WIRE)), _binder)
+    assert ev == Progress(
+        value=0.42,
+        message="KSampler 21/50",
+        nodes_done=11,
+        nodes_total=31,
+        current_node="12",
+        step=21,
+        steps=50,
+        current_node_class="KSampler",
+    )
+
+
+def test_progress_from_model_matches_the_stream_decoder():
+    # `job.progress` lifts the generated model; `job.events()` decodes the SSE
+    # frame. Both produce the SDK's `Progress`, and the contract serves the
+    # same schema down both paths — so for the same payload they must agree,
+    # field for field. A field added to one lift and not the other fails here.
+    from comfy_low.models import Progress as LowProgress
+
+    from_stream = event_from_raw(RawEvent(event="progress", data=dict(_PROGRESS_WIRE)), _binder)
+    from_model = progress_from_model(LowProgress.model_validate(dict(_PROGRESS_WIRE)))
+    assert from_model == from_stream