Upgrade pipeline observability UI

Add final result reduction
2026-07-24 15:13:59 +03:00 · 2026-07-24 14:58:18 +03:00
36 changed files with 1197 additions and 693 deletions
@@ -3,10 +3,10 @@
 SciMesh is a scientific-workload framework for molecular datasets. Its public CLI
 runs exact similarity search and sparse similarity-graph construction locally in
 one Python process; it creates no dense similarity matrix. The Go/PostgreSQL
-coordinator and Python worker can run a diagnostic, shard-based
-`similarity-search` pipeline locally. Its CSV artifacts are not a global result
-until CTX-07--09 add planning and reduction; use the local CLI for scientific
-results today. See [`STATUS.md`](STATUS.md).
+coordinator and Python worker can run a shard-based `similarity-search`
+pipeline locally. After every shard succeeds, the coordinator deterministically
+merges its candidates into one final global top-k CSV. See
+[`STATUS.md`](STATUS.md).

 The ChEMBL TSV database is intentionally not included in this repository. Download it separately and pass its path to the commands below. The expected columns are `chembl_id` and `canonical_smiles`.

@@ -1,7 +1,7 @@
 # SciMesh Status

 **Updated:** 2026-07-24
-**Branch baseline:** `main` at `f953112` (distributed pipeline hardening)
+**Branch baseline:** `main` at `6e67daa` (distributed similarity-search)

 ## Current state

@@ -18,8 +18,10 @@ the reference behaviour for future distributed execution:
 The Go coordinator and its PostgreSQL-backed task lifecycle are implemented:
 registration, atomic claiming, lease renewal, artifact storage, dataset
 chunking, result/failure reporting, and job progress. The Python worker now
-uses the live coordinator contract; its HTTP path was exercised against a real
-Docker PostgreSQL stack on 2026-07-23.
+uses the live coordinator contract. Completed similarity-search shard results
+are reduced once into a checksum-protected final CSV, which is downloadable
+through the coordinator. The full Go checks (including a fresh migration and
+real PostgreSQL smoke test) passed on 2026-07-24.

 ## Milestone tracker

@@ -33,30 +35,24 @@ Docker PostgreSQL stack on 2026-07-23.
 | CTX-05 Artifact storage | Implemented | Coordinator-owned inputs/results, checksum verification, and upload flow. |
 | CTX-06 Python Worker live-contract alignment | Implemented | Worker completed a real uploaded shard via HTTP on 2026-07-23. |
 | CTX-07 Distributed workload protocol | Implemented | Versioned Python contract models, registry, strict plan validation, and deterministic reduction ordering are in `scimesh/distributed/`. |
-| CTX-08 Distributed similarity-search | Implemented (scientific layer) | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. Coordinator persistence/orchestration remains CTX-09. |
-| CTX-09 Reducer and final-result API | Not started | Depends on CTX-07 and CTX-08. |
+| CTX-08 Distributed similarity-search | Implemented | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. |
+| CTX-09 Reducer and final-result API | Implemented | Atomic `reducing` claim, deterministic coordinator-side top-k reducer, sanitized reducer failure, final artifact persistence, `result_uri`, and final CSV download. |
 | CTX-10 Distributed similarity-graph | Not started | Local reference exists. |
-| CTX-11 Dashboard/operator view | Implemented (diagnostic scope) | Protected local view: job/task/worker status, validated similarity-search upload, diagnostic partial-artifact download, and bounded polling. Final-result reduction remains CTX-09. |
+| CTX-11 Dashboard/operator view | Implemented | Protected live control room: recent-run/worker overview, real pipeline-stage visualization, shard attempts and safe failures, validated similarity-search upload, coordinator artifacts, final-result download, and bounded polling. |
 | CTX-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. |

 ## Next recommended assignment

-Assign **CTX-09** to the coordinator role: materialize planned shards,
-persist them transactionally, invoke the registered reducer once, and expose a
-durable final artifact.
+Assign **CTX-10** to the distributed-science role: implement deterministic
+block-pair planning and reduction for `similarity-graph`.

 ## Known constraints

- The Python `similarity-search` planner/reducer is implemented, but the Go
-  coordinator does not yet invoke it or persist its final artifact. The
-  operator UI labels `partial_result` files as diagnostic and cannot present
-  them as final output.
-  Use the local `scimesh` CLI for complete workload results.
 - The worker/coordinator flow currently accepts both underscore API workload
  names and hyphenated CLI names while the contract is consolidated.
 - A real-stack worker test uses a small `query_smiles` shard. The Python
-  planner resolves `query_id` once and shares `query_smiles`; connecting that
-  planner to uploaded coordinator jobs belongs to CTX-09.
+  planner resolves `query_id` once and shares `query_smiles`; the upload UI
+  currently accepts `query_smiles` only.
 - The coordinator accepts uploaded distributed jobs only for
  `similarity-search` with `query_smiles`. It rejects `similarity-graph` until
  CTX-10 supplies cross-shard pair planning.
@@ -76,9 +76,19 @@ UI_AUTH_TOKEN='local-ui-secret' make up
 ```

 The UI is disabled by default and never accepts the worker bearer token.
-It shows recent jobs, task/worker state, and the per-job partial artifacts.
-Those files are explicitly diagnostic until the CTX-09 reducer creates a final
-result; the UI does not present them as final scientific output.
+The **control room** shows live workers, recent runs, shard state/attempts,
+safe failures, coordinator artifacts, and the final CSV for completed
+similarity-search jobs. The job page follows the real stages: TSV accepted →
+shards execute → workers return CSVs → `reducing` → final deterministic global
+top-k result. It polls only its own coordinator read-model and never controls
+or exposes worker processes.
+
+For a hands-on run, open `/ui`, choose **New similarity search**, select a
+small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes
+running in separate terminals. The detail page updates every two seconds and
+stops polling after a completed, failed, or cancelled job. Download the
+`final_result` artifact only after the job reaches **Completed**; shard partial
+CSVs remain available as diagnostics.

 `up` starts three services in order: Postgres waits until `pg_isready` passes, a
 one-shot `migrate` container applies the schema and exits, and only then does the
@@ -75,14 +75,15 @@ func run() error {
 		ClaimTask:        usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration),
 		RenewLease:       usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration),
 		CompleteTask:     usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, tx, clk),
+		ReduceJob:        usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk),
 		FailTask:         usecase.NewFailTask(taskRepo, jobRepo, tx, clk),
 		GetJobStatus:     usecase.NewGetJobStatus(jobRepo, taskRepo),
 		CancelJob:        usecase.NewCancelJob(jobRepo, taskRepo, tx, clk),
 		UploadArtifact:   usecase.NewUploadArtifact(taskRepo, 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),
-		PreviewArtifact:  usecase.NewPreviewArtifact(uiReadRepo, blobStore),
 	}

 	// Background reapers are tracked so shutdown can wait for them. Without this
@@ -11,6 +11,7 @@ type JobStatus string
 const (
 	JobPending   JobStatus = "pending"
 	JobRunning   JobStatus = "running"
+	JobReducing  JobStatus = "reducing"
 	JobCompleted JobStatus = "completed"
 	JobFailed    JobStatus = "failed"
 	JobCancelled JobStatus = "cancelled"
@@ -18,14 +19,18 @@ const (

 // Job is one user submission that fans out into one or more tasks.
 type Job struct {
-	ID              uuid.UUID
-	Workload        string
-	InputURI        string     // external input URI; empty for uploaded datasets
-	InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions
-	Parameters      map[string]any
-	Status          JobStatus
-	CreatedAt       time.Time
-	CompletedAt     *time.Time
+	ID               uuid.UUID
+	Workload         string
+	InputURI         string     // external input URI; empty for uploaded datasets
+	InputArtifactID  *uuid.UUID // uploaded input artifact; nil for URI submissions
+	ResultArtifactID *uuid.UUID
+	Parameters       map[string]any
+	Status           JobStatus
+	CreatedAt        time.Time
+	CompletedAt      *time.Time
+	ReducerStartedAt *time.Time
+	ErrorCode        *string
+	ErrorMessage     *string
 }

 // NewUploadedJob builds a job whose input was uploaded to the coordinator. The
@@ -114,6 +119,8 @@ func (p JobProgress) DeriveStatus() JobStatus {
 	switch {
 	case p.Job.Status == JobCancelled:
 		return JobCancelled
+	case p.Job.Status == JobReducing:
+		return JobReducing
 	case p.Total == 0:
 		return JobPending
 	case p.Done == p.Total:
@@ -216,6 +216,51 @@ func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID, status domain.
 	return nil
 }

+func (r *JobRepo) ClaimReduction(_ context.Context, id uuid.UUID, startedAt time.Time) (bool, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	j, ok := r.jobs[id]
+	if !ok {
+		return false, domain.ErrJobNotFound
+	}
+	if j.Status != domain.JobReducing || j.ReducerStartedAt != nil {
+		return false, nil
+	}
+	j.ReducerStartedAt = &startedAt
+	return true, nil
+}
+
+func (r *JobRepo) CompleteWithResult(_ context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	j, ok := r.jobs[id]
+	if !ok {
+		return domain.ErrJobNotFound
+	}
+	j.ResultArtifactID = &resultArtifactID
+	j.Status = domain.JobCompleted
+	j.CompletedAt = &completedAt
+	j.ReducerStartedAt = nil
+	j.ErrorCode = nil
+	j.ErrorMessage = nil
+	return nil
+}
+
+func (r *JobRepo) FailReduction(_ context.Context, id uuid.UUID, code, message string, completedAt time.Time) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	j, ok := r.jobs[id]
+	if !ok {
+		return domain.ErrJobNotFound
+	}
+	j.Status = domain.JobFailed
+	j.CompletedAt = &completedAt
+	j.ReducerStartedAt = nil
+	j.ErrorCode = &code
+	j.ErrorMessage = &message
+	return nil
+}
+
 // --- WorkerRepo ----------------------------------------------------------

 type WorkerRepo struct {
@@ -0,0 +1,195 @@
+// Package reducer contains deterministic, coordinator-side result reductions.
+package reducer
+
+import (
+	"bytes"
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"sort"
+	"strconv"
+)
+
+var searchHeader = []string{"rank", "chembl_id", "canonical_smiles", "similarity"}
+
+type similarityMatch struct {
+	similarity float64
+	id         string
+	smiles     string
+}
+
+// ReduceSimilaritySearch streams worker-local top-k CSVs into the exact global
+// top-k. Each partial is validated before it can affect the final artifact.
+func ReduceSimilaritySearch(partials []io.Reader, parameters map[string]any) ([]byte, error) {
+	topK, err := positiveInt(parameters["top_k"], 20)
+	if err != nil {
+		return nil, err
+	}
+	direction, err := thresholdDirection(parameters["threshold_direction"])
+	if err != nil {
+		return nil, err
+	}
+	h := &matchHeap{direction: direction}
+	for _, partial := range partials {
+		if err := readPartial(partial, direction, func(match similarityMatch) {
+			if len(h.items) < topK {
+				heapPush(h, match)
+				return
+			}
+			if better(match, h.items[0], direction) {
+				h.items[0] = match
+				heapDown(h, 0)
+			}
+		}); err != nil {
+			return nil, err
+		}
+	}
+
+	matches := append([]similarityMatch(nil), h.items...)
+	sort.Slice(matches, func(i, j int) bool { return better(matches[i], matches[j], direction) })
+	var out bytes.Buffer
+	writer := csv.NewWriter(&out)
+	if err := writer.Write(searchHeader); err != nil {
+		return nil, err
+	}
+	for index, match := range matches {
+		if err := writer.Write([]string{
+			strconv.Itoa(index + 1), match.id, match.smiles, fmt.Sprintf("%.6f", match.similarity),
+		}); err != nil {
+			return nil, err
+		}
+	}
+	writer.Flush()
+	if err := writer.Error(); err != nil {
+		return nil, err
+	}
+	return out.Bytes(), nil
+}
+
+func readPartial(input io.Reader, direction string, consume func(similarityMatch)) error {
+	reader := csv.NewReader(input)
+	header, err := reader.Read()
+	if err != nil {
+		return fmt.Errorf("read partial header: %w", err)
+	}
+	if !equalStrings(header, searchHeader) {
+		return fmt.Errorf("partial result has an invalid CSV header")
+	}
+	var previous *similarityMatch
+	for rank := 1; ; rank++ {
+		row, err := reader.Read()
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+		if err != nil {
+			return fmt.Errorf("read partial row: %w", err)
+		}
+		if len(row) != len(searchHeader) || row[0] != strconv.Itoa(rank) {
+			return fmt.Errorf("partial result has an invalid rank")
+		}
+		score, err := strconv.ParseFloat(row[3], 64)
+		if err != nil || math.IsNaN(score) || math.IsInf(score, 0) || score < 0 || score > 1 {
+			return fmt.Errorf("partial result has an invalid similarity")
+		}
+		match := similarityMatch{similarity: score, id: row[1], smiles: row[2]}
+		if previous != nil && better(match, *previous, direction) {
+			return fmt.Errorf("partial result is not sorted deterministically")
+		}
+		previous = &match
+		consume(match)
+	}
+}
+
+func positiveInt(value any, fallback int) (int, error) {
+	if value == nil {
+		return fallback, nil
+	}
+	switch n := value.(type) {
+	case int:
+		if n > 0 {
+			return n, nil
+		}
+	case int64:
+		if n > 0 && n <= math.MaxInt {
+			return int(n), nil
+		}
+	case float64:
+		if n > 0 && n == math.Trunc(n) && n <= math.MaxInt {
+			return int(n), nil
+		}
+	}
+	return 0, fmt.Errorf("top_k must be a positive integer")
+}
+
+func thresholdDirection(value any) (string, error) {
+	if value == nil {
+		return "greater", nil
+	}
+	direction, ok := value.(string)
+	if !ok || (direction != "greater" && direction != "less") {
+		return "", fmt.Errorf("threshold_direction must be greater or less")
+	}
+	return direction, nil
+}
+
+func better(left, right similarityMatch, direction string) bool {
+	if left.similarity != right.similarity {
+		if direction == "less" {
+			return left.similarity < right.similarity
+		}
+		return left.similarity > right.similarity
+	}
+	if left.id != right.id {
+		return left.id < right.id
+	}
+	return left.smiles < right.smiles
+}
+
+func equalStrings(left, right []string) bool {
+	if len(left) != len(right) {
+		return false
+	}
+	for index := range left {
+		if left[index] != right[index] {
+			return false
+		}
+	}
+	return true
+}
+
+// matchHeap keeps the worst retained match at index zero.
+type matchHeap struct {
+	items     []similarityMatch
+	direction string
+}
+
+func heapPush(h *matchHeap, value similarityMatch) {
+	h.items = append(h.items, value)
+	for child := len(h.items) - 1; child > 0; {
+		parent := (child - 1) / 2
+		if !better(h.items[parent], h.items[child], h.direction) {
+			break
+		}
+		h.items[parent], h.items[child] = h.items[child], h.items[parent]
+		child = parent
+	}
+}
+
+func heapDown(h *matchHeap, parent int) {
+	for {
+		child := parent*2 + 1
+		if child >= len(h.items) {
+			return
+		}
+		if right := child + 1; right < len(h.items) && better(h.items[child], h.items[right], h.direction) {
+			child = right
+		}
+		if !better(h.items[parent], h.items[child], h.direction) {
+			return
+		}
+		h.items[parent], h.items[child] = h.items[child], h.items[parent]
+		parent = child
+	}
+}
@@ -0,0 +1,41 @@
+package reducer
+
+import (
+	"io"
+	"strings"
+	"testing"
+)
+
+func TestReduceSimilaritySearchKeepsExactCrossShardRanking(t *testing.T) {
+	first := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,B,CCC,0.50000048\n2,C,CCCC,0.1\n")
+	second := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.50000049\n")
+
+	output, err := ReduceSimilaritySearch([]io.Reader{first, second}, map[string]any{"top_k": 2})
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.500000\n2,B,CCC,0.500000\n"
+	if string(output) != want {
+		t.Fatalf("output = %q, want %q", output, want)
+	}
+}
+
+func TestReduceSimilaritySearchSupportsLeastSimilarDirection(t *testing.T) {
+	partial := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.1\n2,B,CCC,0.8\n")
+	output, err := ReduceSimilaritySearch([]io.Reader{partial}, map[string]any{
+		"top_k": 1, "threshold_direction": "less",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, want := string(output), "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.100000\n"; got != want {
+		t.Fatalf("output = %q, want %q", got, want)
+	}
+}
+
+func TestReduceSimilaritySearchRejectsMalformedPartial(t *testing.T) {
+	partial := strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n2,A,CC,0.1\n")
+	if _, err := ReduceSimilaritySearch([]io.Reader{partial}, nil); err == nil {
+		t.Fatal("expected malformed rank error")
+	}
+}
@@ -90,6 +90,76 @@ func TestCreateJobPersistsEveryTask(t *testing.T) {
 	}
 }

+func TestClaimReductionIsAtomic(t *testing.T) {
+	pool := testPool(t)
+	job, _ := seedJob(t, pool, 1)
+	repo := NewJobRepo(pool)
+	ctx := context.Background()
+	if err := repo.UpdateStatus(ctx, job.ID, domain.JobReducing, nil); err != nil {
+		t.Fatal(err)
+	}
+
+	var (
+		wg      sync.WaitGroup
+		mu      sync.Mutex
+		claimed int
+	)
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ok, err := repo.ClaimReduction(context.Background(), job.ID, time.Now().UTC())
+			if err != nil {
+				t.Errorf("claim reduction: %v", err)
+				return
+			}
+			if ok {
+				mu.Lock()
+				claimed++
+				mu.Unlock()
+			}
+		}()
+	}
+	wg.Wait()
+	if claimed != 1 {
+		t.Fatalf("reducer claims = %d, want 1", claimed)
+	}
+	stored, err := repo.Get(ctx, job.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stored.Status != domain.JobReducing || stored.ReducerStartedAt == nil {
+		t.Fatalf("stored reduction state = %+v", stored)
+	}
+}
+
+func TestUIReadRepoListsReducerFields(t *testing.T) {
+	pool := testPool(t)
+	job, _ := seedJob(t, pool, 1)
+	jobs := NewJobRepo(pool)
+	ctx := context.Background()
+	if err := jobs.UpdateStatus(ctx, job.ID, domain.JobReducing, nil); err != nil {
+		t.Fatal(err)
+	}
+	if claimed, err := jobs.ClaimReduction(ctx, job.ID, time.Now().UTC()); err != nil || !claimed {
+		t.Fatalf("claim reduction = (%v, %v)", claimed, err)
+	}
+	listed, err := NewUIReadRepo(pool).ListJobs(ctx, 20)
+	if err != nil {
+		t.Fatalf("list UI jobs: %v", err)
+	}
+	for _, item := range listed {
+		if item.ID != job.ID {
+			continue
+		}
+		if item.Status != domain.JobReducing || item.ReducerStartedAt == nil {
+			t.Fatalf("UI reducer projection = %+v", item)
+		}
+		return
+	}
+	t.Fatalf("seeded job %s is missing from UI list", job.ID)
+}
+
 // A job must land whole or not at all: a half-created job leaves chunks no
 // worker could ever complete.
 func TestCreateJobRollsBackOnFailure(t *testing.T) {
@@ -25,7 +25,10 @@ func NewJobRepo(pool *pgxpool.Pool) *JobRepo {

 var _ usecase.JobRepository = (*JobRepo)(nil)

-var jobColumns = []string{"id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at"}
+var jobColumns = []string{
+	"id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at",
+	"input_artifact_id", "result_artifact_id", "error_code", "error_message", "reducer_started_at",
+}

 // Insert runs inside the caller's transaction, alongside the job's tasks — that
 // is what makes "all tasks or none" hold.
@@ -55,7 +58,8 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
 		status string
 	)
 	err = conn(ctx, r.pool).QueryRow(ctx, sql, args...).Scan(
-		&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt)
+		&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt,
+		&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return nil, domain.ErrJobNotFound
 	}
@@ -66,6 +70,64 @@ func (r *JobRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
 	return &j, nil
 }

+func (r *JobRepo) ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error) {
+	sql, args, err := psql.Update("jobs").
+		Set("reducer_started_at", startedAt).
+		Where(sq.Eq{"id": id, "status": string(domain.JobReducing), "reducer_started_at": nil}).
+		ToSql()
+	if err != nil {
+		return false, err
+	}
+	tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() == 1, nil
+}
+
+func (r *JobRepo) CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error {
+	sql, args, err := psql.Update("jobs").
+		SetMap(map[string]any{
+			"status":             string(domain.JobCompleted),
+			"result_artifact_id": resultArtifactID,
+			"completed_at":       completedAt,
+			"reducer_started_at": nil,
+			"error_code":         nil,
+			"error_message":      nil,
+		}).
+		Where(sq.Eq{"id": id, "status": string(domain.JobReducing)}).
+		ToSql()
+	if err != nil {
+		return err
+	}
+	tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...)
+	if err != nil {
+		return err
+	}
+	if tag.RowsAffected() == 0 {
+		return domain.ErrJobNotFound
+	}
+	return nil
+}
+
+func (r *JobRepo) FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error {
+	sql, args, err := psql.Update("jobs").
+		SetMap(map[string]any{
+			"status":             string(domain.JobFailed),
+			"completed_at":       completedAt,
+			"error_code":         code,
+			"error_message":      message,
+			"reducer_started_at": nil,
+		}).
+		Where(sq.Eq{"id": id, "status": string(domain.JobReducing)}).
+		ToSql()
+	if err != nil {
+		return err
+	}
+	_, err = conn(ctx, r.pool).Exec(ctx, sql, args...)
+	return err
+}
+
 func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID,
 	status domain.JobStatus, completedAt *time.Time) error {

@@ -41,13 +41,12 @@ func (r *UIReadRepo) ListJobs(ctx context.Context, limit int) ([]domain.Job, err
 	for rows.Next() {
 		var j domain.Job
 		var status string
-		var inputURI *string
-		if err := rows.Scan(&j.ID, &j.Workload, &inputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt); err != nil {
+		if err := rows.Scan(
+			&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt,
+			&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt,
+		); err != nil {
 			return nil, err
 		}
-		if inputURI != nil {
-			j.InputURI = *inputURI
-		}
 		j.Status = domain.JobStatus(status)
 		jobs = append(jobs, j)
 	}
@@ -119,6 +119,8 @@ type jobProgressResponse struct {
 	Done      int       `json:"completed"`
 	Failed    int       `json:"failed"`
 	Cancelled int       `json:"cancelled"`
+	ResultURI string    `json:"result_uri,omitempty"`
+	ErrorCode string    `json:"error_code,omitempty"`
 }

 type uploadArtifactResponse struct {
@@ -153,7 +155,7 @@ func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse {
 }

 func toJobProgressResponse(p domain.JobProgress) jobProgressResponse {
-	return jobProgressResponse{
+	out := jobProgressResponse{
 		ID:        p.Job.ID,
 		Status:    string(p.DeriveStatus()),
 		Total:     p.Total,
@@ -163,4 +165,11 @@ func toJobProgressResponse(p domain.JobProgress) jobProgressResponse {
 		Failed:    p.Failed,
 		Cancelled: p.Cancelled,
 	}
+	if p.Job.ResultArtifactID != nil && out.Status == string(domain.JobCompleted) {
+		out.ResultURI = "/jobs/" + p.Job.ID.String() + "/result"
+	}
+	if p.Job.ErrorCode != nil {
+		out.ErrorCode = *p.Job.ErrorCode
+	}
+	return out
 }
@@ -0,0 +1,25 @@
+package http
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+func TestJobProgressResponseExposesFinalResultOnlyWhenCompleted(t *testing.T) {
+	id := uuid.New()
+	result := uuid.New()
+	progress := domain.JobProgress{Job: domain.Job{
+		ID: id, Status: domain.JobCompleted, ResultArtifactID: &result,
+	}, Total: 1, Done: 1}
+	if got, want := toJobProgressResponse(progress).ResultURI, "/jobs/"+id.String()+"/result"; got != want {
+		t.Fatalf("result URI = %q, want %q", got, want)
+	}
+
+	progress.Job.Status = domain.JobReducing
+	if got := toJobProgressResponse(progress).ResultURI; got != "" {
+		t.Fatalf("reducing job exposes result URI %q", got)
+	}
+}
@@ -150,6 +150,12 @@ func (s *Server) handleResult(w http.ResponseWriter, r *http.Request) {
 		s.writeError(w, r, err)
 		return
 	}
+	if s.uc.ReduceJob != nil {
+		if err := s.uc.ReduceJob.Execute(ctx, task.JobID); err != nil {
+			s.writeError(w, r, err)
+			return
+		}
+	}
 	writeJSON(w, http.StatusOK, taskResponse{ID: task.ID, JobID: task.JobID, Status: string(task.Status)})
 }

@@ -399,6 +405,25 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, toJobProgressResponse(progress))
 }

+func (s *Server) handleGetJobResult(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+	jobID, ok := s.pathUUID(w, r, "job_id")
+	if !ok {
+		return
+	}
+	art, body, err := s.uc.GetJobResult.Execute(ctx, jobID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	defer func() { _ = body.Close() }()
+	w.Header().Set("Content-Type", art.ContentType)
+	w.Header().Set("Content-Length", strconv.FormatInt(art.SizeBytes, 10))
+	w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", art.Filename))
+	_, _ = io.Copy(w, body)
+}
+
 // handleCancelJob stops all non-terminal shards for an operator-requested job.
 // It is available to both the bearer API and the separately authenticated UI.
 func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
@@ -22,14 +22,15 @@ type UseCases struct {
 	ClaimTask        *usecase.ClaimTask
 	RenewLease       *usecase.RenewLease
 	CompleteTask     *usecase.CompleteTask
+	ReduceJob        *usecase.ReduceJob
 	FailTask         *usecase.FailTask
 	GetJobStatus     *usecase.GetJobStatus
+	GetJobResult     *usecase.GetJobResult
 	CancelJob        *usecase.CancelJob
 	UploadArtifact   *usecase.UploadArtifact
 	DownloadArtifact *usecase.DownloadArtifact
 	GetTaskInput     *usecase.GetTaskInput
 	Dashboard        *usecase.Dashboard
-	PreviewArtifact  *usecase.PreviewArtifact
 }

 type Server struct {
@@ -63,6 +64,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 	protected.HandleFunc("POST /jobs", s.handleCreateJob)
 	protected.HandleFunc("POST /jobs/upload", s.handleUploadDataset)
 	protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
+	protected.HandleFunc("GET /jobs/{job_id}/result", s.handleGetJobResult)
 	protected.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob)
 	protected.HandleFunc("POST /tasks/claim", s.handleClaim)
 	protected.HandleFunc("GET /tasks/{task_id}/input", s.handleGetTaskInput)
@@ -79,11 +81,11 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 		ui.HandleFunc("GET /ui", s.handleUIHome)
 		ui.HandleFunc("GET /ui/jobs/new", s.handleUINewJob)
 		ui.HandleFunc("GET /ui/jobs/{job_id}", s.handleUIJob)
+		ui.HandleFunc("GET /ui/api/overview", s.handleUIOverviewJSON)
 		ui.HandleFunc("GET /ui/api/jobs/{job_id}", s.handleUIJobJSON)
 		ui.HandleFunc("POST /ui/api/jobs/{job_id}/cancel", s.handleCancelJob)
 		ui.HandleFunc("POST /ui/api/jobs/upload", s.handleUploadDataset)
 		ui.HandleFunc("GET /ui/jobs/{job_id}/artifacts/{artifact_id}", s.handleUIArtifactDownload)
-		ui.HandleFunc("GET /ui/jobs/{job_id}/artifacts/{artifact_id}/preview", s.handleUIArtifactPreview)
 		mux.Handle("/ui", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
 		mux.Handle("/ui/", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
 	} else {
@@ -42,7 +42,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
 	clk := memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC))
 	tx := memstore.Tx{}
 	lease := 2 * time.Minute
-	uiRead := memstore.NewUIReadRepo(jobs, tasks, work, arts)
+	downloadArtifact := usecase.NewDownloadArtifact(arts, blobs)

 	uc := coordhttp.UseCases{
 		RegisterWorker:   usecase.NewRegisterWorker(work, clk),
@@ -51,14 +51,15 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
 		ClaimTask:        usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease),
 		RenewLease:       usecase.NewRenewLease(tasks, work, tx, clk, lease),
 		CompleteTask:     usecase.NewCompleteTask(tasks, jobs, arts, tx, clk),
+		ReduceJob:        usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk),
 		FailTask:         usecase.NewFailTask(tasks, jobs, tx, clk),
 		GetJobStatus:     usecase.NewGetJobStatus(jobs, tasks),
+		GetJobResult:     usecase.NewGetJobResult(jobs, downloadArtifact),
 		CancelJob:        usecase.NewCancelJob(jobs, tasks, tx, clk),
 		UploadArtifact:   usecase.NewUploadArtifact(tasks, arts, blobs, tx, clk),
-		DownloadArtifact: usecase.NewDownloadArtifact(arts, blobs),
+		DownloadArtifact: downloadArtifact,
 		GetTaskInput:     usecase.NewGetTaskInput(tasks, arts, blobs),
-		Dashboard:        usecase.NewDashboard(uiRead),
-		PreviewArtifact:  usecase.NewPreviewArtifact(uiRead, blobs),
+		Dashboard:        usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),
 	}
 	worker, err := uc.RegisterWorker.Execute(context.Background(), usecase.RegisterWorkerInput{
 		Name: "test-worker", Capabilities: []string{"w", "similarity-search"},
@@ -151,11 +152,36 @@ func TestUIRequiresDistinctCredentialAndRendersDashboard(t *testing.T) {
 		t.Fatalf("UI status: %d", resp.StatusCode)
 	}
 	body, _ := io.ReadAll(resp.Body)
-	if !strings.Contains(string(body), "SciMesh operator dashboard") {
+	if !strings.Contains(string(body), "SciMesh control room") {
 		t.Errorf("dashboard body missing title")
 	}
 }

+func TestUIOverviewReturnsLiveSafeProjection(t *testing.T) {
+	e := newEnv(t, healthy)
+	code, _ := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
+	if code != http.StatusCreated {
+		t.Fatalf("create job: %d", code)
+	}
+	req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/overview", nil)
+	req.SetBasicAuth("operator", uiToken)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	var overview map[string]any
+	if err := json.NewDecoder(resp.Body).Decode(&overview); err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusOK || overview["active_jobs"].(float64) != 1 || overview["online_workers"].(float64) != 1 {
+		t.Fatalf("overview = (%d, %v)", resp.StatusCode, overview)
+	}
+	if _, leaked := overview["worker_auth_token"]; leaked {
+		t.Fatal("overview must not expose authentication configuration")
+	}
+}
+
 func TestUIDisabledReturnsNotFound(t *testing.T) {
 	e := newEnvWithUIToken(t, healthy, "")
 	resp := e.get(t, "/ui")
@@ -289,134 +315,6 @@ func TestUIArtifactDownloadRejectsAnotherJobsArtifact(t *testing.T) {
 	}
 }

-func TestUIArtifactPreviewRequiresAuth(t *testing.T) {
-	e := newEnv(t, healthy)
-	code, job := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
-	if code != http.StatusCreated {
-		t.Fatalf("create: %d", code)
-	}
-	_, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["w"]}`)
-	artifactID := e.putArtifact(t, claim["task_id"].(string), "w1", int(claim["attempt"].(float64)), "a,b\n1,2\n")
-
-	req, _ := http.NewRequestWithContext(context.Background(), "GET",
-		e.ts.URL+"/ui/jobs/"+job["id"].(string)+"/artifacts/"+artifactID+"/preview", nil)
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusUnauthorized {
-		t.Errorf("unauthenticated preview = %d, want 401", resp.StatusCode)
-	}
-}
-
-func TestUIArtifactPreviewRendersEscapedCSVRows(t *testing.T) {
-	e := newEnv(t, healthy)
-	code, job := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
-	if code != http.StatusCreated {
-		t.Fatalf("create: %d", code)
-	}
-	_, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["w"]}`)
-	csv := "chembl_id,note\nCHEMBL1,<script>alert(1)</script>\n"
-	artifactID := e.putArtifact(t, claim["task_id"].(string), "w1", int(claim["attempt"].(float64)), csv)
-
-	req, _ := http.NewRequestWithContext(context.Background(), "GET",
-		e.ts.URL+"/ui/jobs/"+job["id"].(string)+"/artifacts/"+artifactID+"/preview", nil)
-	req.SetBasicAuth("operator", uiToken)
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		t.Fatalf("preview: %d", resp.StatusCode)
-	}
-	body, _ := io.ReadAll(resp.Body)
-	if strings.Contains(string(body), "<script>alert(1)</script>") {
-		t.Error("preview must escape HTML-like CSV values, found raw <script> tag")
-	}
-	if !strings.Contains(string(body), "&lt;script&gt;") {
-		t.Errorf("expected escaped script tag in preview body: %s", body)
-	}
-	if !strings.Contains(string(body), "CHEMBL1") {
-		t.Error("preview missing expected cell value")
-	}
-}
-
-func TestUIArtifactPreviewRejectsAnotherJobsArtifact(t *testing.T) {
-	e := newEnv(t, healthy)
-	code, _ := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
-	if code != http.StatusCreated {
-		t.Fatalf("first job: %d", code)
-	}
-	_, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["w"]}`)
-	artifactID := e.putArtifact(t, claim["task_id"].(string), "w1", int(claim["attempt"].(float64)), "a,b\n1,2\n")
-	code, second := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
-	if code != http.StatusCreated {
-		t.Fatalf("second job: %d", code)
-	}
-	req, _ := http.NewRequestWithContext(context.Background(), "GET",
-		e.ts.URL+"/ui/jobs/"+second["id"].(string)+"/artifacts/"+artifactID+"/preview", nil)
-	req.SetBasicAuth("operator", uiToken)
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusNotFound {
-		t.Errorf("cross-job preview = %d, want 404", resp.StatusCode)
-	}
-}
-
-func TestUIArtifactPreviewIsFriendlyForNonCSV(t *testing.T) {
-	e := newEnv(t, healthy)
-	code, job := e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in","chunks":[{"chunk_index":0,"input_uri":"s3://c","input_sha256":"sha"}]}`)
-	if code != http.StatusCreated {
-		t.Fatalf("create: %d", code)
-	}
-	_, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["w"]}`)
-	taskID := claim["task_id"].(string)
-	attempt := int(claim["attempt"].(float64))
-
-	req, _ := http.NewRequestWithContext(context.Background(), "PUT",
-		e.ts.URL+"/tasks/"+taskID+"/artifacts/notes.bin", strings.NewReader("\x00\x01binary garbage"))
-	req.Header.Set("Authorization", "Bearer "+token)
-	req.Header.Set("Content-Type", "application/octet-stream")
-	req.Header.Set("X-Worker-ID", e.workerID)
-	req.Header.Set("X-Task-Attempt", strconv.Itoa(attempt))
-	putResp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer putResp.Body.Close()
-	if putResp.StatusCode != http.StatusOK {
-		t.Fatalf("put non-csv artifact: %d", putResp.StatusCode)
-	}
-	var m map[string]any
-	b, _ := io.ReadAll(putResp.Body)
-	_ = json.Unmarshal(b, &m)
-	artifactID := m["artifact_id"].(string)
-
-	previewReq, _ := http.NewRequestWithContext(context.Background(), "GET",
-		e.ts.URL+"/ui/jobs/"+job["id"].(string)+"/artifacts/"+artifactID+"/preview", nil)
-	previewReq.SetBasicAuth("operator", uiToken)
-	resp, err := http.DefaultClient.Do(previewReq)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer resp.Body.Close()
-	if resp.StatusCode != http.StatusOK {
-		t.Fatalf("preview status: %d", resp.StatusCode)
-	}
-	body, _ := io.ReadAll(resp.Body)
-	if strings.Contains(string(body), "binary garbage") {
-		t.Error("non-CSV bytes must not be rendered as text")
-	}
-	if !strings.Contains(string(body), "not a CSV file") {
-		t.Errorf("expected a friendly non-CSV explanation, got: %s", body)
-	}
-}
-
 func TestHealthUnavailableWhenDBDown(t *testing.T) {
 	e := newEnv(t, func(context.Context) error { return context.DeadlineExceeded })
 	resp := e.get(t, "/health")
@@ -530,6 +428,69 @@ func TestFullLifecycle(t *testing.T) {
 	}
 }

+func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) {
+	e := newEnv(t, healthy)
+	code, job := e.uploadDataset(t, "similarity-search", 10, "chembl_id\tcanonical_smiles\nA\tCC\n")
+	if code != http.StatusCreated {
+		t.Fatalf("upload job: %d", code)
+	}
+	jobID := job["job_id"].(string)
+	code, claim := e.do(t, "POST", "/tasks/claim", `{"worker_id":"w1","capabilities":["similarity-search"]}`)
+	if code != http.StatusOK {
+		t.Fatalf("claim: %d", code)
+	}
+	taskID := claim["task_id"].(string)
+	attempt := int(claim["attempt"].(float64))
+	artifactID := e.putArtifact(t, taskID, "w1", attempt, "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n")
+	if code, _ := e.do(t, "POST", "/tasks/"+taskID+"/result",
+		`{"worker_id":"w1","attempt":`+itoa(attempt)+`,"result":{"artifact_id":"`+artifactID+`"}}`); code != http.StatusOK {
+		t.Fatalf("complete: %d", code)
+	}
+
+	code, progress := e.do(t, "GET", "/jobs/"+jobID, "")
+	if code != http.StatusOK || progress["status"] != "completed" || progress["result_uri"] != "/jobs/"+jobID+"/result" {
+		t.Fatalf("progress = (%d, %v)", code, progress)
+	}
+	req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+progress["result_uri"].(string), nil)
+	req.Header.Set("Authorization", "Bearer "+token)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK || string(body) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n" {
+		t.Fatalf("final result = (%d, %q)", resp.StatusCode, body)
+	}
+
+	uiRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID, nil)
+	uiRequest.SetBasicAuth("operator", uiToken)
+	uiResponse, err := http.DefaultClient.Do(uiRequest)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer uiResponse.Body.Close()
+	uiBody, _ := io.ReadAll(uiResponse.Body)
+	if uiResponse.StatusCode != http.StatusOK || !strings.Contains(string(uiBody), "Final result ready") {
+		t.Fatalf("final UI = (%d, %q)", uiResponse.StatusCode, uiBody)
+	}
+
+	jsonRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/jobs/"+jobID, nil)
+	jsonRequest.SetBasicAuth("operator", uiToken)
+	jsonResponse, err := http.DefaultClient.Do(jsonRequest)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer jsonResponse.Body.Close()
+	var detail map[string]any
+	if err := json.NewDecoder(jsonResponse.Body).Decode(&detail); err != nil {
+		t.Fatal(err)
+	}
+	if jsonResponse.StatusCode != http.StatusOK || detail["final_result_available"] != true {
+		t.Fatalf("final UI JSON = (%d, %v)", jsonResponse.StatusCode, detail)
+	}
+}
+
 func TestForeignArtifactResultConflict(t *testing.T) {
 	e := newEnv(t, healthy)
 	e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in",
@@ -1,31 +0,0 @@
-{{define "artifact-preview.html"}}
-<!doctype html>
-<html lang="en">
-<head>
-  <meta charset="utf-8">
-  <meta name="viewport" content="width=device-width, initial-scale=1">
-  <title>SciMesh artifact preview</title>
-  <style>
-    :root{color:#172033;background:#f6f8fc;font:16px/1.5 system-ui,sans-serif}body{margin:0}.page{max-width:1180px;margin:auto;padding:32px 20px 56px}a{color:#174ecf}.back{text-decoration:none}h1{margin:18px 0 4px;font-size:1.6rem;word-break:break-word}.muted{color:#68758b}.notice{margin:16px 0;padding:15px 17px;border:1px solid #f2cb72;border-radius:10px;background:#fff8e6}.table-wrap{overflow-x:auto;border:1px solid #dfe5f0;border-radius:10px;background:#fff;margin-top:16px}table{width:100%;border-collapse:collapse}td,th{padding:10px 12px;border-bottom:1px solid #e8ecf4;text-align:left;vertical-align:top;white-space:pre-wrap;word-break:break-word}th{color:#50617d;font-size:.78rem;text-transform:uppercase;letter-spacing:.06em;background:#f6f8fc}tr:last-child td{border:0}.empty{padding:24px;text-align:center;color:#68758b}
-  </style>
-</head>
-<body>
-  <main class="page">
-    <a class="back" href="/ui/jobs/{{.JobID}}">← Back to job</a>
-    <h1>Preview: {{.Filename}}</h1>
-    <p class="muted">Diagnostic preview only — a partial shard result, not a final molecular-search answer. At most {{.RowLimit}} rows and {{.ByteLimit}} bytes are read from storage.</p>
-    {{if not .Previewable}}
-      <div class="notice">{{.Reason}}</div>
-    {{else}}
-      {{if .Truncated}}<div class="notice">Truncated: showing at most the first {{.RowLimit}} rows or {{.ByteLimit}} bytes of this artifact. Download it for the full contents.</div>{{end}}
-      <div class="table-wrap">
-        <table>
-          <tr>{{range .Headers}}<th>{{.}}</th>{{end}}</tr>
-          {{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{else}}<tr><td class="empty" colspan="99">No data rows.</td></tr>{{end}}
-        </table>
-      </div>
-    {{end}}
-  </main>
-</body>
-</html>
-{{end}}
@@ -4,20 +4,36 @@
 <head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
-  <title>SciMesh operator dashboard</title>
+  <title>SciMesh control room</title>
  <style>
-    :root{color:#172033;background:#f6f8fc;font:16px/1.5 system-ui,sans-serif}body{margin:0}.page{max-width:1180px;margin:auto;padding:32px 20px 56px}.top{display:flex;justify-content:space-between;gap:24px;align-items:start}.eyebrow{margin:0;color:#50617d;font-size:.86rem;font-weight:700;text-transform:uppercase;letter-spacing:.08em}h1{margin:.2rem 0;font-size:2rem}h2{margin:32px 0 12px;font-size:1.28rem}.lead{margin:0;color:#56657c}.button{display:inline-block;border:0;border-radius:8px;padding:11px 15px;background:#1f5eff;color:#fff;font-weight:700;text-decoration:none;white-space:nowrap}.notice{margin-top:24px;padding:16px 18px;border:1px solid #f2cb72;border-radius:10px;background:#fff8e6}.notice strong{display:block}.steps{display:grid;grid-template-columns:repeat(3,1fr);gap:12px;margin-top:14px}.step,.card{padding:16px;border:1px solid #dfe5f0;border-radius:10px;background:#fff}.step b{display:block;color:#1f5eff}.table-wrap{overflow-x:auto;background:#fff;border:1px solid #dfe5f0;border-radius:10px}table{width:100%;border-collapse:collapse}td,th{padding:13px 14px;border-bottom:1px solid #e8ecf4;text-align:left;vertical-align:top}th{color:#50617d;font-size:.78rem;text-transform:uppercase;letter-spacing:.06em}tr:last-child td{border:0}a{color:#174ecf}small,.muted{color:#68758b}.status{display:inline-block;border-radius:999px;padding:3px 9px;font-size:.84rem;font-weight:700}.status-success{background:#dff6e9;color:#126b3d}.status-danger{background:#ffe4e6;color:#a31135}.status-active{background:#e4edff;color:#174ecf}.status-waiting{background:#edf0f5;color:#50617d}.bar{height:7px;min-width:120px;margin-top:7px;overflow:hidden;border-radius:999px;background:#e6eaf1}.bar>span{display:block;height:100%;background:#1f5eff}.kicker{font-variant-numeric:tabular-nums}.empty{padding:28px;text-align:center;color:#68758b}.worker{display:grid;grid-template-columns:1.3fr .8fr 2fr 1fr;gap:12px;align-items:center}.worker+.worker{border-top:1px solid #e8ecf4;padding-top:12px;margin-top:12px}@media(max-width:760px){.top,.steps{display:block}.button{margin-top:12px}.step{margin-top:10px}.worker{grid-template-columns:1fr}.hide-mobile{display:none}}
+    :root{color:#dce8ff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 15% -10%,#163d77 0,transparent 32rem),radial-gradient(circle at 95% 5%,#123e39 0,transparent 29rem),#08111f}.page{max-width:1240px;margin:auto;padding:28px 22px 64px}.top{display:flex;align-items:flex-start;justify-content:space-between;gap:22px}.eyebrow{margin:0;color:#78a9ff;font-size:.77rem;font-weight:750;letter-spacing:.14em;text-transform:uppercase}.title{max-width:700px;margin:8px 0;font-size:clamp(2rem,5vw,3.6rem);line-height:1.04;letter-spacing:-.055em}.lead{max-width:690px;margin:0;color:#aabbd5;font-size:1.05rem}.button{display:inline-flex;align-items:center;gap:8px;border:0;border-radius:10px;padding:12px 16px;background:#4f8cff;color:#071224;font:inherit;font-weight:800;text-decoration:none;box-shadow:0 12px 30px #163d7766}.live{display:inline-flex;align-items:center;gap:7px;margin-top:18px;color:#8ba2c2;font-size:.87rem}.pulse{width:8px;height:8px;border-radius:50%;background:#5ee6a6;box-shadow:0 0 0 5px #5ee6a622}.summary{display:grid;grid-template-columns:1.4fr repeat(3,1fr);gap:13px;margin:32px 0}.panel,.metric,.flow-step,.job,.worker{border:1px solid #26415f;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #0000001f}.panel{padding:20px}.pipeline{display:grid;grid-template-columns:repeat(4,1fr);gap:10px;margin-top:16px}.flow-step{position:relative;padding:14px;min-height:116px}.flow-step:not(:last-child):after{content:"";position:absolute;top:35px;right:-10px;width:10px;height:2px;background:#3c5d83}.flow-step b{display:block;color:#f2f7ff}.flow-step small{display:block;margin-top:6px;color:#91a8c6}.flow-step .dot{display:inline-block;width:9px;height:9px;margin-right:7px;border-radius:50%;background:#5ee6a6}.metric{padding:16px}.metric b{display:block;margin-top:7px;color:#f5f8ff;font-size:2rem;line-height:1}.metric span{color:#9bb0cc;font-size:.84rem}.section-head{display:flex;align-items:baseline;justify-content:space-between;gap:12px;margin:36px 0 12px}.section-head h2{margin:0;color:#f3f7ff;font-size:1.18rem}.section-head p{margin:0;color:#8fa6c3;font-size:.9rem}.jobs{display:grid;gap:10px}.job{display:grid;grid-template-columns:minmax(210px,1.35fr) minmax(150px,.9fr) minmax(180px,1fr) auto;gap:18px;align-items:center;padding:17px 18px;text-decoration:none;color:inherit;transition:border-color .15s,transform .15s}.job:hover{border-color:#5d96ee;transform:translateY(-1px)}.job-name{color:#f3f7ff;font-weight:750}.job-id{margin-top:3px;color:#8196b3;font-family:ui-monospace,SFMono-Regular,monospace;font-size:.76rem}.badge{display:inline-flex;align-items:center;border-radius:999px;padding:4px 9px;font-size:.78rem;font-weight:800}.badge-waiting{background:#23344d;color:#b9cce9}.badge-active{background:#173d77;color:#9fc7ff}.badge-success{background:#123f34;color:#76efb5}.badge-danger{background:#552334;color:#ff9bad}.bar{height:7px;margin-top:8px;overflow:hidden;border-radius:999px;background:#20344e}.bar span{display:block;height:100%;border-radius:inherit;background:linear-gradient(90deg,#5a92ff,#61e1bd)}.numbers{color:#afc0d9;font-size:.88rem}.arrow{color:#79aaff;font-size:1.35rem}.workers{display:grid;grid-template-columns:repeat(3,1fr);gap:11px}.worker{padding:15px}.worker-head{display:flex;justify-content:space-between;gap:8px}.worker strong{color:#f3f7ff}.worker small,.worker p{color:#95a9c4}.worker p{margin:12px 0 0}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #365576;border-radius:5px;padding:2px 6px;color:#a9c9f4;font:.75rem ui-monospace,SFMono-Regular,monospace}.empty{padding:30px;border:1px dashed #35516f;border-radius:14px;color:#9ab0cb;text-align:center}.offline{color:#faafbd}.sr{position:absolute;width:1px;height:1px;clip:rect(0,0,0,0);overflow:hidden;white-space:nowrap}@media(max-width:820px){.top,.section-head{display:block}.button{margin-top:18px}.summary{grid-template-columns:1fr 1fr}.panel{grid-column:span 2}.pipeline{grid-template-columns:1fr 1fr}.flow-step:not(:last-child):after{display:none}.job{grid-template-columns:1fr 1fr}.arrow{display:none}.workers{grid-template-columns:1fr 1fr}}@media(max-width:540px){.page{padding:22px 14px}.summary,.workers{grid-template-columns:1fr}.panel{grid-column:auto}.pipeline,.job{grid-template-columns:1fr}.title{font-size:2.35rem}}
  </style>
 </head>
 <body>
  <main class="page">
-    <header class="top"><div><p class="eyebrow">Local coordinator</p><h1>SciMesh operator dashboard</h1><p class="lead">See where a computation is and what should happen next.</p></div><a class="button" href="/ui/jobs/new">Start a check</a></header>
-    <section class="notice" aria-label="Current pipeline limitation"><strong>This screen currently diagnoses shard jobs.</strong><span>Workers upload partial CSVs to the coordinator. Until a reducer is implemented, those files are not one final scientific result.</span><div class="steps"><div class="step"><b>1. Upload TSV</b>The coordinator splits the file into shard tasks.</div><div class="step"><b>2. Wait for a worker</b>A worker claims a shard, calculates similarity, and returns a CSV.</div><div class="step"><b>3. Inspect artifacts</b>Download a partial result from the job page.</div></div></section>
-    <h2>Recent jobs</h2>
-    <div class="table-wrap"><table><tr><th>Computation</th><th>State</th><th>Progress</th><th class="hide-mobile">Created</th></tr>{{range .Jobs}}<tr><td><a href="/ui/jobs/{{.ID}}"><strong>{{workloadLabel .Workload}}</strong></a><br><small>Open job details</small></td><td><span class="status status-{{statusClass .Status}}">{{statusLabel .Status}}</span><br><small>{{statusHint .Status}}</small></td><td class="kicker"><strong>{{.Completed}} / {{.Total}}</strong> complete{{if gt .Failed 0}} · <span style="color:#a31135">failed: {{.Failed}}</span>{{end}}{{if gt .Cancelled 0}} · <span>stopped: {{.Cancelled}}</span>{{end}}<div class="bar"><span style="width:{{progressPercent .Completed .Failed .Cancelled .Total}}%"></span></div></td><td class="hide-mobile"><small>{{time .CreatedAt}}</small></td></tr>{{else}}<tr><td colspan="4" class="empty"><strong>No jobs yet.</strong><br>Click “Start a check”, upload a small TSV, and leave a worker running.</td></tr>{{end}}</table></div>
-    <h2>Workers</h2>
-    <section class="card">{{range .Workers}}<div class="worker"><div><strong>{{.Name}}</strong><br><small>{{.ID}}</small></div><div><span class="status status-{{if eq .Status "online"}}success{{else}}waiting{{end}}">{{workerStatusLabel .Status}}</span></div><div>{{range .Capabilities}}<code>{{.}}</code> {{end}}</div><div class="muted">Last signal<br>{{time .LastHeartbeatAt}}</div></div>{{else}}<div class="empty"><strong>No worker is registered yet.</strong><br>Run <code>scimesh-worker</code> with the coordinator URL and worker token.</div>{{end}}</section>
+    <header class="top">
+      <div><p class="eyebrow">Local scientific compute</p><h1 class="title">SciMesh control room</h1><p class="lead">Follow the real path from a molecular TSV to a globally reduced similarity result—without reading coordinator logs.</p><div class="live"><i class="pulse"></i><span id="refresh-state">Live overview · refreshes every 2 seconds</span></div></div>
+      <a class="button" href="/ui/jobs/new">＋ New similarity search</a>
+    </header>
+    <section class="summary" aria-label="Pipeline summary">
+      <div class="panel"><strong>How a search becomes a result</strong><div class="pipeline"><div class="flow-step"><span><i class="dot"></i>01</span><b>Upload TSV</b><small>The coordinator validates and slices the dataset.</small></div><div class="flow-step"><span><i class="dot"></i>02</span><b>Run shards</b><small>Workers fingerprint molecules and return shard top-k CSVs.</small></div><div class="flow-step"><span><i class="dot"></i>03</span><b>Merge exactly</b><small>The coordinator ranks retained candidates deterministically.</small></div><div class="flow-step"><span><i class="dot"></i>04</span><b>Download CSV</b><small>A checksum-protected global result is ready.</small></div></div></div>
+      <div class="metric"><span>Active runs</span><b id="active-jobs">{{.ActiveJobs}}</b><small>waiting, running, or merging</small></div>
+      <div class="metric"><span>Available workers</span><b id="online-workers">{{.OnlineWorkers}}</b><small>recently registered</small></div>
+      <div class="metric"><span>Finished runs</span><b id="finished-jobs">{{.FinishedJobs}}</b><small>in the latest 20</small></div>
+    </section>
+
+    <section><div class="section-head"><h2>Recent computations</h2><p id="job-count">{{len .Jobs}} shown · newest first</p></div><div id="jobs" class="jobs">{{range .Jobs}}<a class="job" href="/ui/jobs/{{.ID}}"><div><div class="job-name">{{workloadLabel .Workload}}</div><div class="job-id">{{.ID}}</div></div><div><span class="badge badge-{{statusClass .Status}}">{{statusLabel .Status}}</span><div class="job-id">{{statusHint .Status}}</div></div><div><div class="numbers"><b>{{.Completed}}</b> / {{.Total}} shards complete{{if gt .Failed 0}} · <span class="offline">{{.Failed}} failed</span>{{end}}</div><div class="bar"><span style="width:{{progressPercent .Completed .Failed .Cancelled .Total}}%"></span></div></div><span class="arrow" aria-hidden="true">→</span></a>{{else}}<div class="empty"><strong>No computations yet.</strong><br>Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.</div>{{end}}</div></section>
+    <section><div class="section-head"><h2>Worker fleet</h2><p>Workers register themselves; this page never controls their processes.</p></div><div id="workers" class="workers">{{range .Workers}}<article class="worker"><div class="worker-head"><strong>{{.Name}}</strong><span class="badge badge-{{workerStatusClass .Status}}">{{workerStatusLabel .Status}}</span></div><small>{{.ID}}</small><p>{{range .Capabilities}}<span class="cap">{{.}}</span>{{end}}</p><p>Last signal · {{time .LastHeartbeatAt}}</p></article>{{else}}<div class="empty"><strong>No worker is registered.</strong><br>Start <code>scimesh-worker</code> in another terminal, then return here.</div>{{end}}</div></section>
  </main>
+  <script>
+    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 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){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()));box.append(card)}};
+    let timer;const refresh=async()=>{try{const response=await fetch('/ui/api/overview',{headers:{Accept:'application/json'}});if(!response.ok)throw Error();const view=await response.json();document.querySelector('#active-jobs').textContent=view.active_jobs;document.querySelector('#online-workers').textContent=view.online_workers;document.querySelector('#finished-jobs').textContent=view.finished_jobs;document.querySelector('#job-count').textContent=view.jobs.length+' shown · newest first';renderJobs(view.jobs);renderWorkers(view.workers);document.querySelector('#refresh-state').textContent='Live overview · updated just now'}catch(_){document.querySelector('#refresh-state').textContent='Connection interrupted · trying again automatically'}};
+    const start=()=>{if(!timer&&!document.hidden)timer=setInterval(refresh,2000)};document.addEventListener('visibilitychange',()=>{if(document.hidden&&timer){clearInterval(timer);timer=undefined}else start()});start();
+  </script>
 </body>
 </html>
 {{end}}
@@ -4,27 +4,20 @@
 <head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
-  <title>Create a check — SciMesh</title>
+  <title>New similarity search · SciMesh</title>
  <style>
-    :root{color:#172033;background:#f6f8fc;font:16px/1.5 system-ui,sans-serif}body{margin:0}.page{max-width:760px;margin:auto;padding:32px 20px 56px}a{color:#174ecf}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#50617d;font-size:.86rem;font-weight:700;text-transform:uppercase;letter-spacing:.08em}h1{margin:0;font-size:2rem}.lead{color:#56657c}.notice{margin:22px 0;padding:16px 18px;border:1px solid #f2cb72;border-radius:10px;background:#fff8e6}.notice strong{display:block}.card{padding:22px;border:1px solid #dfe5f0;border-radius:12px;background:#fff}label{display:block;margin:18px 0 4px;font-weight:700}input{box-sizing:border-box;width:100%;padding:10px;border:1px solid #bac5d8;border-radius:7px;font:inherit}input[type=file]{padding:8px;background:#f8faff}.hint{margin:4px 0;color:#68758b;font-size:.9rem}.button{margin-top:22px;border:0;border-radius:8px;padding:11px 16px;background:#1f5eff;color:#fff;font:inherit;font-weight:700;cursor:pointer}.button:disabled{opacity:.6;cursor:wait}.error{margin-top:16px;color:#a31135}.working{margin-top:16px;color:#174ecf}.checklist{margin:8px 0;padding-left:20px;color:#56657c}.checklist li{margin:5px 0}
+    :root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:980px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:720px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}.card,.aside,.notice{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021}.card{padding:22px}.aside,.notice{padding:18px}.aside h2,.notice h2{margin:0;color:#f1f6ff;font-size:1rem}.aside p,.notice p{color:#9fb3cf}.aside ol{margin:13px 0 0;padding-left:20px;color:#aebfda}.aside li{margin:10px 0}label{display:block;margin:18px 0 5px;color:#eaf2ff;font-weight:750}input,select{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus,select:focus{outline:2px solid #5d97f5;outline-offset:1px}input[type=file]{padding:8px}.hint{margin:5px 0;color:#92a9c6;font-size:.87rem}.split{display:grid;grid-template-columns:1fr 1fr;gap:12px}.run-preview{margin-top:18px;border:1px solid #2c8c70;border-radius:11px;padding:13px;background:#0c2b2a;color:#a8f1d0}.run-preview strong{color:#e6fff4}.button{display:inline-flex;margin-top:22px;border:0;border-radius:10px;padding:12px 16px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.button:disabled{opacity:.6;cursor:wait}.working{margin:14px 0 0;color:#9fc5ff}.error{margin:12px 0 0;color:#ffacba}.hidden{display:none}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 6px;color:#b5d3f5;font:.76rem ui-monospace,SFMono-Regular,monospace}@media(max-width:720px){.layout,.split{grid-template-columns:1fr}.page{padding:22px 14px}}
  </style>
 </head>
 <body>
  <main class="page">
-    <a class="back" href="/ui">← Back to jobs</a><p class="eyebrow">Guided run</p><h1>Search for similar molecules</h1><p class="lead">Creates a diagnostic <code>similarity-search</code> job: a worker finds the top-k molecules most similar to a target SMILES.</p>
-    <section class="notice"><strong>Before starting</strong><ul class="checklist"><li>Keep at least one <code>scimesh-worker</code> running.</li><li>Use a small TSV for a hands-on check.</li><li><b>“Rows per shard” does not limit the file size.</b> It splits the entire upload into tasks: a full ChEMBL TSV at 1,000 rows per shard creates thousands of tasks.</li></ul></section>
-    <form id="run" class="card">
-      <label for="file">ChEMBL TSV</label><input id="file" type="file" name="file" required accept=".tsv,.txt,text/tab-separated-values"><p class="hint">Expected columns: <code>chembl_id</code> and <code>canonical_smiles</code>.</p>
-      <label for="query-smiles">Target molecule (SMILES)</label><input id="query-smiles" name="query_smiles" required maxlength="200" value="CCO" autocomplete="off"><p class="hint"><code>CCO</code> is ethanol. For gefitinib, use its SMILES here or the local CLI with <code>--query-id</code>.</p>
-      <label for="top-k">Matches to return</label><input id="top-k" name="top_k" type="number" min="1" max="100000" value="20" required><p class="hint">This is the top-k within each shard, not a global top-k for the whole dataset yet.</p>
-      <label for="chunk-rows">Rows per shard</label><input id="chunk-rows" name="chunk_rows" type="number" min="1" max="100000" value="1000" required><p class="hint">Fewer rows mean more tasks and more visible progress; more rows mean fewer, longer tasks.</p>
-      <label for="max-rows">Maximum dataset rows to process <small>(optional)</small></label><input id="max-rows" name="max_rows" type="number" min="1" max="10000000" placeholder="For example: 500"><p class="hint">Useful for a quick check of a large TSV. The coordinator creates shards from only the first N data rows; it still stores the original upload.</p>
-      <button class="button" id="submit" type="submit">Upload file and create job</button><p id="working" class="working" hidden aria-live="polite">Uploading the file and creating shard tasks… Keep this page open.</p><p id="error" class="error" role="alert"></p>
-    </form>
+    <a class="back" href="/ui">← Back to control room</a><p class="eyebrow">New computation</p><h1>Similarity search, end to end</h1><p class="lead">Upload a ChEMBL-style TSV. Workers calculate shard candidates; after every shard succeeds, SciMesh merges the exact global top-k into one final CSV.</p>
+    <div class="layout"><form id="run" class="card" novalidate><label for="file">ChEMBL TSV</label><input id="file" type="file" name="file" required accept=".tsv,.txt,text/tab-separated-values"><p class="hint">Required columns: <code>chembl_id</code> and <code>canonical_smiles</code>.</p><label for="query-smiles">Target molecule (SMILES)</label><input id="query-smiles" name="query_smiles" required maxlength="200" value="CCO" autocomplete="off"><p class="hint">Use a valid SMILES. The coordinator shares this exact query with every shard.</p><div class="split"><div><label for="top-k">Global top-k</label><input id="top-k" name="top_k" type="number" min="1" max="100000" value="20" required><p class="hint">How many final molecules to retain.</p></div><div><label for="chunk-rows">Rows per shard</label><input id="chunk-rows" name="chunk_rows" type="number" min="1" max="100000" value="1000" required><p class="hint">Smaller shards make more visible tasks.</p></div></div><div class="split"><div><label for="threshold">Similarity threshold <small>(optional)</small></label><input id="threshold" name="threshold" type="number" min="0" max="1" step="0.01" placeholder="For example: 0.70"><p class="hint">Leave blank to rank every valid candidate.</p></div><div><label for="direction">Keep molecules</label><select id="direction" name="threshold_direction"><option value="greater">more similar (≥ threshold)</option><option value="less">less similar (≤ threshold)</option></select><p class="hint">“Less” helps explore dissimilar molecules.</p></div></div><label for="max-rows">Maximum dataset rows <small>(optional quick run)</small></label><input id="max-rows" name="max_rows" type="number" min="1" max="10000000" placeholder="For example: 500"><p class="hint">Only the first N data rows become shards; the original upload remains stored by the coordinator.</p><div id="preview" class="run-preview"><strong>Ready to plan a run.</strong><br>Select a TSV to see the file that will be sent to the coordinator.</div><button class="button" id="submit" type="submit">Create pipeline run →</button><p id="working" class="working hidden" aria-live="polite">Uploading TSV and creating coordinator-owned shards…</p><p id="error" class="error" role="alert"></p></form><aside class="aside"><h2>What you will observe</h2><ol><li><strong>Input accepted</strong><br>TSV is validated and split into durable shard artifacts.</li><li><strong>Workers claim tasks</strong><br>Each worker downloads one shard, fingerprints it, and uploads a partial CSV.</li><li><strong>Global reduction</strong><br>The coordinator compares exact scores from all partial results.</li><li><strong>Final download</strong><br>The result page exposes a checksum-protected global CSV.</li></ol><h2 style="margin-top:26px">Before you submit</h2><p>Keep at least one worker running in another terminal. The browser cannot start or control worker processes.</p><p><span class="cap">similarity-search</span> is currently the only distributed workload available here.</p></aside></div>
  </main>
  <script>
-    const form=document.querySelector('#run'),button=document.querySelector('#submit'),working=document.querySelector('#working'),error=document.querySelector('#error');
-    form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';const fields=new FormData(form),file=fields.get('file'),maxRows=String(fields.get('max_rows')||'').trim();if(!(file instanceof File)||file.size===0){error.textContent='Choose a non-empty TSV file.';return}const parameters={query_smiles:fields.get('query_smiles'),top_k:Number(fields.get('top_k')),progress_every:0},upload=new FormData();upload.append('workload','similarity-search');upload.append('parameters',JSON.stringify(parameters));upload.append('chunk_rows',fields.get('chunk_rows'));if(maxRows)upload.append('max_rows',maxRows);upload.append('file',file,file.name);button.disabled=true;working.hidden=false;try{const response=await fetch('/ui/api/jobs/upload',{method:'POST',body:upload}),data=await response.json();if(!response.ok)throw Error(data.error||'Unable to create the job.');location.href='/ui/jobs/'+data.job_id}catch(err){error.textContent=err.message==='invalid input'?'Check the TSV and fields: the coordinator could not accept this request.':err.message;button.disabled=false;working.hidden=true}});
+    const form=document.querySelector('#run'),button=document.querySelector('#submit'),working=document.querySelector('#working'),error=document.querySelector('#error'),preview=document.querySelector('#preview'),fileInput=document.querySelector('#file');
+    const bytes=size=>size<1024?size+' B':size<1024*1024?(size/1024).toFixed(1)+' KiB':(size/(1024*1024)).toFixed(1)+' MiB';fileInput.addEventListener('change',()=>{const file=fileInput.files&&fileInput.files[0];preview.replaceChildren();if(!file){preview.append(document.createTextNode('Select a TSV to see the file that will be sent to the coordinator.'));return}const strong=document.createElement('strong');strong.textContent='Source ready: '+file.name;preview.append(strong,document.createElement('br'),document.createTextNode(bytes(file.size)+' · the coordinator will validate its header before creating tasks.'))});
+    form.addEventListener('submit',async event=>{event.preventDefault();error.textContent='';const fields=new FormData(form),file=fields.get('file'),query=String(fields.get('query_smiles')||'').trim(),topK=Number(fields.get('top_k')),chunkRows=Number(fields.get('chunk_rows')),threshold=String(fields.get('threshold')||'').trim(),maxRows=String(fields.get('max_rows')||'').trim();if(!(file instanceof File)||file.size===0){error.textContent='Choose a non-empty TSV file.';return}if(!query||query.length>200||!Number.isInteger(topK)||topK<1||!Number.isInteger(chunkRows)||chunkRows<1){error.textContent='Enter a target SMILES, a positive global top-k, and a positive rows-per-shard value.';return}if(threshold&&(Number.isNaN(Number(threshold))||Number(threshold)<0||Number(threshold)>1)){error.textContent='Similarity threshold must be between 0 and 1.';return}const parameters={query_smiles:query,top_k:topK,threshold_direction:fields.get('threshold_direction'),progress_every:0};if(threshold)parameters.threshold=Number(threshold);const upload=new FormData();upload.append('workload','similarity-search');upload.append('parameters',JSON.stringify(parameters));upload.append('chunk_rows',String(chunkRows));if(maxRows)upload.append('max_rows',maxRows);upload.append('file',file,file.name);button.disabled=true;working.classList.remove('hidden');try{const response=await fetch('/ui/api/jobs/upload',{method:'POST',body:upload}),data=await response.json();if(!response.ok)throw Error(data.error||'Unable to create the job.');location.href='/ui/jobs/'+encodeURIComponent(data.job_id)}catch(err){error.textContent=err.message==='invalid input'?'The coordinator could not accept this run. Check the TSV columns and form values.':err.message;button.disabled=false;working.classList.add('hidden')}});
  </script>
 </body>
 </html>
@@ -26,6 +26,7 @@ var uiTemplates = template.Must(template.New("ui").Funcs(template.FuncMap{
 	"taskErrorLabel":    uiTaskErrorLabel,
 	"taskErrorHint":     uiTaskErrorHint,
 	"workerStatusLabel": uiWorkerStatusLabel,
+	"workerStatusClass": uiWorkerStatusClass,
 	"workloadLabel":     uiWorkloadLabel,
 	"progressPercent":   uiProgressPercent,
 	"cancellable":       uiCancellable,
@@ -48,8 +49,10 @@ func uiStatusLabel(status string) string {
 		return "Assigned to a worker"
 	case "running":
 		return "Running"
+	case "reducing":
+		return "Merging results"
 	case "completed":
-		return "Tasks complete"
+		return "Completed"
 	case "failed":
 		return "Needs attention"
 	case "cancelled":
@@ -67,8 +70,10 @@ func uiStatusHint(status string) string {
 		return "A worker has claimed the task and should begin processing shortly."
 	case "running":
 		return "A worker is reading a shard, calculating fingerprints, and uploading its result through the coordinator."
+	case "reducing":
+		return "All shards are complete. The coordinator is merging their candidates into one final CSV."
 	case "completed":
-		return "Every shard task is complete. Files below are still partial results."
+		return "The final result is ready to download."
 	case "failed":
 		return "One or more shard tasks failed. Open the task list below for details."
 	case "cancelled":
@@ -86,7 +91,7 @@ func uiStatusClass(status string) string {
 		return "danger"
 	case "cancelled":
 		return "waiting"
-	case "running", "leased":
+	case "running", "leased", "reducing":
 		return "active"
 	default:
 		return "waiting"
@@ -97,6 +102,8 @@ func uiWorkerStatusLabel(status string) string {
 	switch status {
 	case "online":
 		return "Available"
+	case "busy":
+		return "Busy"
 	case "offline":
 		return "Offline"
 	default:
@@ -104,6 +111,17 @@ func uiWorkerStatusLabel(status string) string {
 	}
 }

+func uiWorkerStatusClass(status string) string {
+	switch status {
+	case "online":
+		return "success"
+	case "busy":
+		return "active"
+	default:
+		return "waiting"
+	}
+}
+
 // uiTaskErrorLabel deliberately maps worker implementation errors to an
 // operator-facing diagnosis. Raw subprocess commands and local paths belong in
 // the worker terminal, not in the web UI.
@@ -203,6 +221,20 @@ func (s *Server) handleUIHome(w http.ResponseWriter, r *http.Request) {
 	s.renderUI(w, "dashboard.html", view)
 }

+// handleUIOverviewJSON is the bounded polling projection used by the operator
+// dashboard. It intentionally returns only the safe UI read model, never
+// worker tokens, storage keys, or database entities.
+func (s *Server) handleUIOverviewJSON(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+	view, err := s.uc.Dashboard.Overview(ctx, 20)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, view)
+}
+
 func (s *Server) handleUINewJob(w http.ResponseWriter, r *http.Request) {
 	s.renderUI(w, "new-job.html", nil)
 }
@@ -279,26 +311,3 @@ func (s *Server) handleUIArtifactDownload(w http.ResponseWriter, r *http.Request
 	w.Header().Set("X-Checksum-SHA256", art.SHA256)
 	_, _ = io.Copy(w, body)
 }
-
-// handleUIArtifactPreview renders a bounded, job-scoped CSV preview. The use
-// case enforces the same ownership and downloadable rule as the download
-// proxy above; nothing here trusts the artifact ID beyond that check.
-func (s *Server) handleUIArtifactPreview(w http.ResponseWriter, r *http.Request) {
-	jobID, ok := s.uiJobID(w, r)
-	if !ok {
-		return
-	}
-	artifactID, err := uuid.Parse(r.PathValue("artifact_id"))
-	if err != nil {
-		s.writeError(w, r, domain.ErrInvalidInput)
-		return
-	}
-	ctx, cancel := s.reqCtx(r)
-	defer cancel()
-	view, err := s.uc.PreviewArtifact.Execute(ctx, jobID, artifactID)
-	if err != nil {
-		s.writeError(w, r, err)
-		return
-	}
-	s.renderUI(w, "artifact-preview.html", view)
-}
@@ -10,7 +10,8 @@ func TestUIStatusPresentation(t *testing.T) {
 	}{
 		{"pending", "Waiting for a worker", "waiting"},
 		{"running", "Running", "active"},
-		{"completed", "Tasks complete", "success"},
+		{"reducing", "Merging results", "active"},
+		{"completed", "Completed", "success"},
 		{"failed", "Needs attention", "danger"},
 	}
 	for _, test := range tests {
@@ -42,3 +43,12 @@ func TestUITaskErrorPresentationDoesNotExposeCommand(t *testing.T) {
 		t.Error("error hint must explain the failure")
 	}
 }
+
+func TestUIWorkerStatusPresentation(t *testing.T) {
+	if got := uiWorkerStatusLabel("busy"); got != "Busy" {
+		t.Errorf("busy worker label = %q", got)
+	}
+	if got := uiWorkerStatusClass("busy"); got != "active" {
+		t.Errorf("busy worker class = %q", got)
+	}
+}
@@ -100,7 +100,7 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error
 		if job.Status == domain.JobCancelled {
 			return nil
 		}
-		if job.Status == domain.JobCompleted || job.Status == domain.JobFailed {
+		if job.Status == domain.JobReducing || job.Status == domain.JobCompleted || job.Status == domain.JobFailed {
 			return domain.ErrJobNotCancellable
 		}
 		// The lease reaper can be the transition that exhausted the final task.
@@ -111,7 +111,7 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error
 			return err
 		}
 		derived := progressFrom(*job, counts).DeriveStatus()
-		if derived == domain.JobCompleted || derived == domain.JobFailed {
+		if derived == domain.JobReducing || derived == domain.JobCompleted || derived == domain.JobFailed {
 			return domain.ErrJobNotCancellable
 		}
 		cancelled, err = uc.tasks.CancelByJob(ctx, jobID, now)
@@ -226,10 +226,20 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository
 	if err != nil {
 		return err
 	}
-	status := progressFrom(domain.Job{}, counts).DeriveStatus()
+	job, err := jobs.Get(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	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" {
+		status = domain.JobReducing
+	}

 	var completedAt *time.Time
-	if status == domain.JobCompleted || status == domain.JobFailed {
+	if status == domain.JobFailed {
 		completedAt = &now
 	}
 	return jobs.UpdateStatus(ctx, jobID, status, completedAt)
@@ -70,6 +70,9 @@ type JobRepository interface {
 	Insert(ctx context.Context, j *domain.Job) error
 	Get(ctx context.Context, id uuid.UUID) (*domain.Job, error)
 	UpdateStatus(ctx context.Context, id uuid.UUID, status domain.JobStatus, completedAt *time.Time) error
+	ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error)
+	CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error
+	FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error
 }

 // WorkerRepository persists the worker registry.
@@ -1,150 +0,0 @@
-package usecase
-
-import (
-	"context"
-	"encoding/csv"
-	"errors"
-	"io"
-	"strings"
-
-	"github.com/google/uuid"
-
-	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
-)
-
-// previewMaxRows and previewMaxBytes bound how much of an artifact the
-// diagnostic preview ever reads or renders: a partial shard CSV can be large,
-// and this is a diagnostic aid, not a viewer for the full file.
-const (
-	previewMaxRows  = 30
-	previewMaxBytes = 64 * 1024
-)
-
-// ArtifactPreviewView is what the UI renders for a diagnostic CSV preview. It
-// never carries a storage path, database error, or worker-local detail.
-type ArtifactPreviewView struct {
-	JobID       string
-	ArtifactID  string
-	Filename    string
-	Previewable bool
-	Reason      string
-	Headers     []string
-	Rows        [][]string
-	Truncated   bool
-	RowLimit    int
-	ByteLimit   int64
-}
-
-// PreviewArtifact renders at most the first previewMaxRows rows of a CSV
-// artifact, reading at most previewMaxBytes from storage. It reuses the same
-// job-scoped, downloadable-artifact rule as the download proxy so an artifact
-// ID from another job is never previewable.
-type PreviewArtifact struct {
-	read  UIReadRepository
-	blobs BlobStore
-}
-
-func NewPreviewArtifact(read UIReadRepository, blobs BlobStore) *PreviewArtifact {
-	return &PreviewArtifact{read: read, blobs: blobs}
-}
-
-func (p *PreviewArtifact) Execute(ctx context.Context, jobID, artifactID uuid.UUID) (ArtifactPreviewView, error) {
-	job, err := p.read.GetJob(ctx, jobID)
-	if err != nil {
-		return ArtifactPreviewView{}, err
-	}
-	tasks, err := p.read.ListTasksByJob(ctx, jobID)
-	if err != nil {
-		return ArtifactPreviewView{}, err
-	}
-	// Same status derivation the dashboard uses, so a final artifact previews
-	// exactly when it would also be offered for download.
-	status := jobCard(*job, tasks).Status
-
-	artifacts, err := p.read.ListArtifactsByJob(ctx, jobID)
-	if err != nil {
-		return ArtifactPreviewView{}, err
-	}
-	var art *domain.Artifact
-	for i := range artifacts {
-		if artifacts[i].ID == artifactID {
-			art = &artifacts[i]
-			break
-		}
-	}
-	if art == nil {
-		return ArtifactPreviewView{}, domain.ErrArtifactNotFound
-	}
-	downloadable := art.Kind == domain.ArtifactPartialResult ||
-		(art.Kind == domain.ArtifactFinalResult && status == string(domain.JobCompleted))
-	if !downloadable {
-		return ArtifactPreviewView{}, domain.ErrArtifactNotFound
-	}
-
-	view := ArtifactPreviewView{
-		JobID:      jobID.String(),
-		ArtifactID: art.ID.String(),
-		Filename:   art.Filename,
-		RowLimit:   previewMaxRows,
-		ByteLimit:  previewMaxBytes,
-	}
-	if !isCSVArtifact(art) {
-		view.Reason = "This artifact is not a CSV file, so it cannot be shown as text here. Download it instead."
-		return view, nil
-	}
-	if art.SizeBytes == 0 {
-		view.Reason = "This artifact is empty."
-		return view, nil
-	}
-
-	rc, err := p.blobs.Open(ctx, art.StorageKey)
-	if err != nil {
-		return ArtifactPreviewView{}, err
-	}
-	defer func() { _ = rc.Close() }()
-
-	// LimitedReader caps the bytes read from storage regardless of how many
-	// rows are found within that window — the artifact is never loaded whole.
-	limited := &io.LimitedReader{R: rc, N: previewMaxBytes}
-	reader := csv.NewReader(limited)
-	reader.FieldsPerRecord = -1 // a byte-limited cut mid-row must not look like a schema error
-
-	header, err := reader.Read()
-	if err != nil {
-		view.Reason = "This artifact could not be read as CSV."
-		return view, nil
-	}
-	view.Headers = append([]string(nil), header...)
-
-	rows := make([][]string, 0, previewMaxRows)
-	for len(rows) < previewMaxRows {
-		record, err := reader.Read()
-		if err != nil {
-			if !errors.Is(err, io.EOF) {
-				// Malformed content further into the stream: keep what parsed
-				// cleanly and say the preview stopped early.
-				view.Truncated = true
-			}
-			break
-		}
-		rows = append(rows, append([]string(nil), record...))
-	}
-	view.Rows = rows
-
-	if art.SizeBytes > previewMaxBytes {
-		view.Truncated = true
-	} else if len(rows) == previewMaxRows {
-		if _, err := reader.Read(); err == nil {
-			view.Truncated = true
-		}
-	}
-	view.Previewable = true
-	return view, nil
-}
-
-func isCSVArtifact(a *domain.Artifact) bool {
-	if a.ContentType == "text/csv" {
-		return true
-	}
-	return strings.HasSuffix(strings.ToLower(a.Filename), ".csv")
-}
@@ -1,226 +0,0 @@
-package usecase_test
-
-import (
-	"context"
-	"errors"
-	"strconv"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/google/uuid"
-
-	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
-	"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
-	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
-)
-
-func newPreviewHarness() (*usecase.PreviewArtifact, *memstore.JobRepo, *memstore.TaskRepo, *memstore.ArtifactRepo, *memstore.BlobStore) {
-	jobs := memstore.NewJobRepo()
-	tasks := memstore.NewTaskRepo()
-	work := memstore.NewWorkerRepo()
-	arts := memstore.NewArtifactRepo()
-	blobs := memstore.NewBlobStore()
-	read := memstore.NewUIReadRepo(jobs, tasks, work, arts)
-	return usecase.NewPreviewArtifact(read, blobs), jobs, tasks, arts, blobs
-}
-
-func mustInsertJob(t *testing.T, jobs *memstore.JobRepo, status domain.JobStatus) uuid.UUID {
-	t.Helper()
-	job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: status, CreatedAt: time.Now()}
-	if err := jobs.Insert(context.Background(), job); err != nil {
-		t.Fatalf("insert job: %v", err)
-	}
-	return job.ID
-}
-
-func mustCompleteJob(t *testing.T, jobs *memstore.JobRepo, tasks *memstore.TaskRepo) uuid.UUID {
-	t.Helper()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	task := &domain.Task{
-		ID: uuid.New(), JobID: jobID, ChunkIndex: 0, Workload: "similarity-search",
-		Status: domain.TaskCompleted, MaxAttempts: 3, CreatedAt: time.Now(),
-	}
-	if err := tasks.InsertBatch(context.Background(), []*domain.Task{task}); err != nil {
-		t.Fatalf("insert task: %v", err)
-	}
-	return jobID
-}
-
-func mustInsertArtifact(t *testing.T, arts *memstore.ArtifactRepo, blobs *memstore.BlobStore,
-	jobID uuid.UUID, kind domain.ArtifactKind, filename, contentType, body string) uuid.UUID {
-	t.Helper()
-	id := uuid.New()
-	sha, size, err := blobs.Put(context.Background(), id.String(), strings.NewReader(body))
-	if err != nil {
-		t.Fatalf("put blob: %v", err)
-	}
-	art := &domain.Artifact{
-		ID: id, JobID: jobID, Kind: kind, Filename: filename,
-		StorageKey: id.String(), ContentType: contentType,
-		SizeBytes: size, SHA256: sha, CreatedAt: time.Now(),
-	}
-	if err := arts.Insert(context.Background(), art); err != nil {
-		t.Fatalf("insert artifact: %v", err)
-	}
-	return id
-}
-
-func TestPreviewArtifactRendersCSVRows(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult, "result.csv", "text/csv",
-		"chembl_id,score\nCHEMBL1,0.9\nCHEMBL2,0.8\n")
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if !view.Previewable {
-		t.Fatalf("expected previewable, reason=%q", view.Reason)
-	}
-	if view.Truncated {
-		t.Error("small CSV should not be truncated")
-	}
-	if len(view.Headers) != 2 || view.Headers[0] != "chembl_id" {
-		t.Errorf("headers = %v", view.Headers)
-	}
-	if len(view.Rows) != 2 || view.Rows[0][0] != "CHEMBL1" {
-		t.Errorf("rows = %v", view.Rows)
-	}
-}
-
-func TestPreviewArtifactTruncatesAt30Rows(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-
-	var sb strings.Builder
-	sb.WriteString("id,value\n")
-	for i := 0; i < 40; i++ {
-		sb.WriteString("R" + strconv.Itoa(i) + ",v\n")
-	}
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult, "result.csv", "text/csv", sb.String())
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if len(view.Rows) != 30 {
-		t.Fatalf("rows = %d, want 30", len(view.Rows))
-	}
-	if !view.Truncated {
-		t.Error("expected truncated for more than 30 data rows")
-	}
-}
-
-func TestPreviewArtifactTruncatesAt64KiB(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-
-	var sb strings.Builder
-	sb.WriteString("id,value\n")
-	row := "row," + strings.Repeat("x", 200) + "\n"
-	for sb.Len() < 70*1024 {
-		sb.WriteString(row)
-	}
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult, "result.csv", "text/csv", sb.String())
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if !view.Truncated {
-		t.Error("expected truncated for an artifact bigger than 64KiB")
-	}
-	if len(view.Rows) > 30 {
-		t.Errorf("rows = %d, want <= 30", len(view.Rows))
-	}
-}
-
-func TestPreviewArtifactRejectsNonCSV(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult,
-		"shard-0.tsv", "text/tab-separated-values", "chembl_id\tcanonical_smiles\nCHEMBL1\tCCO\n")
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if view.Previewable {
-		t.Error("non-CSV artifact must not be previewable as text")
-	}
-	if view.Reason == "" {
-		t.Error("expected a friendly reason")
-	}
-}
-
-func TestPreviewArtifactFailsSafelyOnEmptyArtifact(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult, "result.csv", "text/csv", "")
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if view.Previewable {
-		t.Error("empty artifact must not be previewable")
-	}
-	if view.Reason == "" {
-		t.Error("expected a friendly reason")
-	}
-}
-
-func TestPreviewArtifactFailsSafelyOnMalformedCSV(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	// An unterminated quote makes even the header row unparsable.
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactPartialResult, "result.csv", "text/csv", `"unterminated`)
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if view.Previewable {
-		t.Error("malformed CSV must not be previewable")
-	}
-	if view.Reason == "" {
-		t.Error("expected a friendly reason")
-	}
-}
-
-func TestPreviewArtifactRejectsCrossJobArtifact(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobA := mustInsertJob(t, jobs, domain.JobRunning)
-	jobB := mustInsertJob(t, jobs, domain.JobRunning)
-	artID := mustInsertArtifact(t, arts, blobs, jobA, domain.ArtifactPartialResult, "result.csv", "text/csv", "a,b\n1,2\n")
-
-	if _, err := preview.Execute(context.Background(), jobB, artID); !errors.Is(err, domain.ErrArtifactNotFound) {
-		t.Fatalf("err = %v, want ErrArtifactNotFound", err)
-	}
-}
-
-func TestPreviewArtifactRejectsUncompletedFinalResult(t *testing.T) {
-	preview, jobs, _, arts, blobs := newPreviewHarness()
-	jobID := mustInsertJob(t, jobs, domain.JobRunning)
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactFinalResult, "final.csv", "text/csv", "a,b\n1,2\n")
-
-	if _, err := preview.Execute(context.Background(), jobID, artID); !errors.Is(err, domain.ErrArtifactNotFound) {
-		t.Fatalf("err = %v, want ErrArtifactNotFound", err)
-	}
-}
-
-func TestPreviewArtifactAllowsFinalResultOnceJobIsCompleted(t *testing.T) {
-	preview, jobs, tasks, arts, blobs := newPreviewHarness()
-	jobID := mustCompleteJob(t, jobs, tasks)
-	artID := mustInsertArtifact(t, arts, blobs, jobID, domain.ArtifactFinalResult, "final.csv", "text/csv", "a,b\n1,2\n")
-
-	view, err := preview.Execute(context.Background(), jobID, artID)
-	if err != nil {
-		t.Fatalf("Execute: %v", err)
-	}
-	if !view.Previewable {
-		t.Fatalf("expected previewable, reason=%q", view.Reason)
-	}
-}
@@ -0,0 +1,133 @@
+package usecase
+
+import (
+	"bytes"
+	"context"
+	"io"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/reducer"
+)
+
+// ReduceJob turns completed coordinator-owned partial artifacts into one final
+// artifact. It performs no worker I/O and never trusts a worker URI or path.
+type ReduceJob struct {
+	jobs      JobRepository
+	tasks     TaskRepository
+	artifacts ArtifactRepository
+	blobs     BlobStore
+	tx        TxManager
+	clock     Clock
+}
+
+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}
+}
+
+// Execute is idempotent for jobs that are not currently reducing. The worker
+// completion path may call it after every result; only the last task changes a
+// similarity-search job into reducing state.
+func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error {
+	claimed, err := uc.jobs.ClaimReduction(ctx, jobID, uc.clock.Now())
+	if err != nil || !claimed {
+		return err
+	}
+	job, err := uc.jobs.Get(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	if job.Status != domain.JobReducing {
+		return nil
+	}
+	if job.Workload != "similarity-search" {
+		return uc.fail(ctx, jobID)
+	}
+
+	completed, err := uc.tasks.ListCompleted(ctx, jobID)
+	if err != nil {
+		return uc.fail(ctx, jobID)
+	}
+	if len(completed) == 0 {
+		return uc.fail(ctx, jobID)
+	}
+	readers := make([]io.Reader, 0, len(completed))
+	closers := make([]io.Closer, 0, len(completed))
+	for _, task := range completed {
+		if task.ResultArtifactID == nil {
+			closeAll(closers)
+			return uc.fail(ctx, jobID)
+		}
+		artifact, err := uc.artifacts.Get(ctx, *task.ResultArtifactID)
+		if err != nil || artifact.JobID != jobID || artifact.TaskID == nil || *artifact.TaskID != task.ID || artifact.Kind != domain.ArtifactPartialResult {
+			closeAll(closers)
+			return uc.fail(ctx, jobID)
+		}
+		body, err := uc.blobs.Open(ctx, artifact.StorageKey)
+		if err != nil {
+			closeAll(closers)
+			return uc.fail(ctx, jobID)
+		}
+		readers = append(readers, body)
+		closers = append(closers, body)
+	}
+	output, reduceErr := reducer.ReduceSimilaritySearch(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())
+	if err != nil {
+		return uc.fail(ctx, jobID)
+	}
+	sum, size, err := uc.blobs.Put(ctx, final.StorageKey, bytes.NewReader(output))
+	if err != nil {
+		return uc.fail(ctx, jobID)
+	}
+	final.SetContent(sum, size)
+	if err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		if err := uc.artifacts.Insert(ctx, final); err != nil {
+			return err
+		}
+		return uc.jobs.CompleteWithResult(ctx, jobID, final.ID, uc.clock.Now())
+	}); err != nil {
+		_ = uc.blobs.Delete(ctx, final.StorageKey)
+		return err
+	}
+	return nil
+}
+
+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.
+	return uc.jobs.FailReduction(ctx, jobID, "reducer_failed", "final result reduction failed", uc.clock.Now())
+}
+
+func closeAll(closers []io.Closer) {
+	for _, closer := range closers {
+		_ = closer.Close()
+	}
+}
+
+type GetJobResult struct {
+	jobs     JobRepository
+	download *DownloadArtifact
+}
+
+func NewGetJobResult(jobs JobRepository, download *DownloadArtifact) *GetJobResult {
+	return &GetJobResult{jobs: jobs, download: download}
+}
+
+func (uc *GetJobResult) Execute(ctx context.Context, jobID uuid.UUID) (*domain.Artifact, io.ReadCloser, error) {
+	job, err := uc.jobs.Get(ctx, jobID)
+	if err != nil {
+		return nil, nil, err
+	}
+	if job.Status != domain.JobCompleted || job.ResultArtifactID == nil {
+		return nil, nil, domain.ErrArtifactNotFound
+	}
+	return uc.download.Execute(ctx, *job.ResultArtifactID)
+}
@@ -2,6 +2,7 @@ package usecase

 import (
 	"context"
+	"fmt"
 	"time"

 	"github.com/google/uuid"
@@ -21,17 +22,21 @@ type UIReadRepository interface {
 }

 type JobCard struct {
-	ID        string    `json:"id"`
-	Workload  string    `json:"workload"`
-	Status    string    `json:"status"`
-	CreatedAt time.Time `json:"created_at"`
-	Total     int       `json:"total"`
-	Pending   int       `json:"pending"`
-	Leased    int       `json:"leased"`
-	Running   int       `json:"running"`
-	Completed int       `json:"completed"`
-	Failed    int       `json:"failed"`
-	Cancelled int       `json:"cancelled"`
+	ID               string     `json:"id"`
+	Workload         string     `json:"workload"`
+	Status           string     `json:"status"`
+	CreatedAt        time.Time  `json:"created_at"`
+	CompletedAt      *time.Time `json:"completed_at,omitempty"`
+	ReducerStartedAt *time.Time `json:"reducer_started_at,omitempty"`
+	ErrorCode        string     `json:"error_code,omitempty"`
+	ErrorMessage     string     `json:"error_message,omitempty"`
+	Total            int        `json:"total"`
+	Pending          int        `json:"pending"`
+	Leased           int        `json:"leased"`
+	Running          int        `json:"running"`
+	Completed        int        `json:"completed"`
+	Failed           int        `json:"failed"`
+	Cancelled        int        `json:"cancelled"`
 }

 type TaskCard struct {
@@ -42,10 +47,20 @@ type TaskCard struct {
 	MaxAttempts    int        `json:"max_attempts"`
 	LeaseOwner     string     `json:"lease_owner,omitempty"`
 	LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
+	StartedAt      *time.Time `json:"started_at,omitempty"`
+	CompletedAt    *time.Time `json:"completed_at,omitempty"`
 	ErrorCode      string     `json:"error_code,omitempty"`
 	ErrorMessage   string     `json:"error_message,omitempty"`
 }

+// ParameterCard is an intentionally small allowlist of run configuration that
+// helps an operator verify what is being computed without exposing arbitrary
+// job payloads to the browser.
+type ParameterCard struct {
+	Label string `json:"label"`
+	Value string `json:"value"`
+}
+
 type ArtifactCard struct {
 	ID           string `json:"id"`
 	Kind         string `json:"kind"`
@@ -65,14 +80,18 @@ type WorkerCard struct {
 }

 type DashboardView struct {
-	Jobs    []JobCard
-	Workers []WorkerCard
+	Jobs          []JobCard    `json:"jobs"`
+	Workers       []WorkerCard `json:"workers"`
+	ActiveJobs    int          `json:"active_jobs"`
+	FinishedJobs  int          `json:"finished_jobs"`
+	OnlineWorkers int          `json:"online_workers"`
 }
 type JobDetailView struct {
 	JobCard
-	Tasks                []TaskCard     `json:"tasks"`
-	Artifacts            []ArtifactCard `json:"artifacts"`
-	FinalResultAvailable bool           `json:"final_result_available"`
+	Tasks                []TaskCard      `json:"tasks"`
+	Artifacts            []ArtifactCard  `json:"artifacts"`
+	Parameters           []ParameterCard `json:"parameters"`
+	FinalResultAvailable bool            `json:"final_result_available"`
 }

 type Dashboard struct{ read UIReadRepository }
@@ -98,10 +117,20 @@ func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, err
 		return DashboardView{}, err
 	}
 	for _, job := range jobs {
-		out.Jobs = append(out.Jobs, jobCard(job, tasksByJob[job.ID]))
+		card := jobCard(job, tasksByJob[job.ID])
+		out.Jobs = append(out.Jobs, card)
+		switch card.Status {
+		case string(domain.JobCompleted), string(domain.JobFailed), string(domain.JobCancelled):
+			out.FinishedJobs++
+		default:
+			out.ActiveJobs++
+		}
 	}
 	for _, worker := range workers {
 		out.Workers = append(out.Workers, WorkerCard{ID: worker.ID.String(), Name: worker.Name, Status: string(worker.Status), Capabilities: worker.Capabilities, LastHeartbeatAt: worker.LastHeartbeatAt})
+		if worker.Status == domain.WorkerOnline || worker.Status == domain.WorkerBusy {
+			out.OnlineWorkers++
+		}
 	}
 	return out, nil
 }
@@ -119,11 +148,27 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
 	if err != nil {
 		return JobDetailView{}, err
 	}
-	out := JobDetailView{JobCard: jobCard(*job, tasks), Tasks: make([]TaskCard, 0, len(tasks)), Artifacts: make([]ArtifactCard, 0, len(artifacts))}
+	workers, err := d.read.ListWorkers(ctx, 100)
+	if err != nil {
+		return JobDetailView{}, err
+	}
+	workerNames := make(map[string]string, len(workers))
+	for _, worker := range workers {
+		workerNames[worker.ID.String()] = worker.Name
+	}
+	out := JobDetailView{
+		JobCard:    jobCard(*job, tasks),
+		Tasks:      make([]TaskCard, 0, len(tasks)),
+		Artifacts:  make([]ArtifactCard, 0, len(artifacts)),
+		Parameters: uiParameters(job.Parameters),
+	}
 	for _, task := range tasks {
-		card := TaskCard{ID: task.ID.String(), ChunkIndex: task.ChunkIndex, Status: string(task.Status), Attempt: task.Attempt, MaxAttempts: task.MaxAttempts, LeaseExpiresAt: task.LeaseExpiresAt}
+		card := TaskCard{ID: task.ID.String(), ChunkIndex: task.ChunkIndex, Status: string(task.Status), Attempt: task.Attempt, MaxAttempts: task.MaxAttempts, LeaseExpiresAt: task.LeaseExpiresAt, StartedAt: task.StartedAt, CompletedAt: task.CompletedAt}
 		if task.LeaseOwner != nil {
-			card.LeaseOwner = *task.LeaseOwner
+			card.LeaseOwner = workerNames[*task.LeaseOwner]
+			if card.LeaseOwner == "" {
+				card.LeaseOwner = "Worker " + shortID(*task.LeaseOwner)
+			}
 		}
 		if task.ErrorCode != nil {
 			card.ErrorCode = *task.ErrorCode
@@ -158,7 +203,13 @@ func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID
 }

 func jobCard(job domain.Job, tasks []domain.Task) JobCard {
-	c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt}
+	c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt, CompletedAt: job.CompletedAt, ReducerStartedAt: job.ReducerStartedAt}
+	if job.ErrorCode != nil {
+		c.ErrorCode = *job.ErrorCode
+	}
+	if job.ErrorMessage != nil {
+		c.ErrorMessage = *job.ErrorMessage
+	}
 	for _, task := range tasks {
 		c.Total++
 		switch task.Status {
@@ -180,3 +231,46 @@ func jobCard(job domain.Job, tasks []domain.Task) JobCard {
 	c.Status = string(p.DeriveStatus())
 	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"},
+	}
+	out := make([]ParameterCard, 0, len(keys))
+	for _, entry := range keys {
+		value, ok := parameters[entry.key]
+		if !ok {
+			continue
+		}
+		formatted, ok := formatUIParameter(value)
+		if ok {
+			out = append(out, ParameterCard{Label: entry.label, Value: formatted})
+		}
+	}
+	return out
+}
+
+func formatUIParameter(value any) (string, bool) {
+	switch typed := value.(type) {
+	case string:
+		return typed, true
+	case int, int64, float64, bool:
+		return fmt.Sprint(typed), true
+	default:
+		return "", false
+	}
+}
+
+func shortID(value string) string {
+	if len(value) <= 8 {
+		return value
+	}
+	return value[:8]
+}
@@ -0,0 +1,19 @@
+package usecase
+
+import "testing"
+
+func TestUIParametersAreAllowlisted(t *testing.T) {
+	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"},
+	})
+	if len(parameters) != 2 {
+		t.Fatalf("parameters = %#v, want only two allowlisted values", parameters)
+	}
+	if parameters[0] != (ParameterCard{Label: "Target SMILES", Value: "CCO"}) ||
+		parameters[1] != (ParameterCard{Label: "Global top-k", Value: "20"}) {
+		t.Fatalf("parameters = %#v", parameters)
+	}
+}
@@ -55,6 +55,8 @@ type harness struct {
 	getInput    *usecase.GetTaskInput
 	expire      *usecase.ExpireLeases
 	cancel      *usecase.CancelJob
+	reduce      *usecase.ReduceJob
+	jobResult   *usecase.GetJobResult
 }

 func newHarness() *harness {
@@ -81,9 +83,88 @@ func newHarness() *harness {
 	h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs)
 	h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk)
 	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.jobResult = usecase.NewGetJobResult(h.jobs, h.downloadArt)
 	return h
 }

+func TestSimilaritySearchReductionCreatesFinalArtifact(t *testing.T) {
+	h := newHarness()
+	jobID := h.seedJob(t, "similarity-search", 2)
+	_, err := h.jobs.Get(ctx, jobID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil {
+		t.Fatal(err)
+	}
+	partials := []string{
+		"rank,chembl_id,canonical_smiles,similarity\n1,B,CCC,0.50000048\n",
+		"rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.50000049\n",
+	}
+	for _, partial := range partials {
+		taskID, attempt := h.leaseOne(t, "w1", "similarity-search")
+		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) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.500000\n2,B,CCC,0.500000\n" {
+		t.Fatalf("unexpected final %q", bytes)
+	}
+}
+
+func TestSimilaritySearchReductionFailureIsSanitized(t *testing.T) {
+	h := newHarness()
+	jobID := h.seedJob(t, "similarity-search", 1)
+	if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil {
+		t.Fatal(err)
+	}
+	taskID, attempt := h.leaseOne(t, "w1", "similarity-search")
+	art, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{
+		TaskID: taskID, WorkerID: "w1", Attempt: attempt, Filename: "partial.csv",
+		ContentType: "text/csv", Body: strings.NewReader("rank,chembl_id,canonical_smiles,similarity\n2,A,CC,0.9\n"),
+	})
+	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)
+	}
+	job, err := h.jobs.Get(ctx, jobID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if job.Status != domain.JobFailed || job.ErrorCode == nil || *job.ErrorCode != "reducer_failed" ||
+		job.ErrorMessage == nil || *job.ErrorMessage != "final result reduction failed" {
+		t.Fatalf("unexpected failed job: %+v", job)
+	}
+	if job.ResultArtifactID != nil {
+		t.Fatal("failed reduction must not expose a final result")
+	}
+}
+
 // seedJob creates a URI-chunked job with n chunks and returns its id.
 func (h *harness) seedJob(t *testing.T, workload string, n int) uuid.UUID {
 	t.Helper()
@@ -0,0 +1,7 @@
+BEGIN;
+
+ALTER TABLE jobs DROP COLUMN IF EXISTS error_message;
+ALTER TABLE jobs DROP COLUMN IF EXISTS error_code;
+ALTER TABLE jobs DROP COLUMN IF EXISTS reducer_started_at;
+
+COMMIT;
@@ -0,0 +1,7 @@
+-- PostgreSQL enum values must be committed before they are used by a later
+-- transaction, so this migration intentionally has no BEGIN/COMMIT wrapper.
+ALTER TYPE job_status ADD VALUE IF NOT EXISTS 'reducing';
+
+ALTER TABLE jobs ADD COLUMN IF NOT EXISTS error_code text;
+ALTER TABLE jobs ADD COLUMN IF NOT EXISTS error_message text;
+ALTER TABLE jobs ADD COLUMN IF NOT EXISTS reducer_started_at timestamptz;
@@ -26,7 +26,8 @@ must be updated in the same change as any behaviour it describes.
 | `POST /tasks/{id}/heartbeat` | renew lease | ✅ done |
 | `POST /tasks/{id}/result` | complete | ✅ done, references `artifact_id` |
 | `POST /tasks/{id}/failure` | fail | ✅ done |
-| `GET /jobs/{id}` | progress | ✅ done |
+| `GET /jobs/{id}` | progress and final-result URI | ✅ done |
+| `GET /jobs/{id}/result` | download final CSV | ✅ done |
 | `PUT /tasks/{id}/artifacts/{name}` | upload partial | ✅ done |
 | `GET /artifacts/{id}/download` | download by id | ✅ done |
 | `POST /jobs/upload` | upload dataset, coordinator chunks it | ✅ done |
@@ -67,6 +68,35 @@ artifact. The coordinator splits the selected TSV rows into shard artifacts
 Each resulting task's claim response carries `input.uri = /tasks/{id}/input`,
 served by §5.4.

+## Job progress and final result
+
+```http
+GET /jobs/{job_id}
+Authorization: Bearer <token>
+```
+
+The response contains task counters and a derived status. A successful
+`similarity-search` enters `reducing` after the last shard completes, then
+becomes `completed` only after the coordinator stores its deterministic final
+CSV. At that point `result_uri` is present:
+
+```json
+{
+  "id": "uuid",
+  "status": "completed",
+  "total": 3,
+  "pending": 0,
+  "leased": 0,
+  "completed": 3,
+  "failed": 0,
+  "cancelled": 0,
+  "result_uri": "/jobs/uuid/result"
+}
+```
+
+`GET /jobs/{job_id}/result` downloads that final coordinator-owned CSV. Before
+the job is completed (or for a failed/cancelled job), it returns `404`.
+
 ## Stop a job

 ```http
@@ -56,9 +56,10 @@ Response: `{ "worker_id": "<uuid>", "heartbeat_interval_seconds": 15 }`.
 - **Keep `worker_id`**. Use it as your identity in every later call. Using the
  registered UUID is what lets the coordinator track your liveness (it marks
  workers offline after they go silent).
- Current diagnostic uploads use `similarity-search` with `query_smiles`. The
-  reference worker accepts the legacy `similarity_search` spelling too. Do not
-  advertise `similarity-graph` until CTX-10 implements cross-shard pair planning.
+- Current distributed uploads use `similarity-search` with `query_smiles`; the
+  coordinator merges completed shard candidates into a final CSV. The reference
+  worker accepts the legacy `similarity_search` spelling too. Do not advertise
+  `similarity-graph` until CTX-10 implements cross-shard pair planning.

 ## 2. Claim a task

@@ -97,8 +97,9 @@ paths:
        multipart/form-data. The text fields (`workload`, `parameters`,
        `chunk_rows`, `max_rows`) MUST precede the `file` part: the file is streamed, not
        buffered, so the fields have to be parsed before it arrives. Currently
-        only diagnostic `similarity-search` with `parameters.query_smiles` is
-        accepted; distributed graph planning is not implemented.
+        only `similarity-search` with `parameters.query_smiles` is accepted.
+        When every shard succeeds, the coordinator merges their candidates into
+        one final CSV; distributed graph planning is not implemented.
      requestBody:
        required: true
        content:
@@ -124,7 +125,9 @@ paths:
        - $ref: "#/components/parameters/JobID"
      responses:
        "200":
-          description: Progress counts and derived status.
+          description: >
+            Progress counts and derived status. A completed similarity-search
+            response includes `result_uri` for its final CSV.
          content:
            application/json:
              schema: { $ref: "#/components/schemas/JobProgress" }
@@ -132,6 +135,21 @@ paths:
        "401": { $ref: "#/components/responses/Unauthorized" }
        "404": { $ref: "#/components/responses/NotFound" }

+  /jobs/{job_id}/result:
+    get:
+      tags: [jobs]
+      summary: Download a completed job's final result
+      parameters:
+        - $ref: "#/components/parameters/JobID"
+      responses:
+        "200":
+          description: Final coordinator-owned CSV.
+          content:
+            text/csv:
+              schema: { type: string, format: binary }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "404": { $ref: "#/components/responses/NotFound" }
+
  /jobs/{job_id}/cancel:
    post:
      tags: [jobs]
@@ -483,6 +501,13 @@ components:
        completed: { type: integer }
        failed: { type: integer }
        cancelled: { type: integer }
+        result_uri:
+          type: string
+          description: Present only when the final result is available.
+          example: /jobs/5a4c3a7f-ccfc-47d6-b78d-2d1fa565bafd/result
+        error_code:
+          type: string
+          description: Sanitized terminal reducer failure code, when applicable.

    ClaimRequest:
      type: object
@@ -577,7 +602,7 @@ components:

    JobStatus:
      type: string
-      enum: [pending, running, completed, failed, cancelled]
+      enum: [pending, running, reducing, completed, failed, cancelled]

    TaskStatus:
      type: string
@@ -12,16 +12,28 @@ for a trusted local team. The coordinator remains the only process with direct
 database and artifact-storage access; the browser never calls PostgreSQL and
 never receives a worker bearer token.

-The first release must be useful before CTX-07--CTX-10 are complete. Therefore
-it has two visibly different modes:
+## Current delivered scope
+
+The initial operator UI and CTX-09 final reduction are now implemented. The
+control room polls a bounded, coordinator-owned read model every two seconds
+while a tab is visible. It shows the worker fleet, recent jobs, safe shard
+diagnostics, the actual `reducing` phase, and final-result availability. A job
+detail page renders the concrete pipeline stages—input accepted, shards,
+worker CSVs, reduction, final CSV—from coordinator state and replaces task and
+artifact views as work changes. All browser mutations remain limited to
+validated dataset upload and operator cancellation.
+
+The interface must distinguish an in-progress distributed search from a run
+whose reducer has produced a durable final result:

 | Mode | What it proves | What it must not claim |
 | --- | --- | --- |
-| **Pipeline check** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and downloads work end-to-end. | That multiple shard results have been scientifically reduced into one answer. |
-| **Final run** | A reducer has produced a durable final CSV for the full job. | Available only after CTX-09, and for graph only after CTX-10. |
+| **In-progress run** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and shard diagnostics work end-to-end. | That the partial CSVs are a global scientific answer. |
+| **Final run** | A reducer has produced a durable final CSV for the full job. | Available for `similarity-search` after CTX-09; graph remains unavailable until CTX-10. |

 Never label a partial artifact as a final molecular result. The UI must show a
-clear `Pipeline check — partial results` badge while a reducer is unavailable.
+clear waiting or `reducing` stage until a final artifact exists and the job is
+`completed`.

 ## 2. Constraints and decisions

@@ -74,7 +86,7 @@ clear `Pipeline check — partial results` badge while a reducer is unavailable.
 | Worker registration/lease flow | Implemented | Add a read-only worker list; no browser worker controls. |
 | Task diagnostics | No public list/detail response | Add sanitized job task list with attempt, status, lease owner, expiry and error. |
 | Artifact download | Worker endpoint exists | Add UI-authorized, job-scoped download proxy. |
-| Final result | Reducer is not implemented | Gate behind CTX-09; show partial diagnostic artifacts meanwhile. |
+| Final result | CTX-09 final artifact and download route exist | Show the `reducing` stage, then make the final CSV prominent only for `completed`. |
 | Distributed graph correctness | Planner/reducer unavailable | Do not advertise a multi-shard graph as final until CTX-10. |

 ## 5. Proposed structure
@@ -186,11 +198,10 @@ Rules:
 Inputs: exactly one `query_smiles` or `query_id`, `top_k`, optional threshold,
 threshold direction, `max_rows`, and `progress_every`.

-For a runnable manual pipeline check before CTX-08, offer `query_smiles` and
-default `chunk_rows` large enough to create one shard. A `query_id` across
-multiple shards is disabled with an explanation until CTX-07 resolves it once
-before fan-out. The detail page calls an artifact a **partial top-k CSV**, not
-a global top-k, until CTX-09 reduction exists.
+The current upload form accepts `query_smiles`, because resolving a
+cross-shard `query_id` has not yet been connected to coordinator uploads. The
+detail page calls an artifact a **partial top-k CSV** until all shards are
+complete and CTX-09 reduction stores the final global result.

 ### 8.3 Similarity graph

@@ -283,7 +294,7 @@ checksum/size metadata display, and prominent partial/final labels.
 file; `Content-Disposition` is safe; preview never loads an unbounded CSV; no
 final-result button exists before CTX-09.

-### WUI-06 — Final-result UX after CTX-09
+### WUI-06 — Final-result UX after CTX-09 — implemented

 **Depends on:** CTX-09 and WUI-05.

@@ -365,8 +376,9 @@ that the interface exists today.

 ## 13. Definition of done for the first hand-testable release

-WUI-00 through WUI-05 are complete when a clean local checkout can run a
-trusted, authenticated local UI; display coordinator readiness, workers, jobs,
-tasks and safe errors; submit a valid small search pipeline check; poll it to a
-terminal task state; and download/preview the coordinator-owned partial CSV.
-The page must make the absence of final reduction impossible to miss.
+The hand-testable release is complete when a clean local checkout can run a
+trusted, authenticated local UI; display workers, jobs, pipeline stages, tasks
+and safe errors; submit a valid small search; poll it through `reducing`; and
+download the coordinator-owned final CSV only after completion. The page must
+make the distinction between partial diagnostics and the final result
+impossible to miss.
Author	SHA1	Message	Date
Emil	d0aeb7fc95	Upgrade pipeline observability UI coordinator / test (push) Waiting to run Details	2026-07-24 15:13:59 +03:00
Emil	a055473706	Add final result reduction coordinator / test (push) Waiting to run Details	2026-07-24 14:58:18 +03:00