Add local operator web interface

coordinator/internal/storage/postgres/integration_test.go

@@ -137,30 +137,37 @@ 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
-	// steal an already-leased row.
+	// More workers than tasks. With SKIP LOCKED, a concurrent caller can
+	// 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()
-			task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{
-				Owner:      fmt.Sprintf("worker-%d", n),
-				Now:        now,
-				LeaseUntil: now.Add(time.Minute),
-			})
-			if err != nil {
-				t.Errorf("claim: %v", err)
+			for attempt := 0; attempt < 20; attempt++ {
+				task, err := repo.ClaimNext(context.Background(), usecase.ClaimFilter{
+					Owner:      fmt.Sprintf("worker-%d", n),
+					Now:        now,
+					LeaseUntil: now.Add(time.Minute),
+				})
+				if err != nil {
+					t.Errorf("claim: %v", err)
+					return
+				}
+				if task == nil || task.JobID != job.ID {
+					time.Sleep(time.Millisecond)
+					continue
+				}
+				mu.Lock()
+				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
 			}
-			if task == nil || task.JobID != job.ID {
-				return // empty queue, or a task from another test's job
-			}
-			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)
 		}(i)
 	}
 	wg.Wait()

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()
+}