# 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://<azure-cache-host>: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.