Add job cancellation and dataset row limit

STATUS.md

@@ -36,7 +36,7 @@ Docker PostgreSQL stack on 2026-07-23.
 | CTX-08 Distributed similarity-search | Not started | Local reference exists. |
 | 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 | Not started | Deferred until API and reducer work. |
+| CTX-11 Dashboard/operator view | In progress | `feat/web-interface` adds a protected local view: job/task/worker status, dataset upload, diagnostic partial-artifact download, and 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
@@ -46,8 +46,9 @@ reduction boundaries before implementing distributed search or graph execution.
 
 ## Known constraints
 
-- Planner/reducer semantics are not implemented; use the local `scimesh` CLI
+- Planner/reducer semantics are not implemented; the operator UI labels
-  for complete workload results.
+  `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. Resolving a

coordinator/.env.example

@@ -6,6 +6,10 @@ DATABASE_URL=postgres://scimesh:scimesh@localhost:5432/scimesh?sslmode=disable
 # Shared bearer token every worker must present. Leave empty to disable auth (dev only).
 WORKER_AUTH_TOKEN=change-me
 
+# Optional local operator UI. Use a separate value; never reuse the worker token.
+# When empty, /ui is disabled.
+UI_AUTH_TOKEN=
+
 # Logging. LOG_LEVEL: debug|info|warn|error. LOG_FILE empty = stdout only;
 # set a path to also write a size-rotated file (kept across restarts).
 LOG_LEVEL=info

coordinator/README.md

@@ -68,6 +68,18 @@ make logs                     # follow the coordinator
 make down                     # stop (add down-clean to drop the DB volume)
 ```
 
+To enable the local operator UI, set a separate credential before starting:
+
+```sh
+UI_AUTH_TOKEN='local-ui-secret' make up
+# Open http://localhost:8080/ui and use any username with this value as password.
+```
+
+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.
+
 `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 start — so it never queries a database that has no tables.

coordinator/cmd/coordinator/main.go

@@ -65,6 +65,7 @@ func run() error {
 		jobRepo      = postgres.NewJobRepo(pool)
 		workerRepo   = postgres.NewWorkerRepo(pool)
 		artifactRepo = postgres.NewArtifactRepo(pool)
+		uiReadRepo   = postgres.NewUIReadRepo(pool)
 	)
 
 	useCases := httptransport.UseCases{
@@ -76,9 +77,11 @@ func run() error {
 		CompleteTask:     usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, 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, clk),
 		DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore),
 		GetTaskInput:     usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
+		Dashboard:        usecase.NewDashboard(uiReadRepo),
 	}
 
 	// Background reapers are tracked so shutdown can wait for them. Without this
@@ -105,7 +108,7 @@ func run() error {
 	// pool.Ping backs /health: readiness means the database answers, not just
 	// that the process is alive.
 	api := httptransport.NewServer(useCases, log, cfg.RequestTimeout, cfg.HeartbeatInterval, cfg.MaxUploadBytes, pool.Ping)
-	err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token))
+	err = infra.RunServer(ctx, log, cfg.Addr, api.Handler(cfg.Token, cfg.UIToken))
 
 	// Shutdown order matters, and defers alone cannot express it (they run
 	// LIFO, so the deferred stop() would fire *after* the wait below).

coordinator/docker-compose.yml

@@ -49,6 +49,8 @@ services:
       # Host is the service name: compose resolves it on the project network.
       DATABASE_URL: postgres://${POSTGRES_USER:-scimesh}:${POSTGRES_PASSWORD:-scimesh}@postgres:5432/${POSTGRES_DB:-scimesh}?sslmode=disable
       WORKER_AUTH_TOKEN: ${WORKER_AUTH_TOKEN:-dev-token}
+      # Empty disables /ui. Set this separately from the worker token.
+      UI_AUTH_TOKEN: ${UI_AUTH_TOKEN:-}
       DB_MAX_CONNS: "10"
       REQUEST_TIMEOUT: "15s"
       LEASE_DURATION: "2m"

coordinator/internal/chunk/tsv.go

@@ -26,9 +26,19 @@ var ErrNoRows = fmt.Errorf("input has no data rows")
 // Only one shard is buffered at a time, so memory is bounded by shard size (a
 // worker-sized slice of the data), not by the size of the whole dataset.
 func SplitTSV(r io.Reader, rowsPerShard int, emit func(index int, shard io.Reader) error) error {
+	return SplitTSVLimit(r, rowsPerShard, 0, emit)
+}
+
+// SplitTSVLimit behaves like SplitTSV but emits no more than maxRows data rows.
+// A maxRows value of zero means unlimited. This lets an operator make a small,
+// representative pipeline check without materialising a second dataset file.
+func SplitTSVLimit(r io.Reader, rowsPerShard, maxRows int, emit func(index int, shard io.Reader) error) error {
 	if rowsPerShard <= 0 {
 		return fmt.Errorf("rowsPerShard must be positive, got %d", rowsPerShard)
 	}
+	if maxRows < 0 {
+		return fmt.Errorf("maxRows must be non-negative, got %d", maxRows)
+	}
 
 	sc := bufio.NewScanner(r)
 	// Allow long lines: a SMILES row can be far wider than bufio's 64 KB default.
@@ -73,6 +83,9 @@ func SplitTSV(r io.Reader, rowsPerShard int, emit func(index int, shard io.Reade
 				return err
 			}
 		}
+		if maxRows > 0 && index*rowsPerShard+rows == maxRows {
+			break
+		}
 	}
 	if err := sc.Err(); err != nil {
 		return fmt.Errorf("read rows: %w", err)

coordinator/internal/chunk/tsv_test.go

@@ -103,6 +103,22 @@ func TestSplitSingleShardWhenSizeExceedsRows(t *testing.T) {
 	}
 }
 
+func TestSplitLimitUsesOnlyLeadingDataRows(t *testing.T) {
+	input := "h\nr1\nr2\nr3\nr4\nr5\n"
+	var shards []string
+	err := SplitTSVLimit(strings.NewReader(input), 2, 3, func(_ int, shard io.Reader) error {
+		b, _ := io.ReadAll(shard)
+		shards = append(shards, string(b))
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, want := strings.Join(shards, ""), "h\nr1\nr2\nh\nr3\n"; got != want {
+		t.Errorf("limited shards = %q, want %q", got, want)
+	}
+}
+
 // The scanned bytes are reused by bufio; the shard buffer must copy them, or a
 // later row would corrupt an earlier one. This guards that copy.
 func TestSplitDoesNotAliasScannerBuffer(t *testing.T) {

coordinator/internal/domain/errors.go

@@ -12,6 +12,7 @@ var (
 	ErrTaskNotFound      = errors.New("task not found")
 	ErrWorkerNotFound    = errors.New("worker not found")
 	ErrArtifactNotFound  = errors.New("artifact not found")
+	ErrJobNotCancellable = errors.New("job cannot be cancelled")
 	ErrLeaseConflict     = errors.New("task leased to another worker")
 	ErrStaleAttempt      = errors.New("attempt does not match lease")
 	ErrResultConflict    = errors.New("different result already recorded")

coordinator/internal/domain/job.go

@@ -105,12 +105,15 @@ type JobProgress struct {
 	Leased    int
 	Done      int
 	Failed    int
+	Cancelled int
 }
 
 // DeriveStatus computes what the job's status should be from its task counts,
 // so the rule lives here rather than in a SQL trigger or a handler.
 func (p JobProgress) DeriveStatus() JobStatus {
 	switch {
+	case p.Job.Status == JobCancelled:
+		return JobCancelled
 	case p.Total == 0:
 		return JobPending
 	case p.Done == p.Total:

coordinator/internal/domain/job_test.go

@@ -81,6 +81,7 @@ func TestDeriveStatus(t *testing.T) {
 		{"all done", JobProgress{Total: 3, Done: 3}, JobCompleted},
 		{"done and failed", JobProgress{Total: 3, Done: 2, Failed: 1}, JobFailed},
 		{"failed but work remains", JobProgress{Total: 3, Pending: 1, Failed: 2}, JobRunning},
+		{"cancelled job wins over task histogram", JobProgress{Job: Job{Status: JobCancelled}, Total: 3, Done: 1, Cancelled: 2}, JobCancelled},
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {

coordinator/internal/domain/task.go

@@ -259,6 +259,24 @@ func (t *Task) ExpireLease(now time.Time) {
 	t.CompletedAt = &now
 }
 
+// Cancel prevents any further worker transition for a task that has not
+// reached a terminal result. A cancelled lease deliberately becomes invalid:
+// a worker still running locally must not upload or complete after its job was
+// stopped by the operator.
+func (t *Task) Cancel(now time.Time) bool {
+	if t.Status == TaskCompleted || t.Status == TaskFailed || t.Status == TaskCancelled {
+		return false
+	}
+	t.Status = TaskCancelled
+	t.LeaseOwner = nil
+	t.LeaseExpiresAt = nil
+	t.ErrorCode = nil
+	t.ErrorMessage = nil
+	t.CompletedAt = &now
+	t.Version++
+	return true
+}
+
 // ClaimedTask is the worker-facing projection of a leased task. Input is either
 // an external URI or a coordinator-stored shard (InputArtifactID set); the
 // transport turns the latter into a coordinator download URL.

coordinator/internal/domain/task_test.go

@@ -168,6 +168,23 @@ func TestExpireLeaseIgnoresUnleasedTasks(t *testing.T) {
 	}
 }
 
+func TestCancelInvalidatesLeaseButPreservesTerminalTask(t *testing.T) {
+	task := leasedTask(1, 3)
+	if !task.Cancel(testNow) {
+		t.Fatal("leased task should be cancelled")
+	}
+	if task.Status != TaskCancelled || task.LeaseOwner != nil || task.LeaseExpiresAt != nil {
+		t.Errorf("cancelled task = %+v", task)
+	}
+	if task.Cancel(testLater) {
+		t.Error("cancelled task must not be changed twice")
+	}
+	completed := &Task{Status: TaskCompleted}
+	if completed.Cancel(testNow) {
+		t.Error("completed task must remain terminal")
+	}
+}
+
 func TestFirstHeartbeatMovesLeasedToRunning(t *testing.T) {
 	task := leasedTask(1, 3)
 	until := testLater.Add(time.Hour)

coordinator/internal/infra/config.go

@@ -24,6 +24,8 @@ type Config struct {
 	DatabaseURL string
 	// Shared bearer token workers must present. Empty disables auth (dev only).
 	Token string
+	// Local operator UI credential. Empty disables the embedded UI entirely.
+	UIToken string
 
 	// Minimum log level: debug, info, warn, error.
 	LogLevel string
@@ -77,6 +79,7 @@ func LoadConfig() (Config, error) {
 		// COORDINATOR_TOKEN is the contract name; WORKER_AUTH_TOKEN is the
 		// former name, still honoured so existing .env files keep working.
 		Token:              getEnv("COORDINATOR_TOKEN", os.Getenv("WORKER_AUTH_TOKEN")),
+		UIToken:            os.Getenv("UI_AUTH_TOKEN"),
 		LogLevel:           getEnv("LOG_LEVEL", "info"),
 		LogFile:            os.Getenv("LOG_FILE"),
 		StorageDir:         getEnv("COORDINATOR_STORAGE_DIR", "./data"),
@@ -94,6 +97,9 @@ func LoadConfig() (Config, error) {
 	if cfg.DatabaseURL == "" {
 		return Config{}, fmt.Errorf("DATABASE_URL is required")
 	}
+	if cfg.UIToken != "" && cfg.Token != "" && cfg.UIToken == cfg.Token {
+		return Config{}, fmt.Errorf("UI_AUTH_TOKEN must differ from the worker auth token")
+	}
 
 	var err error
 	if cfg.DBMaxConns, err = getEnvInt32("DB_MAX_CONNS", cfg.DBMaxConns); err != nil {

coordinator/internal/infra/config_test.go

@@ -0,0 +1,34 @@
+package infra
+
+import (
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestLoadConfigRejectsSharedUIAndWorkerToken(t *testing.T) {
+	t.Setenv("ENV_FILE", filepath.Join(t.TempDir(), "missing.env"))
+	t.Setenv("DATABASE_URL", "postgres://test")
+	t.Setenv("COORDINATOR_TOKEN", "shared-secret")
+	t.Setenv("UI_AUTH_TOKEN", "shared-secret")
+
+	_, err := LoadConfig()
+	if err == nil || !strings.Contains(err.Error(), "must differ") {
+		t.Fatalf("LoadConfig error = %v, want distinct-token error", err)
+	}
+}
+
+func TestLoadConfigAllowsDistinctUIAndWorkerTokens(t *testing.T) {
+	t.Setenv("ENV_FILE", filepath.Join(t.TempDir(), "missing.env"))
+	t.Setenv("DATABASE_URL", "postgres://test")
+	t.Setenv("COORDINATOR_TOKEN", "worker-secret")
+	t.Setenv("UI_AUTH_TOKEN", "ui-secret")
+
+	cfg, err := LoadConfig()
+	if err != nil {
+		t.Fatalf("LoadConfig: %v", err)
+	}
+	if cfg.Token != "worker-secret" || cfg.UIToken != "ui-secret" {
+		t.Fatalf("unexpected tokens: %+v", cfg)
+	}
+}

coordinator/internal/memstore/memstore.go

@@ -148,6 +148,18 @@ func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[doma
 	return counts, nil
 }
 
+func (r *TaskRepo) CancelByJob(_ context.Context, jobID uuid.UUID, now time.Time) (int64, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var cancelled int64
+	for _, task := range r.tasks {
+		if task.JobID == jobID && task.Cancel(now) {
+			cancelled++
+		}
+	}
+	return cancelled, nil
+}
+
 func (r *TaskRepo) ExpireLeases(ctx context.Context, now time.Time) (int64, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()

coordinator/internal/memstore/ui_read.go

@@ -0,0 +1,102 @@
+package memstore
+
+import (
+	"context"
+	"sort"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// UIReadRepo is the in-memory read projection used by HTTP/UI tests.
+type UIReadRepo struct {
+	jobs      *JobRepo
+	tasks     *TaskRepo
+	workers   *WorkerRepo
+	artifacts *ArtifactRepo
+}
+
+func NewUIReadRepo(j *JobRepo, t *TaskRepo, w *WorkerRepo, a *ArtifactRepo) *UIReadRepo {
+	return &UIReadRepo{j, t, w, a}
+}
+
+var _ usecase.UIReadRepository = (*UIReadRepo)(nil)
+
+func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
+	return r.jobs.Get(ctx, id)
+}
+func (r *UIReadRepo) ListJobs(_ context.Context, limit int) ([]domain.Job, error) {
+	if limit < 1 || limit > 100 {
+		return nil, domain.ErrInvalidInput
+	}
+	r.jobs.mu.Lock()
+	defer r.jobs.mu.Unlock()
+	out := make([]domain.Job, 0, len(r.jobs.jobs))
+	for _, job := range r.jobs.jobs {
+		out = append(out, *job)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].CreatedAt.Equal(out[j].CreatedAt) {
+			return out[i].ID.String() > out[j].ID.String()
+		}
+		return out[i].CreatedAt.After(out[j].CreatedAt)
+	})
+	if len(out) > limit {
+		out = out[:limit]
+	}
+	return out, nil
+}
+func (r *UIReadRepo) ListTasksByJob(_ context.Context, jobID uuid.UUID) ([]domain.Task, error) {
+	r.tasks.mu.Lock()
+	defer r.tasks.mu.Unlock()
+	out := []domain.Task{}
+	for _, task := range r.tasks.tasks {
+		if task.JobID == jobID {
+			out = append(out, *clone(task))
+		}
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].ChunkIndex < out[j].ChunkIndex })
+	return out, nil
+}
+func (r *UIReadRepo) ListWorkers(_ context.Context, limit int) ([]domain.Worker, error) {
+	if limit < 1 || limit > 100 {
+		return nil, domain.ErrInvalidInput
+	}
+	r.workers.mu.Lock()
+	defer r.workers.mu.Unlock()
+	out := []domain.Worker{}
+	for _, worker := range r.workers.workers {
+		copy := *worker
+		copy.Capabilities = append([]string(nil), worker.Capabilities...)
+		out = append(out, copy)
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].LastHeartbeatAt.Equal(out[j].LastHeartbeatAt) {
+			return out[i].ID.String() > out[j].ID.String()
+		}
+		return out[i].LastHeartbeatAt.After(out[j].LastHeartbeatAt)
+	})
+	if len(out) > limit {
+		out = out[:limit]
+	}
+	return out, nil
+}
+func (r *UIReadRepo) ListArtifactsByJob(_ context.Context, jobID uuid.UUID) ([]domain.Artifact, error) {
+	r.artifacts.mu.Lock()
+	defer r.artifacts.mu.Unlock()
+	out := []domain.Artifact{}
+	for _, artifact := range r.artifacts.arts {
+		if artifact.JobID == jobID {
+			out = append(out, *artifact)
+		}
+	}
+	sort.Slice(out, func(i, j int) bool {
+		if out[i].CreatedAt.Equal(out[j].CreatedAt) {
+			return out[i].ID.String() < out[j].ID.String()
+		}
+		return out[i].CreatedAt.Before(out[j].CreatedAt)
+	})
+	return out, nil
+}

coordinator/internal/storage/postgres/integration_test.go

@@ -137,12 +137,16 @@ func TestConcurrentClaimGivesEachTaskToExactlyOneWorker(t *testing.T) {
 		claimed = make(map[uuid.UUID]string)
 		wg      sync.WaitGroup
 	)
-	// More workers than tasks, so the surplus must come back empty rather than
+	// More workers than tasks. With SKIP LOCKED, a concurrent caller can
-	// steal an already-leased row.
+	// transiently see no eligible row while every remaining row is locked by a
+	// different claim statement. Poll briefly, as a real worker does, before
+	// treating the queue as empty. This verifies the actual contract: tasks are
+	// unique and all eventually become claimable without lock contention.
 	for i := 0; i < tasks*2; i++ {
 		wg.Add(1)
 		go func(n int) {
 			defer wg.Done()
+			for attempt := 0; attempt < 20; attempt++ {
 				task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{
 					Owner:      fmt.Sprintf("worker-%d", n),
 					Now:        now,
@@ -153,14 +157,17 @@ func TestConcurrentClaimGivesEachTaskToExactlyOneWorker(t *testing.T) {
 					return
 				}
 				if task == nil || task.JobID != job.ID {
-				return // empty queue, or a task from another test's job
+					time.Sleep(time.Millisecond)
+					continue
 				}
 				mu.Lock()
-			defer mu.Unlock()
 				if prev, dup := claimed[task.ID]; dup {
 					t.Errorf("task %s handed to both %s and worker-%d", task.ID, prev, n)
 				}
 				claimed[task.ID] = fmt.Sprintf("worker-%d", n)
+				mu.Unlock()
+				return
+			}
 		}(i)
 	}
 	wg.Wait()
@@ -199,6 +206,27 @@ func TestClaimNextReturnsNilOnEmptyQueue(t *testing.T) {
 	}
 }
 
+func TestCancelJobCancelsEveryUnfinishedTask(t *testing.T) {
+	pool := testPool(t)
+	ctx := context.Background()
+	job, _ := seedJob(t, pool, 3)
+	clk := fixedClock{now: time.Now().UTC()}
+	uc := usecase.NewCancelJob(NewJobRepo(pool), NewTaskRepo(pool), NewTxManager(pool), clk)
+
+	cancelled, err := uc.Execute(ctx, job.ID)
+	if err != nil || cancelled != 3 {
+		t.Fatalf("cancel = (%d, %v), want (3, nil)", cancelled, err)
+	}
+	stored, err := NewJobRepo(pool).Get(ctx, job.ID)
+	if err != nil || stored.Status != domain.JobCancelled {
+		t.Fatalf("job after cancel = (%+v, %v)", stored, err)
+	}
+	counts, err := NewTaskRepo(pool).CountByStatus(ctx, job.ID)
+	if err != nil || counts[domain.TaskCancelled] != 3 {
+		t.Fatalf("cancelled tasks = %d, err = %v", counts[domain.TaskCancelled], err)
+	}
+}
+
 func TestUpdateRejectsStaleVersion(t *testing.T) {
 	pool := testPool(t)
 	ctx := context.Background()

coordinator/internal/storage/postgres/task_repo.go

@@ -295,6 +295,29 @@ func (r *TaskRepo) CountByStatus(ctx context.Context, jobID uuid.UUID) (map[doma
 	return counts, rows.Err()
 }
 
+// cancelByJobSQL mirrors domain.Task.Cancel in one set-based update. It runs in
+// the same transaction as the job-status update, so no claimable shard remains
+// after an operator receives a successful cancellation response.
+const cancelByJobSQL = `
+UPDATE tasks
+SET status           = 'cancelled'::task_status,
+    lease_owner      = NULL,
+    lease_expires_at = NULL,
+    error_code       = NULL,
+    error_message    = NULL,
+    completed_at     = $2,
+    version          = version + 1
+WHERE job_id = $1
+  AND status IN ('pending','leased','running')`
+
+func (r *TaskRepo) CancelByJob(ctx context.Context, jobID uuid.UUID, now time.Time) (int64, error) {
+	tag, err := conn(ctx, r.pool).Exec(ctx, cancelByJobSQL, jobID, now)
+	if err != nil {
+		return 0, err
+	}
+	return tag.RowsAffected(), nil
+}
+
 // expireLeasesSQL applies the lease-expiry rule set-based, mirroring
 // domain.Task.ExpireLease: requeue while attempts remain, otherwise fail.
 //

coordinator/internal/storage/postgres/ui_read_repo.go

@@ -0,0 +1,123 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	sq "github.com/Masterminds/squirrel"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+	"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
+)
+
+// UIReadRepo contains bounded, deterministic read queries for the operator UI.
+type UIReadRepo struct{ pool *pgxpool.Pool }
+
+func NewUIReadRepo(pool *pgxpool.Pool) *UIReadRepo { return &UIReadRepo{pool: pool} }
+
+var _ usecase.UIReadRepository = (*UIReadRepo)(nil)
+
+func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, error) {
+	job, err := NewJobRepo(r.pool).Get(ctx, id)
+	return job, err
+}
+
+func (r *UIReadRepo) ListJobs(ctx context.Context, limit int) ([]domain.Job, error) {
+	if limit < 1 || limit > 100 {
+		return nil, domain.ErrInvalidInput
+	}
+	sql, args, err := psql.Select(jobColumns...).From("jobs").OrderBy("created_at DESC", "id DESC").Limit(uint64(limit)).ToSql()
+	if err != nil {
+		return nil, err
+	}
+	rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
+	if err != nil {
+		return nil, fmt.Errorf("list jobs: %w", err)
+	}
+	defer rows.Close()
+	jobs := make([]domain.Job, 0)
+	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 {
+			return nil, err
+		}
+		if inputURI != nil {
+			j.InputURI = *inputURI
+		}
+		j.Status = domain.JobStatus(status)
+		jobs = append(jobs, j)
+	}
+	return jobs, rows.Err()
+}
+
+func (r *UIReadRepo) ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Task, error) {
+	sql, args, err := psql.Select(taskColumns...).From("tasks").Where(sq.Eq{"job_id": jobID}).OrderBy("chunk_index ASC").ToSql()
+	if err != nil {
+		return nil, err
+	}
+	rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
+	if err != nil {
+		return nil, fmt.Errorf("list tasks: %w", err)
+	}
+	defer rows.Close()
+	tasks := make([]domain.Task, 0)
+	for rows.Next() {
+		task, err := scanTask(rows)
+		if err != nil {
+			return nil, err
+		}
+		tasks = append(tasks, *task)
+	}
+	return tasks, rows.Err()
+}
+
+func (r *UIReadRepo) ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error) {
+	if limit < 1 || limit > 100 {
+		return nil, domain.ErrInvalidInput
+	}
+	sql, args, err := psql.Select(workerColumns...).From("workers").OrderBy("last_heartbeat_at DESC", "id DESC").Limit(uint64(limit)).ToSql()
+	if err != nil {
+		return nil, err
+	}
+	rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
+	if err != nil {
+		return nil, fmt.Errorf("list workers: %w", err)
+	}
+	defer rows.Close()
+	workers := make([]domain.Worker, 0)
+	for rows.Next() {
+		worker, err := scanWorker(rows)
+		if err != nil {
+			return nil, err
+		}
+		workers = append(workers, *worker)
+	}
+	return workers, rows.Err()
+}
+
+func (r *UIReadRepo) ListArtifactsByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Artifact, error) {
+	sql, args, err := psql.Select(artifactColumns...).From("artifacts").Where(sq.Eq{"job_id": jobID}).OrderBy("created_at ASC", "id ASC").ToSql()
+	if err != nil {
+		return nil, err
+	}
+	rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
+	if err != nil {
+		return nil, fmt.Errorf("list artifacts: %w", err)
+	}
+	defer rows.Close()
+	artifacts := make([]domain.Artifact, 0)
+	for rows.Next() {
+		var a domain.Artifact
+		var kind string
+		if err := rows.Scan(&a.ID, &a.JobID, &a.TaskID, &a.Attempt, &kind, &a.Filename, &a.StorageKey, &a.ContentType, &a.SizeBytes, &a.SHA256, &a.CreatedAt); err != nil {
+			return nil, err
+		}
+		a.Kind = domain.ArtifactKind(kind)
+		artifacts = append(artifacts, a)
+	}
+	return artifacts, rows.Err()
+}

coordinator/internal/transport/http/dto.go

@@ -118,6 +118,7 @@ type jobProgressResponse struct {
 	Leased    int       `json:"leased"`
 	Done      int       `json:"completed"`
 	Failed    int       `json:"failed"`
+	Cancelled int       `json:"cancelled"`
 }
 
 type uploadArtifactResponse struct {
@@ -160,5 +161,6 @@ func toJobProgressResponse(p domain.JobProgress) jobProgressResponse {
 		Leased:    p.Leased,
 		Done:      p.Done,
 		Failed:    p.Failed,
+		Cancelled: p.Cancelled,
 	}
 }

coordinator/internal/transport/http/errors.go

@@ -50,7 +50,8 @@ func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) {
 	case errors.Is(err, domain.ErrLeaseConflict),
 		errors.Is(err, domain.ErrStaleAttempt),
 		errors.Is(err, domain.ErrResultConflict),
-		errors.Is(err, domain.ErrTaskNotLeased):
+		errors.Is(err, domain.ErrTaskNotLeased),
+		errors.Is(err, domain.ErrJobNotCancellable):
 		status = http.StatusConflict
 	case errors.Is(err, usecase.ErrNotImplemented):
 		status = http.StatusNotImplemented

coordinator/internal/transport/http/handlers.go

@@ -182,7 +182,7 @@ func (s *Server) handleFailure(w http.ResponseWriter, r *http.Request) {
 const defaultChunkRows = 1000
 
 // handleUploadDataset accepts a multipart submission — the dataset file plus the
-// workload/parameters/chunk_rows fields — and hands the file, streamed, to the
+// workload/parameters/chunk_rows/max_rows fields — and hands the file, streamed, to the
 // chunker. The text fields MUST precede the file part: the file is streamed, not
 // buffered, so by the time it arrives the other fields are already parsed.
 func (s *Server) handleUploadDataset(w http.ResponseWriter, r *http.Request) {
@@ -197,11 +197,13 @@ func (s *Server) handleUploadDataset(w http.ResponseWriter, r *http.Request) {
 		workload    string
 		params      map[string]any
 		rows        = defaultChunkRows
+		maxRows     int
 		result      usecase.SubmitDatasetResult
 		gotDataset  bool
 		gotWorkload bool
 		gotParams   bool
 		gotRows     bool
+		gotMaxRows  bool
 	)
 
 	for {
@@ -249,6 +251,19 @@ func (s *Server) handleUploadDataset(w http.ResponseWriter, r *http.Request) {
 			}
 			rows = n
 			gotRows = true
+		case "max_rows":
+			if gotDataset || gotMaxRows {
+				s.writeError(w, r, domain.ErrInvalidInput)
+				return
+			}
+			b, _ := io.ReadAll(io.LimitReader(part, 32))
+			n, err := strconv.Atoi(strings.TrimSpace(string(b)))
+			if err != nil || n < 1 {
+				s.writeError(w, r, domain.ErrInvalidInput)
+				return
+			}
+			maxRows = n
+			gotMaxRows = true
 		case "file", "dataset":
 			if gotDataset || workload == "" {
 				s.writeError(w, r, domain.ErrInvalidInput)
@@ -262,6 +277,7 @@ func (s *Server) handleUploadDataset(w http.ResponseWriter, r *http.Request) {
 				Workload:     workload,
 				Parameters:   params,
 				RowsPerShard: rows,
+				MaxRows:      maxRows,
 				Filename:     filename,
 				ContentType:  part.Header.Get("Content-Type"),
 				Body:         part,
@@ -379,6 +395,27 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, toJobProgressResponse(progress))
 }
 
+// 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) {
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+	jobID, ok := s.pathUUID(w, r, "job_id")
+	if !ok {
+		return
+	}
+	cancelled, err := s.uc.CancelJob.Execute(ctx, jobID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, map[string]any{
+		"job_id":          jobID,
+		"status":          domain.JobCancelled,
+		"cancelled_tasks": cancelled,
+	})
+}
+
 // --- helpers ---
 
 func (s *Server) reqCtx(r *http.Request) (context.Context, context.CancelFunc) {

coordinator/internal/transport/http/middleware.go

@@ -64,6 +64,48 @@ func withAuth(token string) func(http.Handler) http.Handler {
 	}
 }
 
+// withBasicAuth protects the local operator UI with a credential distinct from
+// the worker bearer token. The username is intentionally ignored; the password
+// is the configured UI token. Basic Auth is suitable only for localhost or a
+// TLS-terminating trusted reverse proxy.
+func withBasicAuth(token string) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			_, password, ok := r.BasicAuth()
+			if !ok || subtle.ConstantTimeCompare([]byte(password), []byte(token)) != 1 {
+				w.Header().Set("WWW-Authenticate", `Basic realm="SciMesh UI", charset="UTF-8"`)
+				writeJSON(w, http.StatusUnauthorized, errorResponse{Error: "unauthorized", RequestID: requestIDFrom(r.Context())})
+				return
+			}
+			next.ServeHTTP(w, r)
+		})
+	}
+}
+
+// withSameOrigin rejects browser form/fetch writes initiated by another origin.
+// A missing Origin is allowed for direct local tools; authenticated UI pages use
+// the browser-supplied Origin header on state-changing requests.
+func withSameOrigin(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodGet || r.Method == http.MethodHead || r.Method == http.MethodOptions {
+			next.ServeHTTP(w, r)
+			return
+		}
+		origin := r.Header.Get("Origin")
+		if origin != "" {
+			scheme := "http"
+			if r.TLS != nil {
+				scheme = "https"
+			}
+			if origin != scheme+"://"+r.Host {
+				writeJSON(w, http.StatusForbidden, errorResponse{Error: "cross-origin request rejected", RequestID: requestIDFrom(r.Context())})
+				return
+			}
+		}
+		next.ServeHTTP(w, r)
+	})
+}
+
 // statusRecorder captures the status code for the access log.
 type statusRecorder struct {
 	http.ResponseWriter

coordinator/internal/transport/http/server.go

@@ -24,9 +24,11 @@ type UseCases struct {
 	CompleteTask     *usecase.CompleteTask
 	FailTask         *usecase.FailTask
 	GetJobStatus     *usecase.GetJobStatus
+	CancelJob        *usecase.CancelJob
 	UploadArtifact   *usecase.UploadArtifact
 	DownloadArtifact *usecase.DownloadArtifact
 	GetTaskInput     *usecase.GetTaskInput
+	Dashboard        *usecase.Dashboard
 }
 
 type Server struct {
@@ -54,12 +56,13 @@ func NewServer(uc UseCases, log *slog.Logger, requestTimeout, heartbeatInterval
 
 // Handler builds the router. Go 1.22's ServeMux matches on method and path
 // wildcards, so no third-party router is needed.
-func (s *Server) Handler(token string) http.Handler {
+func (s *Server) Handler(token string, uiToken ...string) http.Handler {
 	protected := http.NewServeMux()
 	protected.HandleFunc("POST /workers/register", s.handleRegister)
 	protected.HandleFunc("POST /jobs", s.handleCreateJob)
 	protected.HandleFunc("POST /jobs/upload", s.handleUploadDataset)
 	protected.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
+	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)
 	protected.HandleFunc("POST /tasks/{task_id}/heartbeat", s.handleHeartbeat)
@@ -70,6 +73,23 @@ func (s *Server) Handler(token string) http.Handler {
 
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /health", s.handleHealth)
+	if len(uiToken) > 0 && uiToken[0] != "" && s.uc.Dashboard != nil {
+		ui := http.NewServeMux()
+		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/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)
+		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 {
+		// More specific than the protected catch-all: UI absence is not an auth
+		// failure and does not disclose that a UI feature is configured elsewhere.
+		mux.HandleFunc("/ui", http.NotFound)
+		mux.HandleFunc("/ui/", http.NotFound)
+	}
 	mux.Handle("/", chain(protected,
 		withRequestID,        // outermost: every response gets an ID,
 		withAccessLog(s.log), // including the 401s below

coordinator/internal/transport/http/server_test.go

@@ -20,6 +20,7 @@ import (
 )
 
 const token = "secret"
+const uiToken = "ui-secret"
 
 type env struct {
 	ts    *httptest.Server
@@ -27,6 +28,10 @@ type env struct {
 }
 
 func newEnv(t *testing.T, ready func(context.Context) error) *env {
+	return newEnvWithUIToken(t, ready, uiToken)
+}
+
+func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configuredUIToken string) *env {
 	t.Helper()
 	tasks := memstore.NewTaskRepo()
 	jobs := memstore.NewJobRepo()
@@ -46,12 +51,14 @@ func newEnv(t *testing.T, ready func(context.Context) error) *env {
 		CompleteTask:     usecase.NewCompleteTask(tasks, jobs, arts, tx, clk),
 		FailTask:         usecase.NewFailTask(tasks, jobs, tx, clk),
 		GetJobStatus:     usecase.NewGetJobStatus(jobs, tasks),
+		CancelJob:        usecase.NewCancelJob(jobs, tasks, tx, clk),
 		UploadArtifact:   usecase.NewUploadArtifact(tasks, arts, blobs, clk),
 		DownloadArtifact: usecase.NewDownloadArtifact(arts, blobs),
 		GetTaskInput:     usecase.NewGetTaskInput(tasks, arts, blobs),
+		Dashboard:        usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),
 	}
 	srv := coordhttp.NewServer(uc, slog.New(slog.NewTextHandler(io.Discard, nil)), 5*time.Second, 15*time.Second, 1<<30, ready)
-	ts := httptest.NewServer(srv.Handler(token))
+	ts := httptest.NewServer(srv.Handler(token, configuredUIToken))
 	t.Cleanup(ts.Close)
 	return &env{ts: ts, blobs: blobs}
 }
@@ -97,6 +104,181 @@ func TestHealthOK(t *testing.T) {
 	}
 }
 
+func TestUIRequiresDistinctCredentialAndRendersDashboard(t *testing.T) {
+	e := newEnv(t, healthy)
+	request := func() *http.Request {
+		req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui", nil)
+		return req
+	}
+	resp, err := http.DefaultClient.Do(request())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("no UI auth: %d", resp.StatusCode)
+	}
+
+	req := request()
+	req.Header.Set("Authorization", "Bearer "+token)
+	resp, err = http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("worker token authorized UI: %d", resp.StatusCode)
+	}
+
+	req = request()
+	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("UI status: %d", resp.StatusCode)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	if !strings.Contains(string(body), "SciMesh operator dashboard") {
+		t.Errorf("dashboard body missing title")
+	}
+}
+
+func TestUIDisabledReturnsNotFound(t *testing.T) {
+	e := newEnvWithUIToken(t, healthy, "")
+	resp := e.get(t, "/ui")
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("disabled UI = %d, want 404", resp.StatusCode)
+	}
+}
+
+func TestUIRejectsCrossOriginUpload(t *testing.T) {
+	e := newEnv(t, healthy)
+	req, _ := http.NewRequestWithContext(context.Background(), "POST", e.ts.URL+"/ui/api/jobs/upload", strings.NewReader("dataset=x"))
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Set("Origin", "https://attacker.example")
+	req.SetBasicAuth("operator", uiToken)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusForbidden {
+		t.Fatalf("cross-origin upload = %d, want 403", resp.StatusCode)
+	}
+}
+
+func TestUIUploadDatasetCreatesJob(t *testing.T) {
+	e := newEnv(t, healthy)
+	var body bytes.Buffer
+	mw := multipart.NewWriter(&body)
+	_ = mw.WriteField("workload", "similarity-search")
+	_ = mw.WriteField("parameters", `{"query_smiles":"CCO","top_k":20,"progress_every":0}`)
+	_ = mw.WriteField("chunk_rows", "1000")
+	file, err := mw.CreateFormFile("file", "chembl.tsv")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _ = io.WriteString(file, "chembl_id\tcanonical_smiles\nCHEMBL1\tCCO\n")
+	if err := mw.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	req, _ := http.NewRequestWithContext(context.Background(), "POST", e.ts.URL+"/ui/api/jobs/upload", &body)
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+	req.SetBasicAuth("operator", uiToken)
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusCreated {
+		result, _ := io.ReadAll(resp.Body)
+		t.Fatalf("UI upload = %d: %s", resp.StatusCode, result)
+	}
+}
+
+func TestCancelJobStopsUnfinishedTasks(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://c0","input_sha256":"sha"},{"chunk_index":1,"input_uri":"s3://c1","input_sha256":"sha"}]}`)
+	if code != http.StatusCreated {
+		t.Fatalf("create: %d", code)
+	}
+	jobID := job["id"].(string)
+	if code, body := e.do(t, "POST", "/jobs/"+jobID+"/cancel", ""); code != http.StatusOK || body["cancelled_tasks"].(float64) != 2 {
+		t.Fatalf("cancel = (%d, %v)", code, body)
+	}
+	if code, progress := e.do(t, "GET", "/jobs/"+jobID, ""); code != http.StatusOK || progress["status"] != "cancelled" || progress["cancelled"].(float64) != 2 {
+		t.Fatalf("cancelled job progress = (%d, %v)", code, progress)
+	}
+}
+
+func TestUICancelJobUsesOperatorCredential(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://c0","input_sha256":"sha"}]}`)
+	if code != http.StatusCreated {
+		t.Fatalf("create: %d", code)
+	}
+	req, _ := http.NewRequestWithContext(context.Background(), "POST", e.ts.URL+"/ui/api/jobs/"+job["id"].(string)+"/cancel", 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("UI cancel = %d", resp.StatusCode)
+	}
+}
+
+func TestUIJobAndArtifactAreScopedToTheirJob(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)
+	}
+	req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+job["id"].(string), 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("detail: %d", resp.StatusCode)
+	}
+	if got := resp.Header.Get("Content-Security-Policy"); got == "" {
+		t.Error("missing UI CSP")
+	}
+}
+
+func TestUIArtifactDownloadRejectsAnotherJobsArtifact(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)), "result")
+	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, 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 artifact = %d, want 404", resp.StatusCode)
+	}
+}
+
 func TestHealthUnavailableWhenDBDown(t *testing.T) {
 	e := newEnv(t, func(context.Context) error { return context.DeadlineExceeded })
 	resp := e.get(t, "/health")
@@ -244,6 +426,31 @@ func TestUploadDatasetChunksAndServesInput(t *testing.T) {
 	}
 }
 
+func TestUploadDatasetLimitsRows(t *testing.T) {
+	e := newEnv(t, healthy)
+	var buf bytes.Buffer
+	mw := multipart.NewWriter(&buf)
+	_ = mw.WriteField("workload", "w")
+	_ = mw.WriteField("chunk_rows", "2")
+	_ = mw.WriteField("max_rows", "3")
+	fw, _ := mw.CreateFormFile("file", "chembl.tsv")
+	_, _ = io.Copy(fw, strings.NewReader("id\tsmiles\nA\tCC\nB\tCCC\nC\tCCCC\nD\tCCCCC\n"))
+	_ = mw.Close()
+	req, _ := http.NewRequestWithContext(context.Background(), "POST", e.ts.URL+"/jobs/upload", &buf)
+	req.Header.Set("Authorization", "Bearer "+token)
+	req.Header.Set("Content-Type", mw.FormDataContentType())
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	var result map[string]any
+	_ = json.NewDecoder(resp.Body).Decode(&result)
+	if resp.StatusCode != http.StatusCreated || result["task_count"].(float64) != 2 {
+		t.Fatalf("limited upload = (%d, %v)", resp.StatusCode, result)
+	}
+}
+
 func TestErrorMappings(t *testing.T) {
 	e := newEnv(t, healthy)
 	zero := "00000000-0000-0000-0000-000000000000"

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

@@ -0,0 +1,23 @@
+{{define "dashboard.html"}}
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>SciMesh operator dashboard</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}}
+  </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>
+  </main>
+</body>
+</html>
+{{end}}

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

@@ -0,0 +1,29 @@
+{{define "job.html"}}
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>SciMesh job</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}.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}h2{margin:32px 0 12px;font-size:1.3rem}.summary,.card{padding:20px;border:1px solid #dfe5f0;border-radius:12px;background:#fff}.summary-head{display:flex;justify-content:space-between;gap:16px;align-items:start}.status{display:inline-block;border-radius:999px;padding:4px 10px;font-size:.9rem;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}.hint{margin:8px 0 0;color:#56657c}.bar{height:10px;margin:20px 0 8px;overflow:hidden;border-radius:999px;background:#e6eaf1}.bar>span{display:block;height:100%;background:#1f5eff;transition:width .3s}.numbers{display:grid;grid-template-columns:repeat(6,1fr);gap:10px}.number{padding:12px;border-radius:8px;background:#f6f8fc}.number b{display:block;font-size:1.35rem}.stop{display:block;margin-left:auto;border:1px solid #d43b51;border-radius:7px;padding:8px 11px;background:#fff;color:#b2223a;font:inherit;font-weight:700;cursor:pointer}.stop:disabled{opacity:.6}.notice{margin:20px 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}table{width:100%;border-collapse:collapse}td,th{padding:12px 13px;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}small,.muted{color:#68758b}.error{color:#a31135;max-width:360px;word-break:break-word}.download{display:inline-block;padding:7px 10px;border:1px solid #b9c9ee;border-radius:7px;text-decoration:none}.empty{padding:24px;text-align:center;color:#68758b}details{margin-top:18px;color:#56657c}code{word-break:break-all}@media(max-width:700px){.summary-head{display:block}.numbers{grid-template-columns:repeat(2,1fr)}}
+  </style>
+</head>
+<body>
+  <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"><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>
+  </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.'],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()});
+    setInterval(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}catch(_){}} ,2000);
+  </script>
+</body>
+</html>
+{{end}}

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

@@ -0,0 +1,31 @@
+{{define "new-job.html"}}
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>Create a check — 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}
+  </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>
+  </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}});
+  </script>
+</body>
+</html>
+{{end}}

coordinator/internal/transport/http/ui.go

@@ -0,0 +1,281 @@
+package http
+
+import (
+	"embed"
+	"fmt"
+	"html/template"
+	"io"
+	"mime"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+//go:embed templates/*.html
+var uiFiles embed.FS
+
+var uiTemplates = template.Must(template.New("ui").Funcs(template.FuncMap{
+	"time":              formatUITime,
+	"statusLabel":       uiStatusLabel,
+	"statusHint":        uiStatusHint,
+	"statusClass":       uiStatusClass,
+	"taskErrorLabel":    uiTaskErrorLabel,
+	"taskErrorHint":     uiTaskErrorHint,
+	"workerStatusLabel": uiWorkerStatusLabel,
+	"workloadLabel":     uiWorkloadLabel,
+	"progressPercent":   uiProgressPercent,
+	"cancellable":       uiCancellable,
+	"bytes":             uiBytes,
+	"add":               func(a, b int) int { return a + b },
+}).ParseFS(uiFiles, "templates/*.html"))
+
+func formatUITime(t time.Time) string {
+	if t.IsZero() {
+		return "—"
+	}
+	return t.UTC().Format("02.01.2006 15:04 UTC")
+}
+
+func uiStatusLabel(status string) string {
+	switch status {
+	case "pending":
+		return "Waiting for a worker"
+	case "leased":
+		return "Assigned to a worker"
+	case "running":
+		return "Running"
+	case "completed":
+		return "Tasks complete"
+	case "failed":
+		return "Needs attention"
+	case "cancelled":
+		return "Stopped"
+	default:
+		return status
+	}
+}
+
+func uiStatusHint(status string) string {
+	switch status {
+	case "pending":
+		return "Waiting for an available worker with the required capability."
+	case "leased":
+		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 "completed":
+		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":
+		return "The operator stopped this job. No new shards can be claimed."
+	default:
+		return "Status reported by the coordinator."
+	}
+}
+
+func uiStatusClass(status string) string {
+	switch status {
+	case "completed":
+		return "success"
+	case "failed":
+		return "danger"
+	case "cancelled":
+		return "waiting"
+	case "running", "leased":
+		return "active"
+	default:
+		return "waiting"
+	}
+}
+
+func uiWorkerStatusLabel(status string) string {
+	switch status {
+	case "online":
+		return "Available"
+	case "offline":
+		return "Offline"
+	default:
+		return status
+	}
+}
+
+// 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.
+func uiTaskErrorLabel(errorCode string) string {
+	switch errorCode {
+	case "CalledProcessError":
+		return "Local calculation failed"
+	case "ValueError":
+		return "Task input could not be processed"
+	case "CoordinatorTransientError":
+		return "Coordinator connection was interrupted"
+	case "CoordinatorConflictError":
+		return "Worker lease was no longer valid"
+	case "FileNotFoundError":
+		return "Local task file is missing"
+	default:
+		return errorCode
+	}
+}
+
+func uiTaskErrorHint(errorCode string) string {
+	switch errorCode {
+	case "CalledProcessError":
+		return "The local SciMesh command stopped before it could upload a result. Check the worker terminal for the original error."
+	case "ValueError":
+		return "The coordinator task or its downloaded input did not meet the worker validation rules."
+	case "CoordinatorTransientError":
+		return "The worker will retry after the coordinator connection is available again."
+	case "CoordinatorConflictError":
+		return "Another worker or a lease timeout changed this task before completion."
+	case "FileNotFoundError":
+		return "The worker could not find one of its local task files. Restart it with an absolute --work-dir."
+	default:
+		return "Check the worker terminal for the original error details."
+	}
+}
+
+func uiWorkloadLabel(workload string) string {
+	switch workload {
+	case "similarity-search", "similarity_search":
+		return "Molecule similarity search"
+	case "similarity-graph", "similarity_graph":
+		return "Molecular similarity graph"
+	default:
+		return workload
+	}
+}
+
+func uiCancellable(status string) bool {
+	return status == "pending" || status == "running"
+}
+
+func uiProgressPercent(completed, failed, cancelled, total int) int {
+	if total <= 0 {
+		return 0
+	}
+	percent := (completed + failed + cancelled) * 100 / total
+	if percent > 100 {
+		return 100
+	}
+	return percent
+}
+
+func uiBytes(n int64) string {
+	const kib = 1024
+	if n < kib {
+		return fmt.Sprintf("%d B", n)
+	}
+	if n < kib*kib {
+		return fmt.Sprintf("%.1f KiB", float64(n)/kib)
+	}
+	if n < kib*kib*kib {
+		return fmt.Sprintf("%.1f MiB", float64(n)/(kib*kib))
+	}
+	return fmt.Sprintf("%.1f GiB", float64(n)/(kib*kib*kib))
+}
+
+func (s *Server) renderUI(w http.ResponseWriter, name string, data any) {
+	w.Header().Set("Content-Type", "text/html; charset=utf-8")
+	w.Header().Set("Cache-Control", "no-store")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	w.Header().Set("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'; base-uri 'none'; frame-ancestors 'none'")
+	if err := uiTemplates.ExecuteTemplate(w, name, data); err != nil {
+		s.log.Error("render UI", "err", err)
+		http.Error(w, "internal error", http.StatusInternalServerError)
+	}
+}
+
+func (s *Server) handleUIHome(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
+	}
+	s.renderUI(w, "dashboard.html", view)
+}
+
+func (s *Server) handleUINewJob(w http.ResponseWriter, r *http.Request) {
+	s.renderUI(w, "new-job.html", nil)
+}
+
+func (s *Server) uiJobID(w http.ResponseWriter, r *http.Request) (uuid.UUID, bool) {
+	return s.pathUUID(w, r, "job_id")
+}
+
+func (s *Server) handleUIJob(w http.ResponseWriter, r *http.Request) {
+	jobID, ok := s.uiJobID(w, r)
+	if !ok {
+		return
+	}
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+	view, err := s.uc.Dashboard.JobDetail(ctx, jobID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	s.renderUI(w, "job.html", view)
+}
+
+func (s *Server) handleUIJobJSON(w http.ResponseWriter, r *http.Request) {
+	jobID, ok := s.uiJobID(w, r)
+	if !ok {
+		return
+	}
+	ctx, cancel := s.reqCtx(r)
+	defer cancel()
+	view, err := s.uc.Dashboard.JobDetail(ctx, jobID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, view)
+}
+
+func (s *Server) handleUIArtifactDownload(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()
+	belongs, err := s.uc.Dashboard.ArtifactBelongsToJob(ctx, jobID, artifactID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	if !belongs {
+		s.writeError(w, r, domain.ErrArtifactNotFound)
+		return
+	}
+	// Reuse the coordinator-owned blob stream after the job-scoped check above.
+	art, body, err := s.uc.DownloadArtifact.Execute(ctx, artifactID)
+	if err != nil {
+		s.writeError(w, r, err)
+		return
+	}
+	defer func() {
+		if err := body.Close(); err != nil {
+			s.log.Warn("close downloaded UI artifact", "artifact_id", artifactID, "err", err)
+		}
+	}()
+	w.Header().Set("Content-Type", art.ContentType)
+	w.Header().Set("Content-Disposition", mime.FormatMediaType("attachment", map[string]string{"filename": art.Filename}))
+	w.Header().Set("Content-Length", strconv.FormatInt(art.SizeBytes, 10))
+	w.Header().Set("X-Checksum-SHA256", art.SHA256)
+	_, _ = io.Copy(w, body)
+}

coordinator/internal/transport/http/ui_test.go

@@ -0,0 +1,44 @@
+package http
+
+import "testing"
+
+func TestUIStatusPresentation(t *testing.T) {
+	tests := []struct {
+		status string
+		label  string
+		class  string
+	}{
+		{"pending", "Waiting for a worker", "waiting"},
+		{"running", "Running", "active"},
+		{"completed", "Tasks complete", "success"},
+		{"failed", "Needs attention", "danger"},
+	}
+	for _, test := range tests {
+		t.Run(test.status, func(t *testing.T) {
+			if got := uiStatusLabel(test.status); got != test.label {
+				t.Errorf("label = %q, want %q", got, test.label)
+			}
+			if got := uiStatusClass(test.status); got != test.class {
+				t.Errorf("class = %q, want %q", got, test.class)
+			}
+		})
+	}
+}
+
+func TestUIProgressPercent(t *testing.T) {
+	if got := uiProgressPercent(3, 1, 0, 8); got != 50 {
+		t.Errorf("progress = %d, want 50", got)
+	}
+	if got := uiProgressPercent(1, 1, 0, 0); got != 0 {
+		t.Errorf("empty progress = %d, want 0", got)
+	}
+}
+
+func TestUITaskErrorPresentationDoesNotExposeCommand(t *testing.T) {
+	if got := uiTaskErrorLabel("CalledProcessError"); got != "Local calculation failed" {
+		t.Errorf("error label = %q", got)
+	}
+	if got := uiTaskErrorHint("CalledProcessError"); got == "" {
+		t.Error("error hint must explain the failure")
+	}
+}

coordinator/internal/usecase/dto.go

@@ -53,6 +53,9 @@ type SubmitDatasetInput struct {
 	Workload     string
 	Parameters   map[string]any
 	RowsPerShard int
+	// MaxRows limits how many data rows are turned into shards. Zero means the
+	// whole uploaded dataset; the input artifact itself remains stored intact.
+	MaxRows     int
 	Filename    string
 	ContentType string
 	Body        io.Reader

coordinator/internal/usecase/job.go

@@ -62,6 +62,45 @@ type GetJobStatus struct {
 	tasks TaskRepository
 }
 
+// --- CancelJob -----------------------------------------------------------
+
+type CancelJob struct {
+	jobs  JobRepository
+	tasks TaskRepository
+	tx    TxManager
+	clock Clock
+}
+
+func NewCancelJob(jobs JobRepository, tasks TaskRepository, tx TxManager, clock Clock) *CancelJob {
+	return &CancelJob{jobs: jobs, tasks: tasks, tx: tx, clock: clock}
+}
+
+// Execute stops a job atomically. Completed and finally failed tasks are kept
+// as historical evidence; all other tasks are cancelled, including leased and
+// running ones. A repeated cancel of an already cancelled job is idempotent.
+func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error) {
+	now := uc.clock.Now()
+	var cancelled int64
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		job, err := uc.jobs.Get(ctx, jobID)
+		if err != nil {
+			return err
+		}
+		if job.Status == domain.JobCancelled {
+			return nil
+		}
+		if job.Status == domain.JobCompleted || job.Status == domain.JobFailed {
+			return domain.ErrJobNotCancellable
+		}
+		cancelled, err = uc.tasks.CancelByJob(ctx, jobID, now)
+		if err != nil {
+			return err
+		}
+		return uc.jobs.UpdateStatus(ctx, jobID, domain.JobCancelled, &now)
+	})
+	return cancelled, err
+}
+
 func NewGetJobStatus(jobs JobRepository, tasks TaskRepository) *GetJobStatus {
 	return &GetJobStatus{jobs: jobs, tasks: tasks}
 }
@@ -147,6 +186,7 @@ func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobPr
 		Leased:    counts[domain.TaskLeased] + counts[domain.TaskRunning],
 		Done:      counts[domain.TaskCompleted],
 		Failed:    counts[domain.TaskFailed],
+		Cancelled: counts[domain.TaskCancelled],
 	}
 	for _, n := range counts {
 		p.Total += n

coordinator/internal/usecase/ports.go

@@ -56,6 +56,10 @@ type TaskRepository interface {
 	// CountByStatus aggregates a job's tasks for progress reporting.
 	CountByStatus(ctx context.Context, jobID uuid.UUID) (map[domain.TaskStatus]int, error)
 
+	// CancelByJob marks every non-terminal task as cancelled and invalidates its
+	// lease. It returns how many tasks changed.
+	CancelByJob(ctx context.Context, jobID uuid.UUID, now time.Time) (int64, error)
+
 	// ExpireLeases applies the lease-expiry rule to every elapsed task and
 	// reports how many were affected.
 	ExpireLeases(ctx context.Context, now time.Time) (int64, error)

coordinator/internal/usecase/ui.go

@@ -0,0 +1,177 @@
+package usecase
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// UIReadRepository is a read-only projection source for the local operator UI.
+// It intentionally exposes no storage paths or credentials.
+type UIReadRepository interface {
+	GetJob(ctx context.Context, jobID uuid.UUID) (*domain.Job, error)
+	ListJobs(ctx context.Context, limit int) ([]domain.Job, error)
+	ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Task, error)
+	ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error)
+	ListArtifactsByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Artifact, error)
+}
+
+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"`
+}
+
+type TaskCard struct {
+	ID             string     `json:"id"`
+	ChunkIndex     int        `json:"chunk_index"`
+	Status         string     `json:"status"`
+	Attempt        int        `json:"attempt"`
+	MaxAttempts    int        `json:"max_attempts"`
+	LeaseOwner     string     `json:"lease_owner,omitempty"`
+	LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
+	ErrorCode      string     `json:"error_code,omitempty"`
+	ErrorMessage   string     `json:"error_message,omitempty"`
+}
+
+type ArtifactCard struct {
+	ID           string `json:"id"`
+	Kind         string `json:"kind"`
+	Filename     string `json:"filename"`
+	SizeBytes    int64  `json:"size_bytes"`
+	SHA256       string `json:"sha256"`
+	Downloadable bool   `json:"downloadable"`
+	Diagnostic   bool   `json:"diagnostic"`
+}
+
+type WorkerCard struct {
+	ID              string    `json:"id"`
+	Name            string    `json:"name"`
+	Status          string    `json:"status"`
+	Capabilities    []string  `json:"capabilities"`
+	LastHeartbeatAt time.Time `json:"last_heartbeat_at"`
+}
+
+type DashboardView struct {
+	Jobs    []JobCard
+	Workers []WorkerCard
+}
+type JobDetailView struct {
+	JobCard
+	Tasks                []TaskCard     `json:"tasks"`
+	Artifacts            []ArtifactCard `json:"artifacts"`
+	FinalResultAvailable bool           `json:"final_result_available"`
+}
+
+type Dashboard struct{ read UIReadRepository }
+
+func NewDashboard(read UIReadRepository) *Dashboard { return &Dashboard{read: read} }
+
+func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, error) {
+	jobs, err := d.read.ListJobs(ctx, limit)
+	if err != nil {
+		return DashboardView{}, err
+	}
+	workers, err := d.read.ListWorkers(ctx, limit)
+	if err != nil {
+		return DashboardView{}, err
+	}
+	out := DashboardView{Jobs: make([]JobCard, 0, len(jobs)), Workers: make([]WorkerCard, 0, len(workers))}
+	for _, job := range jobs {
+		tasks, err := d.read.ListTasksByJob(ctx, job.ID)
+		if err != nil {
+			return DashboardView{}, err
+		}
+		out.Jobs = append(out.Jobs, jobCard(job, tasks))
+	}
+	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})
+	}
+	return out, nil
+}
+
+func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailView, error) {
+	job, err := d.read.GetJob(ctx, jobID)
+	if err != nil {
+		return JobDetailView{}, err
+	}
+	tasks, err := d.read.ListTasksByJob(ctx, jobID)
+	if err != nil {
+		return JobDetailView{}, err
+	}
+	artifacts, err := d.read.ListArtifactsByJob(ctx, jobID)
+	if err != nil {
+		return JobDetailView{}, err
+	}
+	out := JobDetailView{JobCard: jobCard(*job, tasks), Tasks: make([]TaskCard, 0, len(tasks)), Artifacts: make([]ArtifactCard, 0, len(artifacts))}
+	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}
+		if task.LeaseOwner != nil {
+			card.LeaseOwner = *task.LeaseOwner
+		}
+		if task.ErrorCode != nil {
+			card.ErrorCode = *task.ErrorCode
+		}
+		if task.ErrorMessage != nil {
+			card.ErrorMessage = *task.ErrorMessage
+		}
+		out.Tasks = append(out.Tasks, card)
+	}
+	for _, artifact := range artifacts {
+		diagnostic := artifact.Kind == domain.ArtifactPartialResult
+		downloadable := diagnostic || (artifact.Kind == domain.ArtifactFinalResult && out.Status == string(domain.JobCompleted))
+		out.Artifacts = append(out.Artifacts, ArtifactCard{ID: artifact.ID.String(), Kind: string(artifact.Kind), Filename: artifact.Filename, SizeBytes: artifact.SizeBytes, SHA256: artifact.SHA256, Downloadable: downloadable, Diagnostic: diagnostic})
+		if artifact.Kind == domain.ArtifactFinalResult && downloadable {
+			out.FinalResultAvailable = true
+		}
+	}
+	return out, nil
+}
+
+func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID uuid.UUID) (bool, error) {
+	artifacts, err := d.read.ListArtifactsByJob(ctx, jobID)
+	if err != nil {
+		return false, err
+	}
+	for _, a := range artifacts {
+		if a.ID == artifactID {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+func jobCard(job domain.Job, tasks []domain.Task) JobCard {
+	c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt}
+	for _, task := range tasks {
+		c.Total++
+		switch task.Status {
+		case domain.TaskPending:
+			c.Pending++
+		case domain.TaskLeased:
+			c.Leased++
+		case domain.TaskRunning:
+			c.Running++
+		case domain.TaskCompleted:
+			c.Completed++
+		case domain.TaskFailed:
+			c.Failed++
+		case domain.TaskCancelled:
+			c.Cancelled++
+		}
+	}
+	p := domain.JobProgress{Job: job, Total: c.Total, Pending: c.Pending, Leased: c.Leased + c.Running, Done: c.Completed, Failed: c.Failed, Cancelled: c.Cancelled}
+	c.Status = string(p.DeriveStatus())
+	return c
+}

coordinator/internal/usecase/upload.go

@@ -64,7 +64,7 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
 		cleanup()
 		return SubmitDatasetResult{}, err
 	}
-	splitErr := chunk.SplitTSV(rc, in.RowsPerShard, func(index int, shard io.Reader) error {
+	splitErr := chunk.SplitTSVLimit(rc, in.RowsPerShard, in.MaxRows, func(index int, shard io.Reader) error {
 		art, err := domain.NewArtifact(job.ID, nil, domain.ArtifactShard,
 			fmt.Sprintf("shard-%d.tsv", index), in.ContentType, now)
 		if err != nil {

coordinator/internal/usecase/usecase_test.go

@@ -54,6 +54,7 @@ type harness struct {
 	downloadArt *usecase.DownloadArtifact
 	getInput    *usecase.GetTaskInput
 	expire      *usecase.ExpireLeases
+	cancel      *usecase.CancelJob
 }
 
 func newHarness() *harness {
@@ -79,6 +80,7 @@ func newHarness() *harness {
 	h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs)
 	h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs)
 	h.expire = usecase.NewExpireLeases(h.tasks, h.clk)
+	h.cancel = usecase.NewCancelJob(h.jobs, h.tasks, tx, h.clk)
 	return h
 }
 
@@ -468,6 +470,44 @@ func TestSubmitDatasetChunksAndServesInput(t *testing.T) {
 	}
 }
 
+func TestSubmitDatasetLimitsRowsBeforeCreatingShards(t *testing.T) {
+	h := newHarness()
+	tsv := "id\tsmiles\nA\tCC\nB\tCCC\nC\tCCCC\nD\tCCCCC\nE\tCCCCCC\n"
+	res, err := h.submit.Execute(ctx, usecase.SubmitDatasetInput{
+		Workload: "w", RowsPerShard: 2, MaxRows: 3, Filename: "chembl.tsv",
+		ContentType: "text/tab-separated-values", Body: strings.NewReader(tsv),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.TaskCount != 2 {
+		t.Fatalf("task_count = %d, want 2", res.TaskCount)
+	}
+}
+
+func TestCancelJobInvalidatesClaimedAndPendingTasks(t *testing.T) {
+	h := newHarness()
+	jobID := h.seedJob(t, "w", 3)
+	claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: "w1", Workloads: []string{"w"}})
+	if err != nil || claimed == nil {
+		t.Fatalf("claim: %v", err)
+	}
+	cancelled, err := h.cancel.Execute(ctx, jobID)
+	if err != nil || cancelled != 3 {
+		t.Fatalf("cancel = (%d, %v), want (3, nil)", cancelled, err)
+	}
+	if _, err := h.renew.Execute(ctx, usecase.RenewLeaseInput{TaskID: claimed.TaskID, WorkerID: "w1", Attempt: claimed.Attempt}); !errors.Is(err, domain.ErrTaskNotLeased) {
+		t.Errorf("cancelled lease heartbeat = %v, want ErrTaskNotLeased", err)
+	}
+	progress, err := h.status.Execute(ctx, jobID)
+	if err != nil || progress.DeriveStatus() != domain.JobCancelled || progress.Cancelled != 3 {
+		t.Errorf("cancelled progress = %+v, err = %v", progress, err)
+	}
+	if cancelled, err := h.cancel.Execute(ctx, jobID); err != nil || cancelled != 0 {
+		t.Errorf("second cancel = (%d, %v), want (0, nil)", cancelled, err)
+	}
+}
+
 func TestGetTaskInputMissingForURITask(t *testing.T) {
 	h := newHarness()
 	h.seedJob(t, "w", 1) // URI-based task, no coordinator-stored input

docs/api-contract.md

@@ -52,9 +52,11 @@ Content-Type: multipart/form-data
 ```
 
 Fields, in order (text fields first, file last — the file is streamed):
-`workload`, `parameters` (JSON), `chunk_rows` (int, default 1000), and the file
+`workload`, `parameters` (JSON), `chunk_rows` (int, default 1000), optional
-part `file`. The coordinator stores the input, splits the TSV into shard
+`max_rows` (positive int), and the file part `file`. `max_rows` limits the
-artifacts (header repeated per shard), and creates one task per shard.
+leading data rows that become shards; it does not change the stored source
+artifact. The coordinator splits the selected TSV rows into shard artifacts
+(header repeated per shard) and creates one task per shard.
 
 `201`:
 
@@ -65,6 +67,24 @@ artifacts (header repeated per shard), and creates one task per shard.
 Each resulting task's claim response carries `input.uri = /tasks/{id}/input`,
 served by §5.4.
 
+## Stop a job
+
+```http
+POST /jobs/{job_id}/cancel
+Authorization: Bearer <token>
+```
+
+The coordinator transactionally marks every pending, leased, or running shard
+as `cancelled`, invalidates its lease, and marks the job `cancelled`. Completed
+and terminally failed shards remain as history. Repeating a cancellation of an
+already cancelled job is safe.
+
+`200`:
+
+```json
+{ "job_id": "uuid", "status": "cancelled", "cancelled_tasks": 12 }
+```
+
 ## Register worker
 
 ```http

docs/openapi.yaml

@@ -95,7 +95,7 @@ paths:
       summary: Upload a dataset; the coordinator chunks it into shard tasks
       description: >
         multipart/form-data. The text fields (`workload`, `parameters`,
-        `chunk_rows`) MUST precede the `file` part: the file is streamed, not
+        `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.
       requestBody:
         required: true
@@ -130,6 +130,23 @@ paths:
         "401": { $ref: "#/components/responses/Unauthorized" }
         "404": { $ref: "#/components/responses/NotFound" }
 
+  /jobs/{job_id}/cancel:
+    post:
+      tags: [jobs]
+      summary: Cancel a job and invalidate all unfinished task leases
+      parameters:
+        - $ref: "#/components/parameters/JobID"
+      responses:
+        "200":
+          description: The job is cancelled. Completed and terminally failed tasks remain unchanged.
+          content:
+            application/json:
+              schema: { $ref: "#/components/schemas/CancelJobResponse" }
+        "400": { $ref: "#/components/responses/BadRequest" }
+        "401": { $ref: "#/components/responses/Unauthorized" }
+        "404": { $ref: "#/components/responses/NotFound" }
+        "409": { $ref: "#/components/responses/Conflict" }
+
   /tasks/claim:
     post:
       tags: [tasks]
@@ -429,6 +446,11 @@ components:
           type: integer
           description: Data rows per shard. Default 1000.
           example: 1000
+        max_rows:
+          type: integer
+          minimum: 1
+          description: Optional leading data-row limit for a small pipeline check.
+          example: 500
         file:
           type: string
           format: binary
@@ -441,6 +463,13 @@ components:
         task_count: { type: integer, example: 3 }
         input_artifact_id: { type: string, format: uuid }
 
+    CancelJobResponse:
+      type: object
+      properties:
+        job_id: { type: string, format: uuid }
+        status: { type: string, enum: [cancelled] }
+        cancelled_tasks: { type: integer }
+
     JobProgress:
       type: object
       properties:
@@ -451,6 +480,7 @@ components:
         leased: { type: integer }
         completed: { type: integer }
         failed: { type: integer }
+        cancelled: { type: integer }
 
     ClaimRequest:
       type: object
@@ -549,4 +579,4 @@ components:
 
     TaskStatus:
       type: string
-      enum: [pending, leased, completed, failed, cancelled]
+      enum: [pending, leased, running, completed, failed, cancelled]

docs/web-interface-plan.md

@@ -173,6 +173,9 @@ Rules:
 - TSV file, required, streamed; show expected columns
   `chembl_id` and `canonical_smiles`.
 - `chunk_rows`: integer 1--100000, default 1000.
+- `max_rows`: optional positive integer. The coordinator creates shards only
+  from the first N data rows, so a user can test a large upload without
+  creating thousands of tasks. It does not truncate the stored source blob.
 - optional human-readable run name is a later schema/API addition; v1 does not
   silently store it.
 - display file name and client-side size only as convenience; server limits and

scimesh/worker/config.py

@@ -68,6 +68,10 @@ class WorkerConfig:
             _positive_number(self.cleanup_after_seconds, "cleanup_after_seconds", allow_zero=True)
         if not self.capabilities:
             raise ValueError("capabilities cannot be empty")
+        # Runner subprocesses use a task directory as their cwd. Keep the
+        # configured root absolute so input/output paths remain valid there
+        # even when the CLI received a convenient relative --work-dir value.
+        object.__setattr__(self, "work_dir", self.work_dir.expanduser().resolve())
 
     @classmethod
     def from_environment(

scimesh/worker/runners.py

@@ -18,6 +18,9 @@ class SciMeshRunner:
     """Allowlisted adapter from coordinator workloads to the local SciMesh CLI."""
 
     def run(self, task: ClaimedTask, task_dir: Path) -> RunResult:
+        # The subprocess changes cwd to task_dir. Absolute paths keep a caller
+        # supplied relative work directory from being resolved twice.
+        task_dir = task_dir.resolve()
         input_path = task_dir / "input"
         output_path = task_dir / "result.csv"
         # The coordinator contract historically used underscores while the

tests/test_worker_daemon.py

@@ -314,6 +314,32 @@ def test_environment_overrides_allow_cli_only_configuration(monkeypatch: pytest.
     assert "similarity_search" in config.capabilities
 
 
+def test_relative_work_dir_is_normalized_for_runner_subprocesses(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.chdir(tmp_path)
+    config = WorkerConfig("https://coordinator.example", None, Path("./worker-data"))
+    assert config.work_dir == tmp_path / "worker-data"
+
+    task_dir = config.work_dir / "task" / "1"
+    task_dir.mkdir(parents=True)
+    (task_dir / "input").write_text("fixture", encoding="utf-8")
+    command: list[str] = []
+
+    def fake_run(args: list[str], **_: object) -> None:
+        command.extend(args)
+        output = Path(args[args.index("--output") + 1])
+        output.write_text("id,score\n", encoding="utf-8")
+
+    monkeypatch.setattr("scimesh.worker.runners.subprocess.run", fake_run)
+    task = ClaimedTask(
+        "task", 1, "2026-07-30T00:00:00Z", "similarity-search",
+        InputArtifact("https://example.test/input", "a" * 64), {"query_smiles": "CCO"},
+    )
+    SciMeshRunner().run(task, task_dir)
+    assert command[4] == str(task_dir / "input")
+
+
 def test_worker_registration_sets_returned_identity(tmp_path: Path) -> None:
     worker, _, _, _, _ = daemon(tmp_path, None, b"")
     worker._register_worker()