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.
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.
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.
Shard
State
Attempt
Worker / lease
Outcome
{{range .Tasks}}
#{{.ChunkIndex}}
{{statusLabel .Status}}
{{.Attempt}} / {{.MaxAttempts}}
{{if .LeaseOwner}}{{.LeaseOwner}}{{if .LeaseExpiresAt}} lease until {{time .LeaseExpiresAt}}{{end}}{{else}}—{{end}}
@@ -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.
Choose a workload from the installed library. Upload its dataset; workers compute shards; the coordinator reduces the partials into one final artifact.