From 749396da051e008bbd150da3333367495f722106 Mon Sep 17 00:00:00 2001 From: Emil Date: Sun, 2 Aug 2026 17:47:47 +0300 Subject: [PATCH] Drive coordinator upload and reduction from the workload catalog --- coordinator/Makefile | 2 +- coordinator/cmd/coordinator/main.go | 21 +- coordinator/internal/reducer/concat.go | 57 +++ coordinator/internal/reducer/concat_test.go | 64 +++ .../internal/transport/http/server_test.go | 12 +- .../transport/http/templates/add-worker.html | 2 +- .../transport/http/templates/dashboard.html | 6 +- .../transport/http/templates/job.html | 6 +- .../transport/http/templates/new-job.html | 26 +- .../transport/http/testcatalog_test.go | 15 + coordinator/internal/transport/http/ui.go | 145 ++++++- .../transport/http/ui_newjob_internal_test.go | 44 +++ .../internal/transport/http/ui_workloads.go | 47 +-- coordinator/internal/usecase/job.go | 15 +- coordinator/internal/usecase/reduce.go | 27 +- coordinator/internal/usecase/task.go | 39 +- .../internal/usecase/testcatalog_test.go | 62 +++ coordinator/internal/usecase/ui.go | 78 +++- .../internal/usecase/ui_internal_test.go | 37 +- coordinator/internal/usecase/ui_scope_test.go | 2 +- .../internal/usecase/ui_workers_scope_test.go | 2 +- coordinator/internal/usecase/upload.go | 90 ++--- coordinator/internal/usecase/usecase_test.go | 62 ++- coordinator/internal/workloads/catalog.go | 273 +++++++++++++ coordinator/internal/workloads/schema.go | 367 ++++++++++++++++++ .../http => workloads}/workloads.json | 156 +++++++- 26 files changed, 1473 insertions(+), 184 deletions(-) create mode 100644 coordinator/internal/reducer/concat.go create mode 100644 coordinator/internal/reducer/concat_test.go create mode 100644 coordinator/internal/transport/http/testcatalog_test.go create mode 100644 coordinator/internal/transport/http/ui_newjob_internal_test.go create mode 100644 coordinator/internal/usecase/testcatalog_test.go create mode 100644 coordinator/internal/workloads/catalog.go create mode 100644 coordinator/internal/workloads/schema.go rename coordinator/internal/{transport/http => workloads}/workloads.json (73%) diff --git a/coordinator/Makefile b/coordinator/Makefile index 186b0e0..835f360 100644 --- a/coordinator/Makefile +++ b/coordinator/Makefile @@ -29,7 +29,7 @@ DEMO_DIR ?= .demo # workloads.json is the UI workload catalog, generated from the Python SDK # workload library. It is checked in so the binary embeds it; regenerate it # whenever workloads or their manifests change (requires the Python venv). -WORKLOADS_JSON := internal/transport/http/workloads.json +WORKLOADS_JSON := internal/workloads/workloads.json # The Go worker agent: a static coordinator client that executes SDK # workloads in a Python subprocess per claimed task. diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 32796c7..3607442 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -14,6 +14,7 @@ import ( "github.com/emil28092005/SciMesh/coordinator/internal/storage/postgres" httptransport "github.com/emil28092005/SciMesh/coordinator/internal/transport/http" "github.com/emil28092005/SciMesh/coordinator/internal/usecase" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) func main() { @@ -70,29 +71,35 @@ func run() error { taskResultRepo = postgres.NewTaskResultRepo(pool) ) + catalog, err := workloads.Load() + if err != nil { + log.Error("load workload catalog", "err", err) + return err + } + useCases := httptransport.UseCases{ RegisterWorker: usecase.NewRegisterWorker(workerRepo, clk), CreateJob: usecase.NewCreateJob(jobRepo, taskRepo, tx, clk), - SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts), - ClaimTask: usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration), + SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts, catalog), + ClaimTask: usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration, catalog), RenewLease: usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration), - CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize), - ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk), - FailTask: usecase.NewFailTask(taskRepo, jobRepo, workerRepo, tx, clk), + CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize, catalog), + ReduceJob: usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk, catalog), + FailTask: usecase.NewFailTask(taskRepo, jobRepo, workerRepo, tx, clk, catalog), GetJobStatus: usecase.NewGetJobStatus(jobRepo, taskRepo), CancelJob: usecase.NewCancelJob(jobRepo, taskRepo, tx, clk), UploadArtifact: usecase.NewUploadArtifact(taskRepo, workerRepo, artifactRepo, blobStore, tx, clk), DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore), GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)), GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore), - Dashboard: usecase.NewDashboard(uiReadRepo), + Dashboard: usecase.NewDashboard(uiReadRepo, catalog), PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore), } // Background reapers are tracked so shutdown can wait for them. Without this // the process would exit mid-UPDATE, and the deferred pool.Close() would pull // connections out from under them. - expireLeases := usecase.NewExpireLeases(taskRepo, jobRepo, tx, clk) + expireLeases := usecase.NewExpireLeases(taskRepo, jobRepo, tx, clk, catalog) markOffline := usecase.NewMarkWorkersOffline(workerRepo, clk, cfg.WorkerOfflineAfter) var wg sync.WaitGroup diff --git a/coordinator/internal/reducer/concat.go b/coordinator/internal/reducer/concat.go new file mode 100644 index 0000000..e6977a1 --- /dev/null +++ b/coordinator/internal/reducer/concat.go @@ -0,0 +1,57 @@ +// Package reducer contains deterministic, coordinator-side result reductions. +package reducer + +import ( + "bytes" + "encoding/csv" + "errors" + "fmt" + "io" +) + +// ReduceOrderedConcat concatenates worker partial tables in shard order into a +// single table with one header. Every partial must carry the same header as the +// first partial and rows of the same width; anything else fails the job closed. +func ReduceOrderedConcat(partials []io.Reader) ([]byte, error) { + var out bytes.Buffer + writer := csv.NewWriter(&out) + var firstHeader []string + for _, partial := range partials { + reader := csv.NewReader(partial) + header, err := reader.Read() + if err != nil { + return nil, fmt.Errorf("read partial header: %w", err) + } + if len(header) == 0 { + return nil, fmt.Errorf("partial result has an empty header") + } + if firstHeader == nil { + firstHeader = header + if err := writer.Write(header); err != nil { + return nil, err + } + } else if !equalStrings(header, firstHeader) { + return nil, fmt.Errorf("partial result has an inconsistent header") + } + for { + row, err := reader.Read() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return nil, fmt.Errorf("read partial row: %w", err) + } + if len(row) != len(header) { + return nil, fmt.Errorf("partial result has a row with an inconsistent width") + } + if err := writer.Write(row); err != nil { + return nil, err + } + } + } + writer.Flush() + if err := writer.Error(); err != nil { + return nil, err + } + return out.Bytes(), nil +} diff --git a/coordinator/internal/reducer/concat_test.go b/coordinator/internal/reducer/concat_test.go new file mode 100644 index 0000000..da0a0b2 --- /dev/null +++ b/coordinator/internal/reducer/concat_test.go @@ -0,0 +1,64 @@ +package reducer + +import ( + "io" + "strings" + "testing" +) + +func TestReduceOrderedConcatJoinsPartialsInOrderWithOneHeader(t *testing.T) { + first := strings.NewReader("chembl_id,canonical_smiles\nA,CC\nB,CCC\n") + second := strings.NewReader("chembl_id,canonical_smiles\nC,CCCC\n") + + output, err := ReduceOrderedConcat([]io.Reader{first, second}) + if err != nil { + t.Fatal(err) + } + want := "chembl_id,canonical_smiles\nA,CC\nB,CCC\nC,CCCC\n" + if string(output) != want { + t.Fatalf("output = %q, want %q", output, want) + } +} + +func TestReduceOrderedConcatIsDeterministicAcrossInputOrder(t *testing.T) { + left := strings.NewReader("id,rows\nA,1\nB,2\n") + right := strings.NewReader("id,rows\nC,3\n") + + first, err := ReduceOrderedConcat([]io.Reader{left, right}) + if err != nil { + t.Fatal(err) + } + left, right = strings.NewReader("id,rows\nA,1\nB,2\n"), strings.NewReader("id,rows\nC,3\n") + second, err := ReduceOrderedConcat([]io.Reader{left, right}) + if err != nil { + t.Fatal(err) + } + if string(first) != string(second) { + t.Fatalf("concat is not deterministic: %q != %q", first, second) + } +} + +func TestReduceOrderedConcatRejectsInconsistentHeaders(t *testing.T) { + first := strings.NewReader("a,b\n1,2\n") + second := strings.NewReader("a,c\n1,2\n") + if _, err := ReduceOrderedConcat([]io.Reader{first, second}); err == nil { + t.Fatal("inconsistent headers must fail") + } +} + +func TestReduceOrderedConcatRejectsRaggedRows(t *testing.T) { + partial := strings.NewReader("a,b\n1,2,3\n") + if _, err := ReduceOrderedConcat([]io.Reader{partial}); err == nil { + t.Fatal("ragged rows must fail") + } +} + +func TestReduceOrderedConcatEmptyPartials(t *testing.T) { + output, err := ReduceOrderedConcat(nil) + if err != nil { + t.Fatal(err) + } + if len(output) != 0 { + t.Fatalf("empty input must produce empty output, got %q", output) + } +} diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 3d9748f..fd6b2f8 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -47,19 +47,19 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur uc := coordhttp.UseCases{ RegisterWorker: usecase.NewRegisterWorker(work, clk), CreateJob: usecase.NewCreateJob(jobs, tasks, tx, clk), - SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3), - ClaimTask: usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease), + SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3, testCatalog()), + ClaimTask: usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease, testCatalog()), RenewLease: usecase.NewRenewLease(tasks, work, tx, clk, lease), - CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2), - ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk), - FailTask: usecase.NewFailTask(tasks, jobs, work, tx, clk), + CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2, testCatalog()), + ReduceJob: usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk, testCatalog()), + FailTask: usecase.NewFailTask(tasks, jobs, work, tx, clk, testCatalog()), GetJobStatus: usecase.NewGetJobStatus(jobs, tasks), GetJobResult: usecase.NewGetJobResult(jobs, downloadArtifact), CancelJob: usecase.NewCancelJob(jobs, tasks, tx, clk), UploadArtifact: usecase.NewUploadArtifact(tasks, work, arts, blobs, tx, clk), DownloadArtifact: downloadArtifact, GetTaskInput: usecase.NewGetTaskInput(tasks, arts, blobs), - Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)), + Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts), testCatalog()), PreviewArtifact: usecase.NewPreviewArtifact(memstore.NewUIReadRepo(jobs, tasks, work, arts), blobs), } worker, err := uc.RegisterWorker.Execute(context.Background(), usecase.RegisterWorkerInput{ diff --git a/coordinator/internal/transport/http/templates/add-worker.html b/coordinator/internal/transport/http/templates/add-worker.html index 730f704..486c098 100644 --- a/coordinator/internal/transport/http/templates/add-worker.html +++ b/coordinator/internal/transport/http/templates/add-worker.html @@ -33,7 +33,7 @@

Will my results count?

Your worker is untrusted by default: its results are cross-checked and accepted once a second independent worker computes the same answer (quorum), or once an admin marks your account verified — then your workers are trusted and results count immediately.

-

similarity-search is the only workload a volunteer worker runs today.

+

similarity-search and other SDK workloads from the library run on volunteer workers.

diff --git a/coordinator/internal/transport/http/templates/dashboard.html b/coordinator/internal/transport/http/templates/dashboard.html index 15e12bb..ffce034 100644 --- a/coordinator/internal/transport/http/templates/dashboard.html +++ b/coordinator/internal/transport/http/templates/dashboard.html @@ -13,7 +13,7 @@

Local scientific compute

SciMesh control room

Follow the real path from a molecular TSV to a globally reduced similarity result—without reading coordinator logs.

Live overview · refreshes every 2 seconds
-
{{if .Session}}Signed in · {{.Session.Role}}{{end}}{{if .Session}}Profile{{end}}{{if and .Session (eq .Session.Role "admin")}}Admin{{end}}{{if .Session}}Workloads{{end}}{{if .Session}}Docs{{end}}{{if .Session}}🖥 Add your machine{{end}}＋ New similarity search{{if .Session}}
{{end}}
+
{{if .Session}}Signed in · {{.Session.Role}}{{end}}{{if .Session}}Profile{{end}}{{if and .Session (eq .Session.Role "admin")}}Admin{{end}}{{if .Session}}Workloads{{end}}{{if .Session}}Docs{{end}}{{if .Session}}🖥 Add your machine{{end}}＋ New computation{{if .Session}}
{{end}}
How a search becomes a result
01Upload TSVThe coordinator validates and slices the dataset.
02Run shardsWorkers fingerprint molecules and return shard top-k CSVs.
03Merge exactlyThe coordinator ranks retained candidates deterministically.
04Download CSVA checksum-protected global result is ready.
@@ -22,7 +22,7 @@
Finished runs{{.FinishedJobs}}in the latest 20
-

Recent computations

{{len .Jobs}} shown · newest first

{{range .Jobs}}
{{workloadLabel .Workload}}
{{.ID}}
{{statusLabel .Status}}
{{statusHint .Status}}
{{.Completed}} / {{.Total}} shards complete{{if gt .Failed 0}} · {{.Failed}} failed{{end}}
{{else}}
No computations yet.
Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.
{{end}}
+

Recent computations

{{len .Jobs}} shown · newest first

{{range .Jobs}}
{{workloadLabel .Workload}}
{{.ID}}
{{statusLabel .Status}}
{{statusHint .Status}}
{{.Completed}} / {{.Total}} shards complete{{if gt .Failed 0}} · {{.Failed}} failed{{end}}
{{else}}
No computations yet.
Start a computation, then keep one or more workers running to watch this dashboard come alive.
{{end}}
{{if and .Session (ne .Session.Role "admin")}}

My machines

Workers you registered. Add your machine →

{{range .MyWorkers}}
{{.Name}}{{workerStatusLabel .Status}}
{{.ID}}

{{range .Capabilities}}{{.}}{{end}}

Last signal · {{time .LastHeartbeatAt}}

{{else}}
No machine of yours is connected.
Turn this computer into a worker →
{{end}}
{{end}}

Worker fleet

Workers register themselves; this page never controls their processes.

{{range .Workers}}
{{.Name}}{{workerStatusLabel .Status}}
{{.ID}}

{{range .Capabilities}}{{.}}{{end}}

Last signal · {{time .LastHeartbeatAt}}

{{else}}
No worker is registered.
Start scimesh-worker in another terminal, then return here.
{{end}}
@@ -30,7 +30,7 @@ const statusInfo={pending:['Waiting for a worker','waiting'],leased:['Assigned to a worker','active'],running:['Running','active'],reducing:['Merging results','active'],completed:['Completed','success'],failed:['Needs attention','danger'],cancelled:['Stopped','waiting']}; const pct=j=>j.total?Math.min(100,Math.floor((j.completed+j.failed+j.cancelled)*100/j.total)):0; const node=(tag,text,cls)=>{const n=document.createElement(tag);if(text!==undefined)n.textContent=text;if(cls)n.className=cls;return n}; - const renderJobs=jobs=>{const box=document.querySelector('#jobs');box.replaceChildren();if(!jobs.length){const empty=node('div',undefined,'empty');empty.append(node('strong','No computations yet.'),document.createElement('br'),document.createTextNode('Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.'));box.append(empty);return}for(const job of jobs){const info=statusInfo[job.status]||[job.status,'waiting'],link=node('a',undefined,'job');link.href='/ui/jobs/'+encodeURIComponent(job.id);const intro=node('div');intro.append(node('div',job.workload==='similarity-search'?'Molecule similarity search':job.workload,'job-name'),node('div',job.id,'job-id'));const state=node('div');state.append(node('span',info[0],'badge badge-'+info[1]),node('div',job.status==='reducing'?'Every shard is complete; coordinator is ranking the global top-k.':'Live coordinator state','job-id'));const progress=node('div'),numbers=node('div',undefined,'numbers');numbers.append(node('b',String(job.completed)),document.createTextNode(' / '+job.total+' shards complete'));if(job.failed){numbers.append(document.createTextNode(' · '),node('span',job.failed+' failed','offline'))}const bar=node('div',undefined,'bar'),fill=node('span');fill.style.width=pct(job)+'%';bar.append(fill);progress.append(numbers,bar);link.append(intro,state,progress,node('span','→','arrow'));box.append(link)}}; + const renderJobs=jobs=>{const box=document.querySelector('#jobs');box.replaceChildren();if(!jobs.length){const empty=node('div',undefined,'empty');empty.append(node('strong','No computations yet.'),document.createElement('br'),document.createTextNode('Start a computation, then keep one or more workers running to watch this dashboard come alive.'));box.append(empty);return}for(const job of jobs){const info=statusInfo[job.status]||[job.status,'waiting'],link=node('a',undefined,'job');link.href='/ui/jobs/'+encodeURIComponent(job.id);const intro=node('div');intro.append(node('div',job.workload==='similarity-search'?'Molecule similarity search':job.workload,'job-name'),node('div',job.id,'job-id'));const state=node('div');state.append(node('span',info[0],'badge badge-'+info[1]),node('div',job.status==='reducing'?'Every shard is complete; coordinator is ranking the global top-k.':'Live coordinator state','job-id'));const progress=node('div'),numbers=node('div',undefined,'numbers');numbers.append(node('b',String(job.completed)),document.createTextNode(' / '+job.total+' shards complete'));if(job.failed){numbers.append(document.createTextNode(' · '),node('span',job.failed+' failed','offline'))}const bar=node('div',undefined,'bar'),fill=node('span');fill.style.width=pct(job)+'%';bar.append(fill);progress.append(numbers,bar);link.append(intro,state,progress,node('span','→','arrow'));box.append(link)}}; const workerCard=worker=>{const card=node('article',undefined,'worker'),head=node('div',undefined,'worker-head'),left=node('div'),workerInfo=worker.status==='online'?['Available','success']:worker.status==='busy'?['Busy','active']:['Offline','waiting'];left.append(node('strong',worker.name),node('small',worker.id));head.append(left,node('span',workerInfo[0],'badge badge-'+workerInfo[1]));const caps=node('p');for(const capability of worker.capabilities||[])caps.append(node('span',capability,'cap'));card.append(head,caps,node('p','Last signal · '+new Date(worker.last_heartbeat_at).toLocaleString()));return card}; const renderWorkers=workers=>{const box=document.querySelector('#workers');box.replaceChildren();if(!workers.length){const empty=node('div',undefined,'empty');empty.append(node('strong','No worker is registered.'),document.createElement('br'),document.createTextNode('Start scimesh-worker in another terminal, then return here.'));box.append(empty);return}for(const worker of workers)box.append(workerCard(worker))}; const renderMyWorkers=workers=>{const box=document.querySelector('#my-workers');if(!box)return;box.replaceChildren();if(!workers.length){const empty=node('div',undefined,'empty'),link=node('a','Turn this computer into a worker →');link.href='/ui/workers/new';link.style.color='#79aaff';empty.append(node('strong','No machine of yours is connected.'),document.createElement('br'),link);box.append(empty);return}for(const worker of workers)box.append(workerCard(worker))}; diff --git a/coordinator/internal/transport/http/templates/job.html b/coordinator/internal/transport/http/templates/job.html index c9c77b2..0fa8218 100644 --- a/coordinator/internal/transport/http/templates/job.html +++ b/coordinator/internal/transport/http/templates/job.html @@ -20,7 +20,7 @@

Pipeline stages

Each stage reflects coordinator state, not a simulated progress bar.

1TSV accepted

The coordinator stored the source and created shard tasks.

2Shards execute

{{.Completed}} of {{.Total}} candidate partitions are complete.

3Workers return CSVs

Workers upload a checked partial result for every completed shard.

4Global reduction

The coordinator waits until all shards are complete.

5Final CSV

Available only after deterministic reduction succeeds.

-

Run configuration

Allowlisted scientific parameters.

What is being computed?

{{range .Parameters}}
{{.Label}}{{.Value}}
{{else}}

No displayable parameters were supplied.

{{end}}

Result status

Safe operator guidance.

{{if .FinalResultAvailable}}

Final result ready

The coordinator merged shard candidates with exact scores and stored the global top-k CSV.

{{range .Artifacts}}{{if and (eq .Kind "final_result") .Downloadable}}Preview CSVDownload final CSV{{end}}{{end}}{{else if eq .Status "reducing"}}

Merging completed shards

The final candidate heap is being ranked now. This page will update when the CSV is stored.

{{else if eq .Status "failed"}}

Run needs attention

{{if .ErrorMessage}}{{.ErrorMessage}}{{else}}One or more shards could not produce a final result. Review the task table below.{{end}}

{{else}}

Waiting for the final result

Partial CSVs are diagnostics. They become one global result only after every shard completes and reduction succeeds.

{{end}}
+

Run configuration

Allowlisted scientific parameters.

What is being computed?

{{range .Parameters}}
{{.Label}}{{.Value}}
{{else}}

No displayable parameters were supplied.

{{end}}

Result status

Safe operator guidance.

{{if .FinalResultAvailable}}

Final result ready

The coordinator reduced every completed shard into one checksum-protected result file.

{{range .Artifacts}}{{if and (eq .Kind "final_result") .Downloadable}}Preview CSVDownload final CSV{{end}}{{end}}{{else if eq .Status "reducing"}}

Merging completed shards

Every shard is complete; the coordinator is reducing the partial results now. This page will update when the result file is stored.

{{else if eq .Status "failed"}}

Run needs attention

{{if .ErrorMessage}}{{.ErrorMessage}}{{else}}One or more shards could not produce a final result. Review the task table below.{{end}}

{{else}}

Waiting for the final result

Partial CSVs are diagnostics. They become one final result only after every shard completes and reduction succeeds.

{{end}}

Shard activity

Every task is one input partition. The table refreshes while work is in progress.

{{range .Tasks}}{{else}}{{end}}
ShardStateAttemptWorker / leaseOutcome
#{{.ChunkIndex}}{{statusLabel .Status}}{{.Attempt}} / {{.MaxAttempts}}{{if .LeaseOwner}}{{.LeaseOwner}}{{if .LeaseExpiresAt}}
lease until {{time .LeaseExpiresAt}}{{end}}{{else}}—{{end}}		{{if .ErrorCode}}{{taskErrorLabel .ErrorCode}}
{{taskErrorHint .ErrorCode}}{{else if eq .Status "completed"}}Partial CSV uploaded{{else}}—{{end}}
No shard tasks are present yet.
@@ -31,9 +31,9 @@ const id={{printf "%q" .ID}},statusInfo={pending:['Waiting for a worker','waiting','A compatible worker has not claimed a shard yet.'],leased:['Assigned to a worker','active','A worker has a shard lease and should begin shortly.'],running:['Running','active','Workers are calculating fingerprints and returning shard-level candidates.'],reducing:['Merging results','active','All shards are complete. The coordinator is ranking the global top-k.'],completed:['Completed','success','The final result is stored and ready to download.'],failed:['Needs attention','danger','The job cannot produce a final result. Review the safe diagnosis below.'],cancelled:['Stopped','waiting','The operator stopped unfinished shards.']},terminal=new Set(['completed','failed','cancelled']); const text=(tag,value,cls)=>{const n=document.createElement(tag);if(value!==undefined)n.textContent=value;if(cls)n.className=cls;return n},pct=j=>j.total?Math.min(100,Math.floor((j.completed+j.failed+j.cancelled)*100/j.total)):0,fmtTime=value=>value?new Date(value).toLocaleString():'—',fmtBytes=n=>n<1024?n+' B':n<1024*1024?(n/1024).toFixed(1)+' KiB':(n/(1024*1024)).toFixed(1)+' MiB',taskError={CalledProcessError:['Local calculation failed','The local SciMesh command stopped before it could upload a result. Check the worker terminal for the original error.'],ValueError:['Task input could not be processed','The shard or its parameters did not meet the worker validation rules.'],CoordinatorTransientError:['Coordinator connection was interrupted','The worker will retry when the coordinator is available again.'],CoordinatorConflictError:['Worker lease was no longer valid','Another worker or a lease timeout changed this shard before completion.'],FileNotFoundError:['Local task file is missing','Restart the worker with an absolute --work-dir.']}; const stage=(index,title,description,state)=>{const card=text('article',undefined,'stage stage-'+state);card.append(text('span',String(index),'index'),text('b',title),text('p',description));return card}; - const renderStages=job=>{const holder=document.querySelector('#stages'),allDone=job.completed===job.total&&job.total>0,reducerFailed=job.status==='failed'&&job.error_code==='reducer_failed',shardFailed=job.status==='failed'&&!reducerFailed;holder.replaceChildren(stage(1,'TSV accepted','The coordinator stored the source and created shard tasks.','done'),stage(2,'Shards execute',job.completed+' of '+job.total+' candidate partitions are complete.',shardFailed?'failed':allDone?'done':(job.running||job.leased)?'active':'waiting'),stage(3,'Workers return CSVs',allDone?'Every completed shard has a coordinator-owned partial CSV.':(job.running||job.leased)?'Workers are actively claiming and processing partitions.':'Waiting for a worker to claim a shard.',shardFailed?'failed':allDone?'done':(job.running||job.leased)?'active':'waiting'),stage(4,'Global reduction',reducerFailed?'The coordinator could not safely reduce partial results.':job.status==='reducing'?'The coordinator is merging exact candidate scores.':job.status==='completed'?'The global top-k has been merged deterministically.':'Reduction begins only after every shard completes.',reducerFailed?'failed':job.status==='reducing'?'active':job.status==='completed'?'done':'waiting'),stage(5,'Final CSV',job.final_result_available?'The checksum-protected global result is ready.':'Available only after deterministic reduction succeeds.',job.final_result_available?'done':reducerFailed?'failed':'waiting'))}; + const renderStages=job=>{const holder=document.querySelector('#stages'),allDone=job.completed===job.total&&job.total>0,reducerFailed=job.status==='failed'&&job.error_code==='reducer_failed',shardFailed=job.status==='failed'&&!reducerFailed;holder.replaceChildren(stage(1,'TSV accepted','The coordinator stored the source and created shard tasks.','done'),stage(2,'Shards execute',job.completed+' of '+job.total+' candidate partitions are complete.',shardFailed?'failed':allDone?'done':(job.running||job.leased)?'active':'waiting'),stage(3,'Workers return CSVs',allDone?'Every completed shard has a coordinator-owned partial CSV.':(job.running||job.leased)?'Workers are actively claiming and processing partitions.':'Waiting for a worker to claim a shard.',shardFailed?'failed':allDone?'done':(job.running||job.leased)?'active':'waiting'),stage(4,'Global reduction',reducerFailed?'The coordinator could not safely reduce partial results.':job.status==='reducing'?'The coordinator is merging exact candidate scores.':job.status==='completed'?'The partial results have been reduced deterministically.':'Reduction begins only after every shard completes.',reducerFailed?'failed':job.status==='reducing'?'active':job.status==='completed'?'done':'waiting'),stage(5,'Final CSV',job.final_result_available?'The checksum-protected global result is ready.':'Available only after deterministic reduction succeeds.',job.final_result_available?'done':reducerFailed?'failed':'waiting'))}; const renderParameters=parameters=>{const holder=document.querySelector('#parameters');holder.replaceChildren();if(!parameters.length){holder.append(text('p','No displayable parameters were supplied.'));return}for(const parameter of parameters){const row=text('div',undefined,'parameter');row.append(text('span',parameter.label),text('code',parameter.value));holder.append(row)}}; - const renderResult=job=>{const card=document.querySelector('#result-card');card.className='run-note';card.replaceChildren();if(job.final_result_available){card.classList.add('result');card.append(text('h3','Final result ready'),text('p','The coordinator merged shard candidates with exact scores and stored the global top-k CSV.'));const final=(job.artifacts||[]).find(a=>a.kind==='final_result'&&a.downloadable);if(final){const preview=text('a','Preview CSV','download');preview.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id)+'/preview';const download=text('a','Download final CSV','download');download.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id);card.append(preview,download)}}else if(job.status==='reducing'){card.append(text('h3','Merging completed shards'),text('p','The final candidate heap is being ranked now. This page will update when the CSV is stored.'))}else if(job.status==='failed'){card.classList.add('alert');card.append(text('h3','Run needs attention'),text('p',job.error_message||'One or more shards could not produce a final result. Review the task table below.'))}else{card.append(text('h3','Waiting for the final result'),text('p','Partial CSVs are diagnostics. They become one global result only after every shard completes and reduction succeeds.'))}}; + const renderResult=job=>{const card=document.querySelector('#result-card');card.className='run-note';card.replaceChildren();if(job.final_result_available){card.classList.add('result');card.append(text('h3','Final result ready'),text('p','The coordinator reduced every completed shard into one checksum-protected result file.'));const final=(job.artifacts||[]).find(a=>a.kind==='final_result'&&a.downloadable);if(final){const preview=text('a','Preview CSV','download');preview.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id)+'/preview';const download=text('a','Download final CSV','download');download.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id);card.append(preview,download)}}else if(job.status==='reducing'){card.append(text('h3','Merging completed shards'),text('p','Every shard is complete; the coordinator is reducing the partial results now. This page will update when the result file is stored.'))}else if(job.status==='failed'){card.classList.add('alert');card.append(text('h3','Run needs attention'),text('p',job.error_message||'One or more shards could not produce a final result. Review the task table below.'))}else{card.append(text('h3','Waiting for the final result'),text('p','Partial CSVs are diagnostics. They become one final result only after every shard completes and reduction succeeds.'))}}; const renderTasks=tasks=>{const holder=document.querySelector('#tasks');holder.replaceChildren();if(!tasks.length){const row=document.createElement('tr'),cell=text('td','No shard tasks are present yet.','empty');cell.colSpan=5;row.append(cell);holder.append(row);return}for(const task of tasks){const row=document.createElement('tr'),info=statusInfo[task.status]||[task.status,'waiting',''];row.append(text('td','#'+task.chunk_index));const state=text('td'),badge=text('span',info[0],'badge badge-'+info[1]);state.append(badge);row.append(state,text('td',task.attempt+' / '+task.max_attempts));const worker=text('td');if(task.lease_owner){worker.append(text('strong',task.lease_owner));if(task.lease_expires_at){worker.append(document.createElement('br'),text('small','lease until '+fmtTime(task.lease_expires_at),'muted'))}}else worker.append(text('span','—','muted'));row.append(worker);const outcome=text('td',undefined,'error');if(task.error_code){const explanation=taskError[task.error_code]||['Task needs attention','Check the worker terminal for the original error.'];outcome.append(text('strong',explanation[0]),document.createElement('br'),text('small',explanation[1]))}else if(task.status==='completed')outcome.append(text('span','Partial CSV uploaded','muted'));else outcome.append(text('span','—','muted'));row.append(outcome);holder.append(row)}}; const renderArtifacts=job=>{const holder=document.querySelector('#artifacts');holder.replaceChildren();const artifacts=job.artifacts||[];if(!artifacts.length){holder.append(text('div','Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.','empty'));return}for(const artifact of artifacts){const card=text('article',undefined,'artifact'+(artifact.kind==='final_result'?' artifact-final':''));card.append(text('div',artifact.diagnostic?'Partial result · diagnostic':artifact.kind,'artifact-type'),text('strong',artifact.filename),text('span',fmtBytes(artifact.size_bytes),'muted'),text('code','SHA-256 '+artifact.sha256));if(artifact.downloadable){const preview=text('a','Preview CSV');preview.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(artifact.id)+'/preview';const download=text('a',artifact.kind==='final_result'?'Download final CSV':'Download CSV');download.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(artifact.id);card.append(preview,download)}holder.append(card)}}; const speedHistory=[{at:Date.now(),completed:Number({{.Completed}})}],speedWindow=15,speedLimit=90,svgNS='http://www.w3.org/2000/svg',speedSVG=(tag,attrs)=>{const node=document.createElementNS(svgNS,tag);for(const [key,value] of Object.entries(attrs))node.setAttribute(key,String(value));return node},rateLabel=rate=>rate.toFixed(1)+' shards/min'; diff --git a/coordinator/internal/transport/http/templates/new-job.html b/coordinator/internal/transport/http/templates/new-job.html index 49eba44..da7c43e 100644 --- a/coordinator/internal/transport/http/templates/new-job.html +++ b/coordinator/internal/transport/http/templates/new-job.html @@ -4,20 +4,32 @@ - New similarity search · SciMesh + New computation · SciMesh
- ← Back to control room

New computation

Similarity search, end to end

Upload a ChEMBL-style TSV. Workers calculate shard candidates; after every shard succeeds, SciMesh merges the exact global top-k into one final CSV.

-

Required columns: chembl_id and canonical_smiles.

Use a valid SMILES. The coordinator shares this exact query with every shard.

How many final molecules to retain.

Smaller shards make more visible tasks.

Leave blank to rank every valid candidate.

“Less” helps explore dissimilar molecules.

Only the first N data rows become shards; the original upload remains stored by the coordinator.

Ready to plan a run.
Select a TSV to see the file that will be sent to the coordinator.
+ ← Back to control room

New computation

Any workload, end to end

Choose a workload from the installed library. Upload its dataset; workers compute shards; the coordinator reduces the partials into one final artifact.

+

Smaller shards make more visible tasks.

Only the first N data rows become shards; the upload stays stored.

A delimited table with a header row. The workload defines the required columns.

Ready to plan a run.
Select a dataset to see the file that will be sent to the coordinator.
diff --git a/coordinator/internal/transport/http/testcatalog_test.go b/coordinator/internal/transport/http/testcatalog_test.go new file mode 100644 index 0000000..b4cff30 --- /dev/null +++ b/coordinator/internal/transport/http/testcatalog_test.go @@ -0,0 +1,15 @@ +package http_test + +import ( + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" +) + +// testCatalog loads the embedded workload catalog for http tests, the same +// catalog the server binary loads at startup. +func testCatalog() *workloads.Catalog { + catalog, err := workloads.Load() + if err != nil { + panic(err) + } + return catalog +} diff --git a/coordinator/internal/transport/http/ui.go b/coordinator/internal/transport/http/ui.go index 8812c15..5367877 100644 --- a/coordinator/internal/transport/http/ui.go +++ b/coordinator/internal/transport/http/ui.go @@ -2,17 +2,20 @@ package http import ( "embed" + "encoding/json" "fmt" "html/template" "io" "mime" "net/http" + "sort" "strconv" "time" "github.com/google/uuid" "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) //go:embed templates/*.html @@ -165,6 +168,10 @@ func uiWorkloadLabel(workload string) string { return "Molecule similarity search" case "similarity-graph", "similarity_graph": return "Molecular similarity graph" + case "molwt-filter", "molwt_filter": + return "Molecular weight filter" + case "descriptor-batch", "descriptor_batch": + return "Descriptor batch" default: return workload } @@ -236,7 +243,143 @@ func (s *Server) handleUIOverviewJSON(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleUINewJob(w http.ResponseWriter, r *http.Request) { - s.renderUI(w, "new-job.html", nil) + catalog, err := workloads.Load() + if err != nil { + s.log.Error("load workload catalog", "err", err) + http.Error(w, "internal error", http.StatusInternalServerError) + return + } + s.renderUI(w, "new-job.html", uiNewJobView{Payload: newJobPayload(catalog)}) +} + +// uiNewJobWorkloadJSON is the per-workload data handed to the page script. It +// is presentation metadata from the embedded catalog; the coordinator re- +// validates everything server-side on upload. +type uiNewJobWorkloadJSON struct { + Name string `json:"name"` + Description string `json:"description"` + Schema map[string]any `json:"schema"` + UI []uiNewJobElementJSON `json:"ui"` + Required []string `json:"required"` + OneOf [][]string `json:"one_of"` + UploadReady bool `json:"upload_ready"` + Reduction string `json:"reduction"` + Defaults map[string]any `json:"defaults"` + InputMedia map[string]string `json:"input_media"` +} + +type uiNewJobElementJSON struct { + Field string `json:"field"` + Widget string `json:"widget"` + Label string `json:"label"` + Help string `json:"help"` + Placeholder string `json:"placeholder"` + Options []string `json:"options"` + Default any `json:"default"` + Order int `json:"order"` + Required bool `json:"required"` +} + +type uiNewJobView struct { + Payload template.JS +} + +func newJobPayload(catalog *workloads.Catalog) template.JS { + payload := struct { + Workloads []uiNewJobWorkloadJSON `json:"workloads"` + }{Workloads: make([]uiNewJobWorkloadJSON, 0, len(catalog.Enabled()))} + for _, workload := range catalog.Enabled() { + required := map[string]bool{} + if entries, ok := workload.Parameters["required"].([]any); ok { + for _, entry := range entries { + if name, ok := entry.(string); ok { + required[name] = true + } + } + } + elementViews := make([]uiNewJobElementJSON, 0, len(workload.UIElements)) + for _, element := range workload.UIElements { + elementViews = append(elementViews, uiNewJobElementJSON{ + Field: element.Field, + Widget: element.Widget, + Label: element.Label, + Help: element.Help, + Placeholder: element.Placeholder, + Options: element.Options, + Default: element.Default, + Order: element.Order, + Required: required[element.Field], + }) + } + payload.Workloads = append(payload.Workloads, uiNewJobWorkloadJSON{ + Name: workload.Name, + Description: workload.Description, + Schema: workload.Parameters, + UI: elementViews, + Required: sortedKeys(required), + OneOf: oneOfGroups(workload.Parameters), + UploadReady: workload.UploadReady, + Reduction: workload.Reduction, + Defaults: catalog.ParameterDefaults(workload.Name), + InputMedia: inputMediaViews(catalog, workload.Name), + }) + } + encoded, err := json.Marshal(payload) + if err != nil { + return template.JS("null") + } + return template.JS(encoded) +} + +func sortedKeys(values map[string]bool) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + +func oneOfGroups(schema map[string]any) [][]string { + rawOneOf, ok := schema["oneOf"].([]any) + if !ok { + return nil + } + // The clean shape is "exactly one of these single fields" (e.g. the search + // query id vs SMILES choice): every branch requires exactly one distinct + // field. Anything more complex is left to server-side validation. + var fields []string + seen := map[string]bool{} + for _, rawOption := range rawOneOf { + option, ok := rawOption.(map[string]any) + if !ok { + return nil + } + required, _ := option["required"].([]any) + if len(required) != 1 { + return nil + } + field, ok := required[0].(string) + if !ok || seen[field] { + return nil + } + seen[field] = true + fields = append(fields, field) + } + if len(fields) != len(rawOneOf) { + return nil + } + return [][]string{fields} +} + +func inputMediaViews(catalog *workloads.Catalog, name string) map[string]string { + views := map[string]string{} + for _, port := range catalog.InputPortNames(name) { + if mediaType := catalog.InputMediaType(name, port); mediaType != "" { + views[port] = mediaType + } + } + return views } func (s *Server) uiJobID(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) { diff --git a/coordinator/internal/transport/http/ui_newjob_internal_test.go b/coordinator/internal/transport/http/ui_newjob_internal_test.go new file mode 100644 index 0000000..754d4c2 --- /dev/null +++ b/coordinator/internal/transport/http/ui_newjob_internal_test.go @@ -0,0 +1,44 @@ +package http + +import ( + "strings" + "testing" + + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" +) + +func TestNewJobPageCarriesWorkloadCatalogPayload(t *testing.T) { + catalog, err := workloads.Load() + if err != nil { + t.Fatalf("load workload catalog: %v", err) + } + view := uiNewJobView{Payload: newJobPayload(catalog)} + var builder strings.Builder + if err := uiTemplates.ExecuteTemplate(&builder, "new-job.html", view); err != nil { + t.Fatalf("render new-job page: %v", err) + } + page := builder.String() + for _, expected := range []string{"const DATA=", "similarity-search", "molwt-filter", "descriptor-batch", "one_of", "query_id", "min_molwt", "skip_invalid", "upload_ready"} { + if !strings.Contains(page, expected) { + t.Errorf("new-job page is missing %q", expected) + } + } +} + +func TestNewJobPayloadDeclaresUploadReadiness(t *testing.T) { + catalog, err := workloads.Load() + if err != nil { + t.Fatalf("load workload catalog: %v", err) + } + payload := newJobPayload(catalog) + text := string(payload) + if !strings.Contains(text, `"upload_ready":false`) { + t.Errorf("catalog payload must mark similarity-graph as not upload-ready") + } + if !strings.Contains(text, `"reduction":"top-k"`) { + t.Errorf("catalog payload is missing the top-k reduction for search") + } + if !strings.Contains(text, `"reduction":"ordered-concat"`) { + t.Errorf("catalog payload is missing the ordered-concat reduction for row workloads") + } +} diff --git a/coordinator/internal/transport/http/ui_workloads.go b/coordinator/internal/transport/http/ui_workloads.go index e61a596..ad56a5e 100644 --- a/coordinator/internal/transport/http/ui_workloads.go +++ b/coordinator/internal/transport/http/ui_workloads.go @@ -1,44 +1,22 @@ package http import ( - "embed" "encoding/json" "net/http" "sort" "sync" + + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) -//go:embed workloads.json -var workloadLibraryFile embed.FS - -// uiWorkloadLibrary is the catalog written by `scimesh workload export`. The -// coordinator never evaluates the schemas in it; it is presentation metadata -// for the operator UI, kept in sync by `make workloads-export`. -type uiWorkloadLibrary struct { - SchemaVersion int `json:"schema_version"` - GeneratedBy string `json:"generated_by"` - Workloads []uiWorkloadRaw `json:"workloads"` -} - -type uiWorkloadRaw struct { - Name string `json:"name"` - Version string `json:"version"` - Description string `json:"description"` - Capabilities []string `json:"capabilities"` - TrustModes []string `json:"trust_modes"` - Determinism string `json:"determinism"` - Verifier string `json:"verifier"` - Enabled bool `json:"enabled"` - Parameters map[string]any `json:"parameters_schema"` - Inputs map[string]map[string]any `json:"inputs"` - Outputs map[string]map[string]any `json:"outputs"` -} - +// uiPortView is one sorted input/output port of a workload, rendered as +// pretty JSON on the library page. type uiPortView struct { Name string Schema string } +// uiWorkloadView is the library page's view of one catalog workload. type uiWorkloadView struct { Name string Version string @@ -48,6 +26,7 @@ type uiWorkloadView struct { Determinism string Verifier string Enabled bool + Reduction string Parameters string Inputs []uiPortView Outputs []uiPortView @@ -65,18 +44,13 @@ var ( func loadWorkloadLibrary() (uiWorkloadsView, error) { workloadLibraryOnce.Do(func() { - data, err := workloadLibraryFile.ReadFile("workloads.json") + catalog, err := workloads.Load() if err != nil { workloadLibraryErr = err return } - var raw uiWorkloadLibrary - if err := json.Unmarshal(data, &raw); err != nil { - workloadLibraryErr = err - return - } - view := uiWorkloadsView{Workloads: make([]uiWorkloadView, 0, len(raw.Workloads))} - for _, item := range raw.Workloads { + view := uiWorkloadsView{Workloads: make([]uiWorkloadView, 0, len(catalog.Enabled()))} + for _, item := range catalog.Enabled() { view.Workloads = append(view.Workloads, uiWorkloadView{ Name: item.Name, Version: item.Version, @@ -86,6 +60,7 @@ func loadWorkloadLibrary() (uiWorkloadsView, error) { Determinism: item.Determinism, Verifier: item.Verifier, Enabled: item.Enabled, + Reduction: item.Reduction, Parameters: prettyJSON(item.Parameters), Inputs: portViews(item.Inputs), Outputs: portViews(item.Outputs), @@ -107,7 +82,7 @@ func prettyJSON(value any) string { return string(encoded) } -func portViews(ports map[string]map[string]any) []uiPortView { +func portViews(ports map[string]any) []uiPortView { names := make([]string, 0, len(ports)) for name := range ports { names = append(names, name) diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go index 2404cfe..437b8f2 100644 --- a/coordinator/internal/usecase/job.go +++ b/coordinator/internal/usecase/job.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) // Job operations: the submitter-facing lifecycle of a whole submission. @@ -227,7 +228,7 @@ func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobPr // Shared by CompleteTask and FailTask so both close a job by the same rule — // the rule itself lives in domain.JobProgress.DeriveStatus. func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository, - jobID uuid.UUID, now time.Time) error { + catalog *workloads.Catalog, jobID uuid.UUID, now time.Time) error { counts, err := tasks.CountByStatus(ctx, jobID) if err != nil { @@ -239,9 +240,11 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository } status := progressFrom(*job, counts).DeriveStatus() // All worker shards being complete means scientific reduction is ready, not - // that the job's final artifact already exists. CTX-09 owns the transition - // from reducing to completed after it persists that artifact. - if status == domain.JobCompleted && job.Workload == "similarity-search" { + // that the job's final artifact already exists. Known catalog workloads + // transition to reducing so ReduceJob can produce the final artifact; the + // reducer then completes the job with the result. Unknown (URI-based) jobs + // complete without a coordinator-owned final artifact. + if status == domain.JobCompleted && catalog != nil && catalog.Reduction(job.Workload) != "" { status = domain.JobReducing } @@ -253,14 +256,14 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository } func syncExpiredJobStatuses(ctx context.Context, jobs JobRepository, tasks TaskRepository, - jobIDs []uuid.UUID, now time.Time) error { + catalog *workloads.Catalog, jobIDs []uuid.UUID, now time.Time) error { seen := make(map[uuid.UUID]struct{}, len(jobIDs)) for _, jobID := range jobIDs { if _, duplicate := seen[jobID]; duplicate { continue } seen[jobID] = struct{}{} - if err := syncJobStatus(ctx, jobs, tasks, jobID, now); err != nil { + if err := syncJobStatus(ctx, jobs, tasks, catalog, jobID, now); err != nil { return err } } diff --git a/coordinator/internal/usecase/reduce.go b/coordinator/internal/usecase/reduce.go index c1faf71..efbd57f 100644 --- a/coordinator/internal/usecase/reduce.go +++ b/coordinator/internal/usecase/reduce.go @@ -9,6 +9,7 @@ import ( "github.com/emil28092005/SciMesh/coordinator/internal/domain" "github.com/emil28092005/SciMesh/coordinator/internal/reducer" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) // ReduceJob turns completed coordinator-owned partial artifacts into one final @@ -20,11 +21,12 @@ type ReduceJob struct { blobs BlobStore tx TxManager clock Clock + catalog *workloads.Catalog } func NewReduceJob(jobs JobRepository, tasks TaskRepository, artifacts ArtifactRepository, - blobs BlobStore, tx TxManager, clock Clock) *ReduceJob { - return &ReduceJob{jobs: jobs, tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clock: clock} + blobs BlobStore, tx TxManager, clock Clock, catalog *workloads.Catalog) *ReduceJob { + return &ReduceJob{jobs: jobs, tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clock: clock, catalog: catalog} } // Execute is idempotent for jobs that are not currently reducing. The worker @@ -42,7 +44,11 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error { if job.Status != domain.JobReducing { return nil } - if job.Workload != "similarity-search" { + if uc.catalog == nil { + return uc.fail(ctx, jobID) + } + reduction := uc.catalog.Reduction(job.Workload) + if reduction == "" { return uc.fail(ctx, jobID) } @@ -73,13 +79,13 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error { readers = append(readers, body) closers = append(closers, body) } - output, reduceErr := reducer.ReduceSimilaritySearch(readers, job.Parameters) + output, reduceErr := reducePartials(reduction, readers, job.Parameters) closeAll(closers) if reduceErr != nil { return uc.fail(ctx, jobID) } - final, err := domain.NewArtifact(jobID, nil, domain.ArtifactFinalResult, "similarity-search.csv", "text/csv", uc.clock.Now()) + final, err := domain.NewArtifact(jobID, nil, domain.ArtifactFinalResult, job.Workload+".csv", "text/csv", uc.clock.Now()) if err != nil { return uc.fail(ctx, jobID) } @@ -100,6 +106,17 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error { return nil } +func reducePartials(reduction string, readers []io.Reader, parameters map[string]any) ([]byte, error) { + switch reduction { + case "top-k": + return reducer.ReduceSimilaritySearch(readers, parameters) + case "ordered-concat": + return reducer.ReduceOrderedConcat(readers) + default: + return nil, domain.ErrInvalidInput + } +} + func (uc *ReduceJob) fail(ctx context.Context, jobID uuid.UUID) error { // The public state carries a stable sanitized failure, never parser/storage // internals that may include local paths or implementation details. diff --git a/coordinator/internal/usecase/task.go b/coordinator/internal/usecase/task.go index 9cb9c40..a2c4393 100644 --- a/coordinator/internal/usecase/task.go +++ b/coordinator/internal/usecase/task.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) // Task operations: the worker-facing lifecycle of a single chunk. @@ -27,10 +28,11 @@ type ClaimTask struct { tx TxManager clock Clock leaseDuration time.Duration + catalog *workloads.Catalog } -func NewClaimTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, leaseDuration time.Duration) *ClaimTask { - return &ClaimTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, leaseDuration: leaseDuration} +func NewClaimTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, leaseDuration time.Duration, catalog *workloads.Catalog) *ClaimTask { + return &ClaimTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, leaseDuration: leaseDuration, catalog: catalog} } // Execute reclaims elapsed leases first, then hands out one task. @@ -77,7 +79,7 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl if err != nil { return err } - if err := syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affectedJobs, now); err != nil { + if err := syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, uc.catalog, affectedJobs, now); err != nil { return err } @@ -159,6 +161,7 @@ func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain. type CompleteTask struct { tasks TaskRepository jobs JobRepository + catalog *workloads.Catalog artifacts ArtifactRepository workers WorkerRepository results TaskResultRepository @@ -170,12 +173,12 @@ type CompleteTask struct { } func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts ArtifactRepository, - workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int) *CompleteTask { + workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int, catalog *workloads.Catalog) *CompleteTask { if quorum < 1 { quorum = 2 } return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, workers: workers, - results: results, tx: tx, clock: clock, quorum: quorum} + results: results, tx: tx, clock: clock, quorum: quorum, catalog: catalog} } // Execute applies the result and, when that was the job's last outstanding @@ -229,7 +232,7 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom if err := uc.tasks.Update(ctx, task); err != nil { return err } - return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) + return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now) }) if err != nil { return nil, err @@ -268,7 +271,7 @@ func (uc *CompleteTask) recordVote(ctx context.Context, task *domain.Task, in Co if err := uc.tasks.Update(ctx, task); err != nil { return err } - return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) + return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now) } // workerTrust reports whether the worker's results are accepted directly, and @@ -324,10 +327,11 @@ type FailTask struct { workers WorkerRepository tx TxManager clock Clock + catalog *workloads.Catalog } -func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock) *FailTask { - return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock} +func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, catalog *workloads.Catalog) *FailTask { + return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, catalog: catalog} } // Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps @@ -351,7 +355,7 @@ func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task return err } out = task - return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now) + return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now) }) if err != nil { return nil, err @@ -362,14 +366,15 @@ func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task // --- ExpireLeases -------------------------------------------------------- type ExpireLeases struct { - tasks TaskRepository - jobs JobRepository - tx TxManager - clock Clock + tasks TaskRepository + jobs JobRepository + tx TxManager + clock Clock + catalog *workloads.Catalog } -func NewExpireLeases(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *ExpireLeases { - return &ExpireLeases{tasks: tasks, jobs: jobs, tx: tx, clock: clock} +func NewExpireLeases(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock, catalog *workloads.Catalog) *ExpireLeases { + return &ExpireLeases{tasks: tasks, jobs: jobs, tx: tx, clock: clock, catalog: catalog} } // Execute reclaims elapsed tasks and persists the state of every affected job. @@ -386,7 +391,7 @@ func (uc *ExpireLeases) Execute(ctx context.Context) (int64, error) { if err != nil { return err } - return syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affected, now) + return syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, uc.catalog, affected, now) }) return int64(len(affected)), err } diff --git a/coordinator/internal/usecase/testcatalog_test.go b/coordinator/internal/usecase/testcatalog_test.go new file mode 100644 index 0000000..0a3aa2e --- /dev/null +++ b/coordinator/internal/usecase/testcatalog_test.go @@ -0,0 +1,62 @@ +package usecase_test + +import ( + "testing" + + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" +) + +// testCatalog loads the embedded workload catalog for usecase tests. The +// catalog is checked in and generated from the SDK library, so tests exercise +// the real validation contract. +func testCatalog() *workloads.Catalog { + catalog, err := workloads.Load() + if err != nil { + panic(err) + } + return catalog +} + +func TestEmbeddedCatalogLoadsAndValidatesSearchParameters(t *testing.T) { + catalog := testCatalog() + if err := catalog.ValidateParameters("similarity-search", map[string]any{ + "query_smiles": "CCO", "top_k": 10, "threshold_direction": "greater", + }); err != nil { + t.Fatalf("valid search parameters rejected: %v", err) + } + if err := catalog.ValidateParameters("molwt-filter", map[string]any{ + "min_molwt": 100, "max_molwt": 600, "skip_invalid": true, + }); err != nil { + t.Fatalf("valid molwt parameters rejected: %v", err) + } + if err := catalog.ValidateParameters("nope", map[string]any{}); err == nil { + t.Error("unknown workload accepted") + } + if err := catalog.ValidateParameters("similarity-graph", map[string]any{"threshold": 0.7}); err != nil { + t.Errorf("graph parameters rejected: %v", err) + } + if !catalog.UploadReady("molwt-filter") { + t.Error("molwt-filter must be upload-ready") + } + if catalog.UploadReady("similarity-graph") { + t.Error("similarity-graph must not be upload-ready") + } + if got := catalog.Reduction("similarity-search"); got != "top-k" { + t.Errorf("search reduction = %q, want top-k", got) + } + if got := catalog.Reduction("descriptor-batch"); got != "ordered-concat" { + t.Errorf("descriptor reduction = %q, want ordered-concat", got) + } + for name, parameters := range map[string]map[string]any{ + "both query fields": {"query_id": "CHEMBL1", "query_smiles": "CCO"}, + "undeclared parameter": {"query_smiles": "CCO", "bogus": 1}, + "bad top_k": {"query_smiles": "CCO", "top_k": -1}, + "bad enum": {"query_smiles": "CCO", "threshold_direction": "sideways"}, + "missing query": {}, + "non-integer top_k": {"query_smiles": "CCO", "top_k": 1.5}, + } { + if err := catalog.ValidateParameters("similarity-search", parameters); err == nil { + t.Errorf("%s accepted", name) + } + } +} diff --git a/coordinator/internal/usecase/ui.go b/coordinator/internal/usecase/ui.go index f26d102..4d19f7a 100644 --- a/coordinator/internal/usecase/ui.go +++ b/coordinator/internal/usecase/ui.go @@ -3,12 +3,15 @@ package usecase import ( "context" "fmt" + "sort" + "strings" "time" "github.com/google/uuid" "github.com/emil28092005/SciMesh/coordinator/internal/authctx" "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) // UIReadRepository is a read-only projection source for the local operator UI. @@ -125,9 +128,14 @@ type JobDetailView struct { Session *SessionView `json:"-"` } -type Dashboard struct{ read UIReadRepository } +type Dashboard struct { + read UIReadRepository + catalog *workloads.Catalog +} -func NewDashboard(read UIReadRepository) *Dashboard { return &Dashboard{read: read} } +func NewDashboard(read UIReadRepository, catalog *workloads.Catalog) *Dashboard { + return &Dashboard{read: read, catalog: catalog} +} func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, error) { jobs, err := d.read.ListJobs(ctx, uiOwnerFilter(ctx), limit) @@ -219,7 +227,7 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi JobCard: jobCard(*job, tasks), Tasks: make([]TaskCard, 0, len(tasks)), Artifacts: make([]ArtifactCard, 0, len(artifacts)), - Parameters: uiParameters(job.Parameters), + Parameters: uiParameters(job.Parameters, d.catalog, job.Workload), Session: sessionViewFrom(ctx), } for _, task := range tasks { @@ -304,31 +312,67 @@ func jobCard(job domain.Job, tasks []domain.Task) JobCard { return c } -func uiParameters(parameters map[string]any) []ParameterCard { - keys := []struct { - key string - label string - }{ - {"query_smiles", "Target SMILES"}, - {"query_id", "Target ChEMBL ID"}, - {"top_k", "Global top-k"}, - {"threshold", "Similarity threshold"}, - {"threshold_direction", "Threshold direction"}, +func uiParameters(parameters map[string]any, catalog *workloads.Catalog, workload string) []ParameterCard { + labels := map[string]string{ + "query_smiles": "Target SMILES", + "query_id": "Target ChEMBL ID", + "top_k": "Global top-k", + "threshold": "Similarity threshold", + "threshold_direction": "Threshold direction", + "min_molwt": "Minimum molecular weight", + "max_molwt": "Maximum molecular weight", + "skip_invalid": "Skip invalid molecules", + "block_size": "Block size", } + keys := make([]string, 0, len(parameters)) + declared := declaredParameterNames(catalog, workload) + for key := range parameters { + if declared != nil && !declared[key] { + // Only schema-declared scientific parameters may reach the browser; + // anything else could carry internal coordinator state. + continue + } + keys = append(keys, key) + } + sort.Strings(keys) out := make([]ParameterCard, 0, len(keys)) - for _, entry := range keys { - value, ok := parameters[entry.key] + for _, key := range keys { + value, ok := parameters[key] if !ok { continue } formatted, ok := formatUIParameter(value) - if ok { - out = append(out, ParameterCard{Label: entry.label, Value: formatted}) + if !ok { + continue } + label := labels[key] + if label == "" { + label = strings.ReplaceAll(key, "_", " ") + } + out = append(out, ParameterCard{Label: label, Value: formatted}) } return out } +func declaredParameterNames(catalog *workloads.Catalog, workload string) map[string]bool { + if catalog == nil || workload == "" { + return nil + } + item := catalog.ByName(workload) + if item == nil { + return nil + } + properties, ok := item.Parameters["properties"].(map[string]any) + if !ok { + return nil + } + declared := make(map[string]bool, len(properties)) + for name := range properties { + declared[name] = true + } + return declared +} + func formatUIParameter(value any) (string, bool) { switch typed := value.(type) { case string: diff --git a/coordinator/internal/usecase/ui_internal_test.go b/coordinator/internal/usecase/ui_internal_test.go index b16df70..6324abc 100644 --- a/coordinator/internal/usecase/ui_internal_test.go +++ b/coordinator/internal/usecase/ui_internal_test.go @@ -1,14 +1,22 @@ package usecase -import "testing" +import ( + "testing" + + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" +) func TestUIParametersAreAllowlisted(t *testing.T) { + catalog, err := workloads.Load() + if err != nil { + t.Fatal(err) + } parameters := uiParameters(map[string]any{ "query_smiles": "CCO", "top_k": float64(20), "internal_storage_key": "must-not-reach-browser", "nested": map[string]any{"secret": "no"}, - }) + }, catalog, "similarity-search") if len(parameters) != 2 { t.Fatalf("parameters = %#v, want only two allowlisted values", parameters) } @@ -17,3 +25,28 @@ func TestUIParametersAreAllowlisted(t *testing.T) { t.Fatalf("parameters = %#v", parameters) } } + +func TestUIParametersRenderEverySchemaDeclaredField(t *testing.T) { + catalog, err := workloads.Load() + if err != nil { + t.Fatal(err) + } + parameters := uiParameters(map[string]any{ + "min_molwt": 100, + "max_molwt": 600, + "skip_invalid": true, + "secret_key": "no", + }, catalog, "molwt-filter") + if len(parameters) != 3 { + t.Fatalf("parameters = %#v, want three declared values", parameters) + } + labels := map[string]bool{} + for _, card := range parameters { + labels[card.Label] = true + } + for _, expected := range []string{"Minimum molecular weight", "Maximum molecular weight", "Skip invalid molecules"} { + if !labels[expected] { + t.Errorf("missing parameter card %q in %#v", expected, parameters) + } + } +} diff --git a/coordinator/internal/usecase/ui_scope_test.go b/coordinator/internal/usecase/ui_scope_test.go index de3a618..7499b75 100644 --- a/coordinator/internal/usecase/ui_scope_test.go +++ b/coordinator/internal/usecase/ui_scope_test.go @@ -19,7 +19,7 @@ func newDashboard() (*usecase.Dashboard, *memstore.JobRepo) { tasks := memstore.NewTaskRepo() workers := memstore.NewWorkerRepo() artifacts := memstore.NewArtifactRepo() - return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), jobs + return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), testCatalog()), jobs } func ownedJob(t *testing.T, jobs *memstore.JobRepo, owner uuid.UUID) uuid.UUID { diff --git a/coordinator/internal/usecase/ui_workers_scope_test.go b/coordinator/internal/usecase/ui_workers_scope_test.go index 2da6547..5991e28 100644 --- a/coordinator/internal/usecase/ui_workers_scope_test.go +++ b/coordinator/internal/usecase/ui_workers_scope_test.go @@ -17,7 +17,7 @@ func newDashboardWithWorkers() (*usecase.Dashboard, *memstore.WorkerRepo) { tasks := memstore.NewTaskRepo() workers := memstore.NewWorkerRepo() artifacts := memstore.NewArtifactRepo() - return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), workers + return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), testCatalog()), workers } func seedWorker(t *testing.T, workers *memstore.WorkerRepo, owner *uuid.UUID, name string) { diff --git a/coordinator/internal/usecase/upload.go b/coordinator/internal/usecase/upload.go index 4bd5caf..8fa4798 100644 --- a/coordinator/internal/usecase/upload.go +++ b/coordinator/internal/usecase/upload.go @@ -4,12 +4,12 @@ import ( "context" "fmt" "io" - "math" "github.com/google/uuid" "github.com/emil28092005/SciMesh/coordinator/internal/chunk" "github.com/emil28092005/SciMesh/coordinator/internal/domain" + "github.com/emil28092005/SciMesh/coordinator/internal/workloads" ) // SubmitDataset accepts an uploaded dataset, splits it into shard artifacts, and @@ -23,15 +23,16 @@ type SubmitDataset struct { tx TxManager clk Clock maxAttempts int + catalog *workloads.Catalog } func NewSubmitDataset(blobs BlobStore, artifacts ArtifactRepository, jobs JobRepository, - tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int) *SubmitDataset { - return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts} + tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int, catalog *workloads.Catalog) *SubmitDataset { + return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts, catalog: catalog} } func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (SubmitDatasetResult, error) { - if err := validateUploadedWorkload(in.Workload, in.Parameters); err != nil { + if err := validateUploadedWorkload(uc.catalog, in.Workload, in.Parameters); err != nil { return SubmitDatasetResult{}, err } if uc.maxAttempts < 1 { @@ -86,7 +87,8 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su putKeys = append(putKeys, art.StorageKey) art.SetContent(ssum, ssize) - task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum, in.Parameters, uc.maxAttempts, now) + task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum, + taskParameterSubset(in.Parameters), uc.maxAttempts, now) if err != nil { return err } @@ -128,74 +130,38 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su }, nil } -// validateUploadedWorkload is deliberately narrow until CTX-07/08/10 adds a -// typed distributed-workload registry. In particular, running similarity-graph -// independently per TSV shard is scientifically wrong: cross-shard pairs would -// be absent from the apparent graph. -func validateUploadedWorkload(workload string, parameters map[string]any) error { - if workload != "similarity-search" { +// validateUploadedWorkload checks the submitted workload against the embedded +// catalog: it must be an enabled, upload-ready workload whose parameters +// satisfy the declared JSON schema. Workloads that need planner-produced +// inputs (such as the graph block-pair shards) declare upload_ready=false and +// cannot be driven from a single uploaded dataset. +func validateUploadedWorkload(catalog *workloads.Catalog, workload string, parameters map[string]any) error { + if catalog == nil { return domain.ErrInvalidInput } - allowed := map[string]struct{}{ - "query_smiles": {}, "top_k": {}, "threshold": {}, - "threshold_direction": {}, "progress_every": {}, - } - for key := range parameters { - if _, ok := allowed[key]; !ok { - return domain.ErrInvalidInput - } - } - query, ok := parameters["query_smiles"].(string) - if !ok || query == "" || len(query) > 200 { + if err := catalog.ValidateParameters(workload, parameters); err != nil { return domain.ErrInvalidInput } - if value, ok := parameters["top_k"]; ok && !isPositiveJSONInteger(value) { - return domain.ErrInvalidInput - } - if value, ok := parameters["progress_every"]; ok && !isNonNegativeJSONInteger(value) { - return domain.ErrInvalidInput - } - if value, ok := parameters["threshold"]; ok && !isUnitIntervalNumber(value) { - return domain.ErrInvalidInput - } - if value, ok := parameters["threshold_direction"]; ok && value != "greater" && value != "less" { + if !catalog.UploadReady(workload) { return domain.ErrInvalidInput } return nil } -func isPositiveJSONInteger(value any) bool { return isJSONInteger(value, false) } -func isNonNegativeJSONInteger(value any) bool { return isJSONInteger(value, true) } - -func isJSONInteger(value any, allowZero bool) bool { - var n int64 - switch v := value.(type) { - case int: - n = int64(v) - case int64: - n = v - case float64: - if math.Trunc(v) != v || v > math.MaxInt64 || v < math.MinInt64 { - return false +// taskParameterSubset drops coordinator-level keys from the parameters that +// are handed to workers. max_rows is a plan-time bound applied by the chunker +// here; a worker would reject it as outside its stage projection. +func taskParameterSubset(parameters map[string]any) map[string]any { + if _, present := parameters["max_rows"]; !present { + return parameters + } + subset := make(map[string]any, len(parameters)-1) + for key, value := range parameters { + if key != "max_rows" { + subset[key] = value } - n = int64(v) - default: - return false - } - return n >= 0 && (allowZero || n > 0) -} - -func isUnitIntervalNumber(value any) bool { - switch v := value.(type) { - case float64: - return !math.IsNaN(v) && !math.IsInf(v, 0) && v >= 0 && v <= 1 - case int: - return v >= 0 && v <= 1 - case int64: - return v >= 0 && v <= 1 - default: - return false } + return subset } // GetTaskInput resolves a task's input shard and opens it for streaming. The diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index b64111f..842eca9 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -73,20 +73,20 @@ func newHarness() *harness { } tx := memstore.Tx{} h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk) - h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3) - h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease) + h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3, testCatalog()) + h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease, testCatalog()) h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease) - h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2) - h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk) + h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2, testCatalog()) + h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk, testCatalog()) h.status = usecase.NewGetJobStatus(h.jobs, h.tasks) h.results = usecase.NewListResults(h.tasks) h.register = usecase.NewRegisterWorker(h.work, h.clk) h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.work, h.arts, h.blobs, tx, h.clk) h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs) h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs) - h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk) + h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk, testCatalog()) h.cancel = usecase.NewCancelJob(h.jobs, h.tasks, tx, h.clk) - h.reduce = usecase.NewReduceJob(h.jobs, h.tasks, h.arts, h.blobs, tx, h.clk) + h.reduce = usecase.NewReduceJob(h.jobs, h.tasks, h.arts, h.blobs, tx, h.clk, testCatalog()) h.jobResult = usecase.NewGetJobResult(h.jobs, h.downloadArt) return h } @@ -133,6 +133,44 @@ func TestSimilaritySearchReductionCreatesFinalArtifact(t *testing.T) { } } +func TestMolwtFilterReductionConcatenatesPartialsInOrder(t *testing.T) { + h := newHarness() + jobID := h.seedJob(t, "molwt-filter", 2) + if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil { + t.Fatal(err) + } + partials := []string{ + "chembl_id,canonical_smiles\nA,CC\n", + "chembl_id,canonical_smiles\nB,CCCC\n", + } + for _, partial := range partials { + taskID, attempt := h.leaseOne(t, "w1", "molwt-filter") + art, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, Filename: "partial.csv", ContentType: "text/csv", Body: strings.NewReader(partial)}) + if err != nil { + t.Fatal(err) + } + if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, ResultArtifactID: art.ID}); err != nil { + t.Fatal(err) + } + } + if err := h.reduce.Execute(ctx, jobID); err != nil { + t.Fatal(err) + } + progress, err := h.status.Execute(ctx, jobID) + if err != nil || progress.Job.Status != domain.JobCompleted { + t.Fatalf("status=%s err=%v", progress.Job.Status, err) + } + art, body, err := h.jobResult.Execute(ctx, jobID) + if err != nil { + t.Fatal(err) + } + defer body.Close() + bytes, _ := io.ReadAll(body) + if art.Kind != domain.ArtifactFinalResult || string(bytes) != "chembl_id,canonical_smiles\nA,CC\nB,CCCC\n" { + t.Fatalf("unexpected final %q", bytes) + } +} + func TestSimilaritySearchReductionFailureIsSanitized(t *testing.T) { h := newHarness() jobID := h.seedJob(t, "similarity-search", 1) @@ -817,8 +855,18 @@ func TestSubmitDatasetRejectsUnsupportedDistributedWorkloads(t *testing.T) { Filename: "chembl.tsv", ContentType: "text/tab-separated-values", Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"), }) + if err != nil { + t.Errorf("query_id submission err = %v, want nil", err) + } + _, err = h.submit.Execute(ctx, usecase.SubmitDatasetInput{ + Workload: "similarity-search", Parameters: map[string]any{ + "query_id": "CHEMBL1", "query_smiles": "CCO", + }, RowsPerShard: 2, + Filename: "chembl.tsv", ContentType: "text/tab-separated-values", + Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"), + }) if !errors.Is(err, domain.ErrInvalidInput) { - t.Errorf("query_id submission err = %v, want ErrInvalidInput", err) + t.Errorf("both query fields submission err = %v, want ErrInvalidInput", err) } } diff --git a/coordinator/internal/workloads/catalog.go b/coordinator/internal/workloads/catalog.go new file mode 100644 index 0000000..2dbcb86 --- /dev/null +++ b/coordinator/internal/workloads/catalog.go @@ -0,0 +1,273 @@ +// Package workloads provides the coordinator-side view of the SDK workload +// library. The catalog is generated by `scimesh workload export` and embedded +// into the binary; it is presentation and orchestration metadata, never +// executable code. Every field that drives coordinator behaviour (reduction +// mode, parameter schema, required input columns) is validated at load time +// so a bad export fails fast instead of misbehaving at runtime. +package workloads + +import ( + "embed" + "encoding/json" + "fmt" + "sort" +) + +//go:embed workloads.json +var catalogFile embed.FS + +// Catalog is the parsed and validated workload library. +type Catalog struct { + workloads []*Workload +} + +// Workload is one entry of the embedded catalog. +type Workload struct { + Name string `json:"name"` + Version string `json:"version"` + Description string `json:"description"` + Capabilities []string `json:"capabilities"` + TrustModes []string `json:"trust_modes"` + Determinism string `json:"determinism"` + Verifier string `json:"verifier"` + Enabled bool `json:"enabled"` + Reduction string `json:"reduction"` + UploadReady bool `json:"upload_ready"` + Parameters map[string]any `json:"parameters_schema"` + UIElements []UIElement `json:"ui_elements"` + Inputs map[string]any `json:"inputs"` + Outputs map[string]any `json:"outputs"` + requiredColumns map[string]bool // derived from input validator configuration + inputMediaTypes map[string]string // port name -> media type + parameterDefaults map[string]any // derived from the schema +} + +// UIElement is one workload-declared form control for the "new job" page. +type UIElement struct { + Field string `json:"field"` + Widget string `json:"widget"` + Label string `json:"label"` + Help string `json:"help"` + Placeholder string `json:"placeholder"` + Options []string `json:"options"` + Default any `json:"default"` + Order int `json:"order"` + Group string `json:"group"` +} + +type libraryFile struct { + SchemaVersion int `json:"schema_version"` + GeneratedBy string `json:"generated_by"` + Workloads []*Workload `json:"workloads"` +} + +// Load reads and validates the embedded catalog. +func Load() (*Catalog, error) { + raw, err := catalogFile.ReadFile("workloads.json") + if err != nil { + return nil, fmt.Errorf("read embedded workload catalog: %w", err) + } + return Parse(raw) +} + +// Parse validates and builds a Catalog from catalog JSON bytes. +func Parse(raw []byte) (*Catalog, error) { + var file libraryFile + if err := json.Unmarshal(raw, &file); err != nil { + return nil, fmt.Errorf("parse workload catalog: %w", err) + } + if file.SchemaVersion != 2 { + return nil, fmt.Errorf("workload catalog schema_version must be 2, got %d", file.SchemaVersion) + } + if len(file.Workloads) == 0 { + return nil, fmt.Errorf("workload catalog contains no workloads") + } + catalog := &Catalog{} + names := make(map[string]bool, len(file.Workloads)) + for _, workload := range file.Workloads { + if err := validateWorkload(workload); err != nil { + return nil, err + } + if names[workload.Name] { + return nil, fmt.Errorf("workload catalog lists %q more than once", workload.Name) + } + names[workload.Name] = true + catalog.workloads = append(catalog.workloads, workload) + } + sort.Slice(catalog.workloads, func(i, j int) bool { + return catalog.workloads[i].Name < catalog.workloads[j].Name + }) + return catalog, nil +} + +func validateWorkload(workload *Workload) error { + if workload.Name == "" || workload.Version == "" { + return fmt.Errorf("workload catalog entry must have a name and version") + } + switch workload.Reduction { + case "top-k", "ordered-concat": + default: + return fmt.Errorf("workload %q declares unknown reduction %q", workload.Name, workload.Reduction) + } + if err := validateSchema(workload.Name, workload.Parameters); err != nil { + return err + } + workload.parameterDefaults = schemaDefaults(workload.Parameters) + workload.requiredColumns = map[string]bool{} + for portName, port := range workload.Inputs { + config, ok := port.(map[string]any) + if !ok { + continue + } + validator, _ := config["validator_configuration"].(map[string]any) + columns, _ := validator["required_columns"].([]any) + for _, column := range columns { + text, ok := column.(string) + if ok { + workload.requiredColumns[text] = true + } + } + if mediaType, ok := config["media_type"].(string); ok { + if workload.inputMediaTypes == nil { + workload.inputMediaTypes = map[string]string{} + } + workload.inputMediaTypes[portName] = mediaType + } + } + for _, element := range workload.UIElements { + if err := validateUIElement(workload, element); err != nil { + return err + } + } + return nil +} + +func validateUIElement(workload *Workload, element UIElement) error { + switch element.Widget { + case "text", "textarea", "number", "select", "checkbox": + default: + return fmt.Errorf("workload %q ui element %q has unknown widget %q", workload.Name, element.Field, element.Widget) + } + if element.Widget == "select" && len(element.Options) == 0 { + return fmt.Errorf("workload %q ui element %q is a select without options", workload.Name, element.Field) + } + properties, ok := workload.Parameters["properties"].(map[string]any) + if !ok { + return nil + } + property, declared := properties[element.Field].(map[string]any) + if !declared { + return fmt.Errorf("workload %q ui element %q does not name a declared parameter", workload.Name, element.Field) + } + if schemaType(property) == "boolean" && element.Widget != "checkbox" { + return fmt.Errorf("workload %q ui element %q must use the checkbox widget for a boolean parameter", workload.Name, element.Field) + } + return nil +} + +// Enabled returns the enabled workloads, sorted by name. +func (c *Catalog) Enabled() []*Workload { + result := make([]*Workload, 0, len(c.workloads)) + for _, workload := range c.workloads { + if workload.Enabled { + result = append(result, workload) + } + } + return result +} + +// ByName returns the workload with the given name, or nil. +func (c *Catalog) ByName(name string) *Workload { + for _, workload := range c.workloads { + if workload.Name == name { + return workload + } + } + return nil +} + +// ValidateParameters checks job parameters against the workload schema. It +// enforces the strict subset of JSON Schema used by the SDK manifests: object +// shape, types, required, enum, numeric bounds, string lengths, and oneOf. +func (c *Catalog) ValidateParameters(name string, parameters map[string]any) error { + workload := c.ByName(name) + if workload == nil { + return fmt.Errorf("unknown workload %q", name) + } + if !workload.Enabled { + return fmt.Errorf("workload %q is not enabled", name) + } + return validateParameters(name, workload.Parameters, parameters) +} + +// UploadReady reports whether the workload can be driven from a single +// uploaded dataset file. Workloads that need planner-produced inputs (such as +// the graph block-pair shards) declare upload_ready=false. +func (c *Catalog) UploadReady(name string) bool { + workload := c.ByName(name) + if workload == nil { + return false + } + return workload.UploadReady +} + +// RequiredColumns reports every column the workload's input port requires. +func (c *Catalog) RequiredColumns(name string) map[string]bool { + workload := c.ByName(name) + if workload == nil { + return nil + } + return workload.requiredColumns +} + +// InputMediaType returns the declared media type of the named input port. +func (c *Catalog) InputMediaType(name, port string) string { + workload := c.ByName(name) + if workload == nil { + return "" + } + return workload.inputMediaTypes[port] +} + +// InputPortNames returns the sorted input port names of the workload. +func (c *Catalog) InputPortNames(name string) []string { + workload := c.ByName(name) + if workload == nil { + return nil + } + names := make([]string, 0, len(workload.Inputs)) + for port := range workload.Inputs { + names = append(names, port) + } + sort.Strings(names) + return names +} + +// Reduction returns the reduction mode for the workload, or "" if unknown. +func (c *Catalog) Reduction(name string) string { + workload := c.ByName(name) + if workload == nil { + return "" + } + return workload.Reduction +} + +// ParameterDefaults returns the schema-declared defaults for the workload. +func (c *Catalog) ParameterDefaults(name string) map[string]any { + workload := c.ByName(name) + if workload == nil { + return nil + } + return workload.parameterDefaults +} + +// ParameterSchema returns the schema property for a workload parameter, and +// whether the parameter exists. +func (w *Workload) PropertySchema(field string) (map[string]any, bool) { + properties, ok := w.Parameters["properties"].(map[string]any) + if !ok { + return nil, false + } + property, ok := properties[field].(map[string]any) + return property, ok +} diff --git a/coordinator/internal/workloads/schema.go b/coordinator/internal/workloads/schema.go new file mode 100644 index 0000000..ea60f53 --- /dev/null +++ b/coordinator/internal/workloads/schema.go @@ -0,0 +1,367 @@ +package workloads + +import ( + "fmt" + "math" + "sort" +) + +// validateSchema checks the strict JSON Schema subset used by SDK manifests. +// It mirrors what the SDK registry enforces on the Python side: an object +// schema with additionalProperties=false, typed properties, and the keyword +// subset the coordinator understands (type, enum, required, minimum/maximum, +// minLength/maxLength, oneOf, not). +func validateSchema(workloadName string, schema map[string]any) error { + if schema == nil { + return fmt.Errorf("workload %q has no parameter schema", workloadName) + } + if err := validateSchemaNode(workloadName+".parameters_schema", schema); err != nil { + return err + } + if schemaType(schema) != "object" { + return fmt.Errorf("workload %q parameter schema must be an object schema", workloadName) + } + if additional, ok := schema["additionalProperties"].(bool); !ok || additional { + return fmt.Errorf("workload %q parameter schema must set additionalProperties=false", workloadName) + } + properties, ok := schema["properties"].(map[string]any) + if !ok { + return fmt.Errorf("workload %q parameter schema must declare properties", workloadName) + } + for name, property := range properties { + child, ok := property.(map[string]any) + if !ok { + return fmt.Errorf("workload %q parameter %q must be a schema object", workloadName, name) + } + if err := validateSchemaNode(name, child); err != nil { + return err + } + } + return nil +} + +func validateSchemaNode(field string, node map[string]any) error { + for keyword := range node { + switch keyword { + case "type", "enum", "required", "minimum", "maximum", "exclusiveMinimum", + "exclusiveMaximum", "minLength", "maxLength", "properties", + "additionalProperties", "oneOf", "not", "default", "description", + "items", "minItems", "maxItems": + default: + return fmt.Errorf("%s uses unsupported JSON Schema keyword %q", field, keyword) + } + } + if rawType, ok := node["type"]; ok { + schemaType, ok := rawType.(string) + if !ok { + return fmt.Errorf("%s type must be a string", field) + } + switch schemaType { + case "string", "number", "integer", "boolean", "object", "array": + default: + return fmt.Errorf("%s has unknown type %q", field, schemaType) + } + } + for _, keyword := range []string{"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum"} { + if value, ok := node[keyword]; ok { + if number, ok := value.(float64); !ok || math.IsNaN(number) || math.IsInf(number, 0) { + return fmt.Errorf("%s %s must be a finite number", field, keyword) + } + } + } + for _, keyword := range []string{"minLength", "maxLength", "minItems", "maxItems"} { + if value, ok := node[keyword]; ok { + if number, ok := value.(float64); !ok || number < 0 || number != math.Trunc(number) { + return fmt.Errorf("%s %s must be a non-negative integer", field, keyword) + } + } + } + if required, ok := node["required"]; ok { + entries, ok := required.([]any) + if !ok { + return fmt.Errorf("%s required must be an array of strings", field) + } + for _, entry := range entries { + if _, ok := entry.(string); !ok { + return fmt.Errorf("%s required must be an array of strings", field) + } + } + } + if enums, ok := node["enum"]; ok { + entries, ok := enums.([]any) + if !ok || len(entries) == 0 { + return fmt.Errorf("%s enum must be a non-empty array", field) + } + } + if oneOf, ok := node["oneOf"]; ok { + entries, ok := oneOf.([]any) + if !ok || len(entries) == 0 { + return fmt.Errorf("%s oneOf must be a non-empty array", field) + } + for index, entry := range entries { + child, ok := entry.(map[string]any) + if !ok { + return fmt.Errorf("%s oneOf[%d] must be a schema object", field, index) + } + if err := validateSchemaNode(fmt.Sprintf("%s.oneOf[%d]", field, index), child); err != nil { + return err + } + } + } + if child, ok := node["not"]; ok { + not, ok := child.(map[string]any) + if !ok { + return fmt.Errorf("%s not must be a schema object", field) + } + if err := validateSchemaNode(field+".not", not); err != nil { + return err + } + } + if items, ok := node["items"]; ok { + child, ok := items.(map[string]any) + if !ok { + return fmt.Errorf("%s items must be a schema object", field) + } + if err := validateSchemaNode(field+".items", child); err != nil { + return err + } + } + if properties, ok := node["properties"]; ok { + entries, ok := properties.(map[string]any) + if !ok { + return fmt.Errorf("%s properties must be an object", field) + } + for name, property := range entries { + child, ok := property.(map[string]any) + if !ok { + return fmt.Errorf("%s property %q must be a schema object", field, name) + } + if err := validateSchemaNode(field+"."+name, child); err != nil { + return err + } + } + } + return nil +} + +func schemaType(node map[string]any) string { + rawType, _ := node["type"].(string) + return rawType +} + +// validateParameters checks values against the strict schema subset. +func validateParameters(workloadName string, schema map[string]any, parameters map[string]any) error { + properties, _ := schema["properties"].(map[string]any) + for name := range parameters { + if _, declared := properties[name]; !declared { + return fmt.Errorf("workload %q does not accept parameter %q", workloadName, name) + } + } + required, _ := schema["required"].([]any) + for _, name := range required { + field, _ := name.(string) + if _, present := parameters[field]; !present { + return fmt.Errorf("workload %q requires parameter %q", workloadName, field) + } + } + for name, value := range parameters { + property, declared := properties[name].(map[string]any) + if !declared { + continue + } + if err := validateProperty(workloadName+"."+name, property, value); err != nil { + return err + } + } + if oneOf, ok := schema["oneOf"].([]any); ok && len(oneOf) > 0 { + if err := validateOneOf(workloadName, oneOf, parameters); err != nil { + return err + } + } + return nil +} + +func validateProperty(field string, property map[string]any, value any) error { + if enums, ok := property["enum"].([]any); ok { + for _, candidate := range enums { + if valuesEqual(candidate, value) { + return nil + } + } + return fmt.Errorf("%s must be one of the declared enum values", field) + } + switch schemaType(property) { + case "string": + text, ok := value.(string) + if !ok { + return fmt.Errorf("%s must be a string", field) + } + if minimum, ok := lengthBound(property["minLength"]); ok && len([]rune(text)) < minimum { + return fmt.Errorf("%s is shorter than the minimum length", field) + } + if maximum, ok := lengthBound(property["maxLength"]); ok && len([]rune(text)) > maximum { + return fmt.Errorf("%s exceeds the maximum length", field) + } + case "number", "integer": + number, ok := asFloat(value) + if !ok { + return fmt.Errorf("%s must be a number", field) + } + if schemaType(property) == "integer" && number != math.Trunc(number) { + return fmt.Errorf("%s must be an integer", field) + } + if minimum, ok := numberBound(property["minimum"]); ok && number < minimum { + return fmt.Errorf("%s is below the minimum", field) + } + if maximum, ok := numberBound(property["maximum"]); ok && number > maximum { + return fmt.Errorf("%s exceeds the maximum", field) + } + case "boolean": + if _, ok := value.(bool); !ok { + return fmt.Errorf("%s must be a boolean", field) + } + case "object": + child, ok := value.(map[string]any) + if !ok { + return fmt.Errorf("%s must be an object", field) + } + properties, _ := property["properties"].(map[string]any) + for name := range child { + if _, declared := properties[name]; !declared { + return fmt.Errorf("%s has undeclared field %q", field, name) + } + } + case "array": + items, ok := value.([]any) + if !ok { + return fmt.Errorf("%s must be an array", field) + } + if itemSchema, ok := property["items"].(map[string]any); ok { + for index, item := range items { + if err := validateProperty(fmt.Sprintf("%s[%d]", field, index), itemSchema, item); err != nil { + return err + } + } + } + case "": + // No type keyword: enum-only properties are handled above. + return fmt.Errorf("%s has no JSON Schema type", field) + } + return nil +} + +func validateOneOf(workloadName string, oneOf []any, parameters map[string]any) error { + satisfied := 0 + for _, candidate := range oneOf { + option, ok := candidate.(map[string]any) + if !ok { + continue + } + if optionSatisfied(option, parameters) { + satisfied++ + } + } + if satisfied != 1 { + return fmt.Errorf("workload %q requires exactly one of the declared parameter alternatives", workloadName) + } + return nil +} + +func optionSatisfied(option map[string]any, parameters map[string]any) bool { + if required, ok := option["required"].([]any); ok { + for _, name := range required { + field, _ := name.(string) + if _, present := parameters[field]; !present { + return false + } + } + } + if not, ok := option["not"].(map[string]any); ok { + if required, ok := not["required"].([]any); ok { + for _, name := range required { + field, _ := name.(string) + if _, present := parameters[field]; present { + return false + } + } + } + } + return true +} + +func lengthBound(value any) (int, bool) { + number, ok := value.(float64) + if !ok || number != math.Trunc(number) { + return 0, false + } + return int(number), true +} + +func numberBound(value any) (float64, bool) { + number, ok := asFloat(value) + if !ok || math.IsNaN(number) || math.IsInf(number, 0) { + return 0, false + } + return number, true +} + +func asFloat(value any) (float64, bool) { + switch v := value.(type) { + case float64: + return v, true + case float32: + return float64(v), true + case int: + return float64(v), true + case int32: + return float64(v), true + case int64: + return float64(v), true + } + return 0, false +} + +func valuesEqual(left, right any) bool { + switch l := left.(type) { + case float64: + r, ok := right.(float64) + return ok && l == r + case string: + r, ok := right.(string) + return ok && l == r + case bool: + r, ok := right.(bool) + return ok && l == r + case nil: + return right == nil + } + return false +} + +// schemaDefaults collects the declared default for each property. The UI uses +// these to pre-fill controls that have no workload-declared UI default. +func schemaDefaults(schema map[string]any) map[string]any { + properties, _ := schema["properties"].(map[string]any) + defaults := map[string]any{} + for name, property := range properties { + child, ok := property.(map[string]any) + if !ok { + continue + } + if value, present := child["default"]; present { + defaults[name] = value + } + } + return defaults +} + +// SortedFields returns the sorted declared parameter names. +func SortedFields(schema map[string]any) []string { + properties, _ := schema["properties"].(map[string]any) + names := make([]string, 0, len(properties)) + for name := range properties { + names = append(names, name) + } + sort.Strings(names) + return names +} diff --git a/coordinator/internal/transport/http/workloads.json b/coordinator/internal/workloads/workloads.json similarity index 73% rename from coordinator/internal/transport/http/workloads.json rename to coordinator/internal/workloads/workloads.json index a4fff29..15f0948 100644 --- a/coordinator/internal/transport/http/workloads.json +++ b/coordinator/internal/workloads/workloads.json @@ -1,6 +1,6 @@ { "generated_by": "scimesh workload export", - "schema_version": 1, + "schema_version": 2, "workloads": [ { "capabilities": [ @@ -146,10 +146,25 @@ }, "type": "object" }, + "reduction": "ordered-concat", "trust_modes": [ "trusted", "untrusted_quorum" ], + "ui_elements": [ + { + "default": true, + "field": "skip_invalid", + "group": "", + "help": "Skip rows with invalid SMILES instead of failing the shard.", + "label": "Skip invalid molecules", + "options": [], + "order": 1, + "placeholder": "", + "widget": "checkbox" + } + ], + "upload_ready": true, "verifier": "exact-artifact@1", "version": "1.0.0" }, @@ -227,10 +242,47 @@ }, "type": "object" }, + "reduction": "ordered-concat", "trust_modes": [ "trusted", "untrusted_quorum" ], + "ui_elements": [ + { + "default": null, + "field": "min_molwt", + "group": "", + "help": "Keep molecules with MolWt at least this value. Optional.", + "label": "Minimum molecular weight", + "options": [], + "order": 1, + "placeholder": "e.g. 100", + "widget": "number" + }, + { + "default": null, + "field": "max_molwt", + "group": "", + "help": "Keep molecules with MolWt at most this value. Optional.", + "label": "Maximum molecular weight", + "options": [], + "order": 2, + "placeholder": "e.g. 600", + "widget": "number" + }, + { + "default": true, + "field": "skip_invalid", + "group": "", + "help": "Skip rows with invalid SMILES instead of failing the shard.", + "label": "Skip invalid molecules", + "options": [], + "order": 3, + "placeholder": "", + "widget": "checkbox" + } + ], + "upload_ready": true, "verifier": "exact-artifact@1", "version": "1.0.0" }, @@ -315,10 +367,50 @@ ], "type": "object" }, + "reduction": "ordered-concat", "trust_modes": [ "trusted", "untrusted_quorum" ], + "ui_elements": [ + { + "default": null, + "field": "threshold", + "group": "", + "help": "Minimum (greater) or maximum (less) edge similarity. Required.", + "label": "Similarity threshold", + "options": [], + "order": 1, + "placeholder": "", + "widget": "number" + }, + { + "default": "greater", + "field": "threshold_direction", + "group": "", + "help": "Whether to keep edges above (greater) or below (less) the threshold.", + "label": "Direction", + "options": [ + "greater", + "less" + ], + "order": 2, + "placeholder": "", + "widget": "select" + }, + { + "default": 100, + "field": "block_size", + "group": "", + "help": "Deterministic block size for pair sharding.", + "label": "Block size", + "options": [], + "order": 3, + "placeholder": "", + "widget": "number" + } + ], + "upload_ready": false, "verifier": "exact-artifact@1", "version": "1.0.0" }, @@ -437,10 +529,72 @@ }, "type": "object" }, + "reduction": "top-k", "trust_modes": [ "trusted", "untrusted_quorum" ], + "ui_elements": [ + { + "default": null, + "field": "query_id", + "group": "", + "help": "ChEMBL id of the query molecule. Provide exactly one of id or SMILES.", + "label": "Query molecule id", + "options": [], + "order": 1, + "placeholder": "", + "widget": "text" + }, + { + "default": null, + "field": "query_smiles", + "group": "", + "help": "SMILES of the query molecule. Provide exactly one of id or SMILES.", + "label": "Query molecule SMILES", + "options": [], + "order": 2, + "placeholder": "", + "widget": "text" + }, + { + "default": 20, + "field": "top_k", + "group": "", + "help": "Number of most similar molecules to keep per shard (global merge keeps the best of these).", + "label": "Top k", + "options": [], + "order": 3, + "placeholder": "", + "widget": "number" + }, + { + "default": "greater", + "field": "threshold_direction", + "group": "", + "help": "Keep molecules with similarity greater or less than the threshold.", + "label": "Direction", + "options": [ + "greater", + "less" + ], + "order": 4, + "placeholder": "", + "widget": "select" + }, + { + "default": null, + "field": "threshold", + "group": "", + "help": "Optional similarity bound: results are filtered to this direction.", + "label": "Similarity threshold", + "options": [], + "order": 5, + "placeholder": "e.g. 0.8", + "widget": "number" + } + ], + "upload_ready": true, "verifier": "exact-artifact@1", "version": "1.0.0" }