Harden distributed pipeline

coordinator/internal/usecase/artifact.go

@@ -14,12 +14,13 @@ type UploadArtifact struct {
 	tasks     TaskRepository
 	artifacts ArtifactRepository
 	blobs     BlobStore
+	tx        TxManager
 	clk       Clock
 }
 
 func NewUploadArtifact(tasks TaskRepository, artifacts ArtifactRepository,
-	blobs BlobStore, clk Clock) *UploadArtifact {
-	return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, clk: clk}
+	blobs BlobStore, tx TxManager, clk Clock) *UploadArtifact {
+	return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clk: clk}
 }
 
 func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (*domain.Artifact, error) {
@@ -32,6 +33,15 @@ func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (
 	if !task.IsLeaseHeldBy(in.WorkerID, in.Attempt, uc.clk.Now()) {
 		return nil, domain.ErrLeaseConflict
 	}
+	// A client can retry a PUT after losing the response. Return the one durable
+	// result for this lease attempt instead of storing duplicate artifacts.
+	existing, err := uc.artifacts.FindPartialResult(ctx, in.TaskID, in.Attempt)
+	if err != nil {
+		return nil, err
+	}
+	if existing != nil {
+		return existing, nil
+	}
 
 	taskID := task.ID
 	art, err := domain.NewArtifact(task.JobID, &taskID, domain.ArtifactPartialResult,
@@ -50,25 +60,41 @@ func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (
 	}
 	art.SetContent(sum, size)
 
-	// The stream may take longer than the lease. Re-check after it finishes so
-	// an expired worker cannot leave a durable result record behind. Completion
-	// performs the same ownership check under its transaction.
-	current, err := uc.tasks.Get(ctx, in.TaskID)
+	// The stream may take longer than the lease. Lock the task while re-checking
+	// ownership and inserting metadata: completion or another upload cannot race
+	// this final decision. The database unique index is a second line of defence.
+	var durable *domain.Artifact
+	err = uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		current, err := uc.tasks.GetForUpdate(ctx, in.TaskID)
+		if err != nil {
+			return err
+		}
+		if !current.IsLeaseHeldBy(in.WorkerID, in.Attempt, uc.clk.Now()) {
+			return domain.ErrLeaseConflict
+		}
+		existing, err := uc.artifacts.FindPartialResult(ctx, in.TaskID, in.Attempt)
+		if err != nil {
+			return err
+		}
+		if existing != nil {
+			durable = existing
+			return nil
+		}
+		if err := uc.artifacts.Insert(ctx, art); err != nil {
+			return err
+		}
+		durable = art
+		return nil
+	})
 	if err != nil {
 		_ = uc.blobs.Delete(ctx, art.StorageKey)
 		return nil, err
 	}
-	if !current.IsLeaseHeldBy(in.WorkerID, in.Attempt, uc.clk.Now()) {
+	if durable != art {
+		// Another request won the race while this stream was being written.
 		_ = uc.blobs.Delete(ctx, art.StorageKey)
-		return nil, domain.ErrLeaseConflict
 	}
-
-	// Persist the record. If that fails the blob would be an orphan, so remove it.
-	if err := uc.artifacts.Insert(ctx, art); err != nil {
-		_ = uc.blobs.Delete(ctx, art.StorageKey)
-		return nil, err
-	}
-	return art, nil
+	return durable, nil
 }
 
 // DownloadArtifact returns an artifact's metadata together with a reader over

coordinator/internal/usecase/job.go

@@ -33,6 +33,17 @@ func NewCreateJob(jobs JobRepository, tasks TaskRepository, tx TxManager, clock
 // The all-or-none guarantee comes from TxManager: a half-created job would
 // leave chunks no worker could ever complete.
 func (uc *CreateJob) Execute(ctx context.Context, in CreateJobInput) (*domain.Job, error) {
+	if in.Workload == "similarity-graph" || in.Workload == "similarity_graph" {
+		// CTX-10 must plan triangular block pairs; ordinary independent input
+		// chunks would silently omit every cross-chunk molecular pair.
+		return nil, domain.ErrInvalidInput
+	}
+	if (in.Workload == "similarity-search" || in.Workload == "similarity_search") &&
+		len(in.Chunks) > 1 && in.Parameters["query_id"] != nil {
+		// Resolving once against the source dataset belongs to CTX-07. Letting
+		// each shard resolve it would make most tasks fail or use inconsistent data.
+		return nil, domain.ErrInvalidInput
+	}
 	chunks := make([]domain.ChunkSpec, 0, len(in.Chunks))
 	for _, c := range in.Chunks {
 		chunks = append(chunks, domain.ChunkSpec(c))
@@ -92,6 +103,17 @@ func (uc *CancelJob) Execute(ctx context.Context, jobID uuid.UUID) (int64, error
 		if job.Status == domain.JobCompleted || job.Status == domain.JobFailed {
 			return domain.ErrJobNotCancellable
 		}
+		// The lease reaper can be the transition that exhausted the final task.
+		// Check the authoritative task histogram as well as the cached job status,
+		// so a stale status can never turn a failed/completed job into cancelled.
+		counts, err := uc.tasks.CountByStatus(ctx, jobID)
+		if err != nil {
+			return err
+		}
+		derived := progressFrom(*job, counts).DeriveStatus()
+		if derived == domain.JobCompleted || derived == domain.JobFailed {
+			return domain.ErrJobNotCancellable
+		}
 		cancelled, err = uc.tasks.CancelByJob(ctx, jobID, now)
 		if err != nil {
 			return err
@@ -212,3 +234,18 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository
 	}
 	return jobs.UpdateStatus(ctx, jobID, status, completedAt)
 }
+
+func syncExpiredJobStatuses(ctx context.Context, jobs JobRepository, tasks TaskRepository,
+	jobIDs []uuid.UUID, now time.Time) error {
+	seen := make(map[uuid.UUID]struct{}, len(jobIDs))
+	for _, jobID := range jobIDs {
+		if _, duplicate := seen[jobID]; duplicate {
+			continue
+		}
+		seen[jobID] = struct{}{}
+		if err := syncJobStatus(ctx, jobs, tasks, jobID, now); err != nil {
+			return err
+		}
+	}
+	return nil
+}

coordinator/internal/usecase/ports.go

@@ -60,9 +60,9 @@ type TaskRepository interface {
 	// 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)
+	// ExpireLeases applies the lease-expiry rule to every elapsed task and returns
+	// the distinct jobs whose aggregate status may have changed.
+	ExpireLeases(ctx context.Context, now time.Time) ([]uuid.UUID, error)
 }
 
 // JobRepository persists jobs.
@@ -89,6 +89,9 @@ type WorkerRepository interface {
 type ArtifactRepository interface {
 	Insert(ctx context.Context, a *domain.Artifact) error
 	Get(ctx context.Context, id uuid.UUID) (*domain.Artifact, error)
+	// FindPartialResult returns the durable result already uploaded for one task
+	// attempt. A nil artifact means the attempt has not uploaded one yet.
+	FindPartialResult(ctx context.Context, taskID uuid.UUID, attempt int) (*domain.Artifact, error)
 }
 
 // BlobStore holds artifact bytes, addressed by an opaque storage key. It streams

coordinator/internal/usecase/task.go

@@ -21,12 +21,15 @@ import (
 
 type ClaimTask struct {
 	tasks         TaskRepository
+	jobs          JobRepository
+	workers       WorkerRepository
+	tx            TxManager
 	clock         Clock
 	leaseDuration time.Duration
 }
 
-func NewClaimTask(tasks TaskRepository, clock Clock, leaseDuration time.Duration) *ClaimTask {
-	return &ClaimTask{tasks: tasks, clock: clock, leaseDuration: leaseDuration}
+func NewClaimTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, leaseDuration time.Duration) *ClaimTask {
+	return &ClaimTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, leaseDuration: leaseDuration}
 }
 
 // Execute reclaims elapsed leases first, then hands out one task.
@@ -42,27 +45,46 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl
 	if in.WorkerID == "" {
 		return nil, domain.ErrInvalidInput
 	}
-	now := uc.clock.Now()
-
-	if _, err := uc.tasks.ExpireLeases(ctx, now); err != nil {
-		return nil, err
+	workloads := in.Workloads
+	if workerID, err := uuid.Parse(in.WorkerID); err == nil {
+		worker, err := uc.workers.Get(ctx, workerID)
+		if err != nil {
+			return nil, err
+		}
+		// Never trust caller-supplied capabilities: registration is the durable
+		// worker identity and its allowlist.
+		workloads = worker.Capabilities
 	}
+	var claimed *domain.ClaimedTask
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		now := uc.clock.Now()
+		affectedJobs, err := uc.tasks.ExpireLeases(ctx, now)
+		if err != nil {
+			return err
+		}
+		if err := syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affectedJobs, now); err != nil {
+			return err
+		}
 
-	task, err := uc.tasks.ClaimNext(ctx, ClaimFilter{
-		Workloads:  in.Workloads,
-		Owner:      in.WorkerID,
-		Now:        now,
-		LeaseUntil: now.Add(uc.leaseDuration),
+		task, err := uc.tasks.ClaimNext(ctx, ClaimFilter{
+			Workloads:  workloads,
+			Owner:      in.WorkerID,
+			Now:        now,
+			LeaseUntil: now.Add(uc.leaseDuration),
+		})
+		if err != nil {
+			return err
+		}
+		if task != nil {
+			value := task.AsClaimed()
+			claimed = &value
+		}
+		return nil
 	})
 	if err != nil {
 		return nil, err
 	}
-	if task == nil {
-		return nil, nil // empty queue is a normal state, not an error
-	}
-
-	claimed := task.AsClaimed()
-	return &claimed, nil
+	return claimed, nil // nil means an empty queue
 }
 
 // --- RenewLease ----------------------------------------------------------
@@ -230,18 +252,30 @@ func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task
 
 type ExpireLeases struct {
 	tasks TaskRepository
+	jobs  JobRepository
+	tx    TxManager
 	clock Clock
 }
 
-func NewExpireLeases(tasks TaskRepository, clock Clock) *ExpireLeases {
-	return &ExpireLeases{tasks: tasks, clock: clock}
+func NewExpireLeases(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *ExpireLeases {
+	return &ExpireLeases{tasks: tasks, jobs: jobs, tx: tx, clock: clock}
 }
 
-// Execute reports how many tasks were reclaimed.
+// Execute reclaims elapsed tasks and persists the state of every affected job.
 //
 // The sweep is one set-based statement rather than a load-decide-save loop:
 // several coordinators run it concurrently, and a single atomic UPDATE makes
 // the duplicate work harmless — the loser simply updates 0 rows.
 func (uc *ExpireLeases) Execute(ctx context.Context) (int64, error) {
-	return uc.tasks.ExpireLeases(ctx, uc.clock.Now())
+	var affected []uuid.UUID
+	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
+		now := uc.clock.Now()
+		var err error
+		affected, err = uc.tasks.ExpireLeases(ctx, now)
+		if err != nil {
+			return err
+		}
+		return syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affected, now)
+	})
+	return int64(len(affected)), err
 }

coordinator/internal/usecase/ui.go

@@ -15,6 +15,7 @@ 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)
+	ListTasksByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID][]domain.Task, error)
 	ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error)
 	ListArtifactsByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Artifact, error)
 }
@@ -88,12 +89,16 @@ func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, err
 		return DashboardView{}, err
 	}
 	out := DashboardView{Jobs: make([]JobCard, 0, len(jobs)), Workers: make([]WorkerCard, 0, len(workers))}
+	jobIDs := make([]uuid.UUID, 0, len(jobs))
 	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))
+		jobIDs = append(jobIDs, job.ID)
+	}
+	tasksByJob, err := d.read.ListTasksByJobs(ctx, jobIDs)
+	if err != nil {
+		return DashboardView{}, err
+	}
+	for _, job := range jobs {
+		out.Jobs = append(out.Jobs, jobCard(job, tasksByJob[job.ID]))
 	}
 	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})

coordinator/internal/usecase/upload.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 
 	"github.com/google/uuid"
 
@@ -15,20 +16,27 @@ import (
 // creates the job with one task per shard — the coordinator-side counterpart of
 // a client submitting pre-chunked URIs.
 type SubmitDataset struct {
-	blobs     BlobStore
-	artifacts ArtifactRepository
-	jobs      JobRepository
-	tasks     TaskRepository
-	tx        TxManager
-	clk       Clock
+	blobs       BlobStore
+	artifacts   ArtifactRepository
+	jobs        JobRepository
+	tasks       TaskRepository
+	tx          TxManager
+	clk         Clock
+	maxAttempts int
 }
 
 func NewSubmitDataset(blobs BlobStore, artifacts ArtifactRepository, jobs JobRepository,
-	tasks TaskRepository, tx TxManager, clk Clock) *SubmitDataset {
-	return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk}
+	tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int) *SubmitDataset {
+	return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts}
 }
 
 func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (SubmitDatasetResult, error) {
+	if err := validateUploadedWorkload(in.Workload, in.Parameters); err != nil {
+		return SubmitDatasetResult{}, err
+	}
+	if uc.maxAttempts < 1 {
+		return SubmitDatasetResult{}, domain.ErrInvalidInput
+	}
 	now := uc.clk.Now()
 
 	job, err := domain.NewUploadedJob(in.Workload, in.Parameters, now)
@@ -64,7 +72,7 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
 		cleanup()
 		return SubmitDatasetResult{}, err
 	}
-	splitErr := chunk.SplitTSVLimit(rc, in.RowsPerShard, in.MaxRows, func(index int, shard io.Reader) error {
+	splitErr := chunk.SplitChEMBLTSVLimit(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 {
@@ -77,7 +85,7 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
 		putKeys = append(putKeys, art.StorageKey)
 		art.SetContent(ssum, ssize)
 
-		task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum, in.Parameters, 0, now)
+		task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum, in.Parameters, uc.maxAttempts, now)
 		if err != nil {
 			return err
 		}
@@ -88,7 +96,8 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
 	_ = rc.Close()
 	if splitErr != nil {
 		cleanup()
-		return SubmitDatasetResult{}, splitErr
+		// Dataset shape is caller input, not an internal coordinator failure.
+		return SubmitDatasetResult{}, domain.ErrInvalidInput
 	}
 
 	// 3. Persist job + all artifacts + all tasks atomically.
@@ -118,6 +127,76 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
 	}, nil
 }
 
+// validateUploadedWorkload is deliberately narrow until CTX-07/08/10 adds a
+// typed distributed-workload registry. In particular, running similarity-graph
+// independently per TSV shard is scientifically wrong: cross-shard pairs would
+// be absent from the apparent graph.
+func validateUploadedWorkload(workload string, parameters map[string]any) error {
+	if workload != "similarity-search" {
+		return domain.ErrInvalidInput
+	}
+	allowed := map[string]struct{}{
+		"query_smiles": {}, "top_k": {}, "threshold": {},
+		"threshold_direction": {}, "progress_every": {},
+	}
+	for key := range parameters {
+		if _, ok := allowed[key]; !ok {
+			return domain.ErrInvalidInput
+		}
+	}
+	query, ok := parameters["query_smiles"].(string)
+	if !ok || query == "" || len(query) > 200 {
+		return domain.ErrInvalidInput
+	}
+	if value, ok := parameters["top_k"]; ok && !isPositiveJSONInteger(value) {
+		return domain.ErrInvalidInput
+	}
+	if value, ok := parameters["progress_every"]; ok && !isNonNegativeJSONInteger(value) {
+		return domain.ErrInvalidInput
+	}
+	if value, ok := parameters["threshold"]; ok && !isUnitIntervalNumber(value) {
+		return domain.ErrInvalidInput
+	}
+	if value, ok := parameters["threshold_direction"]; ok && value != "greater" && value != "less" {
+		return domain.ErrInvalidInput
+	}
+	return nil
+}
+
+func isPositiveJSONInteger(value any) bool    { return isJSONInteger(value, false) }
+func isNonNegativeJSONInteger(value any) bool { return isJSONInteger(value, true) }
+
+func isJSONInteger(value any, allowZero bool) bool {
+	var n int64
+	switch v := value.(type) {
+	case int:
+		n = int64(v)
+	case int64:
+		n = v
+	case float64:
+		if math.Trunc(v) != v || v > math.MaxInt64 || v < math.MinInt64 {
+			return false
+		}
+		n = int64(v)
+	default:
+		return false
+	}
+	return n >= 0 && (allowZero || n > 0)
+}
+
+func isUnitIntervalNumber(value any) bool {
+	switch v := value.(type) {
+	case float64:
+		return !math.IsNaN(v) && !math.IsInf(v, 0) && v >= 0 && v <= 1
+	case int:
+		return v >= 0 && v <= 1
+	case int64:
+		return v >= 0 && v <= 1
+	default:
+		return false
+	}
+}
+
 // GetTaskInput resolves a task's input shard and opens it for streaming. The
 // caller closes the reader.
 type GetTaskInput struct {

coordinator/internal/usecase/usecase_test.go

@@ -68,18 +68,18 @@ func newHarness() *harness {
 	}
 	tx := memstore.Tx{}
 	h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk)
-	h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk)
-	h.claim = usecase.NewClaimTask(h.tasks, h.clk, lease)
+	h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3)
+	h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease)
 	h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease)
 	h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, tx, h.clk)
 	h.fail = usecase.NewFailTask(h.tasks, h.jobs, tx, h.clk)
 	h.status = usecase.NewGetJobStatus(h.jobs, h.tasks)
 	h.results = usecase.NewListResults(h.tasks)
 	h.register = usecase.NewRegisterWorker(h.work, h.clk)
-	h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.arts, h.blobs, h.clk)
+	h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.arts, h.blobs, tx, h.clk)
 	h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs)
 	h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs)
-	h.expire = usecase.NewExpireLeases(h.tasks, h.clk)
+	h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk)
 	h.cancel = usecase.NewCancelJob(h.jobs, h.tasks, tx, h.clk)
 	return h
 }
@@ -138,6 +138,44 @@ func TestClaimLeasesAndAdvancesAttempt(t *testing.T) {
 	}
 }
 
+func TestRegisteredWorkerCannotBroadenItsCapabilitiesAtClaim(t *testing.T) {
+	h := newHarness()
+	h.seedJob(t, "restricted", 1)
+	worker, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
+		Name: "search-only", Capabilities: []string{"similarity-search"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{
+		WorkerID: worker.ID.String(), Workloads: []string{"restricted"},
+	})
+	if err != nil || claimed != nil {
+		t.Fatalf("claim = (%v, %v), want no compatible task", claimed, err)
+	}
+}
+
+func TestCreateJobRejectsUnsafeDistributedScientificPlans(t *testing.T) {
+	h := newHarness()
+	_, err := h.createJob.Execute(ctx, usecase.CreateJobInput{
+		Workload: "similarity-graph", InputURI: "s3://input",
+		Chunks: []usecase.ChunkInput{{ChunkIndex: 0, InputURI: "s3://chunk", InputSHA256: "sha"}},
+	})
+	if !errors.Is(err, domain.ErrInvalidInput) {
+		t.Errorf("graph job err = %v, want ErrInvalidInput", err)
+	}
+	_, err = h.createJob.Execute(ctx, usecase.CreateJobInput{
+		Workload: "similarity-search", InputURI: "s3://input", Parameters: map[string]any{"query_id": "CHEMBL1"},
+		Chunks: []usecase.ChunkInput{
+			{ChunkIndex: 0, InputURI: "s3://chunk0", InputSHA256: "sha"},
+			{ChunkIndex: 1, InputURI: "s3://chunk1", InputSHA256: "sha"},
+		},
+	})
+	if !errors.Is(err, domain.ErrInvalidInput) {
+		t.Errorf("sharded query_id job err = %v, want ErrInvalidInput", err)
+	}
+}
+
 func TestClaimEmptyQueueReturnsNil(t *testing.T) {
 	h := newHarness()
 	c, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: "w1", Workloads: []string{"w"}})
@@ -281,7 +319,7 @@ func TestUploadRejectsLeaseThatExpiresDuringStreaming(t *testing.T) {
 	h.seedJob(t, "w", 1)
 	taskID, attempt := h.leaseOne(t, "w1", "w")
 	h.uploadArt = usecase.NewUploadArtifact(
-		h.tasks, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, h.clk,
+		h.tasks, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, memstore.Tx{}, h.clk,
 	)
 
 	_, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{
@@ -414,6 +452,23 @@ func TestUploadArtifactRejectsForeignWorker(t *testing.T) {
 	}
 }
 
+func TestUploadArtifactIsIdempotentPerTaskAttempt(t *testing.T) {
+	h := newHarness()
+	h.seedJob(t, "w", 1)
+	taskID, attempt := h.leaseOne(t, "w1", "w")
+	first := h.uploadResult(t, taskID, "w1", attempt)
+	second, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{
+		TaskID: taskID, WorkerID: "w1", Attempt: attempt,
+		Filename: "retry.csv", ContentType: "text/csv", Body: strings.NewReader("different bytes"),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if second.ID != first {
+		t.Errorf("retry artifact = %s, want existing %s", second.ID, first)
+	}
+}
+
 func TestDownloadArtifactRoundTrips(t *testing.T) {
 	h := newHarness()
 	h.seedJob(t, "w", 1)
@@ -434,10 +489,10 @@ func TestDownloadArtifactRoundTrips(t *testing.T) {
 
 func TestSubmitDatasetChunksAndServesInput(t *testing.T) {
 	h := newHarness()
-	tsv := "id\tsmiles\nA\tCC\nB\tCCC\nC\tCCCC\nD\tCCCCC\nE\tCCCCCC\n"
+	tsv := "chembl_id\tcanonical_smiles\nA\tCC\nB\tCCC\nC\tCCCC\nD\tCCCCC\nE\tCCCCCC\n"
 
 	res, err := h.submit.Execute(ctx, usecase.SubmitDatasetInput{
-		Workload: "w", RowsPerShard: 2, Filename: "chembl.tsv",
+		Workload: "similarity-search", Parameters: map[string]any{"query_smiles": "CCO"}, RowsPerShard: 2, Filename: "chembl.tsv",
 		ContentType: "text/tab-separated-values", Body: strings.NewReader(tsv),
 	})
 	if err != nil {
@@ -453,7 +508,7 @@ func TestSubmitDatasetChunksAndServesInput(t *testing.T) {
 		t.Errorf("job total = %d, want 3", prog.Total)
 	}
 
-	c, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: "w1", Workloads: []string{"w"}})
+	c, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: "w1", Workloads: []string{"similarity-search"}})
 	if err != nil || c == nil {
 		t.Fatalf("claim shard: %v", err)
 	}
@@ -472,9 +527,9 @@ 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"
+	tsv := "chembl_id\tcanonical_smiles\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",
+		Workload: "similarity-search", Parameters: map[string]any{"query_smiles": "CCO"}, RowsPerShard: 2, MaxRows: 3, Filename: "chembl.tsv",
 		ContentType: "text/tab-separated-values", Body: strings.NewReader(tsv),
 	})
 	if err != nil {
@@ -485,6 +540,26 @@ func TestSubmitDatasetLimitsRowsBeforeCreatingShards(t *testing.T) {
 	}
 }
 
+func TestSubmitDatasetRejectsUnsupportedDistributedWorkloads(t *testing.T) {
+	h := newHarness()
+	_, err := h.submit.Execute(ctx, usecase.SubmitDatasetInput{
+		Workload: "similarity-graph", Parameters: map[string]any{"threshold": 0.7}, RowsPerShard: 2,
+		Filename: "chembl.tsv", ContentType: "text/tab-separated-values",
+		Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"),
+	})
+	if !errors.Is(err, domain.ErrInvalidInput) {
+		t.Errorf("graph submission err = %v, want ErrInvalidInput", err)
+	}
+	_, err = h.submit.Execute(ctx, usecase.SubmitDatasetInput{
+		Workload: "similarity-search", Parameters: map[string]any{"query_id": "CHEMBL1"}, RowsPerShard: 2,
+		Filename: "chembl.tsv", ContentType: "text/tab-separated-values",
+		Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"),
+	})
+	if !errors.Is(err, domain.ErrInvalidInput) {
+		t.Errorf("query_id submission err = %v, want ErrInvalidInput", err)
+	}
+}
+
 func TestCancelJobInvalidatesClaimedAndPendingTasks(t *testing.T) {
 	h := newHarness()
 	jobID := h.seedJob(t, "w", 3)
@@ -531,3 +606,22 @@ func TestExpireLeasesReclaims(t *testing.T) {
 		t.Errorf("expire = (%d, %v), want (1, nil)", n, err)
 	}
 }
+
+func TestFinalLeaseExpiryPersistsFailedJobAndCannotBeCancelled(t *testing.T) {
+	h := newHarness()
+	jobID := h.seedJob(t, "w", 1)
+	for attempt := 1; attempt <= domain.DefaultMaxAttempts; attempt++ {
+		h.leaseOne(t, "w1", "w")
+		h.clk.Advance(lease + time.Second)
+		if _, err := h.expire.Execute(ctx); err != nil {
+			t.Fatalf("expire attempt %d: %v", attempt, err)
+		}
+	}
+	progress, err := h.status.Execute(ctx, jobID)
+	if err != nil || progress.Job.Status != domain.JobFailed || progress.DeriveStatus() != domain.JobFailed {
+		t.Fatalf("progress = %+v, err = %v; want persisted failed job", progress, err)
+	}
+	if _, err := h.cancel.Execute(ctx, jobID); !errors.Is(err, domain.ErrJobNotCancellable) {
+		t.Errorf("cancel terminal lease failure = %v, want ErrJobNotCancellable", err)
+	}
+}