From d0aeb7fc95cfa5c465ac816744b9b81621babfef Mon Sep 17 00:00:00 2001 From: Emil Date: Fri, 24 Jul 2026 15:13:59 +0300 Subject: [PATCH] Upgrade pipeline observability UI --- STATUS.md | 2 +- coordinator/README.md | 17 ++- .../storage/postgres/integration_test.go | 27 ++++ .../internal/storage/postgres/ui_read_repo.go | 9 +- coordinator/internal/transport/http/server.go | 1 + .../internal/transport/http/server_test.go | 54 ++++++- .../transport/http/templates/dashboard.html | 32 +++-- .../transport/http/templates/job.html | 37 +++-- .../transport/http/templates/new-job.html | 21 +-- coordinator/internal/transport/http/ui.go | 28 ++++ .../internal/transport/http/ui_test.go | 9 ++ coordinator/internal/usecase/ui.go | 136 +++++++++++++++--- .../internal/usecase/ui_internal_test.go | 19 +++ docs/web-interface-plan.md | 46 +++--- 14 files changed, 355 insertions(+), 83 deletions(-) create mode 100644 coordinator/internal/usecase/ui_internal_test.go diff --git a/STATUS.md b/STATUS.md index 1555d37..cd13285 100644 --- a/STATUS.md +++ b/STATUS.md @@ -38,7 +38,7 @@ real PostgreSQL smoke test) passed on 2026-07-24. | CTX-08 Distributed similarity-search | Implemented | Python planner resolves `query_id` once, creates deterministic shard plans, worker adapter emits exact partial top-k CSVs/metrics, and reducer matches the local reference. | | CTX-09 Reducer and final-result API | Implemented | Atomic `reducing` claim, deterministic coordinator-side top-k reducer, sanitized reducer failure, final artifact persistence, `result_uri`, and final CSV download. | | CTX-10 Distributed similarity-graph | Not started | Local reference exists. | -| CTX-11 Dashboard/operator view | Implemented | Protected local view: job/task/worker status, validated similarity-search upload, partial-artifact diagnostics, final-result download, and bounded polling. | +| CTX-11 Dashboard/operator view | Implemented | Protected live control room: recent-run/worker overview, real pipeline-stage visualization, shard attempts and safe failures, validated similarity-search upload, coordinator artifacts, final-result download, and bounded polling. | | CTX-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. | ## Next recommended assignment diff --git a/coordinator/README.md b/coordinator/README.md index eeebca2..02ef5d7 100644 --- a/coordinator/README.md +++ b/coordinator/README.md @@ -76,10 +76,19 @@ UI_AUTH_TOKEN='local-ui-secret' make up ``` The UI is disabled by default and never accepts the worker bearer token. -It shows recent jobs, task/worker state, diagnostic shard artifacts, and the -final CSV for completed similarity-search jobs. The coordinator enters -`reducing` after the last shard completes, then exposes the final deterministic -global top-k result when merging succeeds. +The **control room** shows live workers, recent runs, shard state/attempts, +safe failures, coordinator artifacts, and the final CSV for completed +similarity-search jobs. The job page follows the real stages: TSV accepted → +shards execute → workers return CSVs → `reducing` → final deterministic global +top-k result. It polls only its own coordinator read-model and never controls +or exposes worker processes. + +For a hands-on run, open `/ui`, choose **New similarity search**, select a +small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes +running in separate terminals. The detail page updates every two seconds and +stops polling after a completed, failed, or cancelled job. Download the +`final_result` artifact only after the job reaches **Completed**; shard partial +CSVs remain available as diagnostics. `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 diff --git a/coordinator/internal/storage/postgres/integration_test.go b/coordinator/internal/storage/postgres/integration_test.go index 7628a50..3252b00 100644 --- a/coordinator/internal/storage/postgres/integration_test.go +++ b/coordinator/internal/storage/postgres/integration_test.go @@ -133,6 +133,33 @@ func TestClaimReductionIsAtomic(t *testing.T) { } } +func TestUIReadRepoListsReducerFields(t *testing.T) { + pool := testPool(t) + job, _ := seedJob(t, pool, 1) + jobs := NewJobRepo(pool) + ctx := context.Background() + if err := jobs.UpdateStatus(ctx, job.ID, domain.JobReducing, nil); err != nil { + t.Fatal(err) + } + if claimed, err := jobs.ClaimReduction(ctx, job.ID, time.Now().UTC()); err != nil || !claimed { + t.Fatalf("claim reduction = (%v, %v)", claimed, err) + } + listed, err := NewUIReadRepo(pool).ListJobs(ctx, 20) + if err != nil { + t.Fatalf("list UI jobs: %v", err) + } + for _, item := range listed { + if item.ID != job.ID { + continue + } + if item.Status != domain.JobReducing || item.ReducerStartedAt == nil { + t.Fatalf("UI reducer projection = %+v", item) + } + return + } + t.Fatalf("seeded job %s is missing from UI list", job.ID) +} + // A job must land whole or not at all: a half-created job leaves chunks no // worker could ever complete. func TestCreateJobRollsBackOnFailure(t *testing.T) { diff --git a/coordinator/internal/storage/postgres/ui_read_repo.go b/coordinator/internal/storage/postgres/ui_read_repo.go index 81270b1..c759b2a 100644 --- a/coordinator/internal/storage/postgres/ui_read_repo.go +++ b/coordinator/internal/storage/postgres/ui_read_repo.go @@ -41,13 +41,12 @@ func (r *UIReadRepo) ListJobs(ctx context.Context, limit int) ([]domain.Job, err 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 { + if err := rows.Scan( + &j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt, + &j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt, + ); err != nil { return nil, err } - if inputURI != nil { - j.InputURI = *inputURI - } j.Status = domain.JobStatus(status) jobs = append(jobs, j) } diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 560c85b..a159ab2 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -81,6 +81,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler { 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/overview", s.handleUIOverviewJSON) 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) diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 97c7681..1146344 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -152,11 +152,36 @@ func TestUIRequiresDistinctCredentialAndRendersDashboard(t *testing.T) { t.Fatalf("UI status: %d", resp.StatusCode) } body, _ := io.ReadAll(resp.Body) - if !strings.Contains(string(body), "SciMesh operator dashboard") { + if !strings.Contains(string(body), "SciMesh control room") { t.Errorf("dashboard body missing title") } } +func TestUIOverviewReturnsLiveSafeProjection(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("create job: %d", code) + } + req, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/overview", nil) + req.SetBasicAuth("operator", uiToken) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var overview map[string]any + if err := json.NewDecoder(resp.Body).Decode(&overview); err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusOK || overview["active_jobs"].(float64) != 1 || overview["online_workers"].(float64) != 1 { + t.Fatalf("overview = (%d, %v)", resp.StatusCode, overview) + } + if _, leaked := overview["worker_auth_token"]; leaked { + t.Fatal("overview must not expose authentication configuration") + } +} + func TestUIDisabledReturnsNotFound(t *testing.T) { e := newEnvWithUIToken(t, healthy, "") resp := e.get(t, "/ui") @@ -437,6 +462,33 @@ func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) { if resp.StatusCode != http.StatusOK || string(body) != "rank,chembl_id,canonical_smiles,similarity\n1,A,CC,0.900000\n" { t.Fatalf("final result = (%d, %q)", resp.StatusCode, body) } + + uiRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID, nil) + uiRequest.SetBasicAuth("operator", uiToken) + uiResponse, err := http.DefaultClient.Do(uiRequest) + if err != nil { + t.Fatal(err) + } + defer uiResponse.Body.Close() + uiBody, _ := io.ReadAll(uiResponse.Body) + if uiResponse.StatusCode != http.StatusOK || !strings.Contains(string(uiBody), "Final result ready") { + t.Fatalf("final UI = (%d, %q)", uiResponse.StatusCode, uiBody) + } + + jsonRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/api/jobs/"+jobID, nil) + jsonRequest.SetBasicAuth("operator", uiToken) + jsonResponse, err := http.DefaultClient.Do(jsonRequest) + if err != nil { + t.Fatal(err) + } + defer jsonResponse.Body.Close() + var detail map[string]any + if err := json.NewDecoder(jsonResponse.Body).Decode(&detail); err != nil { + t.Fatal(err) + } + if jsonResponse.StatusCode != http.StatusOK || detail["final_result_available"] != true { + t.Fatalf("final UI JSON = (%d, %v)", jsonResponse.StatusCode, detail) + } } func TestForeignArtifactResultConflict(t *testing.T) { diff --git a/coordinator/internal/transport/http/templates/dashboard.html b/coordinator/internal/transport/http/templates/dashboard.html index e643cc7..af31dd1 100644 --- a/coordinator/internal/transport/http/templates/dashboard.html +++ b/coordinator/internal/transport/http/templates/dashboard.html @@ -4,20 +4,36 @@ - SciMesh operator dashboard + SciMesh control room
-

Local coordinator

SciMesh operator dashboard

See where a computation is and what should happen next.

Start a check
-
Similarity-search jobs produce a final CSV.Workers return shard-level candidates; after every shard succeeds, the coordinator deterministically merges them into one global top-k 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. Download resultWhen merging finishes, download the final CSV 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}}
+
+

Local scientific compute

SciMesh control room

Follow the real path from a molecular TSV to a globally reduced similarity result—without reading coordinator logs.

Live overview · refreshes every 2 seconds
+ ＋ New similarity search +
+
+
How a search becomes a result
01Upload TSVThe coordinator validates and slices the dataset.
02Run shardsWorkers fingerprint molecules and return shard top-k CSVs.
03Merge exactlyThe coordinator ranks retained candidates deterministically.
04Download CSVA checksum-protected global result is ready.
+
Active runs{{.ActiveJobs}}waiting, running, or merging
+
Available workers{{.OnlineWorkers}}recently registered
+
Finished runs{{.FinishedJobs}}in the latest 20
+
+ +

Recent computations

{{len .Jobs}} shown · newest first

{{range .Jobs}}
{{workloadLabel .Workload}}
{{.ID}}
{{statusLabel .Status}}
{{statusHint .Status}}
{{.Completed}} / {{.Total}} shards complete{{if gt .Failed 0}} · {{.Failed}} failed{{end}}
{{else}}
No computations yet.
Start a small similarity search, then keep one or more workers running to watch this dashboard come alive.
{{end}}
+

Worker fleet

Workers register themselves; this page never controls their processes.

{{range .Workers}}
{{.Name}}{{workerStatusLabel .Status}}
{{.ID}}

{{range .Capabilities}}{{.}}{{end}}

Last signal · {{time .LastHeartbeatAt}}

{{else}}
No worker is registered.
Start scimesh-worker in another terminal, then return here.
{{end}}
+ {{end}} diff --git a/coordinator/internal/transport/http/templates/job.html b/coordinator/internal/transport/http/templates/job.html index 596a12d..43a390a 100644 --- a/coordinator/internal/transport/http/templates/job.html +++ b/coordinator/internal/transport/http/templates/job.html @@ -4,25 +4,38 @@ - SciMesh job + SciMesh pipeline
- ← 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}}

-
{{if .FinalResultAvailable}}Final result ready
Download the final_result CSV below. It is the deterministic global top-k across all completed shards. The partial_result files remain available for diagnostics.{{else if eq .Status "reducing"}}Merging completed shards
Every shard has finished. The coordinator is building one deterministic global CSV; refresh in a moment to download it.{{else}}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 yet.{{end}}
-

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.
+ ← Back to control room +

{{workloadLabel .Workload}}

Live pipeline

One job, shown from accepted input through its final coordinator-owned scientific result.

Live · refreshes every 2 seconds
+
{{statusLabel .Status}}

{{statusHint .Status}}

Completed shards are preserved.

{{.Completed}} of {{.Total}} shards complete

{{.Total}}total shards
{{.Completed}}completed
{{.Pending}}waiting
{{add .Leased .Running}}with workers
{{.Failed}}failed
{{.Cancelled}}stopped
+ +

Pipeline stages

Each stage reflects coordinator state, not a simulated progress bar.

1TSV accepted

The coordinator stored the source and created shard tasks.

2Shards execute

{{.Completed}} of {{.Total}} candidate partitions are complete.

3Workers return CSVs

Workers upload a checked partial result for every completed shard.

4Global reduction

The coordinator waits until all shards are complete.

5Final CSV

Available only after deterministic reduction succeeds.

+ +

Run configuration

Allowlisted scientific parameters.

What is being computed?

{{range .Parameters}}
{{.Label}}{{.Value}}
{{else}}

No displayable parameters were supplied.

{{end}}

Result status

Safe operator guidance.

{{if .FinalResultAvailable}}

Final result ready

The coordinator merged shard candidates with exact scores and stored the global top-k CSV.

{{range .Artifacts}}{{if and (eq .Kind "final_result") .Downloadable}}Download final CSV{{end}}{{end}}{{else if eq .Status "reducing"}}

Merging completed shards

The final candidate heap is being ranked now. This page will update when the CSV is stored.

{{else if eq .Status "failed"}}

Run needs attention

{{if .ErrorMessage}}{{.ErrorMessage}}{{else}}One or more shards could not produce a final result. Review the task table below.{{end}}

{{else}}

Waiting for the final result

Partial CSVs are diagnostics. They become one global result only after every shard completes and reduction succeeds.

{{end}}
+ +

Shard activity

Every task is one input partition. The table refreshes while work is in progress.

{{range .Tasks}}{{else}}{{end}}
ShardStateAttemptWorker / leaseOutcome
#{{.ChunkIndex}}{{statusLabel .Status}}{{.Attempt}} / {{.MaxAttempts}}{{if .LeaseOwner}}{{.LeaseOwner}}{{if .LeaseExpiresAt}}
lease until {{time .LeaseExpiresAt}}{{end}}{{else}}—{{end}}		{{if .ErrorCode}}{{taskErrorLabel .ErrorCode}}
{{taskErrorHint .ErrorCode}}{{else if eq .Status "completed"}}Partial CSV uploaded{{else}}—{{end}}
No shard tasks are present yet.
+ +

Coordinator artifacts

All files stay coordinator-owned; checksums make downloads auditable.

{{range .Artifacts}}
{{if .Diagnostic}}Partial result · diagnostic{{else}}{{.Kind}}{{end}}
{{.Filename}}{{bytes .SizeBytes}}SHA-256 {{.SHA256}}{{if .Downloadable}}Download CSV{{end}}
{{else}}
Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.
{{end}}
+
Technical details

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

diff --git a/coordinator/internal/transport/http/templates/new-job.html b/coordinator/internal/transport/http/templates/new-job.html index 365df86..49eba44 100644 --- a/coordinator/internal/transport/http/templates/new-job.html +++ b/coordinator/internal/transport/http/templates/new-job.html @@ -4,27 +4,20 @@ - Create a check — SciMesh + New similarity search · 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
  • Keep at least one scimesh-worker running.
  • Use a small TSV for a hands-on check.
  • “Rows per shard” does not limit the file size. It splits the entire upload into tasks: a full ChEMBL TSV at 1,000 rows per shard creates thousands of tasks.
-
-

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.

- -
+ ← Back to control room

New computation

Similarity search, end to end

Upload a ChEMBL-style TSV. Workers calculate shard candidates; after every shard succeeds, SciMesh merges the exact global top-k into one final CSV.

+

Required columns: chembl_id and canonical_smiles.

Use a valid SMILES. The coordinator shares this exact query with every shard.

How many final molecules to retain.

Smaller shards make more visible tasks.

Leave blank to rank every valid candidate.

“Less” helps explore dissimilar molecules.

Only the first N data rows become shards; the original upload remains stored by the coordinator.

Ready to plan a run.
Select a TSV to see the file that will be sent to the coordinator.
diff --git a/coordinator/internal/transport/http/ui.go b/coordinator/internal/transport/http/ui.go index abcba26..b7203f9 100644 --- a/coordinator/internal/transport/http/ui.go +++ b/coordinator/internal/transport/http/ui.go @@ -26,6 +26,7 @@ var uiTemplates = template.Must(template.New("ui").Funcs(template.FuncMap{ "taskErrorLabel": uiTaskErrorLabel, "taskErrorHint": uiTaskErrorHint, "workerStatusLabel": uiWorkerStatusLabel, + "workerStatusClass": uiWorkerStatusClass, "workloadLabel": uiWorkloadLabel, "progressPercent": uiProgressPercent, "cancellable": uiCancellable, @@ -101,6 +102,8 @@ func uiWorkerStatusLabel(status string) string { switch status { case "online": return "Available" + case "busy": + return "Busy" case "offline": return "Offline" default: @@ -108,6 +111,17 @@ func uiWorkerStatusLabel(status string) string { } } +func uiWorkerStatusClass(status string) string { + switch status { + case "online": + return "success" + case "busy": + return "active" + default: + return "waiting" + } +} + // 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. @@ -207,6 +221,20 @@ func (s *Server) handleUIHome(w http.ResponseWriter, r *http.Request) { s.renderUI(w, "dashboard.html", view) } +// handleUIOverviewJSON is the bounded polling projection used by the operator +// dashboard. It intentionally returns only the safe UI read model, never +// worker tokens, storage keys, or database entities. +func (s *Server) handleUIOverviewJSON(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 + } + writeJSON(w, http.StatusOK, view) +} + func (s *Server) handleUINewJob(w http.ResponseWriter, r *http.Request) { s.renderUI(w, "new-job.html", nil) } diff --git a/coordinator/internal/transport/http/ui_test.go b/coordinator/internal/transport/http/ui_test.go index 034af7c..49fa89c 100644 --- a/coordinator/internal/transport/http/ui_test.go +++ b/coordinator/internal/transport/http/ui_test.go @@ -43,3 +43,12 @@ func TestUITaskErrorPresentationDoesNotExposeCommand(t *testing.T) { t.Error("error hint must explain the failure") } } + +func TestUIWorkerStatusPresentation(t *testing.T) { + if got := uiWorkerStatusLabel("busy"); got != "Busy" { + t.Errorf("busy worker label = %q", got) + } + if got := uiWorkerStatusClass("busy"); got != "active" { + t.Errorf("busy worker class = %q", got) + } +} diff --git a/coordinator/internal/usecase/ui.go b/coordinator/internal/usecase/ui.go index dac5fe0..8e4e030 100644 --- a/coordinator/internal/usecase/ui.go +++ b/coordinator/internal/usecase/ui.go @@ -2,6 +2,7 @@ package usecase import ( "context" + "fmt" "time" "github.com/google/uuid" @@ -21,17 +22,21 @@ type UIReadRepository interface { } 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"` + ID string `json:"id"` + Workload string `json:"workload"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + ReducerStartedAt *time.Time `json:"reducer_started_at,omitempty"` + ErrorCode string `json:"error_code,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + 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 { @@ -42,10 +47,20 @@ type TaskCard struct { MaxAttempts int `json:"max_attempts"` LeaseOwner string `json:"lease_owner,omitempty"` LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` ErrorCode string `json:"error_code,omitempty"` ErrorMessage string `json:"error_message,omitempty"` } +// ParameterCard is an intentionally small allowlist of run configuration that +// helps an operator verify what is being computed without exposing arbitrary +// job payloads to the browser. +type ParameterCard struct { + Label string `json:"label"` + Value string `json:"value"` +} + type ArtifactCard struct { ID string `json:"id"` Kind string `json:"kind"` @@ -65,14 +80,18 @@ type WorkerCard struct { } type DashboardView struct { - Jobs []JobCard - Workers []WorkerCard + Jobs []JobCard `json:"jobs"` + Workers []WorkerCard `json:"workers"` + ActiveJobs int `json:"active_jobs"` + FinishedJobs int `json:"finished_jobs"` + OnlineWorkers int `json:"online_workers"` } type JobDetailView struct { JobCard - Tasks []TaskCard `json:"tasks"` - Artifacts []ArtifactCard `json:"artifacts"` - FinalResultAvailable bool `json:"final_result_available"` + Tasks []TaskCard `json:"tasks"` + Artifacts []ArtifactCard `json:"artifacts"` + Parameters []ParameterCard `json:"parameters"` + FinalResultAvailable bool `json:"final_result_available"` } type Dashboard struct{ read UIReadRepository } @@ -98,10 +117,20 @@ func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, err return DashboardView{}, err } for _, job := range jobs { - out.Jobs = append(out.Jobs, jobCard(job, tasksByJob[job.ID])) + card := jobCard(job, tasksByJob[job.ID]) + out.Jobs = append(out.Jobs, card) + switch card.Status { + case string(domain.JobCompleted), string(domain.JobFailed), string(domain.JobCancelled): + out.FinishedJobs++ + default: + out.ActiveJobs++ + } } 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}) + if worker.Status == domain.WorkerOnline || worker.Status == domain.WorkerBusy { + out.OnlineWorkers++ + } } return out, nil } @@ -119,11 +148,27 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi if err != nil { return JobDetailView{}, err } - out := JobDetailView{JobCard: jobCard(*job, tasks), Tasks: make([]TaskCard, 0, len(tasks)), Artifacts: make([]ArtifactCard, 0, len(artifacts))} + workers, err := d.read.ListWorkers(ctx, 100) + if err != nil { + return JobDetailView{}, err + } + workerNames := make(map[string]string, len(workers)) + for _, worker := range workers { + workerNames[worker.ID.String()] = worker.Name + } + out := JobDetailView{ + JobCard: jobCard(*job, tasks), + Tasks: make([]TaskCard, 0, len(tasks)), + Artifacts: make([]ArtifactCard, 0, len(artifacts)), + Parameters: uiParameters(job.Parameters), + } 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} + card := TaskCard{ID: task.ID.String(), ChunkIndex: task.ChunkIndex, Status: string(task.Status), Attempt: task.Attempt, MaxAttempts: task.MaxAttempts, LeaseExpiresAt: task.LeaseExpiresAt, StartedAt: task.StartedAt, CompletedAt: task.CompletedAt} if task.LeaseOwner != nil { - card.LeaseOwner = *task.LeaseOwner + card.LeaseOwner = workerNames[*task.LeaseOwner] + if card.LeaseOwner == "" { + card.LeaseOwner = "Worker " + shortID(*task.LeaseOwner) + } } if task.ErrorCode != nil { card.ErrorCode = *task.ErrorCode @@ -158,7 +203,13 @@ func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID } func jobCard(job domain.Job, tasks []domain.Task) JobCard { - c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt} + c := JobCard{ID: job.ID.String(), Workload: job.Workload, CreatedAt: job.CreatedAt, CompletedAt: job.CompletedAt, ReducerStartedAt: job.ReducerStartedAt} + if job.ErrorCode != nil { + c.ErrorCode = *job.ErrorCode + } + if job.ErrorMessage != nil { + c.ErrorMessage = *job.ErrorMessage + } for _, task := range tasks { c.Total++ switch task.Status { @@ -180,3 +231,46 @@ func jobCard(job domain.Job, tasks []domain.Task) JobCard { c.Status = string(p.DeriveStatus()) return c } + +func uiParameters(parameters map[string]any) []ParameterCard { + keys := []struct { + key string + label string + }{ + {"query_smiles", "Target SMILES"}, + {"query_id", "Target ChEMBL ID"}, + {"top_k", "Global top-k"}, + {"threshold", "Similarity threshold"}, + {"threshold_direction", "Threshold direction"}, + } + out := make([]ParameterCard, 0, len(keys)) + for _, entry := range keys { + value, ok := parameters[entry.key] + if !ok { + continue + } + formatted, ok := formatUIParameter(value) + if ok { + out = append(out, ParameterCard{Label: entry.label, Value: formatted}) + } + } + return out +} + +func formatUIParameter(value any) (string, bool) { + switch typed := value.(type) { + case string: + return typed, true + case int, int64, float64, bool: + return fmt.Sprint(typed), true + default: + return "", false + } +} + +func shortID(value string) string { + if len(value) <= 8 { + return value + } + return value[:8] +} diff --git a/coordinator/internal/usecase/ui_internal_test.go b/coordinator/internal/usecase/ui_internal_test.go new file mode 100644 index 0000000..b16df70 --- /dev/null +++ b/coordinator/internal/usecase/ui_internal_test.go @@ -0,0 +1,19 @@ +package usecase + +import "testing" + +func TestUIParametersAreAllowlisted(t *testing.T) { + parameters := uiParameters(map[string]any{ + "query_smiles": "CCO", + "top_k": float64(20), + "internal_storage_key": "must-not-reach-browser", + "nested": map[string]any{"secret": "no"}, + }) + if len(parameters) != 2 { + t.Fatalf("parameters = %#v, want only two allowlisted values", parameters) + } + if parameters[0] != (ParameterCard{Label: "Target SMILES", Value: "CCO"}) || + parameters[1] != (ParameterCard{Label: "Global top-k", Value: "20"}) { + t.Fatalf("parameters = %#v", parameters) + } +} diff --git a/docs/web-interface-plan.md b/docs/web-interface-plan.md index 48b29d7..f6e9801 100644 --- a/docs/web-interface-plan.md +++ b/docs/web-interface-plan.md @@ -12,16 +12,28 @@ 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: +## Current delivered scope + +The initial operator UI and CTX-09 final reduction are now implemented. The +control room polls a bounded, coordinator-owned read model every two seconds +while a tab is visible. It shows the worker fleet, recent jobs, safe shard +diagnostics, the actual `reducing` phase, and final-result availability. A job +detail page renders the concrete pipeline stages—input accepted, shards, +worker CSVs, reduction, final CSV—from coordinator state and replaces task and +artifact views as work changes. All browser mutations remain limited to +validated dataset upload and operator cancellation. + +The interface must distinguish an in-progress distributed search from a run +whose reducer has produced a durable final result: | 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. | +| **In-progress run** | Upload, task creation, claim, heartbeat, artifact upload, task completion, retries, and shard diagnostics work end-to-end. | That the partial CSVs are a global scientific answer. | +| **Final run** | A reducer has produced a durable final CSV for the full job. | Available for `similarity-search` after CTX-09; graph remains unavailable until 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. +clear waiting or `reducing` stage until a final artifact exists and the job is +`completed`. ## 2. Constraints and decisions @@ -74,7 +86,7 @@ clear `Pipeline check — partial results` badge while a reducer is unavailable. | 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. | +| Final result | CTX-09 final artifact and download route exist | Show the `reducing` stage, then make the final CSV prominent only for `completed`. | | Distributed graph correctness | Planner/reducer unavailable | Do not advertise a multi-shard graph as final until CTX-10. | ## 5. Proposed structure @@ -186,11 +198,10 @@ Rules: 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. +The current upload form accepts `query_smiles`, because resolving a +cross-shard `query_id` has not yet been connected to coordinator uploads. The +detail page calls an artifact a **partial top-k CSV** until all shards are +complete and CTX-09 reduction stores the final global result. ### 8.3 Similarity graph @@ -283,7 +294,7 @@ checksum/size metadata display, and prominent partial/final labels. 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 +### WUI-06 — Final-result UX after CTX-09 — implemented **Depends on:** CTX-09 and WUI-05. @@ -365,8 +376,9 @@ that the interface exists today. ## 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. +The hand-testable release is complete when a clean local checkout can run a +trusted, authenticated local UI; display workers, jobs, pipeline stages, tasks +and safe errors; submit a valid small search; poll it through `reducing`; and +download the coordinator-owned final CSV only after completion. The page must +make the distinction between partial diagnostics and the final result +impossible to miss.