platform-common-libraries/docs/jobstore-design.md

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 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

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 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

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:

-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:

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.