diff --git a/PLAN.md b/PLAN.md index e0c34af..81540d2 100644 --- a/PLAN.md +++ b/PLAN.md @@ -780,6 +780,10 @@ for the exact sparse similarity graph. **Goal:** Add a small server-rendered or static HTML UI to inspect jobs, tasks, workers, and download final artifacts. +**Detailed delivery plan:** [`docs/web-interface-plan.md`](docs/web-interface-plan.md). +The plan deliberately starts with a clearly labelled diagnostic UI before +CTX-09 enables final result downloads. + **Depends on:** CTX-04, CTX-09. **Acceptance criteria:** diff --git a/STATUS.md b/STATUS.md index eb33b5b..f6a7a23 100644 --- a/STATUS.md +++ b/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 - for complete workload results. +- Planner/reducer semantics are not implemented; the operator UI labels + `partial_result` files as diagnostic and cannot present them as final output. + Use the local `scimesh` CLI for complete workload results. - The worker/coordinator flow currently accepts both underscore API workload names and hyphenated CLI names while the contract is consolidated. - A real-stack worker test uses a small `query_smiles` shard. Resolving a diff --git a/coordinator/.env.example b/coordinator/.env.example index d20a617..4c1e596 100644 --- a/coordinator/.env.example +++ b/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 diff --git a/coordinator/README.md b/coordinator/README.md index b79b671..43239cc 100644 --- a/coordinator/README.md +++ b/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. diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index dd11a56..1134257 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/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). diff --git a/coordinator/docker-compose.yml b/coordinator/docker-compose.yml index 901ac55..2a6cd6d 100644 --- a/coordinator/docker-compose.yml +++ b/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" diff --git a/coordinator/internal/chunk/tsv.go b/coordinator/internal/chunk/tsv.go index 6b968eb..f020810 100644 --- a/coordinator/internal/chunk/tsv.go +++ b/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) diff --git a/coordinator/internal/chunk/tsv_test.go b/coordinator/internal/chunk/tsv_test.go index ba12917..585b35a 100644 --- a/coordinator/internal/chunk/tsv_test.go +++ b/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) { diff --git a/coordinator/internal/domain/errors.go b/coordinator/internal/domain/errors.go index d64b1f6..fca2979 100644 --- a/coordinator/internal/domain/errors.go +++ b/coordinator/internal/domain/errors.go @@ -8,13 +8,14 @@ import "errors" // // Always compare with errors.Is — outer layers may wrap these with %w. var ( - ErrJobNotFound = errors.New("job not found") - ErrTaskNotFound = errors.New("task not found") - ErrWorkerNotFound = errors.New("worker not found") - ErrArtifactNotFound = errors.New("artifact not found") - ErrLeaseConflict = errors.New("task leased to another worker") - ErrStaleAttempt = errors.New("attempt does not match lease") - ErrResultConflict = errors.New("different result already recorded") - ErrInvalidInput = errors.New("invalid input") - ErrTaskNotLeased = errors.New("task is not currently leased") + ErrJobNotFound = errors.New("job not found") + 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") + ErrInvalidInput = errors.New("invalid input") + ErrTaskNotLeased = errors.New("task is not currently leased") ) diff --git a/coordinator/internal/domain/job.go b/coordinator/internal/domain/job.go index 0aa4827..7d6be3d 100644 --- a/coordinator/internal/domain/job.go +++ b/coordinator/internal/domain/job.go @@ -99,18 +99,21 @@ func NewJobWithTasks(workload, inputURI string, params map[string]any, // JobProgress is the aggregate view of a job and the state of its tasks. type JobProgress struct { - Job Job - Total int - Pending int - Leased int - Done int - Failed int + Job Job + Total int + Pending int + 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: diff --git a/coordinator/internal/domain/job_test.go b/coordinator/internal/domain/job_test.go index 0b58902..54a221d 100644 --- a/coordinator/internal/domain/job_test.go +++ b/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) { diff --git a/coordinator/internal/domain/task.go b/coordinator/internal/domain/task.go index 306fed5..7864660 100644 --- a/coordinator/internal/domain/task.go +++ b/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. diff --git a/coordinator/internal/domain/task_test.go b/coordinator/internal/domain/task_test.go index 9c2ff98..a4c7862 100644 --- a/coordinator/internal/domain/task_test.go +++ b/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) diff --git a/coordinator/internal/infra/config.go b/coordinator/internal/infra/config.go index 8703a27..5414866 100644 --- a/coordinator/internal/infra/config.go +++ b/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 { diff --git a/coordinator/internal/infra/config_test.go b/coordinator/internal/infra/config_test.go new file mode 100644 index 0000000..f9c0945 --- /dev/null +++ b/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) + } +} diff --git a/coordinator/internal/memstore/memstore.go b/coordinator/internal/memstore/memstore.go index 7916951..44db68e 100644 --- a/coordinator/internal/memstore/memstore.go +++ b/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() diff --git a/coordinator/internal/memstore/ui_read.go b/coordinator/internal/memstore/ui_read.go new file mode 100644 index 0000000..ce36e5c --- /dev/null +++ b/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 +} diff --git a/coordinator/internal/storage/postgres/integration_test.go b/coordinator/internal/storage/postgres/integration_test.go index fd461bd..6b3395a 100644 --- a/coordinator/internal/storage/postgres/integration_test.go +++ b/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() @@ -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() diff --git a/coordinator/internal/storage/postgres/task_repo.go b/coordinator/internal/storage/postgres/task_repo.go index 2a4480f..18b2646 100644 --- a/coordinator/internal/storage/postgres/task_repo.go +++ b/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. // diff --git a/coordinator/internal/storage/postgres/ui_read_repo.go b/coordinator/internal/storage/postgres/ui_read_repo.go new file mode 100644 index 0000000..4a24ed2 --- /dev/null +++ b/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() +} diff --git a/coordinator/internal/transport/http/dto.go b/coordinator/internal/transport/http/dto.go index c652b25..f944482 100644 --- a/coordinator/internal/transport/http/dto.go +++ b/coordinator/internal/transport/http/dto.go @@ -111,13 +111,14 @@ type uploadJobResponse struct { } type jobProgressResponse struct { - ID uuid.UUID `json:"id"` - Status string `json:"status"` - Total int `json:"total"` - Pending int `json:"pending"` - Leased int `json:"leased"` - Done int `json:"completed"` - Failed int `json:"failed"` + ID uuid.UUID `json:"id"` + Status string `json:"status"` + Total int `json:"total"` + Pending int `json:"pending"` + Leased int `json:"leased"` + Done int `json:"completed"` + Failed int `json:"failed"` + Cancelled int `json:"cancelled"` } type uploadArtifactResponse struct { @@ -153,12 +154,13 @@ func toClaimedTaskResponse(c domain.ClaimedTask) claimedTaskResponse { func toJobProgressResponse(p domain.JobProgress) jobProgressResponse { return jobProgressResponse{ - ID: p.Job.ID, - Status: string(p.DeriveStatus()), - Total: p.Total, - Pending: p.Pending, - Leased: p.Leased, - Done: p.Done, - Failed: p.Failed, + ID: p.Job.ID, + Status: string(p.DeriveStatus()), + Total: p.Total, + Pending: p.Pending, + Leased: p.Leased, + Done: p.Done, + Failed: p.Failed, + Cancelled: p.Cancelled, } } diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go index 358704e..ae14782 100644 --- a/coordinator/internal/transport/http/errors.go +++ b/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 diff --git a/coordinator/internal/transport/http/handlers.go b/coordinator/internal/transport/http/handlers.go index df61461..4bbf5c8 100644 --- a/coordinator/internal/transport/http/handlers.go +++ b/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) { diff --git a/coordinator/internal/transport/http/middleware.go b/coordinator/internal/transport/http/middleware.go index feb3f4f..2808313 100644 --- a/coordinator/internal/transport/http/middleware.go +++ b/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 diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index cf7c304..00e80d8 100644 --- a/coordinator/internal/transport/http/server.go +++ b/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 diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 9a0c0f0..261197c 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/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" diff --git a/coordinator/internal/transport/http/templates/dashboard.html b/coordinator/internal/transport/http/templates/dashboard.html new file mode 100644 index 0000000..5e24f90 --- /dev/null +++ b/coordinator/internal/transport/http/templates/dashboard.html @@ -0,0 +1,23 @@ +{{define "dashboard.html"}} + + + + + + SciMesh operator dashboard + + + +
+

Local coordinator

SciMesh operator dashboard

See where a computation is and what should happen next.

Start a check
+
This screen currently diagnoses shard jobs.Workers upload partial CSVs to the coordinator. Until a reducer is implemented, those files are not one final scientific result.
1. Upload TSVThe coordinator splits the file into shard tasks.
2. Wait for a workerA worker claims a shard, calculates similarity, and returns a CSV.
3. Inspect artifactsDownload a partial result from the job page.
+

Recent jobs

+
{{range .Jobs}}{{else}}{{end}}
ComputationStateProgressCreated
{{workloadLabel .Workload}}
Open job details		{{statusLabel .Status}}
{{statusHint .Status}}		{{.Completed}} / {{.Total}} complete{{if gt .Failed 0}} · failed: {{.Failed}}{{end}}{{if gt .Cancelled 0}} · stopped: {{.Cancelled}}{{end}}
{{time .CreatedAt}}
No jobs yet.
Click “Start a check”, upload a small TSV, and leave a worker running.
+

Workers

+
{{range .Workers}}
{{.Name}}
{{.ID}}
{{workerStatusLabel .Status}}
{{range .Capabilities}}{{.}} {{end}}
Last signal
{{time .LastHeartbeatAt}}
{{else}}
No worker is registered yet.
Run scimesh-worker with the coordinator URL and worker token.
{{end}}
+
+ + +{{end}} diff --git a/coordinator/internal/transport/http/templates/job.html b/coordinator/internal/transport/http/templates/job.html new file mode 100644 index 0000000..f983e2d --- /dev/null +++ b/coordinator/internal/transport/http/templates/job.html @@ -0,0 +1,29 @@ +{{define "job.html"}} + + + + + + SciMesh job + + + +
+ ← Back to jobs

{{workloadLabel .Workload}}

Execution progress

+
{{statusLabel .Status}}

{{statusHint .Status}}

{{if cancellable .Status}}This cancels every shard that is not finished yet.{{else}}Summary refreshes automatically every two seconds.{{end}}

{{.Completed}} of {{.Total}} tasks complete

{{.Total}}total shards
{{.Completed}}complete
{{.Pending}}waiting
{{add .Leased .Running}}with workers
{{.Failed}}failed
{{.Cancelled}}stopped
Technical details

Job ID: {{.ID}}
Workload: {{.Workload}}
Created: {{time .CreatedAt}}

+
What can be downloaded now?
partial_result 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.
+

Shard tasks

If a task fails, its code and message appear here. Refresh the page to update the detailed rows.

+
{{range .Tasks}}{{else}}{{end}}
ShardStateAttemptWorker / leaseError
#{{.ChunkIndex}}{{statusLabel .Status}}{{.Attempt}} / {{.MaxAttempts}}{{if .LeaseOwner}}{{.LeaseOwner}}{{if .LeaseExpiresAt}}
until {{time .LeaseExpiresAt}}{{end}}{{else}}—{{end}}		{{if .ErrorCode}}{{taskErrorLabel .ErrorCode}}
{{taskErrorHint .ErrorCode}}{{else}}—{{end}}
No tasks have appeared yet.
+

Coordinator artifacts

+
{{range .Artifacts}}{{else}}{{end}}
TypeFileSizeIntegrity check
{{if .Diagnostic}}Partial result
diagnostic{{else}}{{.Kind}}{{end}}		{{.Filename}}{{bytes .SizeBytes}}{{.SHA256}}{{if .Downloadable}}Download CSV{{else}}Unavailable{{end}}
No artifacts yet. The worker uploads a CSV after it completes a shard.
+
+ + + +{{end}} diff --git a/coordinator/internal/transport/http/templates/new-job.html b/coordinator/internal/transport/http/templates/new-job.html new file mode 100644 index 0000000..365df86 --- /dev/null +++ b/coordinator/internal/transport/http/templates/new-job.html @@ -0,0 +1,31 @@ +{{define "new-job.html"}} + + + + + + Create a check — SciMesh + + + +
+ ← Back to jobs

Guided run

Search for similar molecules

Creates a diagnostic similarity-search job: a worker finds the top-k molecules most similar to a target SMILES.

+
Before starting
+
+

Expected columns: chembl_id and canonical_smiles.

+

CCO is ethanol. For gefitinib, use its SMILES here or the local CLI with --query-id.

+

This is the top-k within each shard, not a global top-k for the whole dataset yet.

+

Fewer rows mean more tasks and more visible progress; more rows mean fewer, longer tasks.

+

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.

+ +
+
+ + + +{{end}} diff --git a/coordinator/internal/transport/http/ui.go b/coordinator/internal/transport/http/ui.go new file mode 100644 index 0000000..ddf1168 --- /dev/null +++ b/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) +} diff --git a/coordinator/internal/transport/http/ui_test.go b/coordinator/internal/transport/http/ui_test.go new file mode 100644 index 0000000..a6f82cb --- /dev/null +++ b/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") + } +} diff --git a/coordinator/internal/usecase/dto.go b/coordinator/internal/usecase/dto.go index 6491cbf..17e0498 100644 --- a/coordinator/internal/usecase/dto.go +++ b/coordinator/internal/usecase/dto.go @@ -53,9 +53,12 @@ type SubmitDatasetInput struct { Workload string Parameters map[string]any RowsPerShard int - Filename string - ContentType string - Body io.Reader + // 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 } type SubmitDatasetResult struct { diff --git a/coordinator/internal/usecase/job.go b/coordinator/internal/usecase/job.go index 28e6c6e..748f021 100644 --- a/coordinator/internal/usecase/job.go +++ b/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} } @@ -144,9 +183,10 @@ func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobPr Job: job, Pending: counts[domain.TaskPending], // Leased and running are both "in flight" for progress purposes. - Leased: counts[domain.TaskLeased] + counts[domain.TaskRunning], - Done: counts[domain.TaskCompleted], - Failed: counts[domain.TaskFailed], + 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 diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index a85481b..7b56211 100644 --- a/coordinator/internal/usecase/ports.go +++ b/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) diff --git a/coordinator/internal/usecase/ui.go b/coordinator/internal/usecase/ui.go new file mode 100644 index 0000000..c7283c3 --- /dev/null +++ b/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 +} diff --git a/coordinator/internal/usecase/upload.go b/coordinator/internal/usecase/upload.go index f04f582..20fb899 100644 --- a/coordinator/internal/usecase/upload.go +++ b/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 { diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index 4786b53..68b94ac 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/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 diff --git a/docs/api-contract.md b/docs/api-contract.md index ba16e84..50abf4d 100644 --- a/docs/api-contract.md +++ b/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 -part `file`. The coordinator stores the input, splits the TSV into shard -artifacts (header repeated per shard), and creates one task per shard. +`workload`, `parameters` (JSON), `chunk_rows` (int, default 1000), optional +`max_rows` (positive int), and the file part `file`. `max_rows` limits the +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 +``` + +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 diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 4ae0557..7117a78 100644 --- a/docs/openapi.yaml +++ b/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] diff --git a/docs/web-interface-plan.md b/docs/web-interface-plan.md new file mode 100644 index 0000000..48b29d7 --- /dev/null +++ b/docs/web-interface-plan.md @@ -0,0 +1,372 @@ +# SciMesh: plan for the initial web interface + +## 1. Purpose and outcome + +Build a small, local-first web interface for manually checking the complete +SciMesh pipeline. A person should be able to open one address, upload a small +ChEMBL-style TSV, configure a supported run, observe workers and task progress, +inspect failures, and download artifacts without composing raw HTTP requests. + +This is not a public multi-tenant product. It is an operator and demo interface +for a trusted local team. The coordinator remains the only process with direct +database and artifact-storage access; the browser never calls PostgreSQL and +never receives a worker bearer token. + +The first release must be useful before CTX-07--CTX-10 are complete. Therefore +it has two visibly different modes: + +| Mode | What it proves | What it must not claim | +| --- | --- | --- | +| **Pipeline check** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and downloads work end-to-end. | That multiple shard results have been scientifically reduced into one answer. | +| **Final run** | A reducer has produced a durable final CSV for the full job. | Available only after CTX-09, and for graph only after CTX-10. | + +Never label a partial artifact as a final molecular result. The UI must show a +clear `Pipeline check — partial results` badge while a reducer is unavailable. + +## 2. Constraints and decisions + +- Serve the UI from the Go coordinator at the same origin. No React, Vue, Node + build, CDN, or separate frontend service. +- Use Go `html/template`, `embed`, ordinary CSS, and small vanilla JavaScript + modules. The page works locally with `docker compose up`. +- Keep worker APIs and UI APIs separate. The UI handlers call Go use cases; + they do not make HTTP calls to worker endpoints. +- Add a distinct `UI_AUTH_TOKEN` for browser/operator access. It must never + reuse `WORKER_AUTH_TOKEN`, appear in page HTML, localStorage, logs, URLs, or + error messages. For the initial local UI use HTTP Basic Auth over a trusted + local/reverse-proxied connection. If `UI_AUTH_TOKEN` is unset, `/ui` and + `/ui/api/*` return `404` and the coordinator stays API-only. +- Keep all user-controllable text escaped by `html/template`; JavaScript renders + API fields through `textContent`, never `innerHTML`. +- All downloads go through coordinator-owned UI endpoints with authorization. + Do not expose filesystem paths, `storage_key`, database errors, worker tokens, + or raw worker tracebacks. +- The API vocabulary should be canonicalised before UI forms are implemented. + Choose one external workload spelling (`similarity-search` and + `similarity-graph` recommended, matching the CLI) and keep legacy underscore + aliases only at the worker boundary. Update `docs/api-contract.md` and + `docs/openapi.yaml` in the same change. + +## 3. Target user journey + +1. Start coordinator/PostgreSQL and one or more `scimesh-worker` processes. +2. Open `http://localhost:8080/ui`, authenticate with the UI token, and see + readiness plus the registered-worker table. +3. Choose **New run**, select a workload, fill validated parameters, choose a + TSV, and submit it. +4. The UI redirects to `/ui/jobs/{job_id}` and polls every two seconds. +5. The operator sees counters, per-task attempt/lease/error state, and worker + activity. They may copy the worker launch command but cannot start arbitrary + processes from the browser. +6. During a pipeline check, download an input shard or partial CSV to validate + manually. For a final run, download the final CSV only when the job status is + `completed` and a final artifact exists. +7. A failed run exposes only its sanitized failure reason and retry state. The + browser offers a safe retry action only after an explicit future API supports + it; v1 never fabricates a retry by changing task rows directly. + +## 4. What exists today and required gaps + +| Capability | Current state | UI plan | +| --- | --- | --- | +| Upload/chunk dataset | `POST /jobs/upload` exists | Reuse through a UI handler with server-side multipart validation. | +| Aggregate status | `GET /jobs/{id}` exists | Add list/detail read models for the UI. | +| Worker registration/lease flow | Implemented | Add a read-only worker list; no browser worker controls. | +| Task diagnostics | No public list/detail response | Add sanitized job task list with attempt, status, lease owner, expiry and error. | +| Artifact download | Worker endpoint exists | Add UI-authorized, job-scoped download proxy. | +| Final result | Reducer is not implemented | Gate behind CTX-09; show partial diagnostic artifacts meanwhile. | +| Distributed graph correctness | Planner/reducer unavailable | Do not advertise a multi-shard graph as final until CTX-10. | + +## 5. Proposed structure + +```text +coordinator/ + web/ + templates/ + layout.html + dashboard.html + job_new.html + job_detail.html + error.html + static/ + app.css + dashboard.js + job-detail.js + internal/ + transport/http/ + ui_handlers.go + ui_dto.go + ui_auth.go + ui_handlers_test.go + usecase/ + ui.go # read-only DTO orchestration, no HTML + domain/ + ui.go # only if a shared value object is genuinely needed + storage/postgres/ + ui_read_repo.go # parameterized listing/detail queries +``` + +Embed `web/templates` and `web/static` into the coordinator binary with +`go:embed`. No assets are generated at runtime; `go test ./...` must work +without Node/npm. + +## 6. UI surface and routes + +### HTML routes + +| Route | Purpose | Availability | +| --- | --- | --- | +| `GET /ui` | Dashboard: readiness, recent jobs, workers, quick actions. | WUI-03 | +| `GET /ui/jobs/new` | New-run form and parameter help. | WUI-04 | +| `GET /ui/jobs/{id}` | Job detail and polling shell. | WUI-03 | +| `GET /ui/jobs/{id}/artifacts/{artifact_id}` | Authorized download proxy with attachment headers. | WUI-05 | + +### JSON routes used only by the pages + +| Route | Response / action | Notes | +| --- | --- | --- | +| `GET /ui/api/overview` | readiness, recent jobs, workers | No secrets, no storage paths. | +| `GET /ui/api/jobs` | cursor/page of compact job cards | Default 20, deterministic `created_at DESC, id DESC`. | +| `POST /ui/api/jobs/upload` | validates form, streams dataset, returns `201 {job_id}` | Same input limits as `/jobs/upload`; form fields first, file last. | +| `GET /ui/api/jobs/{id}` | job detail, counters, tasks, allowed artifacts | Polling endpoint, no raw DB model. | +| `GET /ui/api/jobs/{id}/events` | **deferred** | Start with polling; no SSE/WebSocket in v1. | + +All `/ui` routes use UI authentication. Existing worker API routes keep worker +authentication and are not relaxed for the browser. + +## 7. Read models and data minimisation + +Create UI-specific DTOs; do not return domain/database entities directly. + +```text +JobCard: + id, workload, created_at, status, + total, pending, leased, running, completed, failed + +JobDetail: + JobCard fields, + parameters (allowlisted/redacted), + tasks: [{id, chunk_index, status, attempt, max_attempts, + lease_owner_display, lease_expires_at, error_code, error_message}], + artifacts: [{id, kind, filename, size_bytes, sha256, downloadable}] + +WorkerCard: + id, name, capabilities, status, last_heartbeat_at, created_at +``` + +Rules: + +- Display a shortened UUID by default but provide a copy button with the full + value; never interpolate it into HTML. +- Do not expose `storage_key`, absolute artifact paths, raw metrics containing + unexpected values, auth configuration, or worker-local directories. +- An artifact is downloadable only when it belongs to the requested job. A + `final_result` is downloadable only after the job is `completed`; partial + artifacts are marked diagnostic. +- SQL uses explicit columns, pagination/cursors, deterministic ordering, and + joins constrained by `job_id`. + +## 8. Workload forms and validation + +### 8.1 Common fields + +- 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 + validation remain authoritative. + +### 8.2 Similarity search + +Inputs: exactly one `query_smiles` or `query_id`, `top_k`, optional threshold, +threshold direction, `max_rows`, and `progress_every`. + +For a runnable manual pipeline check before CTX-08, offer `query_smiles` and +default `chunk_rows` large enough to create one shard. A `query_id` across +multiple shards is disabled with an explanation until CTX-07 resolves it once +before fan-out. The detail page calls an artifact a **partial top-k CSV**, not +a global top-k, until CTX-09 reduction exists. + +### 8.3 Similarity graph + +Inputs: threshold, threshold direction, block size, `max_rows`, and progress +interval. The form may show a disabled **Experimental — not globally reduced** +card, but it must not submit multi-shard graph jobs until CTX-10 implements +block-pair planning and deterministic reduction. A one-shard pipeline check is +allowed only behind an explicit acknowledgement and produces a diagnostic edge +list, not a final graph. + +Validation exists in three places: HTML constraints for feedback, a small +JavaScript schema for form behaviour, and authoritative Go validation mapped to +typed workload parameters. Never pass arbitrary parameter maps straight from +the browser to workers. + +## 9. Delivery packages + +Each package is a separate PR/task context. Do not start a later package until +its listed dependency and tests are green. + +### WUI-00 — Freeze UI contract and demo scope + +**Depends on:** current `main`. + +**Deliver:** this plan, canonical workload naming decision, and updates to +`docs/api-contract.md`/`docs/openapi.yaml` if names or statuses change. + +**Acceptance:** API has a precise distinction between diagnostic artifacts and +final results; UI security model has a distinct token; CTX-07/09/10 limitations +are visible. + +### WUI-01 — UI read models and PostgreSQL queries + +**Depends on:** WUI-00. + +**Deliver:** repository/use-case methods for deterministic job lists, sanitized +job details, task summaries, artifact metadata, and worker lists. Add indexes +only if `EXPLAIN ANALYZE` on a realistic list query shows need. + +**Acceptance:** no N+1 query path; no storage key/secret leaks; unknown job is +404; artifact lookup is constrained to its job; Go unit plus real-PostgreSQL +integration tests cover ordering, empty lists and ownership boundaries. + +### WUI-02 — UI auth and embedded asset foundation + +**Depends on:** WUI-01. + +**Deliver:** `UI_AUTH_TOKEN` config validation, Basic Auth middleware, embedded +template/static serving, security headers, and an API-only fallback when UI is +disabled. + +**Acceptance:** worker token never authorizes `/ui`; UI token never authorizes +worker routes; `/ui` is 404 when disabled; HTML/content-security headers are +tested; no token appears in logs or errors. + +### WUI-03 — Read-only dashboard and job detail + +**Depends on:** WUI-01, WUI-02. + +**Deliver:** dashboard, worker table, job list, detail page, two-second polling +with pause when the tab is hidden, error/retry display, and accessible empty/ +loading/error states. + +**Acceptance:** a manually created job changes from pending to running on the +page without reload; all text is escaped; polling stops on terminal states; +task attempts and sanitized errors are visible; handler/template tests cover +XSS-shaped names and errors. + +### WUI-04 — New-run upload form + +**Depends on:** WUI-02, WUI-03. + +**Deliver:** workload-specific form, typed Go validation, streamed upload, +progress/submit state, redirect to job detail, and copyable worker launch +instructions. + +**Acceptance:** invalid query combinations fail with 400 and clear UI feedback; +valid small similarity-search upload creates deterministic shard count; upload +limits are enforced; browser never sees worker auth; test covers malformed TSV, +large/invalid fields, duplicate form fields, and failed coordinator storage. + +### WUI-05 — Safe artifact downloads and diagnostic preview + +**Depends on:** WUI-01--04. + +**Deliver:** job-scoped download proxy, CSV preview limited by byte/row count, +checksum/size metadata display, and prominent partial/final labels. + +**Acceptance:** artifact from another job is 404; path traversal cannot select a +file; `Content-Disposition` is safe; preview never loads an unbounded CSV; no +final-result button exists before CTX-09. + +### WUI-06 — Final-result UX after CTX-09 + +**Depends on:** CTX-09 and WUI-05. + +**Deliver:** `reducing` state, final artifact card, final CSV preview/download, +and deterministic result metadata. + +**Acceptance:** final link is shown only for `completed`; reducer failure is +sanitized; page refresh/restart preserves final artifact; end-to-end test +compares downloaded final search CSV with local reference output. + +### WUI-07 — Full similarity-graph UX after CTX-10 + +**Depends on:** CTX-10 and WUI-06. + +**Deliver:** enabled graph form, block-pair planning summary, graph-specific +progress, final edge-list preview/download, and warnings for low thresholds. + +**Acceptance:** graph result equals local brute force on a small fixture; no +dense matrix is introduced; threshold direction is displayed and preserved; +result order is deterministic. + +### WUI-08 — Manual-demo script and CI browser checks + +**Depends on:** WUI-05; extend after WUI-06/07. + +**Deliver:** `make demo-ui` (or documented compose profile), a tiny tracked TSV +fixture, start/stop instructions, and headless browser smoke tests. + +**Acceptance:** clean checkout can start coordinator, a worker, open UI, +submit fixture, observe task completion, and download a diagnostic artifact. +CI covers auth, upload, polling JSON, job isolation, and download permission. + +## 10. Testing strategy + +| Layer | Required checks | +| --- | --- | +| Go domain/use case | status projection, artifact/job ownership, pagination ordering, redaction. | +| Go HTTP | UI auth separation, malformed multipart, CSRF-safe same-origin policy, 404/401, headers, escaping. | +| PostgreSQL | fresh migrations, list/detail query ordering, cross-job artifact denial, completed/final state. | +| Browser | form validation, upload success/failure, polling transition, terminal state, safe text rendering. | +| End-to-end | real PostgreSQL + coordinator + Python worker + small fixture; verify actual bytes/checksum. | + +Use Playwright only if it can run in CI without adding a production runtime +dependency. Otherwise start with Go handler tests and a small `curl`/HTML +smoke script, then add browser automation in WUI-08. + +## 11. Manual verification script after WUI-05 + +```sh +# terminal 1 +cd coordinator +UI_AUTH_TOKEN='local-ui-secret' make up + +# terminal 2: use a separate worker token; do not paste it into the browser +SCIMESH_COORDINATOR_URL=http://localhost:8080 \ +SCIMESH_BEARER_TOKEN='worker-secret' \ +scimesh-worker --work-dir ./worker-data + +# browser +# http://localhost:8080/ui +# authenticate with the UI token, upload a tiny TSV, then watch /ui/jobs/{id} +``` + +The actual configuration variable names, compose wiring, and launch command are +implemented in WUI-02/WUI-08; this block is the acceptance target, not a claim +that the interface exists today. + +## 12. Explicit non-goals for the initial interface + +- no database browser or SQL console; +- no browser-side RDKit or scientific calculation; +- no worker start/stop shell execution from UI; +- no multi-user accounts, RBAC, password reset, public internet exposure, or + user-provided storage credentials; +- no WebSocket/SSE, React/Vue, Docker requirement for the Python package, or + deployment platform; +- no claim that a distributed graph or search result is final before its + planner/reducer acceptance criteria are met. + +## 13. Definition of done for the first hand-testable release + +WUI-00 through WUI-05 are complete when a clean local checkout can run a +trusted, authenticated local UI; display coordinator readiness, workers, jobs, +tasks and safe errors; submit a valid small search pipeline check; poll it to a +terminal task state; and download/preview the coordinator-owned partial CSV. +The page must make the absence of final reduction impossible to miss. diff --git a/scimesh/worker/config.py b/scimesh/worker/config.py index 3732212..a0420eb 100644 --- a/scimesh/worker/config.py +++ b/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( diff --git a/scimesh/worker/runners.py b/scimesh/worker/runners.py index 0ef3e77..46f219a 100644 --- a/scimesh/worker/runners.py +++ b/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 diff --git a/tests/test_worker_daemon.py b/tests/test_worker_daemon.py index 6ad73c9..36f6cf6 100644 --- a/tests/test_worker_daemon.py +++ b/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()