Add safe CSV artifact preview to job detail UI

Adds a Preview action next to eligible partial/final CSV artifacts on
the job detail page. Reads at most 64 KiB and 30 rows via a coordinator-
owned blob open, verifying job ownership and the same downloadable rule
as the existing download proxy so an artifact ID from another job is
never disclosed. Non-CSV and malformed/empty content fail safely with a
sanitized message instead of being rendered as text; all cell values go
through html/template escaping.

README.md

@@ -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 shard-based `similarity-search`
+coordinator and Python worker can run a diagnostic, shard-based
-pipeline locally. After every shard succeeds, the coordinator deterministically
+`similarity-search` pipeline locally. Its CSV artifacts are not a global result
-merges its candidates into one final global top-k CSV. See
+until CTX-07--09 add planning and reduction; use the local CLI for scientific
-[`STATUS.md`](STATUS.md).
+results today. 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`.
 

STATUS.md

@@ -1,7 +1,7 @@
 # SciMesh Status
 
 **Updated:** 2026-07-24
-**Branch baseline:** `main` at `6e67daa` (distributed similarity-search)
+**Branch baseline:** `main` at `f953112` (distributed pipeline hardening)
 
 ## Current state
 
@@ -18,10 +18,8 @@ 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. Completed similarity-search shard results
+uses the live coordinator contract; its HTTP path was exercised against a real
-are reduced once into a checksum-protected final CSV, which is downloadable
+Docker PostgreSQL stack on 2026-07-23.
-through the coordinator. The full Go checks (including a fresh migration and
-real PostgreSQL smoke test) passed on 2026-07-24.
 
 ## Milestone tracker
 
@@ -35,24 +33,30 @@ real PostgreSQL smoke test) passed on 2026-07-24.
 | 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 | 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-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 | Implemented | Atomic `reducing` claim, deterministic coordinator-side top-k reducer, sanitized reducer failure, final artifact persistence, `result_uri`, and final CSV download. |
+| CTX-09 Reducer and final-result API | Not started | Depends on CTX-07 and CTX-08. |
 | CTX-10 Distributed similarity-graph | Not started | Local reference exists. |
-| CTX-11 Dashboard/operator view | Implemented | Protected local view: job/task/worker status, validated similarity-search upload, partial-artifact diagnostics, final-result download, and bounded polling. |
+| 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-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. |
 
 ## Next recommended assignment
 
-Assign **CTX-10** to the distributed-science role: implement deterministic
+Assign **CTX-09** to the coordinator role: materialize planned shards,
-block-pair planning and reduction for `similarity-graph`.
+persist them transactionally, invoke the registered reducer once, and expose a
+durable final artifact.
 
 ## 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`; the upload UI
+  planner resolves `query_id` once and shares `query_smiles`; connecting that
-  currently accepts `query_smiles` only.
+  planner to uploaded coordinator jobs belongs to CTX-09.
 - 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.

coordinator/README.md

@@ -76,10 +76,9 @@ 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, diagnostic shard artifacts, and the
+It shows recent jobs, task/worker state, and the per-job partial artifacts.
-final CSV for completed similarity-search jobs. The coordinator enters
+Those files are explicitly diagnostic until the CTX-09 reducer creates a final
-`reducing` after the last shard completes, then exposes the final deterministic
+result; the UI does not present them as final scientific output.
-global top-k result when merging succeeds.
 
 `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

coordinator/cmd/coordinator/main.go

@@ -75,15 +75,14 @@ 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

coordinator/internal/domain/job.go

@@ -11,7 +11,6 @@ type JobStatus string
 const (
 	JobPending   JobStatus = "pending"
 	JobRunning   JobStatus = "running"
-	JobReducing  JobStatus = "reducing"
 	JobCompleted JobStatus = "completed"
 	JobFailed    JobStatus = "failed"
 	JobCancelled JobStatus = "cancelled"
@@ -19,18 +18,14 @@ const (
 
 // Job is one user submission that fans out into one or more tasks.
 type Job struct {
-	ID               uuid.UUID
+	ID              uuid.UUID
-	Workload         string
+	Workload        string
-	InputURI         string     // external input URI; empty for uploaded datasets
+	InputURI        string     // external input URI; empty for uploaded datasets
-	InputArtifactID  *uuid.UUID // uploaded input artifact; nil for URI submissions
+	InputArtifactID *uuid.UUID // uploaded input artifact; nil for URI submissions
-	ResultArtifactID *uuid.UUID
+	Parameters      map[string]any
-	Parameters       map[string]any
+	Status          JobStatus
-	Status           JobStatus
+	CreatedAt       time.Time
-	CreatedAt        time.Time
+	CompletedAt     *time.Time
-	CompletedAt      *time.Time
-	ReducerStartedAt *time.Time
-	ErrorCode        *string
-	ErrorMessage     *string
 }
 
 // NewUploadedJob builds a job whose input was uploaded to the coordinator. The
@@ -119,8 +114,6 @@ 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:

coordinator/internal/memstore/memstore.go

@@ -216,51 +216,6 @@ 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 {

coordinator/internal/reducer/similarity_search.go

@@ -1,195 +0,0 @@
-// 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
-	}
-}

coordinator/internal/reducer/similarity_search_test.go

@@ -1,41 +0,0 @@
-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")
-	}
-}

coordinator/internal/storage/postgres/integration_test.go

@@ -90,49 +90,6 @@ 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)
-	}
-}
-
 // 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) {

coordinator/internal/storage/postgres/job_repo.go

@@ -25,10 +25,7 @@ func NewJobRepo(pool *pgxpool.Pool) *JobRepo {
 
 var _ usecase.JobRepository = (*JobRepo)(nil)
 
-var jobColumns = []string{
+var jobColumns = []string{"id", "workload", "input_uri", "parameters", "status", "created_at", "completed_at"}
-	"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.
@@ -58,8 +55,7 @@ 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
 	}
@@ -70,64 +66,6 @@ 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 {
 

coordinator/internal/transport/http/dto.go

@@ -119,8 +119,6 @@ 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 {
@@ -155,7 +153,7 @@ func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse {
 }
 
 func toJobProgressResponse(p domain.JobProgress) jobProgressResponse {
-	out := jobProgressResponse{
+	return jobProgressResponse{
 		ID:        p.Job.ID,
 		Status:    string(p.DeriveStatus()),
 		Total:     p.Total,
@@ -165,11 +163,4 @@ 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
 }

coordinator/internal/transport/http/dto_test.go

@@ -1,25 +0,0 @@
-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)
-	}
-}

coordinator/internal/transport/http/handlers.go

@@ -150,12 +150,6 @@ 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)})
 }
 
@@ -405,25 +399,6 @@ 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) {

coordinator/internal/transport/http/server.go

@@ -22,15 +22,14 @@ 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 {
@@ -64,7 +63,6 @@ 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)
@@ -85,6 +83,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 		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 {

coordinator/internal/transport/http/server_test.go

@@ -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
-	downloadArtifact := usecase.NewDownloadArtifact(arts, blobs)
+	uiRead := memstore.NewUIReadRepo(jobs, tasks, work, arts)
 
 	uc := coordhttp.UseCases{
 		RegisterWorker:   usecase.NewRegisterWorker(work, clk),
@@ -51,15 +51,14 @@ 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: downloadArtifact,
+		DownloadArtifact: usecase.NewDownloadArtifact(arts, blobs),
 		GetTaskInput:     usecase.NewGetTaskInput(tasks, arts, blobs),
-		Dashboard:        usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),
+		Dashboard:        usecase.NewDashboard(uiRead),
+		PreviewArtifact:  usecase.NewPreviewArtifact(uiRead, blobs),
 	}
 	worker, err := uc.RegisterWorker.Execute(context.Background(), usecase.RegisterWorkerInput{
 		Name: "test-worker", Capabilities: []string{"w", "similarity-search"},
@@ -290,6 +289,134 @@ 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")
@@ -403,42 +530,6 @@ 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)
-	}
-}
-
 func TestForeignArtifactResultConflict(t *testing.T) {
 	e := newEnv(t, healthy)
 	e.do(t, "POST", "/jobs", `{"workload":"w","input_uri":"s3://in",

coordinator/internal/transport/http/templates/artifact-preview.html

@@ -0,0 +1,31 @@
+{{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}}

coordinator/internal/transport/http/templates/dashboard.html

@@ -12,7 +12,7 @@
 <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="Pipeline flow"><strong>Similarity-search jobs produce a final CSV.</strong><span>Workers return shard-level candidates; after every shard succeeds, the coordinator deterministically merges them into one global top-k 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. Download result</b>When merging finishes, download the final CSV from the job page.</div></div></section>
+    <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>

coordinator/internal/transport/http/templates/job.html

@@ -13,14 +13,14 @@
   <main class="page">
     <a class="back" href="/ui">← Back to jobs</a><p class="eyebrow">{{workloadLabel .Workload}}</p><h1>Execution progress</h1>
     <section class="summary"><div class="summary-head"><div><span id="status" class="status status-{{statusClass .Status}}">{{statusLabel .Status}}</span><p id="hint" class="hint">{{statusHint .Status}}</p></div><div>{{if cancellable .Status}}<button id="stop-job" class="stop" type="button">Stop job</button><small>This cancels every shard that is not finished yet.</small>{{else}}<small>Summary refreshes automatically every two seconds.</small>{{end}}</div></div><div class="bar" aria-label="Progress"><span id="progress-bar" style="width:{{progressPercent .Completed .Failed .Cancelled .Total}}%"></span></div><p id="progress" class="muted">{{.Completed}} of {{.Total}} tasks complete</p><div class="numbers"><div class="number"><b id="total">{{.Total}}</b><small>total shards</small></div><div class="number"><b id="completed">{{.Completed}}</b><small>complete</small></div><div class="number"><b id="pending">{{.Pending}}</b><small>waiting</small></div><div class="number"><b id="active">{{add .Leased .Running}}</b><small>with workers</small></div><div class="number"><b id="failed">{{.Failed}}</b><small>failed</small></div><div class="number"><b id="cancelled">{{.Cancelled}}</b><small>stopped</small></div></div><details><summary>Technical details</summary><p>Job ID: <code>{{.ID}}</code><br>Workload: <code>{{.Workload}}</code><br>Created: {{time .CreatedAt}}</p></details></section>
-    <section class="notice">{{if .FinalResultAvailable}}<strong>Final result ready</strong><br>Download the <code>final_result</code> CSV below. It is the deterministic global top-k across all completed shards. The <code>partial_result</code> files remain available for diagnostics.{{else if eq .Status "reducing"}}<strong>Merging completed shards</strong><br>Every shard has finished. The coordinator is building one deterministic global CSV; refresh in a moment to download it.{{else}}<strong>What can be downloaded now?</strong><br><code>partial_result</code> files come from individual shards. They are useful for checking the pipeline, but are not a merged final CSV yet.{{end}}</section>
+    <section class="notice"><strong>What can be downloaded now?</strong><br><code>partial_result</code> files come from individual shards. They are useful for checking the pipeline, but are not a merged final CSV because the reducer is not implemented yet.</section>
     <h2>Shard tasks</h2><p class="muted">If a task fails, its code and message appear here. Refresh the page to update the detailed rows.</p>
     <div class="table-wrap"><table><tr><th>Shard</th><th>State</th><th>Attempt</th><th>Worker / lease</th><th>Error</th></tr>{{range .Tasks}}<tr><td>#{{.ChunkIndex}}</td><td><span class="status status-{{statusClass .Status}}">{{statusLabel .Status}}</span></td><td>{{.Attempt}} / {{.MaxAttempts}}</td><td>{{if .LeaseOwner}}<code>{{.LeaseOwner}}</code>{{if .LeaseExpiresAt}}<br><small>until {{time .LeaseExpiresAt}}</small>{{end}}{{else}}<span class="muted">—</span>{{end}}</td><td class="error">{{if .ErrorCode}}<strong>{{taskErrorLabel .ErrorCode}}</strong><br><small>{{taskErrorHint .ErrorCode}}</small>{{else}}<span class="muted">—</span>{{end}}</td></tr>{{else}}<tr><td colspan="5" class="empty">No tasks have appeared yet.</td></tr>{{end}}</table></div>
     <h2>Coordinator artifacts</h2>
-    <div class="table-wrap"><table><tr><th>Type</th><th>File</th><th>Size</th><th>Integrity check</th><th></th></tr>{{range .Artifacts}}<tr><td>{{if .Diagnostic}}<strong>Partial result</strong><br><small>diagnostic</small>{{else}}{{.Kind}}{{end}}</td><td>{{.Filename}}</td><td>{{bytes .SizeBytes}}</td><td><code>{{.SHA256}}</code></td><td>{{if .Downloadable}}<a class="download" href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}">Download CSV</a>{{else}}<span class="muted">Unavailable</span>{{end}}</td></tr>{{else}}<tr><td colspan="5" class="empty">No artifacts yet. The worker uploads a CSV after it completes a shard.</td></tr>{{end}}</table></div>
+    <div class="table-wrap"><table><tr><th>Type</th><th>File</th><th>Size</th><th>Integrity check</th><th></th></tr>{{range .Artifacts}}<tr><td>{{if .Diagnostic}}<strong>Partial result</strong><br><small>diagnostic</small>{{else}}{{.Kind}}{{end}}</td><td>{{.Filename}}</td><td>{{bytes .SizeBytes}}</td><td><code>{{.SHA256}}</code></td><td>{{if .Downloadable}}<a class="download" href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}">Download CSV</a> <a class="download" href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}/preview">Preview</a>{{else}}<span class="muted">Unavailable</span>{{end}}</td></tr>{{else}}<tr><td colspan="5" class="empty">No artifacts yet. The worker uploads a CSV after it completes a shard.</td></tr>{{end}}</table></div>
   </main>
   <script>
-    const id={{printf "%q" .ID}},state={pending:['Waiting for a worker','waiting','Waiting for an available worker with the required capability.'],leased:['Assigned to a worker','active','A worker has claimed the task and should begin processing shortly.'],running:['Running','active','A worker is reading a shard, calculating fingerprints, and uploading its result through the coordinator.'],reducing:['Merging results','active','All shards are complete. The coordinator is merging their candidates into one final CSV.'],completed:['Completed','success','The final result is ready to download.'],failed:['Needs attention','danger','One or more shard tasks failed. Open the task list below for details.'],cancelled:['Stopped','waiting','The operator stopped this job. No new shards can be claimed.']};
+    const id={{printf "%q" .ID}},state={pending:['Waiting for a worker','waiting','Waiting for an available worker with the required capability.'],leased:['Assigned to a worker','active','A worker has claimed the task and should begin processing shortly.'],running:['Running','active','A worker is reading a shard, calculating fingerprints, and uploading its result through the coordinator.'],completed:['Tasks complete','success','Every shard task is complete. Files below are still partial results.'],failed:['Needs attention','danger','One or more shard tasks failed. Open the task list below for details.'],cancelled:['Stopped','waiting','The operator stopped this job. No new shards can be claimed.']};
     const stop=document.querySelector('#stop-job');if(stop)stop.addEventListener('click',async()=>{if(!confirm('Stop this job? Unfinished shards will be cancelled.'))return;stop.disabled=true;const response=await fetch('/ui/api/jobs/'+id+'/cancel',{method:'POST'});if(!response.ok){stop.disabled=false;alert('Unable to stop this job.');return}location.reload()});
     const terminal=new Set(['completed','failed','cancelled']);let timer;const poll=async()=>{try{const response=await fetch('/ui/api/jobs/'+id);if(!response.ok)return;const job=await response.json(),info=state[job.status]||[job.status,'waiting','Status reported by the coordinator.'],done=job.completed+job.failed+job.cancelled,percent=job.total?Math.min(100,Math.floor(done*100/job.total)):0,badge=document.querySelector('#status');badge.textContent=info[0];badge.className='status status-'+info[1];document.querySelector('#hint').textContent=info[2];document.querySelector('#progress').textContent=job.completed+' of '+job.total+' tasks complete'+(job.failed?' · failed: '+job.failed:'')+(job.cancelled?' · stopped: '+job.cancelled:'');document.querySelector('#progress-bar').style.width=percent+'%';for(const key of ['total','completed','pending','failed','cancelled'])document.querySelector('#'+key).textContent=job[key];document.querySelector('#active').textContent=job.leased+job.running;if(terminal.has(job.status)&&timer){clearInterval(timer);timer=undefined}}catch(_){}};const start=()=>{if(!timer&&!document.hidden&&!terminal.has(document.querySelector('#status').textContent.toLowerCase()))timer=setInterval(poll,2000)};document.addEventListener('visibilitychange',()=>{if(document.hidden&&timer){clearInterval(timer);timer=undefined}else start()});start();
   </script>

coordinator/internal/transport/http/ui.go

@@ -48,10 +48,8 @@ func uiStatusLabel(status string) string {
 		return "Assigned to a worker"
 	case "running":
 		return "Running"
-	case "reducing":
-		return "Merging results"
 	case "completed":
-		return "Completed"
+		return "Tasks complete"
 	case "failed":
 		return "Needs attention"
 	case "cancelled":
@@ -69,10 +67,8 @@ 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 "The final result is ready to download."
+		return "Every shard task is complete. Files below are still partial results."
 	case "failed":
 		return "One or more shard tasks failed. Open the task list below for details."
 	case "cancelled":
@@ -90,7 +86,7 @@ func uiStatusClass(status string) string {
 		return "danger"
 	case "cancelled":
 		return "waiting"
-	case "running", "leased", "reducing":
+	case "running", "leased":
 		return "active"
 	default:
 		return "waiting"
@@ -283,3 +279,26 @@ 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)
+}

coordinator/internal/transport/http/ui_test.go

@@ -10,8 +10,7 @@ func TestUIStatusPresentation(t *testing.T) {
 	}{
 		{"pending", "Waiting for a worker", "waiting"},
 		{"running", "Running", "active"},
-		{"reducing", "Merging results", "active"},
+		{"completed", "Tasks complete", "success"},
-		{"completed", "Completed", "success"},
 		{"failed", "Needs attention", "danger"},
 	}
 	for _, test := range tests {

coordinator/internal/usecase/job.go

@@ -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.JobReducing || job.Status == domain.JobCompleted || job.Status == domain.JobFailed {
+		if 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.JobReducing || derived == domain.JobCompleted || derived == domain.JobFailed {
+		if derived == domain.JobCompleted || derived == domain.JobFailed {
 			return domain.ErrJobNotCancellable
 		}
 		cancelled, err = uc.tasks.CancelByJob(ctx, jobID, now)
@@ -226,20 +226,10 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository
 	if err != nil {
 		return err
 	}
-	job, err := jobs.Get(ctx, jobID)
+	status := progressFrom(domain.Job{}, counts).DeriveStatus()
-	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.JobFailed {
+	if status == domain.JobCompleted || status == domain.JobFailed {
 		completedAt = &now
 	}
 	return jobs.UpdateStatus(ctx, jobID, status, completedAt)

coordinator/internal/usecase/ports.go

@@ -70,9 +70,6 @@ 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.

coordinator/internal/usecase/preview.go

@@ -0,0 +1,150 @@
+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")
+}

coordinator/internal/usecase/preview_test.go

@@ -0,0 +1,226 @@
+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)
+	}
+}

coordinator/internal/usecase/reduce.go

@@ -1,133 +0,0 @@
-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)
-}

coordinator/internal/usecase/usecase_test.go

@@ -55,8 +55,6 @@ type harness struct {
 	getInput    *usecase.GetTaskInput
 	expire      *usecase.ExpireLeases
 	cancel      *usecase.CancelJob
-	reduce      *usecase.ReduceJob
-	jobResult   *usecase.GetJobResult
 }
 
 func newHarness() *harness {
@@ -83,88 +81,9 @@ 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()

coordinator/migrations/0010_job_reduction.down.sql

@@ -1,7 +0,0 @@
-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;

coordinator/migrations/0010_job_reduction.up.sql

@@ -1,7 +0,0 @@
--- 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;

docs/api-contract.md

@@ -26,8 +26,7 @@ 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 and final-result URI | ✅ done |
+| `GET /jobs/{id}` | progress | ✅ 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 |
@@ -68,35 +67,6 @@ 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

docs/building-workers.md

@@ -56,10 +56,9 @@ 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 distributed uploads use `similarity-search` with `query_smiles`; the
+- Current diagnostic uploads use `similarity-search` with `query_smiles`. The
-  coordinator merges completed shard candidates into a final CSV. The reference
+  reference worker accepts the legacy `similarity_search` spelling too. Do not
-  worker accepts the legacy `similarity_search` spelling too. Do not advertise
+  advertise `similarity-graph` until CTX-10 implements cross-shard pair planning.
-  `similarity-graph` until CTX-10 implements cross-shard pair planning.
 
 ## 2. Claim a task
 

docs/openapi.yaml

@@ -97,9 +97,8 @@ 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 `similarity-search` with `parameters.query_smiles` is accepted.
+        only diagnostic `similarity-search` with `parameters.query_smiles` is
-        When every shard succeeds, the coordinator merges their candidates into
+        accepted; distributed graph planning is not implemented.
-        one final CSV; distributed graph planning is not implemented.
       requestBody:
         required: true
         content:
@@ -125,9 +124,7 @@ paths:
         - $ref: "#/components/parameters/JobID"
       responses:
         "200":
-          description: >
+          description: Progress counts and derived status.
-            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" }
@@ -135,21 +132,6 @@ 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]
@@ -501,13 +483,6 @@ 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
@@ -602,7 +577,7 @@ components:
 
     JobStatus:
       type: string
-      enum: [pending, running, reducing, completed, failed, cancelled]
+      enum: [pending, running, completed, failed, cancelled]
 
     TaskStatus:
       type: string