Emil_Shanaty/SciMesh
1
0
Fork 0
Code Issues Pull Requests Actions 121 Packages Projects Releases Wiki Activity

Merge branch 'feat/coordinator'

Browse Source 
# Conflicts:
#	docs/api-contract.md
This commit is contained in:
Emil
2026-07-23 21:17:53 +03:00
parent abfda35170 8af8ddcf48
commit b4a89dd7c2
75 changed files with 8574 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/api-contract.md
+145 -96
View File
@@ -1,86 +1,150 @@
# SciMesh Coordinator API Contract
# SciMesh coordinator ↔ worker API contract (v1)
**Status:** draft, version 1. This document is the compatibility boundary
between the Go coordinator and the Python Worker. Change it only in the same
pull request as both implementation and contract tests.
**Status marker:** `v1`. This document is the single source of truth for the Go
coordinator and the Python Worker Daemon. It is derived from `PLAN.md` §5 and
must be updated in the same change as any behaviour it describes.
## General rules
> **Machine-readable:** [`openapi.yaml`](openapi.yaml) is the OpenAPI 3.0 mirror
> of this document — feed it to `openapi-python-client` or `datamodel-code-generator`
> to generate the Python client/models. This markdown stays the human-readable
> source; keep the two in sync.
- All worker endpoints require `Authorization: Bearer <token>`.
- Times use UTC RFC 3339, for example `2026-07-23T12:05:00Z`.
- JSON requests and responses use `application/json`.
- `worker_id` and `attempt` identify a lease. The coordinator validates them
transactionally on every task mutation.
- A task becomes `completed` only after a coordinator-owned artifact is durable.
- Identical repeated completion is successful; a different result for the same
attempt is a conflict.
- **Auth:** every endpoint except readiness requires `Authorization: Bearer <token>`.
- **Identity:** every mutating worker request carries `worker_id` and `attempt`;
they are checked against the current task lease in PostgreSQL. A stale attempt
gets `409`.
- **Timestamps:** UTC, RFC 3339 (e.g. `2026-07-22T12:05:00Z`).
- **Unknown JSON fields are rejected** with `400`.
## Worker registration
## Implementation status
| Endpoint | Contract | Coordinator |
| --- | --- | --- |
| `GET /health` | readiness incl. DB | ✅ done |
| `POST /workers/register` | register + capabilities | ✅ done |
| `POST /tasks/claim` | atomic lease | ✅ done |
| `POST /tasks/{id}/heartbeat` | renew lease | ✅ done |
| `POST /tasks/{id}/result` | complete | ✅ done, references `artifact_id` |
| `POST /tasks/{id}/failure` | fail | ✅ done |
| `GET /jobs/{id}` | progress | ✅ done |
| `PUT /tasks/{id}/artifacts/{name}` | upload partial | ✅ done |
| `GET /artifacts/{id}/download` | download by id | ✅ done |
| `POST /jobs/upload` | upload dataset, coordinator chunks it | ✅ done |
| `GET /tasks/{id}/input` | download shard | ✅ done |
---
## Readiness
```http
GET /health
```
`200 {"status":"ok"}` when the database is reachable; `503 {"status":"unavailable"}`
otherwise. Unauthenticated.
## Submit a dataset (submitter-side)
```http
POST /jobs/upload
Authorization: Bearer <token>
Content-Type: multipart/form-data
```
Fields, in order (text fields first, file last — the file is streamed):
`workload`, `parameters` (JSON), `chunk_rows` (int, default 1000), and the file
part `file`. The coordinator stores the input, splits the TSV into shard
artifacts (header repeated per shard), and creates one task per shard.
`201`:
```json
{ "job_id": "uuid", "task_count": 3, "input_artifact_id": "uuid" }
```
Each resulting task's claim response carries `input.uri = /tasks/{id}/input`,
served by §5.4.
## Register worker
```http
POST /workers/register
Authorization: Bearer <token>
Content-Type: application/json
{"name":"lab-worker-01","capabilities":["similarity-search"],"cpu_count":8,"memory_mb":16384}
```
Returns `200 OK`:
```json
{"worker_id":"uuid","heartbeat_interval_seconds":15}
```
## Task lifecycle
### Claim
```http
POST /tasks/claim
{"worker_id":"uuid","capabilities":["similarity-search"],"max_concurrency":1}
```
Returns `204 No Content` when no compatible task exists. A successful atomic
claim returns `200 OK`:
```json
{
"task_id":"uuid",
"attempt":1,
"lease_expires_at":"2026-07-23T12:05:00Z",
"workload":"similarity-search",
"input":{"uri":"https://coordinator.example/tasks/uuid/input","sha256":"hex-sha256"},
"parameters":{"query_id":"CHEMBL939","top_k":20}
"name": "lab-worker-01",
"capabilities": ["similarity-search", "similarity-graph"],
"cpu_count": 8,
"memory_mb": 16384
}
```
The claim is one PostgreSQL transaction using `FOR UPDATE SKIP LOCKED`.
`201`:
### Heartbeat
```json
{ "worker_id": "uuid", "heartbeat_interval_seconds": 15 }
```
`cpu_count`/`memory_mb` are accepted for forward compatibility and not yet
persisted. `capabilities` must be non-empty (an allowlisted workload set).
## Claim task
```http
POST /tasks/claim
Authorization: Bearer <token>
Content-Type: application/json
{ "worker_id": "uuid", "capabilities": ["similarity-search"], "max_concurrency": 1 }
```
- `204 No Content`: no compatible task.
- `200 OK`: a task is leased atomically.
```json
{
"task_id": "uuid",
"attempt": 1,
"lease_expires_at": "2026-07-22T12:05:00Z",
"workload": "similarity-search",
"input": { "uri": "https://coordinator/tasks/uuid/input", "sha256": "hex" },
"parameters": { "query_id": "CHEMBL939", "top_k": 20 }
}
```
`max_concurrency` is accepted; the coordinator leases one task per call for now.
## Renew lease (heartbeat)
```http
POST /tasks/{task_id}/heartbeat
Authorization: Bearer <token>
Content-Type: application/json
{"worker_id":"uuid","attempt":1}
{ "worker_id": "uuid", "attempt": 1 }
```
Returns `200 OK` and the renewed deadline:
Response **must** contain a renewed deadline:
```json
{"lease_expires_at":"2026-07-23T12:10:00Z"}
{ "lease_expires_at": "2026-07-22T12:10:00Z" }
```
The Worker schedules its next heartbeat before half of the returned TTL.
The worker schedules the next heartbeat before half of the returned TTL, never
on a fixed interval alone.
### Input download
## Download input or shard (CTX-05)
`GET /tasks/{task_id}/input` returns the claimed task input. The Worker verifies
its SHA-256 before execution. On a redirect to another origin, it removes the
coordinator bearer token.
`GET /tasks/{task_id}/input` returns the artifact owned by the current task. The
worker verifies its SHA-256 before execution. If the URI redirects to another
origin, the worker removes the coordinator bearer token.
## Artifact upload
## Upload a partial artifact (CTX-05)
```http
PUT /tasks/{task_id}/artifacts/{filename}
Authorization: Bearer <token>
Content-Type: text/csv
X-Worker-ID: uuid
X-Task-Attempt: 1
@@ -88,64 +152,49 @@ X-Task-Attempt: 1
<streamed bytes>
```
The coordinator streams the body to storage, checks lease ownership, records
the checksum and returns `201 Created`:
`200`:
```json
{
"artifact_id":"uuid",
"uri":"https://coordinator.example/artifacts/uuid/download",
"sha256":"hex-sha256",
"size_bytes":1234
}
{ "artifact_id": "uuid", "uri": "https://coordinator/artifacts/uuid/download",
"sha256": "hex", "size_bytes": 1234 }
```
The returned URI is the only URI the Worker may send in task completion.
`worker://` and `file://` are invalid.
## Completion and failure
## Complete or fail task
```http
POST /tasks/{task_id}/result
Authorization: Bearer <token>
Content-Type: application/json
{
"worker_id":"uuid",
"attempt":1,
"result":{
"artifact_id":"uuid",
"uri":"https://coordinator.example/artifacts/uuid/download",
"sha256":"hex-sha256",
"content_type":"text/csv"
},
"metrics":{"elapsed_seconds":12.4,"processed_rows":10000}
"worker_id": "uuid",
"attempt": 1,
"result": { "artifact_id": "uuid", "sha256": "hex", "content_type": "text/csv" },
"metrics": { "elapsed_seconds": 12.4, "processed_rows": 10000 }
}
```
The coordinator returns `200`, `201`, or `202` for a valid completion. It must
verify that the artifact belongs to that task and attempt before completing it.
The worker uploads its partial result first (§5.5), then completes with that
`artifact_id`. The coordinator verifies the artifact was stored for this exact
task before accepting it — a worker cannot complete one task with another task's
artifact. No worker-supplied URI is ever persisted.
Use `POST /tasks/{task_id}/failure` only for a failed attempt:
```json
{"worker_id":"uuid","attempt":1,"error_code":"ValueError","error_message":"input checksum mismatch"}
```http
POST /tasks/{task_id}/failure
```
Messages are sanitised: no token, traceback, absolute local path, or raw input.
Same identity fields, plus sanitized `error_code`, `error_message`, `retryable`.
Never a traceback, token, or absolute worker path.
## Error responses
## Idempotency and errors
| Situation | Response |
| --- | --- |
| Invalid JSON, field, or parameter | `400 Bad Request` |
| Missing or invalid authentication | `401 Unauthorized` / `403 Forbidden` |
| Worker/attempt does not own an active lease | `409 Conflict` |
| Artifact does not belong to the task/attempt | `409 Conflict` |
| Same attempt, different completion manifest | `409 Conflict` |
| Unexpected coordinator failure | `500` without internal details |
## Compatibility tests
Contract tests must cover: registration, `204` claim, successful claim,
heartbeat renewal, foreign worker and stale attempt conflicts, streamed upload,
checksum mismatch, success after upload, failure through `/failure`, and
idempotent completion.
| No compatible task | `204` |
| Worker/attempt does not own lease | `409` |
| Artifact does not belong to task/attempt | `409` |
| Same completion, same manifest | `200` idempotent |
| Same attempt, different manifest | `409` |
| Invalid parameters/input | `400` |
| Auth failure | `401` |
| Unknown job/task | `404` |
docs/building-workers.md
+221
View File
@@ -0,0 +1,221 @@
# Building a SciMesh worker
A worker is a process that pulls tasks from the coordinator, runs them, and
returns results. It talks to the coordinator **only over HTTP** — it never sees
the database, and it needs no inbound port (all requests are outbound). This
guide is what you need to implement one (the reference is a Python daemon, but
nothing here is Python-specific).
**Read alongside:**
[`api-contract.md`](api-contract.md) (the contract in prose) and
[`openapi.yaml`](openapi.yaml) (machine-readable — generate a typed client from
it, see the bottom).
---
## The one loop
A worker is essentially this loop:
```text
register once
loop forever:
task = POST /tasks/claim
if no task (204): sleep, continue
download the task's input, verify its checksum
run the workload ── while running, POST heartbeat before the lease expires
upload the result artifact (PUT)
POST /tasks/{id}/result with the artifact id
on any failure: POST /tasks/{id}/failure
```
Everything below fills in the details.
## 0. Auth
Every request except `GET /health` carries a shared bearer token:
```
Authorization: Bearer <COORDINATOR_TOKEN>
```
The token is handed to you out of band (env var / secret) — the same string the
coordinator was started with. Never log it, never send it in an error body.
## 1. Register (once, at startup)
```http
POST /workers/register
{ "name": "lab-worker-01", "capabilities": ["similarity_search"] }
```
Response: `{ "worker_id": "<uuid>", "heartbeat_interval_seconds": 15 }`.
- `capabilities` are the workload names you can run — the coordinator only hands
you matching tasks.
- **Keep `worker_id`**. Use it as your identity in every later call. Using the
registered UUID is what lets the coordinator track your liveness (it marks
workers offline after they go silent).
## 2. Claim a task
```http
POST /tasks/claim
{ "worker_id": "<uuid>", "capabilities": ["similarity_search"] }
```
- `200` → a leased task (below).
- `204` → nothing to do; back off a little and poll again.
```json
{
"task_id": "<uuid>",
"attempt": 1,
"lease_expires_at": "2026-07-22T12:05:00Z",
"workload": "similarity_search",
"input": { "uri": "/tasks/<uuid>/input", "sha256": "<hex>" },
"parameters": { "query_id": "CHEMBL939", "top_k": 20 }
}
```
**`attempt` matters.** Every later call for this task must echo the exact
`attempt` you were handed. A task requeued after a lost lease comes back with a
higher attempt; an old attempt is rejected with `409`.
## 3. Download the input, verify it
```http
GET {input.uri} # e.g. GET /tasks/<uuid>/input
```
Stream it to disk and **check the SHA-256 against `input.sha256`** before
running. A mismatch means a corrupt shard — fail the task with a clear code,
don't process garbage.
> If `input.uri` ever redirects to another host (object storage), **strip the
> `Authorization` header** on the redirect — never send the coordinator token to
> a third party.
## 4. Run — and heartbeat while you run
Long tasks must prove they are alive, or the coordinator's reaper reclaims the
lease and hands the task to someone else.
```http
POST /tasks/{task_id}/heartbeat
{ "worker_id": "<uuid>", "attempt": 1 }
```
Response: `{ "lease_expires_at": "<new deadline>" }`.
- Schedule the next heartbeat at **less than half** the remaining TTL — don't
rely on a fixed interval. If `lease_expires_at` is 2 minutes out, heartbeat
every ~45s.
- The first heartbeat also moves the task from `leased` to `running` on the
server; you don't have to do anything special for that.
If you miss the deadline, your lease expires: a later `heartbeat`/`result` will
come back `409`, and the task is already back in the queue.
## 5. Upload the result artifact
The coordinator owns results — you upload the bytes, it stores them and computes
the checksum. Identity travels in **headers** here, not the body:
```http
PUT /tasks/{task_id}/artifacts/result.csv
Content-Type: text/csv
X-Worker-ID: <uuid>
X-Task-Attempt: 1
<streamed result bytes>
```
Response: `{ "artifact_id": "<uuid>", "uri": "...", "sha256": "<hex>", "size_bytes": 1234 }`.
Keep the returned `artifact_id`.
## 6. Complete the task
```http
POST /tasks/{task_id}/result
{ "worker_id": "<uuid>", "attempt": 1,
"result": { "artifact_id": "<uuid>" },
"metrics": { "elapsed_seconds": 12.4, "processed_rows": 10000 } }
```
- Reference the `artifact_id` you just uploaded **for this task**. The
coordinator verifies it belongs to this task; another task's artifact → `409`.
- **Idempotent:** if your network dropped and you retry the same `artifact_id`,
you get `200` again, not a conflict. Safe to retry.
## 7. …or fail it
```http
POST /tasks/{task_id}/failure
{ "worker_id": "<uuid>", "attempt": 1,
"error_code": "download_failed", "error_message": "checksum mismatch",
"retryable": true }
```
- `retryable: true` → the task returns to the queue while attempts remain (a new
worker gets it at a higher `attempt`).
- `retryable: false` → it fails terminally.
- Send only a short, sanitized `error_code`/`error_message`. **Never** a Python
traceback, a token, or an absolute local path.
---
## Status-code cheat sheet
| Code | Meaning for the worker |
| --- | --- |
| `204` | claim: queue empty — back off and retry |
| `400` | your request is malformed (bad UUID, unknown field) |
| `401` | bad/missing token |
| `404` | task/job/artifact doesn't exist |
| `409` | you don't hold the lease, or your `attempt` is stale, or a different result was already recorded — **stop working on this task**, it's no longer yours |
A `409` is normal, not a crash: it means the coordinator gave the task to
someone else (usually because your lease expired). Log it and move on to the
next claim.
## Config the worker should expose
Per the worker contract, at minimum:
- `SCIMESH_COORDINATOR_URL` (e.g. `http://coordinator:8080`)
- `SCIMESH_WORKER_ID` (or derive from hostname)
- the bearer token
- poll interval and request timeout
- a working directory for downloaded inputs and generated outputs
## Generate a client from the spec
Instead of hand-writing request code, generate it:
```sh
# typed async client
openapi-python-client generate --path docs/openapi.yaml
# or just the Pydantic models
datamodel-codegen --input docs/openapi.yaml --output scimesh_models.py
```
## Try the endpoints by hand first
`coordinator/api/requests.http` walks the whole flow one request at a time
(register → claim → heartbeat → upload → result), and
`coordinator/scripts/smoke.sh` runs it end to end. Read those to see real
request/response bodies before writing code.
## The rules you must not break
1. Never touch the database — HTTP only.
2. Every mutating call carries `worker_id` **and** `attempt`.
3. Verify the input checksum before running.
4. Upload the result artifact **before** calling `/result`.
5. Never persist a `worker://` or local path as a result — the coordinator owns
artifacts.
6. Strip the bearer token on any cross-origin redirect.
7. Sanitize error output — no tracebacks, tokens, or absolute paths.
docs/openapi.yaml
+552
View File
@@ -0,0 +1,552 @@
openapi: 3.0.3
info:
title: SciMesh Coordinator API
version: 1.0.0
description: >
Durable task-queue server for SciMesh. Workers register, claim tasks one at a
time, heartbeat, upload partial-result artifacts, and complete or fail tasks.
Submitters create jobs — either with pre-chunked input URIs or by uploading a
dataset the coordinator chunks itself.
Machine-readable mirror of `docs/api-contract.md` (v1). All timestamps are
UTC, RFC 3339. Every endpoint except `GET /health` requires a bearer token.
Unknown JSON fields are rejected with 400.
servers:
- url: "{scheme}://{host}"
variables:
scheme:
default: http
enum: [http, https]
host:
default: localhost:8080
security:
- bearerAuth: []
tags:
- name: health
- name: workers
- name: jobs
- name: tasks
- name: artifacts
paths:
/health:
get:
tags: [health]
summary: Readiness (probes the database)
security: []
responses:
"200":
description: The coordinator and its database are ready.
content:
application/json:
schema: { $ref: "#/components/schemas/Health" }
"503":
description: The database is unreachable.
content:
application/json:
schema: { $ref: "#/components/schemas/Health" }
/workers/register:
post:
tags: [workers]
summary: Register a worker
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/RegisterRequest" }
responses:
"201":
description: Registered.
content:
application/json:
schema: { $ref: "#/components/schemas/RegisterResponse" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
/jobs:
post:
tags: [jobs]
summary: Create a job from pre-chunked input URIs
description: >
The submitter supplies each chunk's input URI and checksum. To have the
coordinator split a dataset instead, use `POST /jobs/upload`.
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/CreateJobRequest" }
responses:
"201":
description: Job and its tasks were created transactionally.
content:
application/json:
schema: { $ref: "#/components/schemas/JobCreated" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
/jobs/upload:
post:
tags: [jobs]
summary: Upload a dataset; the coordinator chunks it into shard tasks
description: >
multipart/form-data. The text fields (`workload`, `parameters`,
`chunk_rows`) MUST precede the `file` part: the file is streamed, not
buffered, so the fields have to be parsed before it arrives.
requestBody:
required: true
content:
multipart/form-data:
schema: { $ref: "#/components/schemas/UploadJobForm" }
encoding:
file:
contentType: text/tab-separated-values
responses:
"201":
description: Job, input artifact, shard artifacts, and shard tasks created.
content:
application/json:
schema: { $ref: "#/components/schemas/UploadJobResponse" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
/jobs/{job_id}:
get:
tags: [jobs]
summary: Aggregate job progress
parameters:
- $ref: "#/components/parameters/JobID"
responses:
"200":
description: Progress counts and derived status.
content:
application/json:
schema: { $ref: "#/components/schemas/JobProgress" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
/tasks/claim:
post:
tags: [tasks]
summary: Atomically lease one task
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/ClaimRequest" }
responses:
"200":
description: A task was leased.
content:
application/json:
schema: { $ref: "#/components/schemas/ClaimedTask" }
"204":
description: No compatible task is available.
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
/tasks/{task_id}/heartbeat:
post:
tags: [tasks]
summary: Renew the caller's lease
description: >
The response carries a renewed `lease_expires_at`. Schedule the next
heartbeat before half of the remaining TTL, never on a fixed interval alone.
parameters:
- $ref: "#/components/parameters/TaskID"
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/IdentityRequest" }
responses:
"200":
description: Lease renewed.
content:
application/json:
schema: { $ref: "#/components/schemas/ClaimedTask" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
"409": { $ref: "#/components/responses/Conflict" }
/tasks/{task_id}/input:
get:
tags: [tasks]
summary: Download the task's input shard
description: >
Streams the shard bytes for an uploaded-dataset task. The worker verifies
the `X-Checksum-SHA256` header (also delivered as `input.sha256` on claim)
before executing. URI-based tasks have no coordinator-stored input and
return 404.
parameters:
- $ref: "#/components/parameters/TaskID"
responses:
"200":
description: The shard bytes.
headers:
X-Checksum-SHA256:
schema: { type: string }
description: SHA-256 of the shard.
content:
application/octet-stream:
schema: { type: string, format: binary }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
/tasks/{task_id}/artifacts/{filename}:
put:
tags: [tasks, artifacts]
summary: Upload a partial-result artifact
description: >
Streams the body into blob storage. Identity travels in headers, not the
body. The coordinator measures the size and SHA-256 itself and returns them.
parameters:
- $ref: "#/components/parameters/TaskID"
- name: filename
in: path
required: true
schema: { type: string }
- name: X-Worker-ID
in: header
required: true
schema: { type: string }
- name: X-Task-Attempt
in: header
required: true
schema: { type: integer }
requestBody:
required: true
content:
application/octet-stream:
schema: { type: string, format: binary }
text/csv:
schema: { type: string, format: binary }
responses:
"200":
description: Artifact stored.
content:
application/json:
schema: { $ref: "#/components/schemas/ArtifactUploaded" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
"409": { $ref: "#/components/responses/Conflict" }
/tasks/{task_id}/result:
post:
tags: [tasks]
summary: Complete a task with an uploaded result artifact
description: >
References an artifact previously uploaded for THIS task. The coordinator
verifies ownership before accepting it. Idempotent: replaying the same
artifact_id succeeds; a different one for a completed task is a 409.
parameters:
- $ref: "#/components/parameters/TaskID"
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/ResultRequest" }
responses:
"200":
description: Recorded.
content:
application/json:
schema: { $ref: "#/components/schemas/TaskState" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
"409": { $ref: "#/components/responses/Conflict" }
/tasks/{task_id}/failure:
post:
tags: [tasks]
summary: Report a task failure
description: >
`retryable: true` returns the task to the queue while attempts remain;
otherwise it fails terminally. Send only sanitized error fields — never a
traceback, token, or absolute worker path.
parameters:
- $ref: "#/components/parameters/TaskID"
requestBody:
required: true
content:
application/json:
schema: { $ref: "#/components/schemas/FailureRequest" }
responses:
"200":
description: Recorded.
content:
application/json:
schema: { $ref: "#/components/schemas/TaskState" }
"400": { $ref: "#/components/responses/BadRequest" }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
"409": { $ref: "#/components/responses/Conflict" }
/artifacts/{artifact_id}/download:
get:
tags: [artifacts]
summary: Download an artifact by id
parameters:
- name: artifact_id
in: path
required: true
schema: { type: string, format: uuid }
responses:
"200":
description: The artifact bytes.
headers:
X-Checksum-SHA256:
schema: { type: string }
content:
application/octet-stream:
schema: { type: string, format: binary }
"401": { $ref: "#/components/responses/Unauthorized" }
"404": { $ref: "#/components/responses/NotFound" }
components:
securitySchemes:
bearerAuth:
type: http
scheme: bearer
parameters:
JobID:
name: job_id
in: path
required: true
schema: { type: string, format: uuid }
TaskID:
name: task_id
in: path
required: true
schema: { type: string, format: uuid }
responses:
BadRequest:
description: Invalid input.
content:
application/json:
schema: { $ref: "#/components/schemas/Error" }
Unauthorized:
description: Missing or invalid bearer token.
content:
application/json:
schema: { $ref: "#/components/schemas/Error" }
NotFound:
description: The referenced job, task, or artifact does not exist.
content:
application/json:
schema: { $ref: "#/components/schemas/Error" }
Conflict:
description: Lease not held, stale attempt, or a different result already recorded.
content:
application/json:
schema: { $ref: "#/components/schemas/Error" }
schemas:
Health:
type: object
properties:
status: { type: string, example: ok }
Error:
type: object
properties:
error: { type: string, example: "invalid input" }
request_id: { type: string, description: Correlates with the server logs. }
RegisterRequest:
type: object
required: [capabilities]
properties:
name: { type: string, example: lab-worker-01 }
capabilities:
type: array
minItems: 1
items: { type: string }
example: [similarity_search, similarity_graph]
cpu_count:
type: integer
description: Accepted for forward compatibility; not yet persisted.
memory_mb:
type: integer
description: Accepted for forward compatibility; not yet persisted.
RegisterResponse:
type: object
properties:
worker_id: { type: string, format: uuid }
heartbeat_interval_seconds: { type: integer, example: 15 }
ChunkSpec:
type: object
required: [chunk_index, input_uri, input_sha256]
properties:
chunk_index: { type: integer }
workload:
type: string
description: Empty inherits the job's workload.
input_uri: { type: string }
input_sha256: { type: string }
parameters: { type: object, additionalProperties: true }
max_attempts: { type: integer }
CreateJobRequest:
type: object
required: [workload, input_uri, chunks]
properties:
workload: { type: string, example: similarity_search }
input_uri: { type: string }
parameters: { type: object, additionalProperties: true }
chunks:
type: array
minItems: 1
items: { $ref: "#/components/schemas/ChunkSpec" }
JobCreated:
type: object
properties:
id: { type: string, format: uuid }
status: { $ref: "#/components/schemas/JobStatus" }
UploadJobForm:
type: object
required: [workload, file]
properties:
workload: { type: string, example: similarity_search }
parameters:
type: string
description: JSON object, sent as a string form field.
example: '{"top_k":10}'
chunk_rows:
type: integer
description: Data rows per shard. Default 1000.
example: 1000
file:
type: string
format: binary
description: The dataset (TSV; header repeated into each shard).
UploadJobResponse:
type: object
properties:
job_id: { type: string, format: uuid }
task_count: { type: integer, example: 3 }
input_artifact_id: { type: string, format: uuid }
JobProgress:
type: object
properties:
id: { type: string, format: uuid }
status: { $ref: "#/components/schemas/JobStatus" }
total: { type: integer }
pending: { type: integer }
leased: { type: integer }
completed: { type: integer }
failed: { type: integer }
ClaimRequest:
type: object
required: [worker_id]
properties:
worker_id: { type: string }
capabilities:
type: array
items: { type: string }
description: Workloads this worker can run. Empty means "any".
max_concurrency:
type: integer
description: Accepted; the coordinator leases one task per call.
InputRef:
type: object
properties:
uri:
type: string
description: >
For an uploaded shard, a coordinator path `/tasks/{id}/input`. For a
URI-based task, the external input URI.
sha256: { type: string }
ClaimedTask:
type: object
properties:
task_id: { type: string, format: uuid }
job_id: { type: string, format: uuid }
chunk_index: { type: integer }
workload: { type: string }
input: { $ref: "#/components/schemas/InputRef" }
parameters: { type: object, additionalProperties: true }
attempt: { type: integer }
lease_expires_at: { type: string, format: date-time }
IdentityRequest:
type: object
required: [worker_id, attempt]
properties:
worker_id: { type: string }
attempt: { type: integer }
ResultManifest:
type: object
required: [artifact_id]
properties:
artifact_id:
type: string
format: uuid
description: An artifact previously uploaded for this task.
sha256:
type: string
description: Accepted for the worker's own cross-check; the coordinator trusts its stored metadata.
content_type: { type: string }
ResultRequest:
type: object
required: [worker_id, attempt, result]
properties:
worker_id: { type: string }
attempt: { type: integer }
result: { $ref: "#/components/schemas/ResultManifest" }
metrics: { type: object, additionalProperties: true }
FailureRequest:
type: object
required: [worker_id, attempt, error_code]
properties:
worker_id: { type: string }
attempt: { type: integer }
error_code: { type: string, example: download_failed }
error_message: { type: string }
retryable: { type: boolean }
ArtifactUploaded:
type: object
properties:
artifact_id: { type: string, format: uuid }
uri:
type: string
description: Coordinator download path, `/artifacts/{id}/download`.
sha256: { type: string }
size_bytes: { type: integer, format: int64 }
TaskState:
type: object
properties:
id: { type: string, format: uuid }
job_id: { type: string, format: uuid }
status: { $ref: "#/components/schemas/TaskStatus" }
JobStatus:
type: string
enum: [pending, running, completed, failed, cancelled]
TaskStatus:
type: string
enum: [pending, leased, completed, failed, cancelled]
docs/worker-agent-brief.md
+83
View File
@@ -0,0 +1,83 @@
# Brief: build the SciMesh worker against the coordinator
You are implementing the **worker side**. The **coordinator** (Go/PostgreSQL) is
already built, tested, and running on branch `feat/coordinator`. This brief tells
you what exists, where the contract is, and what to deliver.
## What the coordinator already does (done — do not reimplement)
A durable task-queue server. Over HTTP only (workers never touch the database):
- **Worker registry** — `POST /workers/register` returns a `worker_id`; the
coordinator tracks liveness and marks silent workers offline.
- **Jobs** — created from chunk URIs (`POST /jobs`) or by uploading a dataset
(`POST /jobs/upload`), which the coordinator splits into shard tasks itself.
- **Queue** — atomic claim (`FOR UPDATE SKIP LOCKED`), leases, heartbeats
(`leased → running`), a reaper that requeues expired leases, retry budget.
- **Artifacts** — the worker uploads a partial result (`PUT`), the coordinator
stores it (streamed, checksummed) and owns it; completion references an
`artifact_id`, not a worker URI.
- **Input delivery** — `GET /tasks/{id}/input` streams a task's shard.
Full endpoint list and status: `coordinator/README.md`.
## The contract (read these first)
| File | What it is |
| --- | --- |
| `docs/openapi.yaml` | OpenAPI 3.0 — **generate your client from this** |
| `docs/building-workers.md` | step-by-step guide: the claim→heartbeat→upload→complete loop, auth, lease semantics, status codes, and the rules you must not break |
| `docs/api-contract.md` | the same contract in prose |
| `coordinator/api/requests.http` | real request/response examples for every endpoint |
Generate a typed client instead of hand-writing HTTP:
```sh
openapi-python-client generate --path docs/openapi.yaml
# or just models:
datamodel-codegen --input docs/openapi.yaml --output scimesh_models.py
```
## Run the coordinator locally to develop against it
```sh
cd coordinator && docker compose up -d # listens on :8080, migrations auto-applied
make smoke # exercises the whole flow (should pass)
```
Auth: every request except `GET /health` needs `Authorization: Bearer <token>`
(the compose default is `dev-token`; check `coordinator/.env.example`).
## Your deliverable (CTX-06 in PLAN.md)
A worker daemon that:
1. registers at startup and reuses its `worker_id`;
2. claims one task at a time; backs off on `204`;
3. downloads the input via `input.uri` and **verifies its `sha256`** before running;
4. heartbeats before half the lease TTL elapses;
5. uploads the result artifact, then completes the task with that `artifact_id`;
6. reports failures to `/failure` with sanitized error fields;
7. is configured by env: coordinator URL, worker id, token, poll interval, work dir.
## Acceptance criteria
- A worker registers, claims a shard, downloads and checksum-verifies its input,
heartbeats through a long run, uploads a result, and completes it — end to end
against the real coordinator.
- A lost lease (missed heartbeats) surfaces as a clean `409` and the worker moves
on rather than crashing.
- No result ever references a `worker://` or local path — only uploaded artifacts.
- Contract tests run the worker against the real Go coordinator + Postgres in CI.
## Rules you must not break
1. HTTP only — never the database.
2. Every mutating call carries `worker_id` **and** `attempt`; a stale attempt is `409`.
3. Verify the input checksum before executing.
4. Upload the result artifact **before** calling `/result`.
5. Strip the bearer token on any cross-origin redirect.
6. Sanitize errors — never send a traceback, token, or absolute path.
Anything about the coordinator's behavior that isn't clear here is answered by
`docs/openapi.yaml` (authoritative shapes) and `docs/building-workers.md`.