diff --git a/docs/ctx-07-distributed-workload-protocol.md b/docs/ctx-07-distributed-workload-protocol.md new file mode 100644 index 0000000..c8022c2 --- /dev/null +++ b/docs/ctx-07-distributed-workload-protocol.md @@ -0,0 +1,207 @@ +# CTX-07: distributed workload protocol and planner contract + +## Status and scope + +This document is the implementation contract for CTX-07. It does not implement +a planner, reducer, API endpoint, database migration, or final artifact. Until +CTX-08 and CTX-09 are complete, shard CSVs remain diagnostic partial results. + +The protocol gives local scientific workloads a coordinator-independent way to +validate a job, plan artifact-backed tasks, and later reduce completed outputs. +The Go coordinator owns durable artifacts, transactions, task rows, leases, and +HTTP. A Python workload must never access PostgreSQL or call the coordinator. + +Read `PLAN.md`, `.agents/workloads.md`, and `docs/api-contract.md` before +implementing this CTX. + +## Canonical vocabulary + +- External workload names are lowercase hyphenated names: `similarity-search` + and, later, `similarity-graph`. +- The existing underscore spellings are a temporary compatibility alias at the + Python worker boundary only. Planners, persisted job/task payloads, and new + API examples use the canonical hyphenated spelling. +- A **plan** contains only JSON-compatible values and coordinator artifact + references. It contains no local filesystem path, worker URI, presigned URL, + database connection, or callable. +- `chunk_index` is a non-negative integer, unique within a plan, and sorted + ascending whenever results are enumerated. + +## Python boundary + +CTX-07 adds a small `DistributedWorkload` protocol under `scimesh/distributed/` +and a registry separate from the local CLI registry. Names below are proposed +public types; keep concrete implementation details minimal. + +```python +class DistributedWorkload(Protocol): + name: str + + def validate_job(self, parameters: Mapping[str, object]) -> None: ... + + def plan( + self, + input_path: Path, + input_artifact_id: str, + parameters: Mapping[str, object], + shard_rows: int, + workspace: Path, + ) -> DistributedPlan: ... + + def reduce( + self, + partial_results: Sequence[CompletedPartial], + parameters: Mapping[str, object], + workspace: Path, + ) -> FinalResult: ... +``` + +`input_path` and `workspace` are temporary files supplied by the coordinator +bridge. They are never serialized. `plan()` returns only a `DistributedPlan`; +the bridge validates it, persists artifact/task rows in one coordinator +transaction, and removes its temporary workspace. If validation or planning +fails, no job or task may be written. + +## JSON models + +All objects below are schema version `1`. Future incompatible changes require a +new version; never infer a schema from missing fields. + +### Artifact reference + +```json +{ + "artifact_id": "c4273293-f8b4-4ecb-99df-3b9f5a32b6a6", + "sha256": "3b2d...64-lowercase-hex-characters", + "content_type": "text/tab-separated-values" +} +``` + +The artifact ID is coordinator-owned. The checksum is included so planning and +tests can assert exactly which immutable input was used. A worker receives the +coordinator-generated download URI only through `POST /tasks/claim`. + +### Distributed plan + +```json +{ + "schema_version": 1, + "workload": "similarity-search", + "resolved_parameters": { + "query_smiles": "COc1ccc(Nc2ncnc3cc(OCCCN4CCOCC4)c(OC)c23)cc1", + "query_source": {"kind": "chembl_id", "value": "CHEMBL939"}, + "top_k": 20, + "threshold": 0.7, + "threshold_direction": "greater", + "fingerprint": {"algorithm": "morgan", "radius": 2, "fp_size": 2048} + }, + "tasks": [ + { + "chunk_index": 0, + "input_artifact": { + "artifact_id": "69e41105-d9fb-4c7f-a2db-7dd9e3ba2c76", + "sha256": "4c92...64-lowercase-hex-characters", + "content_type": "text/tab-separated-values" + }, + "parameters": { + "query_smiles": "COc1ccc(Nc2ncnc3cc(OCCCN4CCOCC4)c(OC)c23)cc1", + "top_k": 20, + "threshold": 0.7, + "threshold_direction": "greater", + "fingerprint": {"algorithm": "morgan", "radius": 2, "fp_size": 2048} + } + } + ] +} +``` + +`resolved_parameters` are immutable job metadata. A task copies only the +values required by its worker runner. The coordinator may add its own durable +task ID and generated input URI; it must not alter scientific parameters. + +## Similarity-search planning rules + +1. Accept exactly one of `query_id` and `query_smiles` at the public boundary. +2. Validate a supplied SMILES once. For `query_id`, find and validate that + molecule once against the original uploaded TSV **before** creating shards. +3. Persist the resolved canonical query SMILES and the original query source in + `resolved_parameters`. Workers receive `query_smiles`, never `query_id`. +4. Fingerprint settings are fixed to Morgan radius `2` and `fp_size` `2048`. + Reject a request that tries to override them rather than silently changing + scientific semantics. +5. Split source rows in input order. Every shard includes the original TSV + header and has a contiguous, zero-based `chunk_index`. +6. Each shard uses the global `top_k`, not a smaller local limit. A global + reducer cannot recover a candidate discarded by every shard. +7. Preserve `threshold`, `threshold_direction`, and valid `max_rows` semantics + in the resolved plan. A job-level row limit is applied before sharding, not + independently by every worker. + +Invalid row SMILES are not planner failures. They remain shard data and are +counted by the worker exactly as the local workload does. An invalid query is a +planning failure. + +## Partial-result contract + +A completed similarity-search task owns exactly one coordinator-uploaded CSV +artifact with content type `text/csv` and these columns, in this order: + +```csv +rank,chembl_id,canonical_smiles,similarity +1,CHEMBL123,CCO,0.875000 +``` + +- `rank` is one-based local rank. +- `similarity` uses the local CLI's six-decimal formatting. +- Rows are sorted by `(-similarity, chembl_id, canonical_smiles)` for + `threshold_direction=greater`, or `(similarity, chembl_id, + canonical_smiles)` for `less`. +- The query molecule and every candidate with the same canonical query SMILES + are excluded using the existing local-workload definition. +- Empty valid result files still include the header. + +The worker completion metrics must include JSON numbers for `scanned_rows`, +`valid_molecules`, `invalid_smiles`, `matches_emitted`, and +`elapsed_seconds`. Metrics are observability data; the reducer derives final +scientific output exclusively from coordinator-owned partial artifacts. + +## Reduction boundary + +CTX-09 invokes the registered reducer only after every task is completed. It +passes `CompletedPartial` values ordered by `chunk_index`, each containing its +coordinator artifact reference and validated metrics. + +For similarity-search the reducer: + +1. reads partial CSVs in `chunk_index` order; +2. validates header, row shape, rank, finite similarity in `[0, 1]`, and sort + order; +3. retains a bounded heap of at most the global `top_k` candidates using the + exact local ranking key; +4. writes the same header and deterministic rank numbering as the local CLI. + +It must not deduplicate ordinary records: the local reference keeps input-row +multiplicity. Reduction is independent of worker completion order and uses +`O(top_k + shard_rows)` memory apart from CSV streaming buffers. + +## Required tests for the CTX-07 implementation + +- unknown workload is rejected before any coordinator job/task write; +- invalid public parameters and invalid `query_id` produce no partial plan; +- `query_id` resolution occurs once, before shard construction; +- the same input, parameters, and shard size generate byte-equivalent + JSON plans and identical shard order; +- every task payload is JSON-serializable and contains only artifact references + and validated scalar/object values; +- a two-shard dummy workload proves coordinator transaction rollback on planner + validation failure; +- completed partial artifacts reach the reducer ordered by `chunk_index`, even + when workers finish in a different order; +- the protocol registry never imports the Go coordinator or database code. + +## Deferred work + +CTX-08 implements the similarity-search planner, runner adapter, reducer, and +comparison against the local CLI. CTX-09 persists the final artifact and job +state. CTX-10 defines graph-specific triangular block plans; it must not reuse +the search shard scheme without its pair-coverage invariants.