think
16px
820px
← all files

Gateway P2 — Job API + OCR/classify adapters Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: superpowers:subagent-driven-development (fresh implementer per task, task review between). Steps use - [ ]. This extends the shipped P0/P1 gateway — match its established patterns (the LocalPool/RedisPool duality, the audit Emitter, the proxy.handler admission→lease→emit flow, per-server prometheus registry). Read the referenced P0 files before writing; do not reinvent what P0 already provides.

Goal: Add the asynchronous Job API (POST /jobs, GET /jobs/{id}) and generic HTTP (ocr-http) + classify adapters so OCR (PaddleOCR, Azure DI) and document classification route through the gateway with the same queueing, priority, health, idempotency, durability, and audit coverage as LLM traffic — completing the "all model/OCR egress through one audited choke point" contract.

Architecture: Jobs are durable records (Redis when coordination: redis, in-memory when local — mirror the existing pool.Pool interface + LocalPool/RedisPool split). A job manager pulls queued jobs, admits each through the existing per-upstream slot Pool (interactive outranks batch), invokes a typed adapter for the upstream (ocr-http sync / azure-di async-poll / classify), tracks queued→processing→completed|failed with queue_position/eta_ms/stage/progress, re-queues on worker death (acks-late), and emits one audit event per job (+ per child for fan-out). The Job API endpoints are wire-compatible with the existing gpu-server contract (202 {job_id} → GET /jobs/{id}).

Tech Stack: Go (match the repo's style/version), github.com/klauspost/compress/zstd (already used), redis/go-redis (already used by RedisPool), Prometheus. No new heavy deps without justification.

Global Constraints (copied from the platform contract — every task inherits these)

  • Canonical contract: docs/CONVENTIONS.md v1.0 + design spec §4.3–4.5, §4.1 (registry), §5 (audit). On conflict, CONVENTIONS wins. Audit event fields are fixed by internal/audit/event.go (schema v1) — operation ∈ {chat,embed,ocr,classify,job,job_child}, plus doc_hash, pages, job_id, parent_job_id.
  • On-prem enforcement: ocr-http/classify/azure-di upstreams honor class (on_prem | external_dev) exactly like LLM upstreams; when allow_external_upstreams=false, resolving to an external_dev upstream is refused (same path/behavior as P0).
  • Audit is fire-and-forget and never blocks a request (reuse audit.Emitter); document bodies obey audit.body_cap_bytes — a 100 MB akta is NOT stored: it's truncated at the cap, zstd-compressed, with body_truncated=true + full_body_sha256. doc_hash (from the X-Doc-Hash header or computed from the payload) is the document identity; pages from X-Doc-Pages.
  • Slot-lease correctness is the historically fragile area — P0 review caught multiple leaks (cancel-after-admit, stage-2 cancel, drain TOCTOU). The job worker MUST release its slot on every exit path (success, failure, timeout, panic, worker death). Use defer-based release; add a test that asserts pool Active() returns to baseline after each terminal state.
  • Queued clock ≠ processing clock: poll responses expose queued{queue_position,eta_ms} distinct from processing{stage,progress}. The server tracks these; the engine-side clock rule already lives in the consolidated poll client (OCR repo).
  • Idempotency: POST /jobs dedupes on (tenant, Idempotency-Key) — resubmission returns the EXISTING job (never a duplicate run). A poll for an unknown/expired job returns 404 {code:"JOB_UNKNOWN"}.
  • Durability + drain: jobs survive gateway restart when coordination: redis; graceful drain finishes in-flight jobs (or re-queues them) and returns 503 + Retry-After for new submits — never a silent drop. Reuse the server's existing enter()/leave() drain gate.
  • HA: worker death mid-job → heartbeat/visibility timeout → auto re-queue (acks-late). Multiple gateway replicas share the Redis job store; a job is processed once.
  • TDD; namespace-safe tests (unique ids/tenants, filtered assertions — no bare counts, per the P0 test rule); go test ./... + go vet + go build ./... green before each commit; commit per task; messages end with Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT.
  • Every created/modified .md uploaded to https://x056.think.val.id/upload.
  • Branch feat/gateway-p2-jobs-ocr off master; merge at the end; tag p2-jobs-ocr.

File Structure (new unless noted)

  • internal/config/config.go (MODIFY) — extend Upstream for adapters: Adapter string (""|sync-http|azure-di|classify), SubmitPath/PollPath/ResultPath (adapter-specific), keep Type ocr-http. Validation.
  • internal/jobs/job.go — the Job/JobStatus/JobResult types + status/operation constants.
  • internal/jobs/store.go — Store interface (Create, Get, byIdem, UpdateStatus, ClaimNext, Heartbeat, ListChildren, SetResult) + LocalStore (in-memory).
  • internal/jobs/redisstore.go — RedisStore (durable; Redis lists/hashes; visibility-timeout reclaim).
  • internal/jobs/manager.go — the worker loop: claim → admit via pool.Pool → run adapter → status updates → release → emit audit; heartbeat + acks-late re-queue; fan-out scheduling + merge.
  • internal/jobs/adapter.go — Adapter interface + SyncHTTPAdapter (PaddleOCR/generic), AzureDIAdapter (async-poll), ClassifyAdapter.
  • internal/jobs/api.go — HTTP handlers POST /jobs, GET /jobs/{id} (multipart + JSON), wired into internal/server/server.go.
  • internal/server/server.go (MODIFY) — mount job routes under the drain guard; construct the manager + store; job metrics.
  • deploy/gateway.example.yaml (MODIFY) — add paddleocr, azure-di, doc-classifier upstreams.
  • docs/CONVENTIONS.md (MODIFY, if needed) — only if a field/behavior clarification is required; version-bump per its own rule.

Task 1: Config + registry + audit wiring for ocr-http/azure-di/classify upstreams

Deliverable: Upstream gains adapter fields; config validation accepts and checks them; the registry resolves these upstreams by model id (e.g. paddleocr, azure-di-layout-v4, the classifier id) and by operation; on-prem enforcement applies. No behavior change to existing LLM upstreams.

  • [ ] Add to config.Upstream: Adapter string (sync-http|azure-di|classify; empty invalid when Type==ocr-http/classify), optional SubmitPath, PollPath, ResultPath, OperationValue string (audit operation: ocr|classify). Validate: ocr-http/classify upstreams need ≥1 endpoint + ≥1 model + a known adapter; external_dev still gated by allow_external_upstreams.
  • [ ] Registry: confirm Resolve(model) already returns these (it maps all Models) — add a helper IsJobUpstream(*Upstream) bool (Type ∈ {ocr-http, classify, or llm marked async}) so the API can route sync vs job. Test: resolve paddleocr → the ocr-http upstream; unknown model → ErrUnknownModel; an external_dev OCR upstream with the flag off → refused via the existing enforcement path.
  • [ ] Audit operation constants: ensure event.go exposes ocr/classify/job/job_child (add if missing; keep schema v1). Commit feat(p2): config+registry+audit support for ocr-http/classify upstreams.

Task 2: Durable job store — Store interface + LocalStore

Deliverable: a jobs.Store with an in-memory implementation covering the full lifecycle, idempotency index, and child listing. Mirrors the pool package's Local/Redis split (Redis impl is Task 3).

  • [ ] Job type: ID, Tenant, Surface, UserID, TraceID, IdemKey, Model, Operation string; Class config.Class; Status (queued|processing|completed|failed); QueuePosition int; Stage string; Progress float64; DocHash string; Pages *int; ParentID string; Result json.RawMessage; ErrorCode string; timestamps (CreatedAt, StartedAt, UpdatedAt); Heartbeat time.Time; Attempts int. Payload stored separately (may be large — keep out of the status record; store a ref/blob).
  • [ ] Store interface: Create(ctx, *Job, payload []byte) (existing bool, err error) (dedup on (tenant,IdemKey) — returns existing=true + the found job without inserting), Get(ctx,id), ClaimNext(ctx, class) (*Job, ok) (interactive before batch; sets processing+Heartbeat+Attempts++), Heartbeat(ctx,id), UpdateStatus(ctx,id, fn), SetResult(ctx,id,status,result,errorCode), ListChildren(ctx,parentID), Payload(ctx,id), ReclaimStale(ctx, visibilityTimeout) (processing jobs whose Heartbeat is stale → back to queued, Attempts already ++'d).
  • [ ] LocalStore: mutex-guarded maps; queue ordered by (class-priority, CreatedAt); idem index map[tenant]map[key]id. Test (namespace-safe): create+dedup returns same id; ClaimNext honors interactive>batch ordering; ReclaimStale re-queues a job with a stale heartbeat; ListChildren. Commit feat(p2): job types + Store interface + in-memory LocalStore.

Task 3: RedisStore — durable, HA-safe job state

Deliverable: a Redis-backed Store so jobs survive gateway restart and are processed once across the HA pair; visibility-timeout reclaim for worker death.

  • [ ] Implement every Store method on Redis: job hash job:{id}, payload job:{id}:payload, per-class ready lists jobs:ready:{class} (LPUSH/BRPOPLPUSH into a processing list for acks-late), idem key idem:{tenant}:{key} → id (SETNX for atomic dedup), children set job:{parent}:children. ClaimNext uses BRPOPLPUSH ready → processing + sets heartbeat; ReclaimStale scans the processing list for stale heartbeats and re-queues (LMOVE back). TTLs match the idem/result retention.
  • [ ] Reuse the existing go-redis client construction from internal/pool/redispool.go (same URL/config). Test: gate behind a reachable Redis (skip if absent, like redispool_test.go); dedup atomicity under concurrent Create; claim-once under two concurrent claimers; stale reclaim. Commit feat(p2): RedisStore for durable HA job state.

Task 4: Adapter interface + SyncHTTPAdapter (PaddleOCR / generic ocr-http)

Deliverable: the Adapter abstraction + a synchronous-HTTP adapter that forwards a document payload to a resolved ocr-http endpoint and normalizes the response into a JobResult.

  • [ ] Adapter interface: Run(ctx, *Job, payload []byte, ep string, report func(stage string, progress float64)) (result json.RawMessage, err error). report lets the adapter push processing stage/progress into the store.
  • [ ] SyncHTTPAdapter: POST the payload (JSON or multipart, per the submit content-type recorded on the job) to ep+SubmitPath; on 2xx return the body as result; map upstream 429/503 to a ret/backoff signal; timeouts → failed with error_code. On-prem: never sends an upstream key unless APIKeyEnv is set (server-side injection, like the proxy). Test with a stub upstream: success returns the body; a 500 → failed with code; multipart passthrough intact. Commit feat(p2): Adapter interface + sync-http OCR adapter.

Task 5: AzureDIAdapter — async submit/operation-location poll

Deliverable: an adapter for Azure Document Intelligence's async protocol (submit → Operation-Location → poll until succeeded|failed), normalized to the same JobResult, reporting processing progress while polling.

  • [ ] Submit to ep+SubmitPath (with the DI model id), read Operation-Location, poll PollPath/that URL until terminal, honoring a bounded poll interval + the job's processing timeout; map DI status → job status; return the analyze result JSON. Server-side key injection from APIKeyEnv.
  • [ ] Test with a stub DI upstream: submit→202+Operation-Location→poll running×2→succeeded yields the result and emitted progress; failed → job failed with code; poll timeout → failed. Commit feat(p2): Azure DI async-poll adapter.

Task 6: ClassifyAdapter + interactive classify path

Deliverable: the document-classifier as a first-class upstream. Classify is interactive-priority (spec: "click to classify outranks a 200-doc batch") and returns quickly; route it through the job machinery but with interactive class default and a short hold, operation=classify.

  • [ ] ClassifyAdapter: POST doc/text to the classifier endpoint, return {label, confidence, ...} verbatim as result. Test with a stub classifier: result passthrough; low-confidence still returned (the engine applies its own classifierMinConfidence gate — the gateway does not editorialize). Commit feat(p2): classify adapter (interactive-priority).

Task 7: Job API endpoints + manager (single-job path) + server wiring

Deliverable: POST /jobs and GET /jobs/{id} live, wire-compatible with the gpu-server contract, driven by the job manager running single (non-fan-out) jobs end-to-end through Tasks 2–6, with full audit + drain integration.

  • [ ] POST /jobs: accept multipart (document) or JSON; read the §2 header set (X-Tenant-Id→tenant, X-Surface, X-User-Id, X-Request-Id→trace_id, X-Priority→class, Idempotency-Key, X-Doc-Hash→doc_hash, X-Doc-Pages→pages) via a Meta-style helper (reuse/extend proxy.MetaFrom); resolve model→upstream; enforce on-prem; Store.Create (dedup) → 202 {job_id, status:"queued", queue_position, eta_ms}. Body-size guarded by MaxBodyBytes (documents may be large — this is the multipart ceiling).
  • [ ] GET /jobs/{id}: return {status, queue_position, eta_ms, stage, progress, result?, error_code?}; unknown/expired → 404 {code:"JOB_UNKNOWN"}; tenant-scoped (a tenant can't read another's job).
  • [ ] Manager: loop ClaimNext(class) (interactive first) → admit via the upstream's pool.Pool (reuse P0 admission; release on every path — the fragile area) → dispatch the upstream's Adapter.Run with a report that writes stage/progress → SetResult → emit audit (operation from the upstream, doc_hash/pages, queue_ms=queued duration, upstream_ms=processing duration, body capture obeying body_cap_bytes). Heartbeat during run; ReclaimStale ticker for acks-late. Drain: stop claiming new, let in-flight finish or re-queue; POST /jobs returns 503+Retry-After while draining (reuse enter()).
  • [ ] Mount both routes under the drain guard in server.go; add job metrics (gateway_jobs_total{tenant,upstream,operation,status}, a job-queue-wait histogram). Tests: submit→poll→completed against a stub adapter; idempotent resubmit returns same id + no second run; JOB_UNKNOWN; slot released after completed/failed/timeout (Active() back to baseline); drain returns 503 for new submits. Commit feat(p2): Job API endpoints + manager single-job path + server wiring.

Task 8: Fan-out groups — children + max_parallel + merge

Deliverable: POST /jobs accepts a fan-out group (N children + max_parallel); the manager schedules children across replicas respecting max_parallel and the pool caps, merges child statuses into the parent, and emits job_child audit events per child + a job event for the parent.

  • [ ] Submit shape: {fanout:{children:[{payload_ref|inline, model, doc_hash, pages}], max_parallel:N}} (or multipart with N parts). Parent job operation=job; each child operation=job_child, parent_id=parent. Parent status = derived: queued until any child starts, processing with progress=completed_children/total while running, completed when all succeed, failed if any child fails (configurable: fail-fast vs collect — default collect all, parent failed with per-child results retained).
  • [ ] Manager schedules ≤max_parallel children concurrently, each admitted through the pool like a single job; GET /jobs/{parent} returns merged status + children:[{id,status,...}]; GET /jobs/{childId} works too. Tests: a 4-child group with max_parallel=2 never runs >2 concurrently (assert via a gated stub adapter); parent completes when all children do; one child failing marks parent failed but retains the other results; audit emits 1 job + N job_child. Commit feat(p2): fan-out job groups with max_parallel scheduling + merge.

Task 9: Packaging, example config, docs, live-fire (P2 exit)

Deliverable: deployable P2 — example config with real OCR/classify upstreams, README section, smoke script exercising the Job API, and a live-fire drill; merged + tagged.

  • [ ] deploy/gateway.example.yaml: add paddleocr (ocr-http/sync-http, on_prem, model paddleocr), azure-di (ocr-http/azure-di, on_prem, models azure-di-layout-v4 etc., api_key_env), doc-classifier (classify) upstreams, with slot caps. README: Job API usage (submit/poll, idempotency, fan-out, queue fields).
  • [ ] Smoke script (scripts/): submit a job to a stub/echo ocr-http upstream, poll to completion, verify idempotent resubmit + JOB_UNKNOWN. Live-fire: against the real on-prem PaddleOCR if reachable from the dev box (else the stub), confirm an audit event with operation=ocr + doc_hash reaches the stream.
  • [ ] Update docs/CONVENTIONS.md only if a clarification is needed (version-bump per its rule). go test ./..., go vet, go build ./... green. Commit feat(p2): packaging + example OCR/classify upstreams + Job API smoke + docs; after the final whole-branch review + fix wave, merge to master, git tag p2-jobs-ocr. Write + upload the P2 exit report.

Self-review notes (plan-time)

  • Reuses P0 primitives rather than duplicating: the slot pool.Pool (admission/priority), the audit.Emitter (fire-and-forget + spool + body cap), the go-redis construction, the server drain gate + per-server prometheus registry, MetaFrom header parsing.
  • The Local/Redis Store split mirrors the proven LocalPool/RedisPool pattern (so coordination: local dev works with no Redis; redis gives HA + durability).
  • Slot-lease release is called out as the historically fragile area (P0 caught 4 leak variants) — Task 7 has an explicit Active()-returns-to-baseline test across all terminal states.
  • OCR documents are large: the audit body path relies on the EXISTING body_cap_bytes + truncation + full_body_sha256; doc_hash is the durable document identity (ties to the "audit documents without storing the bytes in the observatory" platform decision).
  • Fan-out is the riskiest task and is last, so it gets the most review scrutiny and can be deferred without blocking the single-job OCR path (which is what the OCR engine integration prompt actually needs first).
  • On-prem enforcement and idempotency are inherited from P0's behavior, applied to the new upstream types — not re-implemented.