From 53c22232fdf968ea69621b6413dcad0dbe0404e7 Mon Sep 17 00:00:00 2001 From: Jonathan Boniface Date: Tue, 9 Jun 2026 17:32:17 +0100 Subject: [PATCH] feat: initial scaffold + JobStore (v0.1.0) --- .gitea/workflows/ci.yml | 78 ++++++++++ .gitignore | 48 ++++++ CHANGELOG.md | 24 +++ README.md | 125 ++++++++++++++++ docs/jobstore-design.md | 165 ++++++++++++++++++++ pyproject.toml | 72 +++++++++ src/platform_common/__init__.py | 8 + src/platform_common/jobs.py | 256 ++++++++++++++++++++++++++++++++ src/platform_common/py.typed | 0 tests/test_jobs.py | 110 ++++++++++++++ 10 files changed, 886 insertions(+) create mode 100644 .gitea/workflows/ci.yml create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 README.md create mode 100644 docs/jobstore-design.md create mode 100644 pyproject.toml create mode 100644 src/platform_common/__init__.py create mode 100644 src/platform_common/jobs.py create mode 100644 src/platform_common/py.typed create mode 100644 tests/test_jobs.py diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..0bc0a74 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,78 @@ +name: ci + +on: + push: + branches: [main] + tags: ["v*"] + pull_request: + branches: [main] + +jobs: + lint-type-test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.12", "3.13"] + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + + - name: Install + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" + + - name: Lint (ruff) + run: ruff check . + + - name: Format check (ruff) + run: ruff format --check . + + - name: Type check (mypy) + run: mypy + + - name: Tests (pytest) + run: pytest -v + + publish: + needs: lint-type-test + if: startsWith(github.ref, 'refs/tags/v') + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install build tooling + run: | + python -m pip install --upgrade pip + pip install build twine + + - name: Verify tag matches pyproject version + run: | + tag="${GITHUB_REF##*/v}" + pkg=$(python -c "import tomllib,pathlib; print(tomllib.loads(pathlib.Path('pyproject.toml').read_text())['project']['version'])") + if [ "$tag" != "$pkg" ]; then + echo "::error::Tag v$tag does not match pyproject.toml version $pkg" + exit 1 + fi + + - name: Build sdist + wheel + run: python -m build + + - name: Publish to Gitea PyPI registry + env: + TWINE_USERNAME: ${{ secrets.GITEA_PYPI_USERNAME }} + TWINE_PASSWORD: ${{ secrets.GITEA_PYPI_TOKEN }} + TWINE_REPOSITORY_URL: https://gitea.kyndemo.live/api/packages/platform/pypi + run: twine upload --non-interactive dist/* diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b1efded --- /dev/null +++ b/.gitignore @@ -0,0 +1,48 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# Virtual environments +.venv/ +venv/ +env/ + +# Caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.tox/ +.coverage +.coverage.* +htmlcov/ +coverage.xml +*.cover + +# Editors +.idea/ +.vscode/ +*.swp + +# OS +.DS_Store diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..01f710a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,24 @@ +# Changelog + +All notable changes to `platform-common` are documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Planned +- `platform_common.checkpoint.make_checkpointer()` — LangGraph checkpointer factory (Redis + Postgres backends). +- `platform_common.blob.BlobStore` — Azure Blob wrapper. +- `platform_common.lease.LeaderLease` — Kubernetes Lease-based singleton coordination. + +## [0.1.0] — 2026-06-09 + +### Added +- Initial package skeleton with `src/` layout, `py.typed` marker, ruff + mypy + pytest config. +- `platform_common.jobs.JobStore` — async protocol for TTL-bounded, replica-safe key/value job state. +- `platform_common.jobs.InMemoryJobStore` — reference / testing backend (LRU + TTL, asyncio-safe). +- `platform_common.jobs.JobRecord` — typed record shape (`id`, `status`, `payload`, `result`, `error`, `created_at`, `updated_at`). +- `platform_common.jobs.JobStatus` — `pending` / `running` / `succeeded` / `failed` / `cancelled`. +- `platform_common.jobs.make_job_store()` — env-driven factory (`JOB_STORE_URL`); defaults to in-memory. +- Gitea Actions CI: ruff, mypy, pytest on every push; build + publish to Gitea PyPI on tag. +- Design doc at `docs/jobstore-design.md`. diff --git a/README.md b/README.md new file mode 100644 index 0000000..0bc50ad --- /dev/null +++ b/README.md @@ -0,0 +1,125 @@ +# platform-common-libraries + +Shared infrastructure plumbing for platform agents. One package, one version, imported by every agent under `cjot-backstage-az/agents/`. + +> **Current version:** `0.1.0` (see [CHANGELOG.md](CHANGELOG.md)) +> **Pairs with:** [`agent-harness-spec`](https://github.com/kyndryl/agent-harness-spec) — the spec mandates externalised state; this library provides the implementation. + +--- + +## Scope + +This package contains **only infrastructure plumbing** that recurs across agents: + +| Module | What it does | Status | +|---|---|---| +| `platform_common.jobs` | `JobStore` — TTL-bounded, replica-safe key/value store for job, session, and conversation state. | **v0.1.0 — interface + in-memory backend** | +| `platform_common.checkpoint` | `make_checkpointer()` — LangGraph checkpointer factory (Redis or Postgres). | Planned — v0.2 | +| `platform_common.blob` | `BlobStore` — Azure Blob wrapper for large artefacts. | Planned — v0.3 | +| `platform_common.lease` | `LeaderLease` — Kubernetes Lease-based singleton coordination. | Planned — v0.4 | +| `platform_common.testing` | Fakes for unit tests in downstream agents. | Tracks each module | + +### Explicit non-goals + +This package does **not** contain: + +- Business logic (catalog models, prompts, LangGraph workflows, domain APIs) +- HTTP framework code (routes, middleware, manifest serialisation — that's `agent-harness-spec/template/`) +- One-off helpers used by a single agent + +If a module would only ever be imported by one agent, it does not belong here. + +--- + +## Install + +### From the Gitea PyPI registry (recommended) + +```bash +pip install --extra-index-url https://gitea.kyndemo.live/api/packages/platform/pypi/simple/ \ + platform-common[redis] +``` + +In an agent's `requirements.txt`: + +``` +--extra-index-url https://gitea.kyndemo.live/api/packages/platform/pypi/simple/ +platform-common[redis]==0.1.0 +``` + +### From git (no registry needed) + +``` +platform-common[redis] @ git+https://gitea.kyndemo.live/platform/platform-common-libraries.git@v0.1.0 +``` + +### Extras + +Each module's third-party dependencies are gated behind an extra so agents only install what they need: + +| Extra | Pulls in | +|---|---| +| `[redis]` | `redis` (sync + asyncio client) | +| `[azure-blob]` | `azure-storage-blob`, `azure-identity` | +| `[langgraph]` | `langgraph` + Redis/Postgres checkpointers | +| `[k8s]` | `kubernetes` (for `LeaderLease`) | +| `[testing]` | `pytest`, `pytest-asyncio`, `fakeredis` | +| `[dev]` | All of the above + `ruff`, `mypy`, `build`, `twine` | + +--- + +## Versioning policy + +Strict [semver](https://semver.org/): + +- **MAJOR** — breaking change to any public symbol (signature, return type, env var name). +- **MINOR** — additive: new module, new optional parameter, new backend. +- **PATCH** — bug fixes, performance, internal refactors. + +Public surface = anything not prefixed with `_`. Deprecations get one minor release with a `DeprecationWarning` before removal. + +The Python floor tracks `agent-harness-spec` — currently `>=3.12`. + +--- + +## Development + +```bash +git clone https://gitea.kyndemo.live/platform/platform-common-libraries.git +cd platform-common-libraries +python -m venv .venv && source .venv/bin/activate +pip install -e ".[dev]" + +ruff check . +mypy +pytest +``` + +### Releasing + +1. Bump `version` in `pyproject.toml`. +2. Move the `[Unreleased]` notes in `CHANGELOG.md` under a new `## [X.Y.Z] — YYYY-MM-DD` heading. +3. Commit, then `git tag vX.Y.Z && git push --tags`. +4. The Gitea Actions `release` workflow builds the wheel + sdist and publishes to the platform PyPI registry. + +--- + +## Repository layout + +``` +platform-common-libraries/ +├── pyproject.toml +├── README.md +├── CHANGELOG.md +├── LICENSE +├── .gitea/workflows/ci.yml # lint, type-check, test, publish on tag +├── src/ +│ └── platform_common/ +│ ├── __init__.py +│ ├── py.typed # PEP 561 marker — agents get type hints +│ └── jobs.py # JobStore (v0.1.0) +├── tests/ +│ └── test_jobs.py +└── docs/ + └── jobstore-design.md # contract & rationale for JobStore +``` diff --git a/docs/jobstore-design.md b/docs/jobstore-design.md new file mode 100644 index 0000000..604641e --- /dev/null +++ b/docs/jobstore-design.md @@ -0,0 +1,165 @@ +# JobStore — design notes + +> **Status:** Implemented in v0.1.0 (interface + in-memory backend). Redis backend lands in v0.2. + +## Why this exists + +The [agent state audit](https://gitea.kyndemo.live/platform/platform-common-libraries) identified that 7+ agents under `cjot-backstage-az/agents/` hold ephemeral state in module-level Python dicts: + +| Agent | Symbol | Holds | +|---|---|---| +| agent-factory | `pipelines` | active build pipelines | +| decomposer | `_active_runs` | LangGraph workflow state (no eviction → leaks memory) | +| discovery-agent | `background_tasks` | scan job status / result / error | +| the-watcher | `background_tasks` | scan job status / result / error | +| golden-path | `conversation_store` | conversation history (LRU + TTL) | +| support-intake-agent | `SessionStore` | intake sessions (LRU + TTL) | +| scaffold-agent | `_ARTIFACT_CACHE` | generated artefacts during scaffold | + +Every one of these: + +1. Is **lost on pod restart** — restart = data loss for in-flight requests. +2. **Pins each Deployment to `replicas: 1`** — a second pod sees an empty dict. +3. **Drifts in implementation** — some have TTL, some don't; some are thread-safe, some aren't; some leak. + +`JobStore` collapses all seven into one contract with one set of semantics. Swap the dict for `JobStore`, the agent becomes safe to scale. + +## Contract + +```python +class JobStore(Protocol): + async def create(self, job_id, payload=None, *, ttl_seconds, status=PENDING) -> JobRecord + async def get(self, job_id) -> JobRecord | None + async def update(self, job_id, *, status=None, payload=None, result=None, error=None, ttl_seconds=None) -> JobRecord + async def delete(self, job_id) -> bool + async def exists(self, job_id) -> bool + async def list_ids(self, *, limit=100) -> list[str] + async def close(self) -> None +``` + +### Why these five verbs and no more + +- **create / get / update / delete** cover every callsite found in the audit. The agents that "list" only do so for a debug `/jobs` endpoint, hence `list_ids` is capped (no pagination, no filtering — backends shouldn't carry that weight). +- **No `set` / `upsert`.** `create` raises on conflict; `update` raises if missing. This catches the bug class "two requests with the same id silently clobber each other". +- **No `query` / `find_by_status`.** Searching across the keyspace is a different concern — if an agent needs that, it has graduated from "job state" to "domain data" and should use Postgres. +- **No `subscribe` / pub-sub.** Out of scope for v0.1. Streaming progress to clients is a future concern; the current pattern (poll `GET /jobs/{id}`) works. + +### Why `JobRecord` is a dataclass, not a dict + +- **Type safety.** `mypy --strict` catches `record.statuz` typos in every agent that imports it. +- **Schema enforcement.** Backends serialise via `to_dict()`/`from_dict()`; ad-hoc keys can't slip in. +- **Future evolution.** Adding fields (`tenant_id`, `trace_id`) is one PR in one place. + +### Why TTL is mandatory on `create` + +The audit found `_active_runs` in [decomposer](cjot-backstage-az/agents/decomposer) has no eviction at all — it leaks until the pod OOMs. Making `ttl_seconds` a required keyword argument means **you cannot accidentally write a leak**. If a caller truly wants long-lived state, they pass `ttl_seconds=86400 * 30` explicitly and the intent is reviewable. + +### Why `JobStatus` is an enum + +The seven agents currently use overlapping but incompatible status strings (`"pending"` vs `"queued"`, `"done"` vs `"complete"` vs `"success"`). Centralising in `JobStatus` (`pending`, `running`, `succeeded`, `failed`, `cancelled`) gives dashboards a single vocabulary. + +## Backend selection + +```python +from platform_common.jobs import make_job_store + +store = make_job_store() # reads JOB_STORE_URL env var +``` + +| `JOB_STORE_URL` | Backend | When to use | +|---|---|---| +| _(unset)_ or `memory://` | `InMemoryJobStore` | Unit tests, local dev, single-replica deployments | +| `redis://host:6379/0` | `RedisJobStore` (v0.2) | Production | +| `rediss://host:6380/0` | `RedisJobStore` (v0.2) with TLS | Azure Cache for Redis | + +## Redis backend (v0.2 sketch) + +Not yet implemented; documented here so the v0.1 contract isn't quietly broken when it arrives. + +### Storage layout + +``` +key: agentjobs:{agent}:{job_id} +type: HASH +TTL: set via EXPIRE at create + on every update (sliding window) +fields: + status "pending" | "running" | ... + payload JSON-encoded dict + result JSON-encoded dict (or absent) + error UTF-8 string (or absent) + created_at float seconds since epoch + updated_at float seconds since epoch +``` + +`{agent}` comes from `OTEL_SERVICE_NAME` so two agents never collide. `list_ids` uses `SCAN` with `MATCH agentjobs:{agent}:*` — bounded by `limit`. + +### Concurrency + +Reads and deletes are single commands. `update` reads + writes — but each agent owns its own `job_id` space (the agent generated the id) and the only concurrent writer is itself across replicas. Two replicas updating the same job is a real-world hazard (e.g. cancel arriving while the worker is updating progress); v0.2 will use a Lua script for the read-modify-write to keep it atomic and document a `version` field for optimistic concurrency in v0.3 if required. + +### Connection pooling + +One module-level `redis.asyncio.ConnectionPool` per process, lazy-initialised on first call. Health-checked on Starlette `lifespan` startup. Closed on shutdown via `store.close()`. + +### Observability + +Each method emits one OTel span (`platform_common.jobs.{verb}`) with `db.system=redis`, `job.id`, `job.agent`, and duration. Counters: `platform_common.jobs.operations` (by `verb`, `outcome`), histogram `platform_common.jobs.operation.duration` (seconds, by `verb`). + +## Migration recipe (per agent) + +Rough shape of the diff when an agent adopts `JobStore`: + +```diff +-from collections import OrderedDict +-import threading +- +-_jobs: dict[str, dict] = {} +-_lock = threading.Lock() ++from platform_common.jobs import JobStatus, make_job_store ++ ++job_store = make_job_store() # reads JOB_STORE_URL + + async def start_scan(req): + job_id = uuid4().hex +- with _lock: +- _jobs[job_id] = {"status": "pending", "result": None} ++ await job_store.create(job_id, {"request": req}, ttl_seconds=3600) + asyncio.create_task(_run(job_id, req)) + return {"job_id": job_id} + + async def _run(job_id, req): ++ await job_store.update(job_id, status=JobStatus.RUNNING) + try: + result = await do_work(req) +- with _lock: +- _jobs[job_id] = {"status": "done", "result": result} ++ await job_store.update(job_id, status=JobStatus.SUCCEEDED, result=result) + except Exception as e: +- with _lock: +- _jobs[job_id] = {"status": "failed", "error": str(e)} ++ await job_store.update(job_id, status=JobStatus.FAILED, error=str(e)) + + async def get_status(job_id): +- with _lock: +- return _jobs.get(job_id) or {"error": "not found"}, 404 ++ record = await job_store.get(job_id) ++ if record is None: ++ return {"error": "not found"}, 404 ++ return record.to_dict(), 200 +``` + +Then in `k8s/configmap.yaml`: + +```yaml +data: + JOB_STORE_URL: rediss://:6380/0 +``` + +…and the agent is replica-safe. + +## Open questions for the v0.2 review + +1. **Eviction semantics on update.** Should `update` reset the TTL (sliding) or preserve it (fixed)? Current proposal: `update(ttl_seconds=...)` explicit only; otherwise preserve. +2. **Tenant scoping.** Should `{agent}` in the Redis key be augmented with `{tenant_id}`? Probably yes, but waits for the multi-tenancy story to land. +3. **List/scan semantics.** `list_ids` is debug-only; do we want a separate admin-API surface, or is `kubectl exec` + `redis-cli` enough? +4. **Distributed locks.** Several agents (`discovery-agent`, `the-watcher`) currently hold an `asyncio.Semaphore(3)` for downstream rate limiting. That's a separate concern from `JobStore` — propose a `platform_common.ratelimit` module in v0.3 rather than overloading this one. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2044423 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,72 @@ +[build-system] +requires = ["hatchling>=1.24"] +build-backend = "hatchling.build" + +[project] +name = "platform-common" +version = "0.1.0" +description = "Shared infrastructure plumbing for platform agents: job store, LangGraph checkpointer, blob store, leader lease." +readme = "README.md" +requires-python = ">=3.12" +license = { text = "Proprietary" } +authors = [{ name = "Platform Engineering" }] +keywords = ["platform", "agents", "redis", "langgraph", "kubernetes"] +classifiers = [ + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Operating System :: OS Independent", + "Typing :: Typed", +] +dependencies = [] + +[project.optional-dependencies] +redis = ["redis>=5.0,<6"] +azure-blob = ["azure-storage-blob>=12.19", "azure-identity>=1.15"] +langgraph = [ + "langgraph>=0.2", + "langgraph-checkpoint-redis>=0.1", + "langgraph-checkpoint-postgres>=2.0", +] +k8s = ["kubernetes>=29.0"] +testing = ["pytest>=8", "pytest-asyncio>=0.23", "fakeredis>=2.21"] +dev = [ + "platform-common[redis,azure-blob,langgraph,k8s,testing]", + "ruff>=0.5", + "mypy>=1.10", + "build>=1.2", + "twine>=5", +] + +[project.urls] +Repository = "https://gitea.kyndemo.live/platform/platform-common-libraries" +Changelog = "https://gitea.kyndemo.live/platform/platform-common-libraries/src/branch/main/CHANGELOG.md" + +[tool.hatch.build.targets.wheel] +packages = ["src/platform_common"] + +[tool.hatch.build.targets.sdist] +include = ["src/platform_common", "README.md", "CHANGELOG.md", "LICENSE"] + +[tool.ruff] +line-length = 100 +target-version = "py312" +src = ["src", "tests"] + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP", "SIM", "PL", "RUF"] +ignore = ["PLR0913"] + +[tool.ruff.lint.per-file-ignores] +"tests/*" = ["PLR2004", "S101"] + +[tool.mypy] +python_version = "3.12" +strict = true +files = ["src/platform_common"] +warn_unused_configs = true + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +addopts = "-ra --strict-markers" diff --git a/src/platform_common/__init__.py b/src/platform_common/__init__.py new file mode 100644 index 0000000..dd4a55b --- /dev/null +++ b/src/platform_common/__init__.py @@ -0,0 +1,8 @@ +"""Shared infrastructure plumbing for platform agents. + +See https://gitea.kyndemo.live/platform/platform-common-libraries for docs. +""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] diff --git a/src/platform_common/jobs.py b/src/platform_common/jobs.py new file mode 100644 index 0000000..77b623d --- /dev/null +++ b/src/platform_common/jobs.py @@ -0,0 +1,256 @@ +"""TTL-bounded, replica-safe key/value store for agent job, session, and conversation state. + +Replaces module-level ``dict``s like ``background_tasks``, ``_active_runs``, ``pipelines``, +``SessionStore``, ``conversation_store``, and ``_ARTIFACT_CACHE`` scattered across agents. + +The public contract is :class:`JobStore`. Two backends ship in this release: + +* :class:`InMemoryJobStore` — asyncio-safe, LRU + TTL, for tests and single-replica dev. +* (Redis backend lands in v0.2 — see ``docs/jobstore-design.md``.) + +Use :func:`make_job_store` to pick a backend from the ``JOB_STORE_URL`` env var. +""" + +from __future__ import annotations + +import asyncio +import os +import time +from collections import OrderedDict +from collections.abc import Iterable, Mapping +from dataclasses import asdict, dataclass, field, replace +from enum import StrEnum +from typing import Any, Protocol, runtime_checkable +from urllib.parse import urlparse + + +class JobStatus(StrEnum): + """Lifecycle state of a job. Stored as the string value in backends.""" + + PENDING = "pending" + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + + +@dataclass(slots=True) +class JobRecord: + """The canonical job record. Backends serialise this to their native format.""" + + id: str + status: JobStatus = JobStatus.PENDING + payload: dict[str, Any] = field(default_factory=dict) + result: dict[str, Any] | None = None + error: str | None = None + created_at: float = field(default_factory=time.time) + updated_at: float = field(default_factory=time.time) + + def to_dict(self) -> dict[str, Any]: + d = asdict(self) + d["status"] = self.status.value + return d + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> JobRecord: + return cls( + id=str(data["id"]), + status=JobStatus(data.get("status", JobStatus.PENDING.value)), + payload=dict(data.get("payload") or {}), + result=dict(data["result"]) if data.get("result") is not None else None, + error=data.get("error"), + created_at=float(data.get("created_at", time.time())), + updated_at=float(data.get("updated_at", time.time())), + ) + + +class JobNotFoundError(KeyError): + """Raised by mutating methods when the job id is unknown.""" + + +@runtime_checkable +class JobStore(Protocol): + """Async contract for a job/session store. All backends MUST implement this.""" + + async def create( + self, + job_id: str, + payload: Mapping[str, Any] | None = None, + *, + ttl_seconds: int, + status: JobStatus = JobStatus.PENDING, + ) -> JobRecord: + """Create a new record. Raises :class:`ValueError` if ``job_id`` already exists.""" + ... + + async def get(self, job_id: str) -> JobRecord | None: + """Return the record or ``None`` if absent or expired.""" + ... + + async def update( + self, + job_id: str, + *, + status: JobStatus | None = None, + payload: Mapping[str, Any] | None = None, + result: Mapping[str, Any] | None = None, + error: str | None = None, + ttl_seconds: int | None = None, + ) -> JobRecord: + """Patch fields and bump ``updated_at``. Raises :class:`JobNotFoundError` if missing.""" + ... + + async def delete(self, job_id: str) -> bool: + """Remove a record. Returns ``True`` if deleted, ``False`` if absent.""" + ... + + async def exists(self, job_id: str) -> bool: ... + + async def list_ids(self, *, limit: int = 100) -> list[str]: + """Return up to ``limit`` ids. Intended for debug endpoints only — not paginated.""" + ... + + async def close(self) -> None: + """Release backend resources (connection pools, etc.). Idempotent.""" + ... + + +class InMemoryJobStore: + """Asyncio-safe in-memory backend with LRU + TTL eviction. + + Suitable for unit tests, local dev, and single-replica deployments. Not safe across + pod restarts (state is lost). Use the Redis backend in production. + """ + + def __init__(self, *, max_records: int = 10_000) -> None: + self._records: OrderedDict[str, JobRecord] = OrderedDict() + self._expiry: dict[str, float] = {} + self._lock = asyncio.Lock() + self._max_records = max_records + + async def create( + self, + job_id: str, + payload: Mapping[str, Any] | None = None, + *, + ttl_seconds: int, + status: JobStatus = JobStatus.PENDING, + ) -> JobRecord: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + async with self._lock: + self._evict_expired_locked() + if job_id in self._records: + raise ValueError(f"job id already exists: {job_id}") + record = JobRecord(id=job_id, status=status, payload=dict(payload or {})) + self._records[job_id] = record + self._expiry[job_id] = time.time() + ttl_seconds + self._enforce_capacity_locked() + return replace(record) + + async def get(self, job_id: str) -> JobRecord | None: + async with self._lock: + if not self._is_live_locked(job_id): + return None + self._records.move_to_end(job_id) + return replace(self._records[job_id]) + + async def update( + self, + job_id: str, + *, + status: JobStatus | None = None, + payload: Mapping[str, Any] | None = None, + result: Mapping[str, Any] | None = None, + error: str | None = None, + ttl_seconds: int | None = None, + ) -> JobRecord: + async with self._lock: + if not self._is_live_locked(job_id): + raise JobNotFoundError(job_id) + record = self._records[job_id] + if status is not None: + record.status = status + if payload is not None: + record.payload = dict(payload) + if result is not None: + record.result = dict(result) + if error is not None: + record.error = error + record.updated_at = time.time() + if ttl_seconds is not None: + if ttl_seconds <= 0: + raise ValueError("ttl_seconds must be positive") + self._expiry[job_id] = time.time() + ttl_seconds + self._records.move_to_end(job_id) + return replace(record) + + async def delete(self, job_id: str) -> bool: + async with self._lock: + existed = self._records.pop(job_id, None) is not None + self._expiry.pop(job_id, None) + return existed + + async def exists(self, job_id: str) -> bool: + async with self._lock: + return self._is_live_locked(job_id) + + async def list_ids(self, *, limit: int = 100) -> list[str]: + async with self._lock: + self._evict_expired_locked() + return list(self._records.keys())[:limit] + + async def close(self) -> None: + async with self._lock: + self._records.clear() + self._expiry.clear() + + def _is_live_locked(self, job_id: str) -> bool: + if job_id not in self._records: + return False + if time.time() >= self._expiry.get(job_id, 0): + self._records.pop(job_id, None) + self._expiry.pop(job_id, None) + return False + return True + + def _evict_expired_locked(self) -> None: + now = time.time() + expired: Iterable[str] = [jid for jid, exp in self._expiry.items() if now >= exp] + for jid in expired: + self._records.pop(jid, None) + self._expiry.pop(jid, None) + + def _enforce_capacity_locked(self) -> None: + while len(self._records) > self._max_records: + jid, _ = self._records.popitem(last=False) + self._expiry.pop(jid, None) + + +def make_job_store(url: str | None = None) -> JobStore: + """Construct a backend from ``url`` (or the ``JOB_STORE_URL`` env var). + + Supported schemes: + + * ``memory://`` — :class:`InMemoryJobStore` (default when env is unset). + * ``redis://`` / ``rediss://`` — **planned for v0.2**; raises :class:`NotImplementedError`. + """ + target = url or os.environ.get("JOB_STORE_URL", "memory://") + scheme = urlparse(target).scheme or "memory" + if scheme == "memory": + return InMemoryJobStore() + if scheme in {"redis", "rediss"}: + raise NotImplementedError( + "Redis backend lands in platform-common v0.2 — see docs/jobstore-design.md" + ) + raise ValueError(f"unsupported JOB_STORE_URL scheme: {scheme!r}") + + +__all__ = [ + "InMemoryJobStore", + "JobNotFoundError", + "JobRecord", + "JobStatus", + "JobStore", + "make_job_store", +] diff --git a/src/platform_common/py.typed b/src/platform_common/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_jobs.py b/tests/test_jobs.py new file mode 100644 index 0000000..0f53bfc --- /dev/null +++ b/tests/test_jobs.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import asyncio + +import pytest + +from platform_common.jobs import ( + InMemoryJobStore, + JobNotFoundError, + JobRecord, + JobStatus, + JobStore, + make_job_store, +) + + +@pytest.fixture +def store() -> InMemoryJobStore: + return InMemoryJobStore() + + +def test_protocol_conformance(store: InMemoryJobStore) -> None: + assert isinstance(store, JobStore) + + +async def test_create_and_get(store: InMemoryJobStore) -> None: + record = await store.create("j1", {"input": "hello"}, ttl_seconds=60) + assert record.id == "j1" + assert record.status is JobStatus.PENDING + assert record.payload == {"input": "hello"} + fetched = await store.get("j1") + assert fetched is not None + assert fetched.payload == {"input": "hello"} + + +async def test_create_duplicate_raises(store: InMemoryJobStore) -> None: + await store.create("j1", ttl_seconds=60) + with pytest.raises(ValueError, match="already exists"): + await store.create("j1", ttl_seconds=60) + + +async def test_get_missing_returns_none(store: InMemoryJobStore) -> None: + assert await store.get("nope") is None + + +async def test_update_patches_fields(store: InMemoryJobStore) -> None: + await store.create("j1", {"a": 1}, ttl_seconds=60) + updated = await store.update("j1", status=JobStatus.RUNNING, result={"out": 42}) + assert updated.status is JobStatus.RUNNING + assert updated.result == {"out": 42} + assert updated.payload == {"a": 1} + assert updated.updated_at >= updated.created_at + + +async def test_update_missing_raises(store: InMemoryJobStore) -> None: + with pytest.raises(JobNotFoundError): + await store.update("nope", status=JobStatus.FAILED) + + +async def test_delete(store: InMemoryJobStore) -> None: + await store.create("j1", ttl_seconds=60) + assert await store.delete("j1") is True + assert await store.delete("j1") is False + assert await store.get("j1") is None + + +async def test_ttl_expiry(store: InMemoryJobStore) -> None: + await store.create("j1", ttl_seconds=1) + assert await store.exists("j1") + await asyncio.sleep(1.1) + assert await store.get("j1") is None + assert await store.exists("j1") is False + + +async def test_invalid_ttl(store: InMemoryJobStore) -> None: + with pytest.raises(ValueError, match="positive"): + await store.create("j1", ttl_seconds=0) + + +async def test_lru_capacity() -> None: + store = InMemoryJobStore(max_records=3) + for i in range(5): + await store.create(f"j{i}", ttl_seconds=60) + ids = await store.list_ids() + assert set(ids) == {"j2", "j3", "j4"} + + +async def test_record_roundtrip() -> None: + record = JobRecord(id="j1", status=JobStatus.SUCCEEDED, payload={"k": "v"}, result={"r": 1}) + restored = JobRecord.from_dict(record.to_dict()) + assert restored == record + + +def test_make_job_store_default(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("JOB_STORE_URL", raising=False) + assert isinstance(make_job_store(), InMemoryJobStore) + + +def test_make_job_store_memory_explicit() -> None: + assert isinstance(make_job_store("memory://"), InMemoryJobStore) + + +def test_make_job_store_redis_not_yet_implemented() -> None: + with pytest.raises(NotImplementedError, match=r"v0\.2"): + make_job_store("redis://localhost:6379/0") + + +def test_make_job_store_unknown_scheme() -> None: + with pytest.raises(ValueError, match="unsupported"): + make_job_store("kafka://broker")