hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'tasks/native-http-sdk' into main

Browse Source 
Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-09-01 10:16:41 -07:00
co-authored by Hanzo Dev
parent 1b0b40fe34 252b839072
commit 7c2cd20df4
9 changed files with 1291 additions and 465 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-tasks/README.md
+90 -15
View File
@@ -1,26 +1,101 @@
# hanzo-tasks
The Hanzo Tasks client: durable workflows over `/v1/tasks`.
Durable activities on the native Hanzo Tasks engine.
Hanzo Cloud embeds the hanzoai/tasks engine and serves it at
`https://api.hanzo.ai/v1/tasks/namespaces/{namespace}/…`, authenticated by your
IAM bearer. This package is that HTTP contract, one method per route. A
workflow's body runs in a worker written with the Go (`tasks/pkg/sdk`) or
TypeScript (`@hanzoai/tasks`) SDK.
The engine is Hanzo's own (`hanzoai/tasks`), reached over JSON at
`api.hanzo.ai/v1/tasks`. It speaks no protobuf and no gRPC, and this package
depends on no workflow runtime — dispatching work and running a worker are both
plain HTTP over `httpx`.
```bash
pip install hanzo-tasks
```
```python
from hanzo_tasks import Client, TasksConfig
## Dispatch
with Client(TasksConfig(token=TOKEN, namespace="acme")) as tasks:
wf = tasks.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"})
tasks.signal(wf.workflow_id, "approve", {"by": "z"})
state = tasks.query(wf.workflow_id, "status")
tasks.cancel(wf.workflow_id, reason="superseded")
```python
from hanzo_tasks import Tasks
async with Tasks.from_env() as tasks:
activity = await tasks.dispatch(
"render",
input={"scene": 3},
task_queue="gpu",
heartbeat_timeout="60s",
)
print(activity.id, activity.run_id, activity.status)
```
Routes covered: start, signal-with-start, list, describe, signal, query,
cancel, terminate, reset, history; schedules create/get/list/delete/trigger.
`from_env()` reads `HANZO_API_KEY`, `HANZO_TASKS_URL` and
`HANZO_TASKS_NAMESPACE`. The org is never named in a request: IAM validates the
bearer token and the edge mints the org from the validated claim, so a caller
cannot reach another tenant's shard by asking to.
## Run a worker
The worker **polls**. It never accepts a push, so it needs no inbound address
and runs behind NAT.
```python
from hanzo_tasks import Tasks, Worker
tasks = Tasks.from_env()
worker = Worker(tasks, task_queue="gpu")
@worker.handler("render")
async def render(activity):
return {"frames": 120}
await worker.run()
```
Return a value to complete the activity; raise to fail it with that cause.
Handlers may be async or plain functions.
## What the engine does, so you don't
These are properties of the engine, not of this client. There is deliberately
no second copy of them here, because a second copy can disagree.
- **A dead worker's work comes back.** The engine reaps expired leases before
every claim, so an activity whose claimant stopped heartbeating returns to
`SCHEDULED` and the next poll picks it up. No client-side timer.
- **Two workers never claim the same activity.** Claims are serialized per
namespace.
- **The lease is the activity's own heartbeat timeout**, else the worker's
`lease_seconds`, else the engine default. `Worker` heartbeats at a third of
the window the server actually granted, so a beat can be missed without
losing the claim.
- **An empty queue is not an error.** It is `204`, and `claim()` returns
`None`.
- **Retries are the engine's.** Pass a `RetryPolicy` to `dispatch()`; the
engine counts attempts and fails the activity when they are exhausted.
## Reading state
```python
page = await tasks.activities(page_size=50)
for activity in page:
print(activity.id, activity.status, activity.attempt)
activity = await tasks.describe(id, run_id)
events, cursor = await tasks.history(id, run_id)
```
## Errors
The engine reports `{"error": ..., "code": <int>}` — `code` is a number here,
not the `status` string the rest of the Hanzo API uses.
| Exception | Meaning |
|---|---|
| `Denied` | 401/403 — no validated principal, or one carrying no org |
| `NotFound` | 404 — no such activity in this namespace |
| `Terminal` | 409 — already completed, failed or canceled |
| `TasksError` | anything else, carrying `.code` |
## License
MIT
pkg/hanzo-tasks/hanzo_tasks/__init__.py
+69 -12
View File
@@ -1,18 +1,75 @@
"""hanzo-tasks — the Hanzo Tasks client.
"""hanzo-tasks — durable activities on the native Hanzo Tasks engine.
Durable workflows in Hanzo Cloud run on hanzoai/tasks, embedded in the cloud
process and served under `/v1/tasks`. From Python you start, signal, query,
cancel, terminate and reset them, and manage schedules; a workflow's body runs
in a worker written with the Go (`tasks/pkg/sdk`) or TypeScript
(`@hanzoai/tasks`) SDK.
The engine is Hanzo's own (`hanzoai/tasks`), reached over JSON at
``api.hanzo.ai/v1/tasks``. It speaks no protobuf and no gRPC, and this package
depends on no workflow runtime — dispatching work and running a worker are
both plain HTTP.
>>> from hanzo_tasks import Client, TasksConfig
>>> with Client(TasksConfig(token=TOKEN, namespace="acme")) as tasks:
... wf = tasks.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"})
... tasks.signal(wf.workflow_id, "approve", {"by": "z"})
Dispatch a unit of work::
from hanzo_tasks import Tasks
async with Tasks.from_env() as tasks:
activity = await tasks.dispatch("render", input={"scene": 3}, task_queue="gpu")
Run a worker that pulls it. The worker polls, so it needs no inbound address
and runs behind NAT::
from hanzo_tasks import Tasks, Worker
tasks = Tasks.from_env()
worker = Worker(tasks, task_queue="gpu")
@worker.handler("render")
async def render(activity):
return {"frames": 120}
await worker.run()
The lease is held for you: the worker heartbeats while your handler runs, and
if it dies the engine returns the activity to the queue for somebody else.
"""
from .client import DEFAULT_HOST, Client, TasksConfig, TasksError, Workflow
from .client import DEFAULT_NAMESPACE, DEFAULT_URL, Tasks
from .errors import Denied, NotFound, TasksError, Terminal
from .types import (
CANCELED,
COMPLETED,
FAILED,
SCHEDULED,
STARTED,
TERMINAL,
Activity,
Event,
Page,
RetryPolicy,
)
from .worker import Worker, default_identity
__version__ = "0.2.0"
__all__ = ["Client", "TasksConfig", "TasksError", "Workflow", "DEFAULT_HOST"]
__all__ = [
# Client
"Tasks",
"DEFAULT_URL",
"DEFAULT_NAMESPACE",
# Worker
"Worker",
"default_identity",
# Types
"Activity",
"Event",
"Page",
"RetryPolicy",
# States
"SCHEDULED",
"STARTED",
"COMPLETED",
"FAILED",
"CANCELED",
"TERMINAL",
# Errors
"TasksError",
"Denied",
"NotFound",
"Terminal",
]
pkg/hanzo-tasks/hanzo_tasks/client.py
+229 -197
View File
@@ -1,237 +1,269 @@
"""The Hanzo Tasks client: `/v1/tasks` over HTTP.
"""The client half: dispatch work, read it back, and settle it.
The durable engine (hanzoai/tasks) is embedded in Hanzo Cloud and answers under
`/v1/tasks/namespaces/{namespace}/...` on the same origin as everything else,
authenticated by the caller's IAM bearer. A workflow is started, signalled,
queried, cancelled, terminated and reset here; its body runs in a worker
written with the Go or TypeScript SDK. This module holds no engine protocol of
its own — it is the HTTP contract, one method per route.
Every call lands on ``/v1/tasks/namespaces/{namespace}/activities`` under the
one Hanzo endpoint, ``api.hanzo.ai``. The org is never named in a request —
IAM validates the bearer token and the edge mints the org from the validated
claim, so a caller cannot reach another tenant's shard by asking to.
"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
import os
from typing import Any
from urllib.parse import quote
import httpx
DEFAULT_HOST = "https://api.hanzo.ai"
from .errors import raise_for
from .types import Activity, Event, Page, RetryPolicy
DEFAULT_URL = "https://api.hanzo.ai"
DEFAULT_NAMESPACE = "default"
@dataclass(frozen=True)
class TasksConfig:
"""Where the engine is and who is calling.
class Tasks:
"""An async client for the Hanzo Tasks activity plane.
`token` is an IAM bearer (a user or service token); the org is read from it
server-side. `namespace` is the engine namespace the caller's workflows live
in — the org's own by convention.
Usable as an async context manager, which closes the transport it owns::
async with Tasks(token=...) as tasks:
await tasks.dispatch("render", input={"scene": 3})
"""
token: str
namespace: str = "default"
host: str = DEFAULT_HOST
timeout: float = 30.0
@dataclass(frozen=True)
class Workflow:
"""One workflow execution as the engine describes it."""
workflow_id: str
run_id: str
status: str = ""
workflow_type: str = ""
raw: dict[str, Any] = field(default_factory=dict)
def __init__(
self,
*,
url: str = DEFAULT_URL,
namespace: str = DEFAULT_NAMESPACE,
token: str | None = None,
identity: str = "",
timeout: float = 30.0,
http: httpx.AsyncClient | None = None,
) -> None:
self.url = url.rstrip("/")
self.namespace = namespace
self.identity = identity
self._token = token
# A caller-supplied transport is borrowed, never closed by us.
self._owned = http is None
self._http = http or httpx.AsyncClient(timeout=timeout)
@classmethod
def of(cls, d: dict[str, Any]) -> "Workflow":
wt = d.get("workflowType")
return cls(
workflow_id=str(d.get("workflowId", "")),
run_id=str(d.get("runId", "")),
status=str(d.get("status", "")),
workflow_type=str(wt.get("name", "") if isinstance(wt, dict) else wt or ""),
raw=d,
def from_env(cls, **kwargs: Any) -> Tasks:
"""Read the endpoint, token and namespace from the environment.
``HANZO_API_KEY`` (or ``HANZO_TASKS_TOKEN``), ``HANZO_TASKS_URL``,
``HANZO_TASKS_NAMESPACE``. Explicit arguments win.
"""
kwargs.setdefault("url", os.environ.get("HANZO_TASKS_URL", DEFAULT_URL))
kwargs.setdefault("namespace", os.environ.get("HANZO_TASKS_NAMESPACE", DEFAULT_NAMESPACE))
kwargs.setdefault(
"token",
os.environ.get("HANZO_API_KEY") or os.environ.get("HANZO_TASKS_TOKEN"),
)
return cls(**kwargs)
class TasksError(RuntimeError):
"""The engine refused or failed a call; `status` is the HTTP status."""
def __init__(self, status: int, message: str) -> None:
super().__init__(f"tasks: HTTP {status}: {message}")
self.status = status
class Client:
"""Synchronous client over `/v1/tasks`. Use as a context manager or call `close()`."""
def __init__(self, config: TasksConfig, transport: httpx.BaseTransport | None = None) -> None:
self.config = config
self._http = httpx.Client(
base_url=config.host.rstrip("/"),
headers={"Authorization": f"Bearer {config.token}", "Content-Type": "application/json"},
timeout=config.timeout,
transport=transport,
)
# ── plumbing ─────────────────────────────────────────────────────────────
def _path(self, *parts: str) -> str:
return "/v1/tasks/namespaces/" + "/".join([self.config.namespace, *parts])
def _call(self, method: str, path: str, *, params: dict[str, Any] | None = None, body: Any = None) -> Any:
r = self._http.request(method, path, params=params or None, json=body)
if r.status_code >= 400:
try:
msg = r.json().get("error") or r.text
except ValueError:
msg = r.text
raise TasksError(r.status_code, str(msg).strip())
if not r.content:
return None
try:
return r.json()
except ValueError:
return r.text
def close(self) -> None:
self._http.close()
def __enter__(self) -> "Client":
async def __aenter__(self) -> Tasks:
return self
def __exit__(self, *_: object) -> None:
self.close()
async def __aexit__(self, *exc: object) -> None:
await self.aclose()
# ── workflows ────────────────────────────────────────────────────────────
async def aclose(self) -> None:
if self._owned:
await self._http.aclose()
def start(
# ── the wire ────────────────────────────────────────────────────────
def _base(self) -> str:
return f"{self.url}/v1/tasks/namespaces/{quote(self.namespace, safe='')}/activities"
def _headers(self) -> dict[str, str]:
if not self._token:
return {}
return {"Authorization": f"Bearer {self._token}"}
async def _call(
self,
workflow_type: str,
method: str,
path: str = "",
*,
task_queue: str,
input: Any = None,
workflow_id: str | None = None,
request_id: str | None = None,
) -> Workflow:
"""Start a workflow. A caller-chosen `workflow_id` makes the start
idempotent; `request_id` deduplicates a retried request."""
body = {
"workflowId": workflow_id or str(uuid.uuid4()),
"workflowType": {"name": workflow_type},
"taskQueue": {"name": task_queue},
"input": input,
"requestId": request_id or str(uuid.uuid4()),
}
return Workflow.of(self._call("POST", self._path("workflows"), body=body) or body)
json: dict[str, Any] | None = None,
params: dict[str, Any] | None = None,
) -> tuple[int, Any]:
response = await self._http.request(
method,
self._base() + path,
json=json,
params=params,
headers=self._headers(),
)
if response.status_code == 204:
return 204, None
body: Any
try:
body = response.json()
except ValueError:
# Refusals on this surface are sometimes plain text, not JSON.
body = response.text
if response.status_code >= 400:
raise_for(response.status_code, body)
return response.status_code, body
def signal_with_start(
# ── dispatch and read ───────────────────────────────────────────────
async def dispatch(
self,
workflow_type: str,
type: str,
*,
task_queue: str,
signal: str,
payload: Any = None,
id: str | None = None,
input: Any = None,
workflow_id: str | None = None,
request_id: str | None = None,
) -> Workflow:
"""Signal a workflow, starting it first if it is not running."""
body = {
"workflowId": workflow_id or str(uuid.uuid4()),
"workflowType": {"name": workflow_type},
"taskQueue": {"name": task_queue},
"input": input,
"signalName": signal,
"signalPayload": payload,
"requestId": request_id or str(uuid.uuid4()),
task_queue: str = "default",
retry: RetryPolicy | None = None,
schedule_to_close_timeout: str = "",
schedule_to_start_timeout: str = "",
start_to_close_timeout: str = "",
heartbeat_timeout: str = "",
request_id: str = "",
) -> Activity:
"""Schedule one activity and return it in SCHEDULED state.
``id`` defaults to a fresh uuid4. ``request_id`` makes the call
idempotent: a retry carrying the same one returns the first activity
unchanged rather than dispatching a second.
``heartbeat_timeout`` doubles as the claim lease — a worker that goes
quiet for longer has its claim reaped and the activity returned to the
queue, so set it to the longest gap between heartbeats you expect.
"""
if id is None:
from uuid import uuid4
id = str(uuid4())
body: dict[str, Any] = {
"activityId": id,
"activityType": {"name": type},
"taskQueue": task_queue,
}
return Workflow.of(self._call("POST", self._path("workflows", "signal-with-start"), body=body) or body)
if input is not None:
body["input"] = input
if retry is not None:
body["retryPolicy"] = retry.wire()
for key, value in (
("scheduleToCloseTimeout", schedule_to_close_timeout),
("scheduleToStartTimeout", schedule_to_start_timeout),
("startToCloseTimeout", start_to_close_timeout),
("heartbeatTimeout", heartbeat_timeout),
("identity", self.identity),
("requestId", request_id),
):
if value:
body[key] = value
_, out = await self._call("POST", json=body)
return Activity.from_wire(out)
def list(self, query: str = "") -> list[Workflow]:
out = self._call("GET", self._path("workflows"), params={"query": query} if query else None)
rows = out.get("workflows", out) if isinstance(out, dict) else out
return [Workflow.of(w) for w in (rows or [])]
async def describe(self, id: str, run_id: str) -> Activity:
"""Read one activity. Raises `NotFound` if this tenant has no such run."""
_, out = await self._call("GET", f"/{_seg(id)}/{_seg(run_id)}")
return Activity.from_wire(out)
def describe(self, workflow_id: str, run_id: str = "") -> Workflow:
out = self._call("GET", self._path("workflows", workflow_id), params={"runId": run_id} if run_id else None)
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
async def activities(self, *, cursor: str = "", page_size: int = 0) -> Page:
"""Read one page of this namespace's activities.
def signal(self, workflow_id: str, name: str, payload: Any = None, run_id: str = "") -> None:
self._call(
"POST",
self._path("workflows", workflow_id, "signal"),
params={"runId": run_id} if run_id else None,
body={"name": name, "payload": payload},
)
Pass the returned `Page.cursor` back as ``cursor`` for the next page;
an empty cursor means the listing is exhausted.
"""
params: dict[str, Any] = {}
if cursor:
params["cursor"] = cursor
if page_size:
params["pageSize"] = page_size
_, out = await self._call("GET", params=params or None)
rows = [Activity.from_wire(row) for row in (out.get("activities") or [])]
return Page(activities=rows, cursor=out.get("nextCursor") or "")
def query(self, workflow_id: str, query_type: str, args: Any = None, run_id: str = "") -> Any:
return self._call(
"POST",
self._path("workflows", workflow_id, "query"),
params={"runId": run_id} if run_id else None,
body={"queryType": query_type, "args": args},
)
def cancel(self, workflow_id: str, reason: str = "", run_id: str = "", identity: str = "") -> Workflow:
out = self._call(
"POST",
self._path("workflows", workflow_id, "cancel"),
params={"runId": run_id} if run_id else None,
body={"reason": reason, "identity": identity},
)
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
def terminate(self, workflow_id: str, reason: str = "", run_id: str = "", identity: str = "") -> Workflow:
out = self._call(
"POST",
self._path("workflows", workflow_id, "terminate"),
params={"runId": run_id} if run_id else None,
body={"reason": reason, "identity": identity},
)
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
def reset(self, workflow_id: str, event_id: int, run_id: str = "", reason: str = "", identity: str = "") -> Workflow:
out = self._call(
"POST",
self._path("workflows", workflow_id, "reset"),
body={"runId": run_id, "eventId": event_id, "reason": reason, "identity": identity},
)
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
def history(self, workflow_id: str, run_id: str = "", after: int = 0, page_size: int = 0, reverse: bool = False) -> Any:
async def history(
self,
id: str,
run_id: str,
*,
after: int = 0,
page_size: int = 0,
reverse: bool = False,
) -> tuple[list[Event], int]:
"""Read an activity's durable history and the cursor that continues it."""
params: dict[str, Any] = {}
if run_id:
params["runId"] = run_id
if after:
params["afterId"] = after
params["after"] = after
if page_size:
params["pageSize"] = page_size
if reverse:
params["reverse"] = "true"
return self._call("GET", self._path("workflows", workflow_id, "history"), params=params)
# ── schedules ────────────────────────────────────────────────────────────
def schedules(self) -> Any:
return self._call("GET", self._path("schedules"))
def create_schedule(self, schedule: dict[str, Any]) -> Any:
"""Create a schedule; `schedule` is the engine's schedule document
(id, spec, action) as the Go and TypeScript SDKs send it."""
return self._call("POST", self._path("schedules"), body=schedule)
def schedule(self, schedule_id: str) -> Any:
return self._call("GET", self._path("schedules", schedule_id))
def delete_schedule(self, schedule_id: str) -> None:
self._call("DELETE", self._path("schedules", schedule_id))
def trigger_schedule(self, schedule_id: str, overlap_policy: str = "", request_id: str | None = None) -> Any:
return self._call(
"POST",
self._path("schedules", schedule_id, "trigger"),
body={"requestId": request_id or str(uuid.uuid4()), "overlapPolicy": overlap_policy},
_, out = await self._call(
"GET", f"/{_seg(id)}/{_seg(run_id)}/history", params=params or None
)
events = [Event.from_wire(row) for row in (out.get("events") or [])]
return events, out.get("nextCursor") or 0
# ── claim and settle ────────────────────────────────────────────────
async def claim(
self,
*,
task_queue: str = "",
identity: str = "",
lease_seconds: int = 0,
) -> Activity | None:
"""Claim the oldest scheduled activity, or None when the queue is empty.
An empty queue is the 204 answer and is not an error — it is the
ordinary result of polling. ``task_queue`` empty claims from any queue.
The engine reaps expired leases before it claims, so a dead worker's
in-flight activity returns to the queue and is picked up here without
any timer on this side.
"""
body: dict[str, Any] = {"taskQueue": task_queue}
who = identity or self.identity
if who:
body["identity"] = who
if lease_seconds:
body["leaseSeconds"] = lease_seconds
status, out = await self._call("POST", "/claim", json=body)
if status == 204:
return None
return Activity.from_wire(out)
async def heartbeat(self, id: str, run_id: str, details: Any = None) -> Activity:
"""Report progress, which also extends the claim's lease."""
return await self._settle(id, run_id, "heartbeat", {"details": details})
async def complete(self, id: str, run_id: str, result: Any = None) -> Activity:
"""Finish the activity successfully, recording ``result``."""
return await self._settle(id, run_id, "complete", {"result": result})
async def fail(self, id: str, run_id: str, cause: str) -> Activity:
"""Finish the activity as failed, recording ``cause``."""
return await self._settle(id, run_id, "fail", {"cause": cause})
async def cancel(self, id: str, run_id: str, reason: str = "") -> Activity:
"""Cancel the activity."""
return await self._settle(id, run_id, "cancel", {"reason": reason})
async def _settle(self, id: str, run_id: str, verb: str, body: dict[str, Any]) -> Activity:
if self.identity:
body.setdefault("identity", self.identity)
_, out = await self._call("POST", f"/{_seg(id)}/{_seg(run_id)}/{verb}", json=body)
return Activity.from_wire(out)
def _seg(value: str) -> str:
"""Percent-encode one path segment.
This keeps a space, ``?`` or ``#`` in an id part of the segment instead of
becoming a query or fragment. It is NOT protection against every input: the
engine matches on Go's decoded ``r.URL.Path``, so an id containing a
literal slash arrives there as two segments and is not addressable at all.
Keep ids slash-free.
"""
return quote(value, safe="")
pkg/hanzo-tasks/hanzo_tasks/errors.py
+56
View File
@@ -0,0 +1,56 @@
"""Refusals the engine states, as exceptions.
The engine answers an error as ``{"error": "...", "code": <int>}`` — ``code``
is a NUMBER here, not the ``status`` string the rest of the Hanzo API uses, so
a generic client that reads ``status`` sees nothing. `raise_for` is the one
place that shape is read.
"""
from __future__ import annotations
class TasksError(Exception):
"""An error the engine reported, carrying its numeric code."""
def __init__(self, message: str, code: int = 0) -> None:
super().__init__(message)
self.message = message
self.code = code
class Denied(TasksError):
"""No validated principal, or one carrying no org (401/403).
The surface fails closed: a token that names no org is refused rather than
served the shared unscoped store.
"""
class NotFound(TasksError):
"""No such activity in this namespace (404)."""
class Terminal(TasksError):
"""The activity already completed, failed or was canceled (409).
Reported rather than swallowed: a second terminal call means two workers
believe they own the same run, which is worth surfacing.
"""
def raise_for(status: int, body: object) -> None:
"""Raise the exception a non-2xx answer stands for."""
message = str(body)
code = status
if isinstance(body, dict):
message = str(body.get("error", body))
raw = body.get("code")
if isinstance(raw, int):
code = raw
if status in (401, 403):
raise Denied(message, code)
if status == 404:
raise NotFound(message, code)
if status == 409:
raise Terminal(message, code)
raise TasksError(message, code)
pkg/hanzo-tasks/hanzo_tasks/types.py
+127
View File
@@ -0,0 +1,127 @@
"""The engine's wire shapes, as Python values.
Field names on the wire are camelCase and an activity is keyed by a pair —
``execution.workflowId`` (the activity id) and ``execution.runId``. These
classes flatten that pair to ``id``/``run_id`` and keep the decoded body in
``raw``, so a field this version does not model is carried rather than lost.
"""
from __future__ import annotations
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Any
SCHEDULED = "ACTIVITY_TASK_STATE_SCHEDULED"
STARTED = "ACTIVITY_TASK_STATE_STARTED"
COMPLETED = "ACTIVITY_TASK_STATE_COMPLETED"
FAILED = "ACTIVITY_TASK_STATE_FAILED"
CANCELED = "ACTIVITY_TASK_STATE_CANCELED"
TERMINAL = frozenset({COMPLETED, FAILED, CANCELED})
@dataclass(frozen=True)
class RetryPolicy:
"""Retry knobs the engine stores against a dispatched activity.
Durations are Go duration strings — ``"5s"``, ``"1m30s"``.
"""
initial_interval: str | None = None
backoff_coefficient: float | None = None
maximum_interval: str | None = None
maximum_attempts: int | None = None
non_retryable_error_types: list[str] | None = None
def wire(self) -> dict[str, Any]:
out: dict[str, Any] = {}
if self.initial_interval is not None:
out["initialInterval"] = self.initial_interval
if self.backoff_coefficient is not None:
out["backoffCoefficient"] = self.backoff_coefficient
if self.maximum_interval is not None:
out["maximumInterval"] = self.maximum_interval
if self.maximum_attempts is not None:
out["maximumAttempts"] = self.maximum_attempts
if self.non_retryable_error_types is not None:
out["nonRetryableErrorTypes"] = list(self.non_retryable_error_types)
return out
@dataclass(frozen=True)
class Activity:
"""One standalone activity: a unit of work the engine tracks durably."""
id: str
run_id: str
type: str
status: str
task_queue: str = ""
attempt: int = 0
maximum_attempts: int = 0
input: Any = None
result: Any = None
failure_cause: str = ""
identity: str = ""
lease_expiry: str = ""
heartbeat_timeout: str = ""
raw: dict[str, Any] = field(default_factory=dict, repr=False)
@property
def terminal(self) -> bool:
return self.status in TERMINAL
@classmethod
def from_wire(cls, body: dict[str, Any]) -> Activity:
execution = body.get("execution") or {}
activity_type = body.get("type") or {}
return cls(
id=execution.get("workflowId", ""),
run_id=execution.get("runId", ""),
type=activity_type.get("name", ""),
status=body.get("status", ""),
task_queue=body.get("taskQueue", ""),
attempt=body.get("attempt", 0),
maximum_attempts=body.get("maximumAttempts", 0),
input=body.get("input"),
result=body.get("result"),
failure_cause=body.get("failureCause", ""),
identity=body.get("identity", ""),
lease_expiry=body.get("leaseExpiry", ""),
heartbeat_timeout=body.get("heartbeatTimeout", ""),
raw=body,
)
@dataclass(frozen=True)
class Event:
"""One durable record in an activity's history."""
id: int
time: str
type: str
attributes: dict[str, Any] = field(default_factory=dict)
@classmethod
def from_wire(cls, body: dict[str, Any]) -> Event:
return cls(
id=body.get("eventId", 0),
time=body.get("eventTime", ""),
type=body.get("eventType", ""),
attributes=body.get("attributes") or {},
)
@dataclass(frozen=True)
class Page:
"""One page of activities, with the cursor that continues it."""
activities: list[Activity]
cursor: str = ""
def __iter__(self) -> Iterator[Activity]:
return iter(self.activities)
def __len__(self) -> int:
return len(self.activities)
pkg/hanzo-tasks/hanzo_tasks/worker.py
+216
View File
@@ -0,0 +1,216 @@
"""The worker half: pull work, hold the lease while it runs, settle it.
The worker POLLS. It never accepts a push, so it needs no inbound address and
runs behind NAT — the reason the claim endpoint exists.
What this does NOT do is keep its own copy of the engine's rules. There is no
client-side reap timer, no retry ladder and no duplicate-claim guard here: the
engine reaps expired leases before every claim, serializes claims per
namespace, and derives the lease from the activity's own heartbeat timeout. A
second opinion on any of that would be a second opinion that can disagree.
"""
from __future__ import annotations
import asyncio
import inspect
import os
import socket
from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any
from .client import Tasks
from .errors import TasksError, Terminal
from .types import Activity
Handler = Callable[[Activity], Any | Awaitable[Any]]
# A fraction of the lease, so a heartbeat is missed twice before the claim is
# reaped. Heartbeating exactly at the lease boundary races the reaper.
HEARTBEAT_FRACTION = 3
def default_identity() -> str:
"""Who this worker is: host and pid, enough to find it in a history."""
return f"{socket.gethostname()}/{os.getpid()}"
class Worker:
"""Polls one task queue and runs the handler registered for each type.
Handlers may be async or plain functions. A plain function is run in a
worker thread rather than inline — a blocking call on the event loop would
stall the heartbeat, the lease would expire mid-run, and the engine would
hand the same activity to somebody else while this one was still working
on it.
"""
def __init__(
self,
tasks: Tasks,
*,
task_queue: str = "default",
identity: str = "",
lease_seconds: int = 60,
poll_interval: float = 1.0,
on_error: Callable[[BaseException, Activity | None], None] | None = None,
) -> None:
self.tasks = tasks
self.task_queue = task_queue
self.identity = identity or tasks.identity or default_identity()
self.lease_seconds = lease_seconds
self.poll_interval = poll_interval
self.on_error = on_error
self._handlers: dict[str, Handler] = {}
# ── registration ────────────────────────────────────────────────────
def handler(self, type: str) -> Callable[[Handler], Handler]:
"""Register the handler for one activity type, as a decorator."""
def register(fn: Handler) -> Handler:
self.register(type, fn)
return fn
return register
def register(self, type: str, fn: Handler) -> None:
self._handlers[type] = fn
# ── the loop ────────────────────────────────────────────────────────
async def run(self, *, stop: asyncio.Event | None = None) -> None:
"""Poll until ``stop`` is set, running whatever is claimed.
Sleeps ``poll_interval`` only when the queue was empty, so a backlog
drains at full speed.
"""
while stop is None or not stop.is_set():
try:
worked = await self.step()
except asyncio.CancelledError:
raise
except BaseException as exc: # a poll failure must not end the loop
self._report(exc, None)
worked = False
if not worked:
if stop is None:
await asyncio.sleep(self.poll_interval)
else:
# Wake immediately when asked to stop, rather than serving
# out the poll interval first.
try:
await asyncio.wait_for(stop.wait(), self.poll_interval)
except TimeoutError:
pass
async def step(self) -> bool:
"""Claim one activity and run it. False when the queue was empty."""
activity = await self.tasks.claim(
task_queue=self.task_queue,
identity=self.identity,
lease_seconds=self.lease_seconds,
)
if activity is None:
return False
await self._execute(activity)
return True
async def _execute(self, activity: Activity) -> None:
handler = self._handlers.get(activity.type)
if handler is None:
# Fail it rather than drop it: an unhandled type left claimed just
# waits out its lease and comes back to the same worker.
await self._settle(
activity, "fail", cause=f"no handler registered for {activity.type!r}"
)
return
beat = asyncio.create_task(self._beat(activity))
try:
result = await self._call(handler, activity)
except asyncio.CancelledError:
beat.cancel()
raise
except BaseException as exc:
self._report(exc, activity)
await self._stop_beat(beat)
await self._settle(activity, "fail", cause=f"{type(exc).__name__}: {exc}")
return
await self._stop_beat(beat)
await self._settle(activity, "complete", result=result)
async def _call(self, handler: Handler, activity: Activity) -> Any:
if inspect.iscoroutinefunction(handler):
return await handler(activity)
# A plain function runs in a thread, never inline: a blocking call on
# the event loop stalls the heartbeat, the lease expires mid-run, and
# the engine hands the same activity to somebody else while this
# worker is still doing it.
outcome = await asyncio.to_thread(handler, activity)
if inspect.isawaitable(outcome):
return await outcome
return outcome
async def _settle(self, activity: Activity, verb: str, **body: Any) -> None:
try:
if verb == "complete":
await self.tasks.complete(activity.id, activity.run_id, body["result"])
else:
await self.tasks.fail(activity.id, activity.run_id, body["cause"])
except Terminal as exc:
# The lease was lost and somebody else settled this run. Report it
# and carry on — the work is done, just not by us.
self._report(exc, activity)
except TasksError as exc:
self._report(exc, activity)
# ── the lease ───────────────────────────────────────────────────────
async def _beat(self, activity: Activity) -> None:
"""Heartbeat until cancelled, holding the claim while work runs."""
interval = self._interval(activity)
while True:
await asyncio.sleep(interval)
try:
await self.tasks.heartbeat(activity.id, activity.run_id)
except asyncio.CancelledError:
raise
except BaseException as exc:
# A missed heartbeat is survivable — the lease has slack — so
# report and keep beating rather than abandoning the run.
self._report(exc, activity)
def _interval(self, activity: Activity) -> float:
"""Beat at a fraction of the lease the SERVER granted.
Read off ``leaseExpiry`` rather than recomputed from a timeout string,
because the engine's own rule — heartbeat timeout, else the requested
lease, else its default — is the one that decides when the reaper
fires, and it already applied it.
"""
window = float(self.lease_seconds or 60)
if activity.lease_expiry:
try:
expiry = datetime.fromisoformat(activity.lease_expiry)
if expiry.tzinfo is None:
expiry = expiry.replace(tzinfo=UTC)
granted = (expiry - datetime.now(UTC)).total_seconds()
if granted > 0:
window = granted
except ValueError:
pass
return max(1.0, window / HEARTBEAT_FRACTION)
@staticmethod
async def _stop_beat(beat: asyncio.Task[None]) -> None:
beat.cancel()
try:
await beat
except (asyncio.CancelledError, Exception):
pass
def _report(self, exc: BaseException, activity: Activity | None) -> None:
if self.on_error is not None:
self.on_error(exc, activity)
pkg/hanzo-tasks/pyproject.toml
+35 -14
View File
@@ -1,42 +1,63 @@
[build-system]
requires = ["setuptools>=61.0.0", "wheel"]
build-backend = "setuptools.build_meta"
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "hanzo-tasks"
version = "0.2.0"
description = "Hanzo Tasks client — start, signal, query and schedule durable workflows over /v1/tasks"
description = "Hanzo Tasks SDK — durable activities on the native Hanzo Tasks engine"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
classifiers = [
"Development Status :: 3 - Alpha",
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Software Development :: Libraries :: Python Modules",
"Typing :: Typed",
]
keywords = ["hanzo", "tasks", "workflow", "durable", "agents"]
keywords = ["hanzo", "tasks", "workflow", "durable", "activities", "agents"]
dependencies = [
"httpx>=0.27",
"httpx>=0.25.0",
]
[project.urls]
Homepage = "https://github.com/hanzoai/python-sdk"
Repository = "https://github.com/hanzoai/python-sdk/tree/main/pkg/hanzo-tasks"
Homepage = "https://hanzo.ai"
Documentation = "https://hanzo.ai/docs/tasks"
Repository = "https://github.com/hanzoai/python-sdk"
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest>=8.0.0",
"pytest-asyncio>=0.24.0",
"ruff>=0.5.0",
"mypy>=1.10.0",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["hanzo_tasks*"]
[tool.hatch.build.targets.wheel]
packages = ["hanzo_tasks"]
[tool.setuptools.package-data]
hanzo_tasks = ["py.typed"]
# Named rather than excluded, so a stray directory beside the package — a local
# virtualenv, a build tree — can never be swept into a release.
[tool.hatch.build.targets.sdist]
include = ["hanzo_tasks", "tests", "README.md", "pyproject.toml"]
[tool.ruff]
target-version = "py312"
line-length = 100
[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "C4", "UP"]
[tool.mypy]
python_version = "3.12"
strict = true
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
pkg/hanzo-tasks/tests/test_tasks.py
+469 -64
View File
@@ -1,79 +1,484 @@
"""Tests against a fake that mirrors the engine's verified wire.
The fake is deliberately faithful to the shapes read out of
`hanzoai/tasks` `pkg/tasks/embed.go` (handleActivities) and `activities.go`:
an empty claim is 204 with no body, a second terminal call is 409, errors are
`{"error", "code"}` with a NUMERIC code, and an activity is keyed by the
`execution.workflowId` / `execution.runId` pair.
"""
from __future__ import annotations
import asyncio
import json
from datetime import UTC
import httpx
import pytest
from hanzo_tasks import Client, TasksConfig, TasksError
from hanzo_tasks import (
Activity,
Denied,
NotFound,
RetryPolicy,
Tasks,
Terminal,
Worker,
)
NS = "default"
BASE = f"/v1/tasks/namespaces/{NS}/activities"
SCHEDULED = "ACTIVITY_TASK_STATE_SCHEDULED"
STARTED = "ACTIVITY_TASK_STATE_STARTED"
COMPLETED = "ACTIVITY_TASK_STATE_COMPLETED"
FAILED = "ACTIVITY_TASK_STATE_FAILED"
def make(handler):
return Client(TasksConfig(token="tok", namespace="acme", host="https://api.example"), transport=httpx.MockTransport(handler))
class Engine:
"""A tiny stand-in for the standalone-activity surface."""
def __init__(self) -> None:
self.rows: dict[tuple[str, str], dict] = {}
self.queue: list[tuple[str, str]] = []
self.calls: list[tuple[str, str]] = []
self.raw: list[tuple[str, str]] = []
self.beats: list[tuple[str, str]] = []
self.headers: list[httpx.Headers] = []
# ── construction ────────────────────────────────────────────────────
def row(self, activity_id: str, run_id: str, **over) -> dict:
row = {
"execution": {"workflowId": activity_id, "runId": run_id},
"type": {"name": over.pop("type", "echo")},
"taskQueue": over.pop("task_queue", "default"),
"status": SCHEDULED,
"attempt": 1,
}
row.update(over)
return row
def schedule(self, activity_id: str, run_id: str = "run-1", **over) -> dict:
row = self.row(activity_id, run_id, **over)
self.rows[(activity_id, run_id)] = row
self.queue.append((activity_id, run_id))
return row
# ── transport ───────────────────────────────────────────────────────
def handler(self, request: httpx.Request) -> httpx.Response:
self.headers.append(request.headers)
path = request.url.path
method = request.method
# Both forms: the engine splits the DECODED path (embed.go:612), while
# the raw one is what actually went on the wire.
self.calls.append((method, path))
self.raw.append((method, request.url.raw_path.decode()))
if not path.startswith(BASE):
return self.err(404, "not found")
rest = path[len(BASE) :].strip("/")
parts = rest.split("/") if rest else []
body = json.loads(request.content) if request.content else {}
if not parts and method == "POST":
return self.start(body)
if not parts and method == "GET":
return self.list(request)
if parts == ["claim"] and method == "POST":
return self.claim(body)
if len(parts) == 2 and method == "GET":
return self.describe(parts[0], parts[1])
if len(parts) == 3 and parts[2] == "history" and method == "GET":
return self.history(parts[0], parts[1])
if len(parts) == 3 and method == "POST":
return self.settle(parts[0], parts[1], parts[2], body)
return self.err(404, "not found")
# ── operations ──────────────────────────────────────────────────────
def start(self, body: dict) -> httpx.Response:
activity_id = body.get("activityId") or ""
if not activity_id:
return self.err(400, "activityId required")
row = self.schedule(
activity_id,
"run-1",
type=(body.get("activityType") or {}).get("name", ""),
task_queue=body.get("taskQueue", "default"),
input=body.get("input"),
heartbeatTimeout=body.get("heartbeatTimeout", ""),
retryPolicy=body.get("retryPolicy"),
)
return httpx.Response(200, json=row)
def list(self, request: httpx.Request) -> httpx.Response:
rows = list(self.rows.values())
return httpx.Response(200, json={"activities": rows, "nextCursor": ""})
def claim(self, body: dict) -> httpx.Response:
want = body.get("taskQueue") or ""
for key in list(self.queue):
row = self.rows[key]
if want and row["taskQueue"] != want:
continue
self.queue.remove(key)
row["status"] = STARTED
row["identity"] = body.get("identity", "")
# The engine stamps the lease it granted; the worker beats off it.
row["leaseExpiry"] = "2099-01-01T00:00:00+00:00"
return httpx.Response(200, json=row)
return httpx.Response(204)
def describe(self, activity_id: str, run_id: str) -> httpx.Response:
row = self.rows.get((activity_id, run_id))
if row is None:
return self.err(404, "activity not found")
return httpx.Response(200, json=row)
def history(self, activity_id: str, run_id: str) -> httpx.Response:
if (activity_id, run_id) not in self.rows:
return self.err(404, "activity not found")
events = [
{
"eventId": 1,
"eventTime": "2026-01-01T00:00:00Z",
"eventType": "ACTIVITY_TASK_SCHEDULED",
"attributes": {"taskQueue": "gpu"},
}
]
return httpx.Response(200, json={"events": events, "nextCursor": 0})
def settle(self, activity_id: str, run_id: str, verb: str, body: dict) -> httpx.Response:
row = self.rows.get((activity_id, run_id))
if row is None:
return self.err(404, "activity not found")
if row["status"] in (COMPLETED, FAILED):
return self.err(409, f"activity terminal: status={row['status']}")
if verb == "heartbeat":
self.beats.append((activity_id, run_id))
elif verb == "complete":
row["status"] = COMPLETED
row["result"] = body.get("result")
elif verb == "fail":
row["status"] = FAILED
row["failureCause"] = body.get("cause", "")
elif verb == "cancel":
row["status"] = "ACTIVITY_TASK_STATE_CANCELED"
else:
return self.err(404, "not found")
return httpx.Response(200, json=row)
@staticmethod
def err(code: int, message: str) -> httpx.Response:
# code is a NUMBER on this surface, unlike the rest of the API.
return httpx.Response(code, json={"error": message, "code": code})
def test_start_posts_the_engine_shape():
seen = {}
def handler(req: httpx.Request) -> httpx.Response:
seen["method"], seen["path"], seen["auth"] = req.method, req.url.path, req.headers.get("authorization")
seen["body"] = json.loads(req.content)
return httpx.Response(200, json={"workflowId": seen["body"]["workflowId"], "runId": "r-1", "status": "Running"})
with make(handler) as c:
wf = c.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"}, workflow_id="wf-1")
assert seen["method"] == "POST" and seen["path"] == "/v1/tasks/namespaces/acme/workflows"
assert seen["auth"] == "Bearer tok"
assert seen["body"]["workflowType"] == {"name": "agent.review"} and seen["body"]["taskQueue"] == {"name": "agents"}
assert seen["body"]["input"] == {"repo": "acme/widgets"} and seen["body"]["requestId"]
assert wf.workflow_id == "wf-1" and wf.run_id == "r-1" and wf.status == "Running"
@pytest.fixture
def engine() -> Engine:
return Engine()
def test_signal_query_cancel_reset_routes():
calls = []
def handler(req: httpx.Request) -> httpx.Response:
calls.append((req.method, req.url.path, req.url.query.decode(), json.loads(req.content) if req.content else None))
return httpx.Response(200, json={"workflowId": "wf-1", "runId": "r-1", "answer": 42})
with make(handler) as c:
c.signal("wf-1", "approve", {"by": "z"}, run_id="r-1")
out = c.query("wf-1", "status")
c.cancel("wf-1", reason="done")
c.terminate("wf-1", reason="hard")
c.reset("wf-1", event_id=7, run_id="r-1", reason="replay")
c.signal_with_start("agent.review", task_queue="agents", signal="kick", payload=1, workflow_id="wf-2")
paths = [(m, p) for m, p, _, _ in calls]
assert paths == [
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/signal"),
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/query"),
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/cancel"),
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/terminate"),
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/reset"),
("POST", "/v1/tasks/namespaces/acme/workflows/signal-with-start"),
]
assert calls[0][2] == "runId=r-1" and calls[0][3] == {"name": "approve", "payload": {"by": "z"}}
assert calls[1][3] == {"queryType": "status", "args": None} and out["answer"] == 42
assert calls[4][3] == {"runId": "r-1", "eventId": 7, "reason": "replay", "identity": ""}
assert calls[5][3]["signalName"] == "kick" and calls[5][3]["signalPayload"] == 1
@pytest.fixture
def tasks(engine: Engine) -> Tasks:
transport = httpx.MockTransport(engine.handler)
return Tasks(
url="https://api.hanzo.ai",
namespace=NS,
token="hk-test",
http=httpx.AsyncClient(transport=transport),
)
def test_schedules_and_errors():
calls = []
# ── the package no longer carries the trap ──────────────────────────────
def handler(req: httpx.Request) -> httpx.Response:
calls.append((req.method, req.url.path))
if req.url.path.endswith("/schedules/nope"):
return httpx.Response(404, json={"error": "schedule not found"})
return httpx.Response(200, json={"ok": True})
with make(handler) as c:
c.create_schedule({"id": "nightly", "spec": {"cron": "0 2 * * *"}, "action": {"workflowType": "agent.digest"}})
c.trigger_schedule("nightly")
c.delete_schedule("nightly")
with pytest.raises(TasksError) as e:
c.schedule("nope")
assert e.value.status == 404 and "schedule not found" in str(e.value)
assert calls[:3] == [
("POST", "/v1/tasks/namespaces/acme/schedules"),
("POST", "/v1/tasks/namespaces/acme/schedules/nightly/trigger"),
("DELETE", "/v1/tasks/namespaces/acme/schedules/nightly"),
]
def test_no_temporal_anywhere() -> None:
"""0.1.0 shipped a Temporal gRPC client aimed at a port we do not serve."""
import pathlib
import hanzo_tasks
root = pathlib.Path(hanzo_tasks.__file__).parent
for path in root.rglob("*.py"):
source = path.read_text()
assert "temporalio" not in source, path
assert "7233" not in source, path
def test_version_is_the_superseding_one() -> None:
import hanzo_tasks
assert hanzo_tasks.__version__ == "0.2.0"
# ── client ──────────────────────────────────────────────────────────────
async def test_dispatch_sends_the_engines_shape(tasks: Tasks, engine: Engine) -> None:
activity = await tasks.dispatch(
"render",
id="job-1",
input={"scene": 3},
task_queue="gpu",
heartbeat_timeout="60s",
retry=RetryPolicy(maximum_attempts=5, backoff_coefficient=2.0),
)
assert ("POST", BASE) in engine.calls
assert activity.id == "job-1"
assert activity.type == "render"
assert activity.task_queue == "gpu"
assert activity.status == SCHEDULED
assert not activity.terminal
stored = engine.rows[("job-1", "run-1")]
assert stored["input"] == {"scene": 3}
assert stored["heartbeatTimeout"] == "60s"
assert stored["retryPolicy"] == {"maximumAttempts": 5, "backoffCoefficient": 2.0}
async def test_dispatch_mints_an_id_when_none_is_given(tasks: Tasks) -> None:
activity = await tasks.dispatch("render")
assert activity.id
async def test_bearer_token_is_sent(tasks: Tasks, engine: Engine) -> None:
await tasks.dispatch("render", id="job-1")
assert engine.headers[-1]["authorization"] == "Bearer hk-test"
async def test_empty_queue_is_none_not_an_error(tasks: Tasks) -> None:
"""204 is the ordinary result of polling, not a failure."""
assert await tasks.claim(task_queue="gpu") is None
async def test_claim_returns_the_started_activity(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1", task_queue="gpu")
claimed = await tasks.claim(task_queue="gpu", identity="spark")
assert claimed is not None
assert claimed.id == "job-1"
assert claimed.status == STARTED
assert claimed.identity == "spark"
async def test_claim_filters_by_queue(tasks: Tasks, engine: Engine) -> None:
engine.schedule("cpu-job", task_queue="cpu")
assert await tasks.claim(task_queue="gpu") is None
assert (await tasks.claim(task_queue="cpu")) is not None
async def test_describe_and_list(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1")
activity = await tasks.describe("job-1", "run-1")
assert activity.id == "job-1"
page = await tasks.activities(page_size=10)
assert len(page) == 1
assert [a.id for a in page] == ["job-1"]
async def test_history(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1")
events, cursor = await tasks.history("job-1", "run-1")
assert [e.type for e in events] == ["ACTIVITY_TASK_SCHEDULED"]
assert events[0].id == 1
assert cursor == 0
async def test_missing_activity_is_not_found(tasks: Tasks) -> None:
with pytest.raises(NotFound) as caught:
await tasks.describe("nope", "run-1")
assert caught.value.code == 404
async def test_second_settle_is_terminal(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1")
await tasks.complete("job-1", "run-1", {"ok": True})
with pytest.raises(Terminal) as caught:
await tasks.complete("job-1", "run-1", {"ok": True})
assert caught.value.code == 409
async def test_refusal_is_denied(engine: Engine) -> None:
def refuse(request: httpx.Request) -> httpx.Response:
return httpx.Response(403, json={"error": "identity required", "code": 403})
tasks = Tasks(http=httpx.AsyncClient(transport=httpx.MockTransport(refuse)))
with pytest.raises(Denied):
await tasks.dispatch("render")
async def test_path_segments_are_percent_encoded(tasks: Tasks, engine: Engine) -> None:
"""Ids go on the wire encoded, so they cannot corrupt the URL.
This is hygiene, not protection against every input: the engine splits
Go's DECODED r.URL.Path (embed.go:612), so an id containing a literal
slash arrives as two segments and is simply not addressable there. What
encoding does buy is that a space, '?' or '#' in an id stays part of the
segment instead of becoming a query or fragment.
"""
engine.rows[("job 1?x", "run-1")] = engine.row("job 1?x", "run-1")
activity = await tasks.describe("job 1?x", "run-1")
assert activity.id == "job 1?x"
# Encoded on the wire, decoded back to one segment by the reader.
assert ("GET", f"{BASE}/job%201%3Fx/run-1") in engine.raw
# ── worker ──────────────────────────────────────────────────────────────
async def test_worker_runs_the_handler_and_completes(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1", type="render", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
seen: list[Activity] = []
@worker.handler("render")
async def render(activity: Activity) -> dict:
seen.append(activity)
return {"frames": 120}
assert await worker.step() is True
assert [a.id for a in seen] == ["job-1"]
assert engine.rows[("job-1", "run-1")]["status"] == COMPLETED
assert engine.rows[("job-1", "run-1")]["result"] == {"frames": 120}
async def test_worker_reports_idle(tasks: Tasks) -> None:
worker = Worker(tasks, task_queue="gpu")
assert await worker.step() is False
async def test_sync_handlers_work(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1", type="render", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
@worker.handler("render")
def render(activity: Activity) -> str:
return "done"
await worker.step()
assert engine.rows[("job-1", "run-1")]["result"] == "done"
async def test_handler_exception_fails_the_activity(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1", type="render", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
@worker.handler("render")
async def render(activity: Activity) -> None:
raise RuntimeError("gpu fell over")
await worker.step()
row = engine.rows[("job-1", "run-1")]
assert row["status"] == FAILED
assert "gpu fell over" in row["failureCause"]
async def test_unknown_type_is_failed_not_dropped(tasks: Tasks, engine: Engine) -> None:
"""Left claimed it would just wait out its lease and come back here."""
engine.schedule("job-1", type="unregistered", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
await worker.step()
row = engine.rows[("job-1", "run-1")]
assert row["status"] == FAILED
assert "no handler registered" in row["failureCause"]
async def test_worker_holds_the_lease_while_the_handler_runs(
tasks: Tasks, engine: Engine, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The heartbeat is what keeps the engine from reaping live work."""
engine.schedule("job-1", type="slow", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
monkeypatch.setattr(worker, "_interval", lambda activity: 0.02)
@worker.handler("slow")
async def slow(activity: Activity) -> str:
await asyncio.sleep(0.25)
return "done"
await worker.step()
assert len(engine.beats) >= 2, engine.beats
assert engine.rows[("job-1", "run-1")]["status"] == COMPLETED
async def test_heartbeat_stops_once_the_handler_returns(
tasks: Tasks, engine: Engine, monkeypatch: pytest.MonkeyPatch
) -> None:
engine.schedule("job-1", type="quick", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu")
monkeypatch.setattr(worker, "_interval", lambda activity: 0.01)
@worker.handler("quick")
async def quick(activity: Activity) -> str:
return "done"
await worker.step()
settled = len(engine.beats)
await asyncio.sleep(0.1)
assert len(engine.beats) == settled
def test_interval_is_a_third_of_the_granted_lease(tasks: Tasks) -> None:
"""Policy is read off the lease the SERVER granted, not recomputed."""
from datetime import datetime, timedelta
worker = Worker(tasks, lease_seconds=60)
expiry = datetime.now(UTC) + timedelta(seconds=30)
activity = Activity.from_wire(
{
"execution": {"workflowId": "a", "runId": "r"},
"type": {"name": "t"},
"status": STARTED,
"leaseExpiry": expiry.isoformat(),
}
)
assert 8.0 < worker._interval(activity) < 11.0
# No lease stamped: fall back to what this worker asked for.
bare = Activity.from_wire(
{"execution": {"workflowId": "a", "runId": "r"}, "type": {"name": "t"}, "status": STARTED}
)
assert worker._interval(bare) == 20.0
async def test_run_loop_stops_when_asked(tasks: Tasks, engine: Engine) -> None:
engine.schedule("job-1", type="render", task_queue="gpu")
worker = Worker(tasks, task_queue="gpu", poll_interval=0.01)
@worker.handler("render")
async def render(activity: Activity) -> str:
return "done"
stop = asyncio.Event()
task = asyncio.create_task(worker.run(stop=stop))
await asyncio.sleep(0.1)
stop.set()
await asyncio.wait_for(task, timeout=2)
assert engine.rows[("job-1", "run-1")]["status"] == COMPLETED
async def test_poll_failure_does_not_end_the_loop(engine: Engine) -> None:
calls = {"n": 0}
def flaky(request: httpx.Request) -> httpx.Response:
calls["n"] += 1
if calls["n"] == 1:
raise httpx.ConnectError("network blip")
return httpx.Response(204)
tasks = Tasks(http=httpx.AsyncClient(transport=httpx.MockTransport(flaky)))
seen: list[BaseException] = []
worker = Worker(tasks, poll_interval=0.01, on_error=lambda exc, activity: seen.append(exc))
stop = asyncio.Event()
task = asyncio.create_task(worker.run(stop=stop))
await asyncio.sleep(0.1)
stop.set()
await asyncio.wait_for(task, timeout=2)
assert len(seen) == 1
assert calls["n"] > 1, "the loop kept polling after the blip"
pkg/hanzo-tasks/uv.lock
Generated
-163
View File
@@ -1,163 +0,0 @@
version = 1
revision = 3
requires-python = ">=3.12"
[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]
[[package]]
name = "hanzo-tasks"
version = "0.1.0"
source = { editable = "." }
dependencies = [
{ name = "temporalio" },
]
[package.optional-dependencies]
dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
]
[package.metadata]
requires-dist = [
{ name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },
{ name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.26.0" },
{ name = "temporalio", specifier = ">=1.9.0" },
]
provides-extras = ["dev"]
[[package]]
name = "iniconfig"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
[[package]]
name = "nexus-rpc"
version = "1.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/2e/f2/d54f5c03d8f4672ccc0875787a385f53dcb61f98a8ae594b5620e85b9cb3/nexus_rpc-1.3.0.tar.gz", hash = "sha256:e56d3b57b60d707ce7a72f83f23f106b86eca1043aa658e44582ab5ff30ab9ad", size = 75650, upload-time = "2025-12-08T22:59:13.002Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d6/74/0afd841de3199c148146c1d43b4bfb5605b2f1dc4c9a9087fe395091ea5a/nexus_rpc-1.3.0-py3-none-any.whl", hash = "sha256:aee0707b4861b22d8124ecb3f27d62dafbe8777dc50c66c91e49c006f971b92d", size = 28873, upload-time = "2025-12-08T22:59:12.024Z" },
]
[[package]]
name = "packaging"
version = "26.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/65/ee/299d360cdc32edc7d2cf530f3accf79c4fca01e96ffc950d8a52213bd8e4/packaging-26.0.tar.gz", hash = "sha256:00243ae351a257117b6a241061796684b084ed1c516a08c48a3f7e147a9d80b4", size = 143416, upload-time = "2026-01-21T20:50:39.064Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "protobuf"
version = "6.33.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/66/70/e908e9c5e52ef7c3a6c7902c9dfbb34c7e29c25d2f81ade3856445fd5c94/protobuf-6.33.6.tar.gz", hash = "sha256:a6768d25248312c297558af96a9f9c929e8c4cee0659cb07e780731095f38135", size = 444531, upload-time = "2026-03-18T19:05:00.988Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fc/9f/2f509339e89cfa6f6a4c4ff50438db9ca488dec341f7e454adad60150b00/protobuf-6.33.6-cp310-abi3-win32.whl", hash = "sha256:7d29d9b65f8afef196f8334e80d6bc1d5d4adedb449971fefd3723824e6e77d3", size = 425739, upload-time = "2026-03-18T19:04:48.373Z" },
{ url = "https://files.pythonhosted.org/packages/76/5d/683efcd4798e0030c1bab27374fd13a89f7c2515fb1f3123efdfaa5eab57/protobuf-6.33.6-cp310-abi3-win_amd64.whl", hash = "sha256:0cd27b587afca21b7cfa59a74dcbd48a50f0a6400cfb59391340ad729d91d326", size = 437089, upload-time = "2026-03-18T19:04:50.381Z" },
{ url = "https://files.pythonhosted.org/packages/5c/01/a3c3ed5cd186f39e7880f8303cc51385a198a81469d53d0fdecf1f64d929/protobuf-6.33.6-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:9720e6961b251bde64edfdab7d500725a2af5280f3f4c87e57c0208376aa8c3a", size = 427737, upload-time = "2026-03-18T19:04:51.866Z" },
{ url = "https://files.pythonhosted.org/packages/ee/90/b3c01fdec7d2f627b3a6884243ba328c1217ed2d978def5c12dc50d328a3/protobuf-6.33.6-cp39-abi3-manylinux2014_aarch64.whl", hash = "sha256:e2afbae9b8e1825e3529f88d514754e094278bb95eadc0e199751cdd9a2e82a2", size = 324610, upload-time = "2026-03-18T19:04:53.096Z" },
{ url = "https://files.pythonhosted.org/packages/9b/ca/25afc144934014700c52e05103c2421997482d561f3101ff352e1292fb81/protobuf-6.33.6-cp39-abi3-manylinux2014_s390x.whl", hash = "sha256:c96c37eec15086b79762ed265d59ab204dabc53056e3443e702d2681f4b39ce3", size = 339381, upload-time = "2026-03-18T19:04:54.616Z" },
{ url = "https://files.pythonhosted.org/packages/16/92/d1e32e3e0d894fe00b15ce28ad4944ab692713f2e7f0a99787405e43533a/protobuf-6.33.6-cp39-abi3-manylinux2014_x86_64.whl", hash = "sha256:e9db7e292e0ab79dd108d7f1a94fe31601ce1ee3f7b79e0692043423020b0593", size = 323436, upload-time = "2026-03-18T19:04:55.768Z" },
{ url = "https://files.pythonhosted.org/packages/c4/72/02445137af02769918a93807b2b7890047c32bfb9f90371cbc12688819eb/protobuf-6.33.6-py3-none-any.whl", hash = "sha256:77179e006c476e69bf8e8ce866640091ec42e1beb80b213c3900006ecfba6901", size = 170656, upload-time = "2026-03-18T19:04:59.826Z" },
]
[[package]]
name = "pygments"
version = "2.19.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/b0/77/a5b8c569bf593b0140bde72ea885a803b82086995367bf2037de0159d924/pygments-2.19.2.tar.gz", hash = "sha256:636cb2477cec7f8952536970bc533bc43743542f70392ae026374600add5b887", size = 4968631, upload-time = "2025-06-21T13:39:12.283Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
]
[[package]]
name = "pytest"
version = "9.0.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
]
[[package]]
name = "pytest-asyncio"
version = "1.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
{ name = "typing-extensions", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/90/2c/8af215c0f776415f3590cac4f9086ccefd6fd463befeae41cd4d3f193e5a/pytest_asyncio-1.3.0.tar.gz", hash = "sha256:d7f52f36d231b80ee124cd216ffb19369aa168fc10095013c6b014a34d3ee9e5", size = 50087, upload-time = "2025-11-10T16:07:47.256Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
]
[[package]]
name = "temporalio"
version = "1.23.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "nexus-rpc" },
{ name = "protobuf" },
{ name = "types-protobuf" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/67/48/ba7413e2fab8dcd277b9df00bafa572da24e9ca32de2f38d428dc3a2825c/temporalio-1.23.0.tar.gz", hash = "sha256:72750494b00eb73ded9db76195e3a9b53ff548780f73d878ec3f807ee3191410", size = 1933051, upload-time = "2026-02-18T17:48:22.353Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/6f/71/26c8f21dca9092201b3b9cb7aff42460b4864b5999aa4c6a4343ac66f1fd/temporalio-1.23.0-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:6b69ac8d75f2d90e66f4edce4316f6a33badc4a30b22efc50e9eddaa9acdc216", size = 12311037, upload-time = "2026-02-18T17:47:47.628Z" },
{ url = "https://files.pythonhosted.org/packages/ec/47/43102816139f2d346680cb7cc1e53da5f6968355ac65b4d35d4edbfca896/temporalio-1.23.0-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:1bbbb2f9c3cdd09451565163f6d741e51f109694c49435d475fdfa42b597219d", size = 11821906, upload-time = "2026-02-18T17:47:55.314Z" },
{ url = "https://files.pythonhosted.org/packages/00/b0/899ff28464a0e17adf17476bdfac8faf4ea41870358ff2d14737e43f9e66/temporalio-1.23.0-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:cf6570e0ee696f99a38d855da4441a890c7187357c16505ed458ac9ef274ed70", size = 12063601, upload-time = "2026-02-18T17:48:03.994Z" },
{ url = "https://files.pythonhosted.org/packages/ed/17/b8c6d2ec3e113c6a788322513a5ff635bdd54b3791d092ed0e273467748a/temporalio-1.23.0-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b82d6cca54c9f376b50e941dd10d12f7fe5b692a314fb087be72cd2898646a79", size = 12394579, upload-time = "2026-02-18T17:48:11.65Z" },
{ url = "https://files.pythonhosted.org/packages/b4/b7/f9ef7fd5ee65aef7d59ab1e95cb1b45df2fe49c17e3aa4d650ae3322f015/temporalio-1.23.0-cp310-abi3-win_amd64.whl", hash = "sha256:43c3b99a46dd329761a256f3855710c4a5b322afc879785e468bdd0b94faace6", size = 12834494, upload-time = "2026-02-18T17:48:19.071Z" },
]
[[package]]
name = "types-protobuf"
version = "6.32.1.20260221"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/5f/e2/9aa4a3b2469508bd7b4e2ae11cbedaf419222a09a1b94daffcd5efca4023/types_protobuf-6.32.1.20260221.tar.gz", hash = "sha256:6d5fb060a616bfb076cbb61b4b3c3969f5fc8bec5810f9a2f7e648ee5cbcbf6e", size = 64408, upload-time = "2026-02-21T03:55:13.916Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2e/e8/1fd38926f9cf031188fbc5a96694203ea6f24b0e34bd64a225ec6f6291ba/types_protobuf-6.32.1.20260221-py3-none-any.whl", hash = "sha256:da7cdd947975964a93c30bfbcc2c6841ee646b318d3816b033adc2c4eb6448e4", size = 77956, upload-time = "2026-02-21T03:55:12.894Z" },
]
[[package]]
name = "typing-extensions"
version = "4.15.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/94/1a15dd82efb362ac84269196e94cf00f187f7ed21c242792a923cdb1c61f/typing_extensions-4.15.0.tar.gz", hash = "sha256:0cea48d173cc12fa28ecabc3b837ea3cf6f38c6d1136f85cbaaf598984861466", size = 109391, upload-time = "2025-08-25T13:49:26.313Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/18/67/36e9267722cc04a6b9f15c7f3441c2363321a3ea07da7ae0c0707beb2a9c/typing_extensions-4.15.0-py3-none-any.whl", hash = "sha256:f0fa19c6845758ab08074a0cfa8b7aecb71c999ca73d62883bc25cc018c4e548", size = 44614, upload-time = "2025-08-25T13:49:24.86Z" },
]