Compare commits

...
Author SHA1 Message Date
Emil ecc8944006 Drop an unused test helper
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 16:36:00 +03:00
Emil dc15e2d04b Warn in the wizard preflight when the installed scimesh version mismatches the binary 2026-08-03 16:35:37 +03:00
Emil ff1fc25d77 Log worker concurrency in the session goal 2026-08-03 15:45:11 +03:00
Emil 6b67326c3b Run claim-execute-upload loops concurrently in the worker agent
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 15:36:22 +03:00
Emil d57f8778ac Log the parallel search workload in the session goal 2026-08-03 15:20:10 +03:00
Emil 5b2d5b5f6e Add similarity-search-parallel: thread-pool scoring, byte-identical to similarity-search
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 15:12:18 +03:00
Emil b63441bb8c Revert "Warn about non-ASCII characters in text fields of the job form"
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
This reverts commit 91d049b4df.
2026-08-03 14:34:46 +03:00
Emil 91d049b4df Warn about non-ASCII characters in text fields of the job form 2026-08-03 14:34:12 +03:00
Emil 70c9e567ca Probe the busy port with a cancellable request
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 14:26:40 +03:00
Emil 81e854eb1d Reopen the already-running setup wizard instead of failing on a busy port 2026-08-03 14:26:17 +03:00
Emil 9a90d59ecf Complete the overnight plan: mark all items done in the session goal 2026-08-03 04:28:43 +03:00
Emil 2b6531fa7e Front the worker-key exchange through the coordinator in serve mode
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 04:24:52 +03:00
Emil ab922356e9 Apply migrations in the admin postgres integration tests regardless of run order 2026-08-03 04:16:37 +03:00
Emil fdea3a7152 Sync STATUS.md with the admin console, wizard, wheel pipeline and CTX-19/20 2026-08-03 04:15:00 +03:00
Emil c90c6b2ef6 Derive the package version from git tags via setuptools_scm 2026-08-03 04:14:23 +03:00
Emil ea0fb8c595 Silence the gosec warning for the wizard's own log path 2026-08-03 04:12:15 +03:00
Emil 357ed34714 Add worker stats to the wizard, artifact pruning and offline-worker removal to the admin console 2026-08-03 04:11:43 +03:00
Emil 7ea7c52325 Cover the admin read model and workload settings with postgres integration tests 2026-08-03 04:07:15 +03:00
Emil 8db1f9846c Restyle the remaining operator pages into the admin design system 2026-08-03 04:05:19 +03:00
Emil 7d1aacb19f Reject blank worker names and probe the real credential and venv in --check 2026-08-03 04:01:45 +03:00
Emil 0be114c8da Write the overnight plan into the session goal 2026-08-03 03:58:19 +03:00
Emil 6956ea6b1f Update the session goal with the final polish round 2026-08-03 03:49:41 +03:00
Emil 5790ae78a7 Create the worker directory before the wizard downloads the release wheel
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 03:46:10 +03:00
Emil 40ff688416 Probe the managed venv in the wizard preflight and document RDKit's X11 deps
coordinator / test (push) Waiting to run
python / test (push) Waiting to run
release / binaries (amd64, darwin) (push) Waiting to run
release / binaries (amd64, linux) (push) Waiting to run
release / binaries (amd64, windows) (push) Waiting to run
release / binaries (arm64, darwin) (push) Waiting to run
release / binaries (arm64, linux) (push) Waiting to run
release / binaries (arm64, windows) (push) Waiting to run
release / wheel (push) Waiting to run
release / release (push) Blocked by required conditions
release / image (push) Waiting to run
users / test (push) Waiting to run
2026-08-03 03:23:28 +03:00
50 changed files with 2356 additions and 87 deletions
+6 -10
View File
@@ -57,24 +57,20 @@ jobs:
wheel:
runs-on: ubuntu-latest
steps:
# Full history: setuptools_scm derives the package version from the
# release tag (v1.1.0-alpha.16 -> 1.1.0a16), so the wheel is
# version-locked to the binaries of the same release.
- uses: actions/checkout@v4
with:
fetch-depth: 0
fetch-tags: true
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: build the scimesh wheel
env:
VERSION: ${{ github.ref_name }}
run: |
# The tag (v1.1.0-alpha.10) becomes the package version in its
# PEP 440 form (1.1.0a10); the wheel is then version-locked to the
# binaries of the same release.
WHEEL_VERSION="${VERSION#v}"
WHEEL_VERSION="${WHEEL_VERSION/-alpha./a}"
WHEEL_VERSION="${WHEEL_VERSION/-beta./b}"
WHEEL_VERSION="${WHEEL_VERSION/-rc./rc}"
sed -i "s/^version = .*/version = \"${WHEEL_VERSION}\"/" pyproject.toml
python -m pip install --quiet build
python -m build --wheel --outdir dist
ls -la dist/
+77 -17
View File
@@ -1,22 +1,82 @@
# Session Goal
COMPLETED
выполни план на ночь (полный план ниже — автономное исполнение, пользователь спит)
выполни полностью намеченный сейчас план. автономно
## Ночная сессия — полный план (пользователь спит, 2026-08-03)
## Plan
1. Фаза 1 — SQLite-хранилище: `coordinator/internal/storage/sqlite` (все порты, TxManager, миграции, `SCIMESH_DB=sqlite|postgres`, тесты).
2. Фаза 2 — Встроенный userservice: перенос `users/internal/*` в `coordinator/internal/userservice/` (sqlite-хранилище), запуск на 127.0.0.1, BootstrapAdmin.
3. Фаза 3 — `coordinator serve` (data-dir, всё-в-одном, --workers N, --open) + subcommand `coordinator agent`.
4. Фаза 4 — Управляемый venv + install.sh/install.ps1 + ассеты релиза.
5. Документация: mkdocs, README, PLAN.md (CTX-17 done, CTX-18), STATUS.md.
6. Проверка: полный E2E без внешних сервисов + все тесты/lint/vet.
### 1. Баги (реальные, найденные в проде)
- [ ] **Пустое имя воркера**: регистрация принимает `name=""` (в БД пользователя было 15 таких). Фикс: валидация в `domain.NewWorker` (TrimSpace != "") + регрессионный тест. Частично начат — в `worker_test.go` сломан тест (`fixedTime` vs `testNow`), доделать.
- [ ] **`worker-agent --check` не видит venv**: проверяет только системный python3; после Install воркер работает через `~/.scimesh-worker/venv`. Выровнять с визардом (пробовать venv, если он есть).
## Progress
- ВСЕ ФАЗЫ ВЫПОЛНЕНЫ И ЗАПУШЕНЫ:
- 9883def — SQLite-бэкенд (SCIMESH_DB=sqlite|postgres, миграции, тесты).
- 1473bbe — встроенный userservice + serve/agent subcommands.
- c06d867 — install.sh/install.ps1 + make serve.
- 215325a — документация (README, mkdocs, PLAN CTX-17/CTX-18, STATUS).
- E2E «чистая машина»: `coordinator serve` → health/login (embedded userservice) → molwt-filter джоб через локального агента → результат byte-точный.
- Верификация: 208 pytest, pyright 0, 18 Go-пакетов ok, gofmt чист, vet чист, golangci 0 issues, postgres integration ok, все CI-раны success.
### 2. Визуальный долг — рестайлинг старых страниц в дизайн-систему админки (#0b0e13, карточки, pill-статусы, кнопки)
- [ ] `new-job.html` (форма запуска вычислений, UIElement-поля сохранить)
- [ ] `job.html` (детали джоба: шарды, артефакты, прогресс, JS-логику сохранить)
- [ ] `workloads.html`
- [ ] `add-worker.html`
- [ ] `profile.html`
- [ ] Браузерная проверка каждой страницы (playwright).
### 3. Технический долг (паритет движков, обещанный планом)
- [ ] **Postgres integration-тесты для admin-методов**: `SetTrust`, `WorkloadSettings` (Get/List/Set), `ListJobsPaginated`, метрики (`JobCountsByDay`, `TaskStats`, `ArtifactSizeByKind`, `DatabaseSizeBytes`) — сейчас покрыт только sqlite. (build tag integration, docker postgres как в CI.)
### 4. Фичи из плана (v2-задел)
- [ ] **Статистика воркера в визарде**: статусная страница — claimed/completed/failed/heartbeat, парсинг из `worker.log` (обещано в docs/ui-admin-worker-plan.md §4).
- [ ] **Admin: prune артефактов** (danger zone): удаление артефактов completed-джобов старше N дней (строки + blob-файлы), освобождает место; кнопка в Settings с подтверждением.
- [ ] **Admin: удаление offline-воркеров** (устаревшие, как 15 мусорных) — кнопкой в Workers вместо SQL.
### 5. Гигиена
- [ ] **`pyproject.toml` версия 0.1.0 → setuptools_scm** (версия из git-тегов), убрать sed-костыль из release.yml; локальный `pip install -e .` покажет правильную версию.
- [ ] Синхронизация STATUS.md / README с фактическим состоянием (админка, визард, wheel, авто-открытие, restyle).
### 6. Верификация и релиз
- [ ] Полный гейт: `go test -race ./...`, golangci-lint `--build-tags=integration`, pytest.
- [ ] Браузерные проверки: админка (все секции), визард (install+start), рестайленные страницы.
- [ ] E2E quorum-флоу (untrusted воркер через worker key → джоб требует quorum) в Docker, если останется время.
- [ ] Релиз `v1.1.0-alpha.16` (бинарники + wheel), проверка ассетов.
- [ ] Обновить этот файл: прогресс по пунктам, COMPLETED в конце.
## Plan (предыдущая задача — выполнена)
17. Docker E2E пайплайна «install как человек → serve → визард → воркер → джоб» — выполнено, см. Progress ниже.
## Progress (день: конкурентность в агенте — параллелизм через нашу архитектуру)
- ✅ По просьбе «реализовать параллелизм через нашу архитектуру»: `WORKER_CONCURRENCY` (поле `concurrency` в конфиге визарда, шаг Machine) — агент регистрируется один раз и ведёт N циклов claim→выполнение→upload под одним worker id. N шардов обрабатываются параллельно на одной машине, используя таски координатора как единицу параллелизма; SDK не менялся.
- ✅ Реализация: `Config.Concurrency` + env; `Daemon.RunForever` → register once → N goroutine-циклов (общий счётчик MaxTasks под мьютексом); визард: поле «Concurrent task loops» + конфиг; тест с маркер-скриптом доказывает 3 параллельных исполнения (race-тесты зелёные после мьютекса в fake).
- ✅ Измерено на релизном коде в Docker (один воркер, 8 шардов): concurrency=1 → 12s; concurrency=4 → 4s (**3×**). Плюс прежний `similarity-search-parallel` (потоки внутри шарда) композируется с конкурентностью.
- ✅ Релиз v1.1.0-alpha.20 (бинарники + wheel); полный гейт: race + lint 0 issues + pytest 213.
- ✅ Бинарник worker-agent на машине пользователя обновлён до alpha.20.
- GIL-высвобождения в RDKit нет (проверено: `RDKIT_ALLOW_GIL_RELEASE` не помогает), поэтому внутришардовые потоки не ускоряют чистый RDKit-путь — конкурентность задач это и компенсирует.
## Progress (утро: similarity-search-parallel)
- ✅ Новый workload `similarity-search-parallel@1.0.0` (отдельная версия, как просил пользователь): подкласс `SimilaritySearchSDKWorkload` + параллельное ядро `search_parallel/core.py` — fingerprinting+скоринг шарда через `ThreadPoolExecutor` (параметр `threads`, default CPU count); `pool.map` сохраняет порядок строк, поэтому merge идентичен последовательному (`_HeapEntry`) и результат **байт-в-байт** равен `similarity-search` при любом числе потоков.
- ✅ Тесты: байт-в-байт vs эталон для threads 1/2/4 с намеренными связями (изомеры, дубликаты), executor-прогон, валидация параметров, регистрация манифеста; 213 pytest зелёные.
- ✅ Экспортирован в каталог координатора (5 ворклоадов), Go-тесты зелёные; release v1.1.0-alpha.19 (бинарники + wheel `scimesh-1.1.0a19`).
- ✅ Распределённый E2E в Docker: воркер с wheel a19, джоб similarity-search-parallel (threads=4) → completed → результат байт-в-байт = локальному эталону.
- ✅ Бинарники на машине пользователя обновлены до alpha.19 (для появления ворклоада в UI нужен рестарт serve + переустановка рантайма воркера).
- Примечание: в CPython потоки не ускоряют чистый RDKit-путь (GIL), но структура готова к ядрам, отпускающим GIL (numpy и т.п.); при желании можно добавить процесс-пул как отдельный workload в будущей версии протокола.
## Progress (ночная сессия)
-**П.1 Пустое имя воркера**: `domain.NewWorker` нормализует/отклоняет пустое имя + `TestNewWorkerRejectsBlankName` (починен `fixedTime``testNow`).
-**П.2 `--check`**: пробует managed venv (если установлен) + реальный пробинг учётки (exchange ключа / claim-пробa) — `CheckAuth` + тесты; на машине пользователя: `✓ auth: credential accepted`, venv python, scimesh installed.
-**П.3 Рестайлинг**: единый CSS-partial `ui-base.html` (дизайн-система админки), все 5 страниц (new-job, job, workloads, add-worker, profile) переведены, проверены в браузере без console-ошибок.
-**П.4 Postgres integration**: admin-методы (SetTrust, ListJobsPaginated, TaskCounts, byDay/byWorkload, TaskStats, ArtifactSize, DB size, WorkloadSettings) + `ensureMigrated` для порядка запуска; весь suite зелёный.
-**П.5 Статистика воркера в визарде**: лог `task claimed` в агента + парсинг registered/claimed/completed/failed → `/api/status.stats` + карточки в статусной странице + тест.
-**П.6 Prune артефактов**: `JobRepository.ListCompletedBefore/Delete` (sqlite+postgres+memstore), usecase `PruneArtifacts` (каскад + blob-файлы), `POST /ui/admin/api/prune`, кнопка в Settings, тесты (sqlite+usecase); E2E: 200, freed bytes.
-**П.7 Удаление offline-воркеров**: `WorkerRepository.Delete` + `Admin.RemoveWorker` (только offline) + `POST /ui/admin/api/workers/{id}/remove` + кнопка в Workers + тест; E2E: 204, строка удалена.
-**П.8 setuptools_scm**: `dynamic = ["version"]`, CI-джоба wheel без sed (fetch-depth 0); локальная проверка: wheel на теге = `scimesh-1.1.0a16-py3-none-any.whl` (совпадает с Go-нормализацией); релизный ассет подтверждён.
-**П.9 Docs**: STATUS.md синхронизирован (админка, визард, wheel, CTX-19/20 implemented).
-**П.10 (доп.) Баг в serve-режиме**: worker-key exchange был недоступен снаружи (userservice на loopback) — добавлен прокси `POST /worker-tokens/exchange` на координаторе, `PublicUserserviceURL=""` + fallback на origin в add-worker. Проверено E2E.
-**П.10 Quorum E2E (Docker)**: 2 untrusted-воркера с разными ключами (alice/bob) → джоб completed 3/3, в task_results по 2 голоса от разных владельцев с одинаковым sha256 → результат байт-в-байт = локальному эталону.
-**Финальный гейт**: `go test -race ./...` ✅, golangci-lint 0 issues ✅, pytest 208 ✅, postgres integration ✅, Windows кросс-сборка ✅.
-**Релиз v1.1.0-alpha.16** (бинарники + wheel `scimesh-1.1.0a16`), все воркфлоу success; бинарники на машине пользователя обновлены до alpha.16.
## Progress (прошлая работа — выполнено)
- ✅ Релизы alpha.12–15: фикс версии визарда, venv task_runner, preflight через venv, MkdirAll при скачивании wheel, кнопка Install в шаблоне.
- ✅ Docker E2E: координатор+воркер контейнеры, установка install.sh, визард (config→install→start), воркер online, джоб completed, результат байт-в-байт = локальному эталону.
- ✅ На машине пользователя: визард alpha.15, правильный токен, venv из wheel, воркер emil-pc online, 15 пустых воркеров вычищены из БД.
- ✅ Гейт: race + lint + pytest 208.
## Completion
COMPLETED — ночной план выполнен полностью (10 пунктов + 2 найденных бага, включая E2E quorum на релизном коде). Все гейты зелёные, релиз v1.1.0-alpha.16 опубликован.
## Completion (предыдущая задача)
COMPLETED — пайплайн доведён до рабочего состояния и проверен на релизных артефактах v1.1.0-alpha.14.
+4
View File
@@ -225,6 +225,10 @@ The coordinator and worker agent are Go modules under `coordinator/` and `users/
cd coordinator && make coordinator agent && go test ./...
```
On headless servers (no desktop environment), RDKit needs a few X11
libraries that desktops already ship: `sudo apt-get install -y libxrender1
libxext6 libxcursor1 libxfixes3 libxi6 libxrandr2`.
`make check` runs the full gate: vet, lint, race tests, the PostgreSQL
integration suite, and the two-worker end-to-end smoke script.
+27 -3
View File
@@ -1,7 +1,8 @@
# SciMesh Status
**Updated:** 2026-08-02
**Branch baseline:** `main`; this revision adds the single-binary platform.
**Updated:** 2026-08-03
**Branch baseline:** `main`; this revision adds the admin console, the worker
setup wizard, and release-shipped Python wheels.
## Current state
@@ -42,6 +43,27 @@ Users can create and revoke worker keys for self-service Worker Agent
enrollment. Untrusted workers require quorum agreement from distinct owners on
the complete result-artifact SHA-256 before a task is accepted.
**Admin console** (`/ui/admin`, admin role only): system/storage/health,
paginated jobs with owner resolution, worker fleet with trust controls and
offline-worker removal, users and worker keys, workload enable/disable
(persisted in both engines, enforced at submit time), 7-day metrics, and
settings with an audited worker-token reveal and artifact pruning for old
finished jobs.
**Worker setup wizard** (`worker-agent setup`, loopback `127.0.0.1`): browser
wizard that configures the coordinator URL and credential, runs a preflight
(coordinator reachability, real credential probe, python, scimesh), installs
the version-locked `scimesh` wheel from the release into a managed venv
(`~/.scimesh-worker/venv`), pins the venv task runner, and starts/stops the
worker with live log and parsed claim/completed/failed counters. The release
pipeline now also ships the Python wheel (`scimesh-<ver>-py3-none-any.whl`,
version derived from the git tag via setuptools_scm), and both `coordinator
serve` and the wizard install it automatically.
The full pipeline was verified end-to-end in containers on release artifacts:
installer → serve → wizard → registration → real similarity-search job →
byte-exact result vs the local reference.
## Milestone tracker
| CTX | Status | Notes |
@@ -57,7 +79,9 @@ the complete result-artifact SHA-256 before a task is accepted.
| CTX-08 Distributed similarity-search | Implemented | Planner/worker/reducer match the local reference byte-exactly. |
| CTX-09 Reducer and final-result API | Implemented | Atomic `reducing` claim, deterministic coordinator-side reducers (`top-k` and `ordered-concat`), sanitized failure, final artifact, `result_uri`. |
| CTX-10 Distributed similarity-graph | Not started | Local reference exists; the SDK-built local graph workload enforces the pair-coverage invariant. |
| CTX-11 Dashboard/operator view | Implemented | Protected live control room, workload library, workload-agnostic "New computation" form (SDK-declared `UIElement`s), MkDocs at `/ui/docs/`, final-result download. |
| CTX-11 Dashboard/operator view | Implemented | The demo control room was removed; `/ui` lands on the admin console. Job form/detail, workload library, add-machine and profile pages share the admin design system; MkDocs at `/ui/docs/`. |
| CTX-19 Coordinator Admin UI | Implemented | `/ui/admin` console (see above), bounded read models, admin-only routes, sqlite+postgres parity with integration tests. |
| CTX-20 Worker Setup UI | Implemented | `worker-agent setup` wizard with runtime installer, `--config`, `--check` (credential + venv probing), live status. |
| CTX-12 Reliability, security, CI | In progress | vet, gofmt, race tests, golangci-lint (0 issues), PostgreSQL integration, and smoke checks exist. |
| CTX-15 User Service and access control | Implemented | User/owner scoping, verified contributors, worker keys, quorum; also embeddable (`coordinator serve`). |
| CTX-16 Workload SDK foundation | Implemented | Strict immutable manifests/plans/artifacts, digest/trust-pinned tasks, negotiation, verifier primitives, conformance harness. |
+1
View File
@@ -173,6 +173,7 @@ func runWithConfig(cfg infra.Config) error {
GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)),
GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
Dashboard: usecase.NewDashboard(uiReadRepo, catalog),
PruneArtifacts: usecase.NewPruneArtifacts(jobRepo, uiReadRepo, blobStore, clk),
PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore),
Admin: usecase.NewAdmin(deps.adminReadRepo, uiReadRepo, workerRepo, deps.settingsRepo, catalog,
usecase.AdminNodeInfo{
+57 -1
View File
@@ -10,11 +10,13 @@ import (
"errors"
"flag"
"fmt"
"io"
"log/slog"
"net/http"
"os"
"os/exec"
"os/signal"
"strings"
"syscall"
"time"
@@ -58,7 +60,17 @@ func main() {
fmt.Println("check: no coordinator URL (pass --coordinator-url or set COORDINATOR_URL)")
os.Exit(1)
}
report := agent.RunCheck(ctx, url)
// Probe the managed venv when the wizard has installed it: workloads
// run with that interpreter, so checking the bare system python3
// would report a false negative.
checkPython, checkToken, checkKey, checkUsers := "", "", "", ""
if configPath := checkConfigPath(); configPath != "" {
checkPython = agent.VenvPython(configPath)
if config, err := agent.LoadConfigFile(configPath); err == nil {
checkToken, checkKey, checkUsers = config.Token, config.WorkerKey, config.UserserviceURL
}
}
report := agent.RunCheck(ctx, url, checkPython, checkToken, checkKey, checkUsers)
printCheck(report)
if !report.Coordinator.OK || !report.Python.OK || !report.Scimesh.OK {
os.Exit(1)
@@ -87,6 +99,15 @@ func main() {
}
}
// checkConfigPath resolves where the wizard's config would be, honouring
// SCIMESH_WORKER_CONFIG like the rest of the agent.
func checkConfigPath() string {
if env := os.Getenv("SCIMESH_WORKER_CONFIG"); env != "" {
return env
}
return agent.DefaultConfigPath()
}
// loadConfig prefers a --config file; environment variables override the file
// (see agent.ConfigFile.Config). Without a file, the plain environment path is
// used exactly as before.
@@ -147,6 +168,16 @@ func runSetup(args []string) int {
})
listener, err := server.Listen()
if err != nil {
// A wizard may already be running on this port (left open, or a
// second terminal). If it is ours, opening the browser is the
// friendlier outcome than failing the command.
if existing := wizardAlreadyRunning(*port); existing != "" {
logger.Info("the setup wizard is already running", "url", existing)
if !*noOpen {
openBrowser(existing)
}
return 0
}
logger.Error("setup wizard could not bind the loopback port", "err", err)
return 1
}
@@ -180,3 +211,28 @@ func openBrowser(url string) {
return
}
}
// wizardAlreadyRunning probes the requested loopback port and returns its URL
// when it serves the setup wizard page, or "" when it does not (another
// process, or nothing at all).
func wizardAlreadyRunning(port int) string {
url := fmt.Sprintf("http://127.0.0.1:%d/", port)
client := http.Client{Timeout: 2 * time.Second}
req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, url, nil)
if err != nil {
return ""
}
resp, err := client.Do(req)
if err != nil {
return ""
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
return ""
}
body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
if err != nil || !strings.Contains(string(body), "SciMesh Worker · Setup") {
return ""
}
return url
}
+113 -11
View File
@@ -9,6 +9,8 @@ import (
"runtime"
"strings"
"time"
guuid "github.com/google/uuid"
)
// CheckItem is one line of the preflight report the setup wizard shows.
@@ -70,18 +72,26 @@ func CheckCoordinator(ctx context.Context, url string, timeout time.Duration) Ch
return report
}
// CheckEnvironment verifies the local runtime: Python present and the scimesh
// package importable.
// CheckEnvironment verifies the local runtime against the python3 found on
// PATH.
func CheckEnvironment(ctx context.Context) CheckReport {
report := CheckReport{Agent: Version}
python, err := exec.LookPath("python3")
if err != nil {
report.Python = CheckItem{Name: "python", OK: false, Detail: "python3 not found on PATH"}
return report
return CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: false, Detail: "python3 not found on PATH"}}
}
report.Python = CheckItem{Name: "python", OK: true, Detail: python}
//nolint:gosec // G204: python comes from LookPath, the argument list is constant
cmd := exec.CommandContext(ctx, python, "-c", "import scimesh; print(scimesh.__version__ if hasattr(scimesh, '__version__') else 'installed')")
return CheckEnvironmentWithPython(ctx, python)
}
// CheckEnvironmentWithPython verifies the local runtime against a specific
// interpreter — the wizard's managed venv python when the runtime installer
// has created one, so the preflight reflects what the worker will actually
// execute with.
func CheckEnvironmentWithPython(ctx context.Context, python string) CheckReport {
report := CheckReport{Agent: Version, Python: CheckItem{Name: "python", OK: true, Detail: python}}
// The version comes from importlib.metadata, so the wizard can compare the
// installed package with the binary version and offer an upgrade.
//nolint:gosec // G204: python is a resolved interpreter path, the argument list is constant
cmd := exec.CommandContext(ctx, python, "-c", "import importlib.metadata as m; print(m.version('scimesh'))")
out, err := cmd.Output()
if err != nil {
// The worker executes workloads by spawning scimesh's task runner, so
@@ -96,10 +106,22 @@ func CheckEnvironment(ctx context.Context) CheckReport {
}
// RunCheck combines the coordinator probe and the local environment probe; it
// is the body behind `worker-agent --check` and the wizard's test step.
func RunCheck(ctx context.Context, coordinatorURL string) CheckReport {
// is the body behind `worker-agent --check` and the wizard's test step. A
// non-empty python overrides the interpreter probed for the scimesh package
// (the managed venv after a runtime install).
// RunCheck combines the coordinator probe, a credential probe and the local
// environment probe; it is the body behind `worker-agent --check` and the
// wizard's test step. A non-empty python overrides the interpreter probed for
// the scimesh package (the managed venv after a runtime install).
func RunCheck(ctx context.Context, coordinatorURL, python, token, workerKey, userserviceURL string) CheckReport {
report := CheckCoordinator(ctx, coordinatorURL, 15*time.Second)
env := CheckEnvironment(ctx)
report.Auth = CheckAuth(ctx, coordinatorURL, token, workerKey, userserviceURL)
var env CheckReport
if python != "" {
env = CheckEnvironmentWithPython(ctx, python)
} else {
env = CheckEnvironment(ctx)
}
report.Python = env.Python
report.Scimesh = env.Scimesh
return report
@@ -111,3 +133,83 @@ var Version = "dev"
// Platform is the host platform string shown on the wizard.
func Platform() string { return runtime.GOOS + "/" + runtime.GOARCH }
// CheckAuth verifies the configured credential against the coordinator
// without mutating anything: with a worker key it first exchanges it at the
// userservice for a short-lived JWT, then it probes /tasks/claim with a
// throwaway worker id and no capabilities. A 401 anywhere means the
// credential was rejected; any other status proves it was accepted.
func CheckAuth(ctx context.Context, url, token, workerKey, userserviceURL string) CheckItem {
item := CheckItem{Name: "auth"}
if token == "" && workerKey == "" {
item.OK = true
item.Detail = "no credential configured — will be checked at registration"
return item
}
client := &http.Client{Timeout: 30 * time.Second}
if workerKey != "" && userserviceURL != "" {
payload, _ := json.Marshal(map[string]string{"key": workerKey})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(userserviceURL, "/")+"/worker-tokens/exchange", strings.NewReader(string(payload)))
if err != nil {
item.OK = false
item.Detail = "invalid userservice URL"
return item
}
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
item.OK = false
item.Detail = "userservice unreachable: " + err.Error()
return item
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode == http.StatusUnauthorized {
item.OK = false
item.Detail = "worker key rejected by the userservice"
return item
}
if resp.StatusCode != http.StatusOK {
item.OK = false
item.Detail = fmt.Sprintf("userservice exchange: HTTP %d", resp.StatusCode)
return item
}
var exchanged struct {
Token string `json:"token"`
}
if err := json.NewDecoder(resp.Body).Decode(&exchanged); err != nil || exchanged.Token == "" {
item.OK = false
item.Detail = "userservice exchange returned no token"
return item
}
token = exchanged.Token
}
if token == "" {
item.OK = false
item.Detail = "no usable credential after the key exchange"
return item
}
payload, _ := json.Marshal(map[string]any{"worker_id": guuid.NewString(), "capabilities": []string{}})
req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(url, "/")+"/tasks/claim", strings.NewReader(string(payload)))
if err != nil {
item.OK = false
item.Detail = "invalid coordinator URL"
return item
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+token)
resp, err := client.Do(req)
if err != nil {
item.OK = false
item.Detail = "coordinator unreachable: " + err.Error()
return item
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode == http.StatusUnauthorized {
item.OK = false
item.Detail = "token rejected by the coordinator"
return item
}
item.OK = true
item.Detail = "credential accepted"
return item
}
+68
View File
@@ -0,0 +1,68 @@
package agent
import (
"context"
"net/http"
"net/http/httptest"
"testing"
)
func TestCheckAuthAcceptsToken(t *testing.T) {
stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/tasks/claim" && r.Header.Get("Authorization") == "Bearer good-token" {
w.WriteHeader(http.StatusNoContent)
return
}
w.WriteHeader(http.StatusUnauthorized)
}))
defer stub.Close()
item := CheckAuth(context.Background(), stub.URL, "good-token", "", "")
if !item.OK {
t.Errorf("good token: %+v", item)
}
item = CheckAuth(context.Background(), stub.URL, "bad-token", "", "")
if item.OK {
t.Error("bad token must fail")
}
}
func TestCheckAuthNoCredentialIsNotAFailure(t *testing.T) {
item := CheckAuth(context.Background(), "http://coord:8080", "", "", "")
if !item.OK {
t.Errorf("no credential must not fail the preflight: %+v", item)
}
}
func TestCheckAuthWorkerKeyExchange(t *testing.T) {
var exchanged bool
users := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/worker-tokens/exchange" {
t.Errorf("unexpected userservice path %q", r.URL.Path)
}
exchanged = true
_, _ = w.Write([]byte(`{"token":"jwt-after-exchange"}`))
}))
defer users.Close()
coord := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") == "Bearer jwt-after-exchange" {
w.WriteHeader(http.StatusNoContent)
return
}
w.WriteHeader(http.StatusUnauthorized)
}))
defer coord.Close()
item := CheckAuth(context.Background(), coord.URL, "", "smk_key", users.URL)
if !item.OK || !exchanged {
t.Errorf("key exchange flow: %+v exchanged=%v", item, exchanged)
}
rejected := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusUnauthorized)
}))
defer rejected.Close()
if item := CheckAuth(context.Background(), coord.URL, "", "smk_bad", rejected.URL); item.OK {
t.Error("rejected key must fail the preflight")
}
}
+17
View File
@@ -32,6 +32,10 @@ type Config struct {
TaskRunner []string // command + args; defaults to python -m scimesh.worker.task
MaxTasks int // 0 = unlimited
ExitWhenIdle bool
// Concurrency is how many claim→execute→upload loops run in parallel
// under one worker id: N shards processed concurrently on one machine,
// using the coordinator's own task pipeline as the parallel unit.
Concurrency int
}
func envList(name string) ([]string, error) {
@@ -145,9 +149,22 @@ func LoadConfig() (*Config, error) {
TaskRunner: runner,
MaxTasks: maxTasks,
ExitWhenIdle: os.Getenv("EXIT_WHEN_IDLE") == "1",
Concurrency: envInt("WORKER_CONCURRENCY", 1),
}, nil
}
func envInt(name string, fallback int) int {
raw := os.Getenv(name)
if raw == "" {
return fallback
}
parsed, err := strconv.Atoi(raw)
if err != nil || parsed < 1 {
return fallback
}
return parsed
}
func durationEnv(name string, fallback time.Duration) (time.Duration, error) {
raw := os.Getenv(name)
if raw == "" {
+22
View File
@@ -21,6 +21,7 @@ type ConfigFile struct {
WorkerName string `json:"worker_name,omitempty"`
CPUCount int `json:"cpu_count"`
MemoryMB int `json:"memory_mb"`
Concurrency int `json:"concurrency,omitempty"`
TaskRunner []string `json:"task_runner,omitempty"`
}
@@ -103,6 +104,10 @@ func (f *ConfigFile) Config() (*Config, error) {
if config.MemoryMB < 0 {
config.MemoryMB = 0
}
config.Concurrency = f.Concurrency
if config.Concurrency < 1 {
config.Concurrency = 1
}
if len(f.TaskRunner) > 0 {
config.TaskRunner = f.TaskRunner
}
@@ -132,3 +137,20 @@ func SaveConfigFile(path string, file ConfigFile) error {
}
return nil
}
// VenvPython returns the managed venv python next to the given config file,
// or "" when the runtime installer has not created one yet. Mirrors the
// wizard's probe so `worker-agent --check` and the preflight agree on what
// interpreter will actually execute workloads.
func VenvPython(configPath string) string {
dir := filepath.Dir(configPath)
for _, candidate := range []string{
filepath.Join(dir, "venv", "bin", "python"),
filepath.Join(dir, "venv", "Scripts", "python.exe"),
} {
if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
return candidate
}
}
return ""
}
+48 -7
View File
@@ -37,15 +37,52 @@ func NewDaemon(config *Config, client *Client, runner *TaskRunner, log *slog.Log
return &Daemon{config: config, client: client, runner: runner, log: log}
}
// RunForever loops until interrupted, idle-exit, or max tasks.
// RunForever registers once, then runs the claim→execute→upload loop
// concurrently under one worker id. With Concurrency > 1, several shards are
// processed in parallel on this machine, using the coordinator's own task
// pipeline as the parallel unit.
func (d *Daemon) RunForever() error {
if !d.registered {
if err := d.register(); err != nil {
return err
}
}
workers := d.config.Concurrency
if workers < 1 {
workers = 1
}
if workers == 1 {
return d.loop()
}
d.log.Info("agent running concurrently", "loops", workers)
var wg sync.WaitGroup
errs := make(chan error, workers)
for i := 0; i < workers; i++ {
wg.Add(1)
go func(loop int) {
defer wg.Done()
if err := d.loop(); err != nil {
errs <- err
return
}
errs <- nil
}(i)
}
wg.Wait()
close(errs)
for err := range errs {
if err != nil {
return err
}
}
return nil
}
// loop is one claim→execute→upload cycle until interrupted, idle-exit, or
// the shared max-tasks budget is consumed.
func (d *Daemon) loop() error {
failures := 0
for {
if !d.registered {
if err := d.register(); err != nil {
return err
}
}
d.cleanupExpiredDirectories()
outcome, err := d.runOnce()
if err != nil {
@@ -63,8 +100,11 @@ func (d *Daemon) RunForever() error {
}
failures = 0
if outcome.Claimed && outcome.Completed {
d.mu.Lock()
d.completed++
if d.config.MaxTasks > 0 && d.completed >= d.config.MaxTasks {
done := d.config.MaxTasks > 0 && d.completed >= d.config.MaxTasks
d.mu.Unlock()
if done {
d.log.Info("max tasks reached")
return nil
}
@@ -158,6 +198,7 @@ func (d *Daemon) runOnce() (Outcome, error) {
if task == nil {
return Outcome{Claimed: false}, nil
}
d.log.Info("task claimed", "task_id", task.TaskID, "attempt", task.Attempt)
started := time.Now()
taskDir := filepath.Join(d.config.WorkDir, task.TaskID, fmt.Sprint(task.Attempt))
if err := os.MkdirAll(taskDir, 0o750); err != nil {
+79
View File
@@ -11,6 +11,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
@@ -55,6 +56,7 @@ type fakeCoordinator struct {
uploadSize int64
inputBytes []byte
conflict bool // 409 on heartbeat/upload/result
mu sync.Mutex
}
func newFakeCoordinator(t *testing.T, task map[string]any) *fakeCoordinator {
@@ -64,6 +66,8 @@ func newFakeCoordinator(t *testing.T, task map[string]any) *fakeCoordinator {
fake.uploadSize = int64(len(fake.inputBytes))
var server *httptest.Server
server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fake.mu.Lock()
defer fake.mu.Unlock()
switch {
case r.Method == http.MethodPost && r.URL.Path == "/workers/register":
writeJSON(w, http.StatusCreated, map[string]any{
@@ -294,3 +298,78 @@ func TestDaemonIdleClaimIsNotCompleted(t *testing.T) {
t.Fatalf("outcome = %+v", outcome)
}
}
func TestDaemonConcurrencyProcessesTasksInParallel(t *testing.T) {
t.Parallel()
marker := filepath.Join(t.TempDir(), "marker")
script := filepath.Join(t.TempDir(), "fake-runner.sh")
content := `#!/bin/sh
out=""
task_dir=""
while [ "$#" -gt 0 ]; do
case "$1" in
--output) out="$2"; shift 2;;
--task-dir) task_dir="$2"; shift 2;;
*) shift;;
esac
done
echo start >> ` + marker + `
sleep 1
echo end >> ` + marker + `
printf 'id,score\n1,1\n' > "$task_dir/result.csv"
printf '{"artifact_path":"%s/result.csv","content_type":"text/csv","metrics":{"rows":1}}' "$task_dir" > "$out"
exit 0
`
if err := os.WriteFile(script, []byte(content), 0o755); err != nil {
t.Fatal(err)
}
fake := newFakeCoordinator(t, validClaimedTaskPayload())
defer fake.close()
config := &Config{
CoordinatorURL: fake.server.URL,
WorkerName: "concurrent-worker",
WorkerID: "22222222-2222-4222-8222-222222222222",
WorkDir: t.TempDir(),
CPUCount: 1,
PollInterval: time.Millisecond,
RequestTimeout: 5 * time.Second,
Heartbeat: 15 * time.Second,
Capabilities: []string{"similarity-search"},
TaskRunner: []string{script},
MaxTasks: 3,
Concurrency: 3,
}
client := NewClient(fake.server.URL, &StaticToken{token: "test-token"}, 5*time.Second)
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
daemon := NewDaemon(config, client, NewTaskRunner(config.TaskRunner), logger)
if err := daemon.RunForever(); err != nil {
t.Fatalf("run: %v", err)
}
raw, err := os.ReadFile(marker)
if err != nil {
t.Fatalf("marker: %v", err)
}
starts := strings.Count(string(raw), "start\n")
ends := strings.Count(string(raw), "end\n")
if starts < 3 || ends < 3 {
t.Fatalf("marker: %d starts / %d ends, want at least 3/3", starts, ends)
}
// With three loops sleeping 1s each, the marker proves all three ran
// concurrently (three starts before the first end completes a 1s sleep).
lines := strings.Split(strings.TrimSpace(string(raw)), "\n")
concurrent := 0
running := 0
for _, line := range lines {
if line == "start" {
running++
if running > concurrent {
concurrent = running
}
} else {
running--
}
}
if concurrent < 3 {
t.Errorf("max concurrent executions = %d, want 3", concurrent)
}
}
@@ -38,6 +38,9 @@ func NormalizePEP440(version string) string {
// with a generous timeout: wheels can be several MB.
func DownloadWheel(ctx context.Context, url, dir string) (string, error) {
target := filepath.Join(dir, wheelNameFromURL(url))
if err := os.MkdirAll(dir, 0o700); err != nil {
return "", fmt.Errorf("download wheel: %w", err)
}
ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+85 -11
View File
@@ -289,15 +289,26 @@ func writeJSON(w http.ResponseWriter, status int, v any) {
// statusView is what the wizard needs to paint the running/stopped state.
type statusView struct {
ConfigPresent bool `json:"config_present"`
ConfigPath string `json:"config_path"`
LogPath string `json:"log_path"`
Running bool `json:"running"`
Pid int `json:"pid"`
WorkerName string `json:"worker_name,omitempty"`
Coordinator string `json:"coordinator,omitempty"`
WorkDir string `json:"work_dir,omitempty"`
TokenSet bool `json:"token_set"`
ConfigPresent bool `json:"config_present"`
ConfigPath string `json:"config_path"`
LogPath string `json:"log_path"`
Running bool `json:"running"`
Pid int `json:"pid"`
WorkerName string `json:"worker_name,omitempty"`
Coordinator string `json:"coordinator,omitempty"`
WorkDir string `json:"work_dir,omitempty"`
TokenSet bool `json:"token_set"`
Stats WorkerStats `json:"stats"`
}
// WorkerStats is parsed from the worker log: the agent reports each claim,
// completion and failure as a structured line, so the wizard can show live
// counters without any coordinator access.
type WorkerStats struct {
Registered bool `json:"registered"`
Claimed int `json:"claimed"`
Completed int `json:"completed"`
Failed int `json:"failed"`
}
// ensureVenvTaskRunner rewrites the saved config so its task runner uses the
@@ -319,8 +330,31 @@ func (s *Server) ensureVenvTaskRunner() {
}
}
// parseWorkerStats counts the structured agent events in the worker log.
func parseWorkerStats(logPath string) WorkerStats {
var stats WorkerStats
//nolint:gosec // G304: logPath is the wizard's own log file next to the config
raw, err := os.ReadFile(logPath)
if err != nil {
return stats
}
for _, line := range strings.Split(string(raw), "\n") {
switch {
case strings.Contains(line, "msg=registered"):
stats.Registered = true
case strings.Contains(line, `msg="task claimed"`):
stats.Claimed++
case strings.Contains(line, `msg="task completed"`):
stats.Completed++
case strings.Contains(line, `msg="task failed"`):
stats.Failed++
}
}
return stats
}
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
view := statusView{ConfigPath: s.cfgPath, LogPath: s.logPath, Running: s.sup.Alive(), Pid: s.sup.Pid()}
view := statusView{ConfigPath: s.cfgPath, LogPath: s.logPath, Running: s.sup.Alive(), Pid: s.sup.Pid(), Stats: parseWorkerStats(s.logPath)}
if raw, err := os.ReadFile(s.cfgPath); err == nil {
var file agent.ConfigFile
if json.Unmarshal(raw, &file) == nil {
@@ -343,6 +377,7 @@ type saveConfigRequest struct {
WorkerName string `json:"worker_name"`
CPUCount int `json:"cpu_count"`
MemoryMB int `json:"memory_mb"`
Concurrency int `json:"concurrency"`
TaskRunner []string `json:"task_runner"`
}
@@ -361,6 +396,7 @@ func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) {
WorkerName: strings.TrimSpace(req.WorkerName),
CPUCount: req.CPUCount,
MemoryMB: req.MemoryMB,
Concurrency: req.Concurrency,
TaskRunner: req.TaskRunner,
}
if file.CoordinatorURL == "" {
@@ -384,6 +420,9 @@ func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) {
if file.CPUCount < 1 {
file.CPUCount = 1
}
if file.Concurrency < 1 {
file.Concurrency = 1
}
// The wizard UI bakes the venv python into the runner after an install;
// an API-driven or scripted flow may not, so the server guarantees it:
// workloads execute through scimesh's task runner, which lives in the venv.
@@ -411,7 +450,13 @@ func (s *Server) handleTest(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "coordinator_url is required"})
return
}
report := agent.RunCheck(r.Context(), url)
// After the runtime installer created the venv, probe that interpreter:
// checking the bare system python3 would keep reporting scimesh as
// missing even though the worker would run with the venv.
report := agent.RunCheck(r.Context(), url, s.venvPython(), req.Token, req.WorkerKey, req.UserserviceURL)
if report.Scimesh.OK {
report.Scimesh = ensureMatchingScimeshVersion(report.Scimesh)
}
writeJSON(w, http.StatusOK, report)
}
@@ -476,6 +521,12 @@ type installRuntimeResponse struct {
// step. The venv python path is returned for the wizard to bake into the
// task runner.
func (s *Server) handleInstallRuntime(w http.ResponseWriter, r *http.Request) {
// The wizard may run the install before any config was saved, so the
// worker directory (venv, wheel) may not exist yet.
if err := os.MkdirAll(s.dir, 0o700); err != nil {
writeJSON(w, http.StatusConflict, map[string]string{"error": "could not create the worker directory"})
return
}
var req installRuntimeRequest
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"})
@@ -580,3 +631,26 @@ func truncate(s string, n int) string {
}
return s[:n] + "…"
}
// ensureMatchingScimeshVersion flips a green scimesh check to a stale one when
// the installed package does not match the worker-agent's own version: a
// version-locked wheel is the only supported runtime, and a mismatch means the
// workload catalog the worker advertises is not what it executes. The wizard
// UI then offers the Install button again. Dev builds have no release wheel,
// so they skip the comparison.
func ensureMatchingScimeshVersion(item agent.CheckItem) agent.CheckItem {
if agent.Version == "" || agent.Version == "dev" {
return item
}
want := agent.NormalizePEP440(agent.Version)
got := strings.TrimSpace(item.Detail)
if got == "" || got == want {
return item
}
item.OK = false
item.Detail = fmt.Sprintf(
"installed scimesh %s, but this worker-agent (%s) needs %s — press Install to upgrade",
got, agent.Version, want,
)
return item
}
@@ -454,3 +454,69 @@ func TestSaveConfigPinsVenvRunnerWhenPresent(t *testing.T) {
t.Errorf("task runner = %v, want the venv python", config.TaskRunner)
}
}
func TestTestProbesTheVenvPythonAfterInstall(t *testing.T) {
sup := &fakeSup{}
server, base := newTestServer(t, sup)
// The runtime installer leaves a venv python; make it a stub that reports
// a fake scimesh version so the preflight goes green through the venv.
venvPython := filepath.Join(server.dir, "venv", "bin", "python")
_ = os.MkdirAll(filepath.Dir(venvPython), 0o755)
_ = os.WriteFile(venvPython, []byte("#!/bin/sh\nif [ \"$1\" = \"-c\" ]; then echo 9.9.9-test; exit 0; fi\nexit 0\n"), 0o755)
req, _ := http.NewRequestWithContext(context.Background(), http.MethodPost, base+"/api/test", strings.NewReader(`{"coordinator_url":"http://127.0.0.1:1"}`))
req.Header.Set("Content-Type", "application/json")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
defer func() { _ = resp.Body.Close() }()
var report agent.CheckReport
if err := json.NewDecoder(resp.Body).Decode(&report); err != nil {
t.Fatal(err)
}
if !report.Scimesh.OK || report.Scimesh.Detail != "9.9.9-test" {
t.Errorf("scimesh check = %+v, want the venv interpreter reporting 9.9.9-test", report.Scimesh)
}
}
func TestParseWorkerStats(t *testing.T) {
logPath := filepath.Join(t.TempDir(), "worker.log")
content := `time=1 level=INFO msg=registered worker_id=w1
time=2 level=INFO msg="task claimed" task_id=t1 attempt=0
time=3 level=INFO msg="task completed" task_id=t1 elapsed_seconds=2
time=4 level=INFO msg="task claimed" task_id=t2 attempt=0
time=5 level=WARN msg="task failed" task_id=t2 error_code=X retryable=true
time=6 level=WARN msg="agent cycle failed" error="boom"
`
if err := os.WriteFile(logPath, []byte(content), 0o600); err != nil {
t.Fatal(err)
}
stats := parseWorkerStats(logPath)
if !stats.Registered || stats.Claimed != 2 || stats.Completed != 1 || stats.Failed != 1 {
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
}
}
func testCheckScimeshVersion(t *testing.T, installed, binary string, wantOK bool, wantDetail string) {
t.Helper()
old := agent.Version
agent.Version = binary
t.Cleanup(func() { agent.Version = old })
item := ensureMatchingScimeshVersion(agent.CheckItem{Name: "scimesh", OK: true, Detail: installed})
if item.OK != wantOK {
t.Errorf("installed=%s binary=%s: ok=%v, want %v (%s)", installed, binary, item.OK, wantOK, item.Detail)
}
if wantDetail != "" && !strings.Contains(item.Detail, wantDetail) {
t.Errorf("detail = %q, want it to contain %q", item.Detail, wantDetail)
}
}
func TestEnsureMatchingScimeshVersion(t *testing.T) {
testCheckScimeshVersion(t, "1.1.0a20", "1.1.0-alpha.20", true, "")
testCheckScimeshVersion(t, "1.1.0a17", "1.1.0-alpha.20", false, "press Install to upgrade")
testCheckScimeshVersion(t, "1.1.0a16.dev7+gea0fb8c59.d20260803", "1.1.0-alpha.20", false, "needs 1.1.0a20")
// Dev builds and unknown versions never block.
testCheckScimeshVersion(t, "anything", "dev", true, "")
testCheckScimeshVersion(t, "1.1.0a20", "", true, "")
}
@@ -141,6 +141,7 @@ code{font-family:var(--mono);font-size:.86em}
</div>
</div>
<div class="field" id="cpu-field" style="display:none"><label>CPU count</label><input id="in-cpu" type="number" min="1" value="1"></div>
<div class="field"><label>Concurrent task loops</label><input id="in-conc" type="number" min="1" max="64" value="1"><p class="hint">Process this many shards in parallel on this machine. Each loop runs its own task runner subprocess.</p></div>
<div class="actions"><button class="btn btn-ghost" id="b2b">← Back</button><button class="btn btn-primary" id="b2">Continue →</button></div>
</div>
</div>
@@ -174,6 +175,12 @@ code{font-family:var(--mono);font-size:.86em}
<div><h1 id="st-title">Worker is working</h1><div class="sub" id="st-sub"></div></div>
<button class="btn btn-danger" id="st-stop" style="margin-left:auto"><svg viewBox="0 0 24 24" fill="none" stroke-width="2"><rect x="6" y="6" width="12" height="12" rx="2"/></svg>Stop</button>
</div>
<div class="stat-row">
<div class="stat"><b id="st-claimed"></b><span>claimed</span></div>
<div class="stat"><b id="st-completed"></b><span>completed</span></div>
<div class="stat bad"><b id="st-failed"></b><span>failed</span></div>
<div class="stat"><b id="st-registered"></b><span>registered</span></div>
</div>
<div class="meta-line" id="st-meta"></div>
<div class="logbox" id="st-log"></div>
<div class="actions"><span class="link" id="st-cfg"></span><button class="btn btn-ghost" id="st-reconfig">Reconfigure…</button></div>
@@ -222,7 +229,8 @@ function draftConfig(){
userservice_url:state.mode==='key'?$('in-users').value.trim():'',
work_dir:$('in-dir').value.trim(),
worker_name:$('in-name').value.trim(),
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0
cpu_count:state.cpu==='custom'?parseInt($('in-cpu').value||'1',10):0,
concurrency:parseInt($('in-conc').value||'1',10)
};
if(state.venvPython)cfg.task_runner=[state.venvPython,'-m','scimesh.worker.task'];
return cfg;
@@ -249,8 +257,8 @@ $('b4').onclick=async()=>{
showStatus();
};
const checkRow=(name,ok,detail,ms)=>
'<div class="check-row"><div class="check-ic '+(ok===null?'check-wait':ok?'check-ok':'check-bad')+'"><svg viewBox="0 0 24 24" fill="none" stroke-linecap="round">'+(ok===null?'<path d="M12 7v5l3 3"/>':ok?'<path d="M4 12l5 5L20 6"/>':'<path d="M6 6l12 12M18 6L6 18"/>')+'</svg></div><div><b>'+name+'</b><span>'+(detail||'')+'</span></div>'+(ms?'<span class="ms">'+ms+' ms</span>':'')+'</div>';
const checkRow=(name,ok,detail,ms,action)=>
'<div class="check-row"><div class="check-ic '+(ok===null?'check-wait':ok?'check-ok':'check-bad')+'"><svg viewBox="0 0 24 24" fill="none" stroke-linecap="round">'+(ok===null?'<path d="M12 7v5l3 3"/>':ok?'<path d="M4 12l5 5L20 6"/>':'<path d="M6 6l12 12M18 6L6 18"/>')+'</svg></div><div><b>'+name+'</b><span>'+(detail||'')+'</span></div>'+(ms?'<span class="ms">'+ms+' ms</span>':'')+(action?action:'')+'</div>';
async function runChecks(){
const box=$('checks');
box.innerHTML=checkRow('Coordinator reachable','',null,null)+checkRow('Python 3','',null,null)+checkRow('scimesh package','',null,null);
@@ -260,8 +268,21 @@ async function runChecks(){
const items=[r.data.coordinator,r.data.python,r.data.scimesh];
for(const item of items){
if(item&&!item.ok)checksOk=false;
box.insertAdjacentHTML('beforeend',checkRow(item?item.name:'?',item?item.ok:null,item?item.detail:'',item?item.latency_ms:null));
let action='';
if(item&&item.name==='scimesh'&&!item.ok){
action='<div style="margin-left:auto;display:flex;gap:8px;align-items:center"><input id="in-pkg" placeholder="wheel path / checkout / index URL (optional)" style="width:230px;padding:7px 10px;font-size:12px"><button class="btn btn-primary" style="padding:6px 12px;font-size:12px" id="install-scimesh">Install</button></div>';
}
box.insertAdjacentHTML('beforeend',checkRow(item?item.name:'?',item?item.ok:null,item?item.detail:'',item?item.latency_ms:null,action));
}
const btn=$('install-scimesh');
if(btn)btn.onclick=async()=>{
const src=$('in-pkg')?$('in-pkg').value.trim():'';
btn.disabled=true;btn.textContent='Installing…';
const ir=await postJSON('/api/runtime/install',{scimesh_package:src});
if(ir.status!==200){err('Could not install scimesh: '+(ir.data.error||'unknown error'));btn.disabled=false;btn.textContent='Install';return}
state.venvPython=ir.data.python;
runChecks();
};
$('b3').disabled=!checksOk;
}
@@ -280,6 +301,11 @@ async function refreshStatus(){
$('st-title').textContent=v.running?(v.worker_name||'Worker')+' is working':(v.worker_name||'Worker')+' is stopped';
$('st-sub').textContent='pid '+(v.pid||'—')+' · config '+(v.config_present?v.config_path:'not saved yet');
$('st-cfg').textContent='Configuration: '+(v.config_present?v.config_path:'—');
const stats=v.stats||{};
$('st-claimed').textContent=stats.claimed!=null?stats.claimed:'—';
$('st-completed').textContent=stats.completed!=null?stats.completed:'—';
$('st-failed').textContent=stats.failed!=null?stats.failed:'—';
$('st-registered').textContent=stats.registered?'yes':'no';
const meta=$('st-meta');
meta.innerHTML='';
if(v.coordinator){const c=document.createElement('span');c.className='chip';c.innerHTML='<svg viewBox="0 0 24 24" fill="none" stroke-width="2"><circle cx="12" cy="12" r="9"/></svg>'+v.coordinator;meta.append(c)}
+5
View File
@@ -1,6 +1,7 @@
package domain
import (
"strings"
"time"
"github.com/google/uuid"
@@ -52,6 +53,10 @@ func NewWorker(name string, capabilities []string, now time.Time) (*Worker, erro
if len(capabilities) == 0 {
return nil, ErrInvalidInput
}
name = strings.TrimSpace(name)
if name == "" {
return nil, ErrInvalidInput
}
return &Worker{
ID: uuid.New(),
Name: name,
@@ -29,3 +29,11 @@ func TestNewWorkerRejectsNoCapabilities(t *testing.T) {
t.Errorf("empty slice: err = %v, want ErrInvalidInput", err)
}
}
func TestNewWorkerRejectsBlankName(t *testing.T) {
for _, name := range []string{"", " ", "\t\n"} {
if _, err := NewWorker(name, []string{"similarity-search"}, testNow); !errors.Is(err, ErrInvalidInput) {
t.Errorf("name %q: got %v, want ErrInvalidInput", name, err)
}
}
}
+29
View File
@@ -461,3 +461,32 @@ func (r *TaskResultRepo) CountAgreeing(_ context.Context, taskID uuid.UUID, sha2
}
return n, nil
}
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
r.mu.Lock()
defer r.mu.Unlock()
var out []domain.Job
for _, j := range r.jobs {
if j.CompletedAt != nil && j.CompletedAt.Before(cutoff) {
out = append(out, *j)
}
}
return out, nil
}
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.jobs, id)
return nil
}
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.workers[id]; !ok {
return domain.ErrWorkerNotFound
}
delete(r.workers, id)
return nil
}
@@ -0,0 +1,278 @@
//go:build integration
package postgres
import (
"context"
"errors"
"os"
"testing"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// ensureMigrated applies the embedded schema first: the admin tests run
// before the dedicated migration test (file order) and need real tables.
func ensureMigrated(t *testing.T) {
t.Helper()
url := os.Getenv("TEST_DATABASE_URL")
if url == "" {
t.Skip("TEST_DATABASE_URL is not set")
}
if err := Migrate(context.Background(), url, nil); err != nil {
t.Fatalf("migrate: %v", err)
}
}
func TestAdminWorkerSetTrust(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
repo := NewWorkerRepo(pool)
w, err := domain.NewWorker("trust-lab", []string{"similarity-search"}, time.Now().UTC())
if err != nil {
t.Fatal(err)
}
if err := repo.Insert(ctx, w); err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM workers WHERE id = $1`, w.ID) })
if err := repo.SetTrust(ctx, w.ID, domain.WorkerUntrusted); err != nil {
t.Fatal(err)
}
got, err := repo.Get(ctx, w.ID)
if err != nil {
t.Fatal(err)
}
if got.TrustLevel != domain.WorkerUntrusted {
t.Errorf("trust = %q, want untrusted", got.TrustLevel)
}
if err := repo.SetTrust(ctx, uuid.New(), domain.WorkerTrusted); !errors.Is(err, domain.ErrWorkerNotFound) {
t.Errorf("unknown worker: got %v, want ErrWorkerNotFound", err)
}
}
func TestAdminListJobsPaginatedAndCounts(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
jobRepo := NewJobRepo(pool)
adminRepo := NewAdminReadRepo(pool)
jobs := make([]*domain.Job, 4)
for i := range jobs {
job, _ := seedJob(t, pool, 1)
jobs[i] = job
}
for _, j := range jobs {
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, j.ID) })
}
if err := jobRepo.UpdateStatus(ctx, jobs[0].ID, domain.JobCompleted, nil); err != nil {
t.Fatal(err)
}
if err := jobRepo.UpdateStatus(ctx, jobs[1].ID, domain.JobCompleted, nil); err != nil {
t.Fatal(err)
}
if err := jobRepo.UpdateStatus(ctx, jobs[2].ID, domain.JobRunning, nil); err != nil {
t.Fatal(err)
}
all, total, err := adminRepo.ListJobsPaginated(ctx, "", 100, 0)
if err != nil {
t.Fatal(err)
}
if total != 4 || len(all) != 4 {
t.Errorf("all: total=%d len=%d, want 4/4", total, len(all))
}
completed, total, err := adminRepo.ListJobsPaginated(ctx, "completed", 100, 0)
if err != nil {
t.Fatal(err)
}
if total != 2 || len(completed) != 2 {
t.Errorf("completed: total=%d len=%d, want 2/2", total, len(completed))
}
page, total, err := adminRepo.ListJobsPaginated(ctx, "", 2, 2)
if err != nil {
t.Fatal(err)
}
if total != 4 || len(page) != 2 {
t.Errorf("page: total=%d len=%d, want 4/2", total, len(page))
}
counts, err := adminRepo.CountJobsByStatus(ctx)
if err != nil {
t.Fatal(err)
}
if counts["completed"] != 2 || counts["running"] != 1 || counts["pending"] != 1 {
t.Errorf("counts = %v", counts)
}
}
func TestAdminTaskCountsByJobs(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
job, tasks := seedJob(t, pool, 3)
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, job.ID) })
resultArtifact := seedArtifact(t, pool, job.ID, &tasks[0].ID, domain.ArtifactPartialResult)
if _, err := pool.Exec(ctx, `UPDATE tasks SET status = 'completed', result_artifact_id = $1 WHERE id = $2`,
resultArtifact.ID, tasks[0].ID); err != nil {
t.Fatal(err)
}
if _, err := pool.Exec(ctx, `UPDATE tasks SET status = 'failed' WHERE id = $1`, tasks[1].ID); err != nil {
t.Fatal(err)
}
counts, err := NewAdminReadRepo(pool).TaskCountsByJobs(ctx, []uuid.UUID{job.ID})
if err != nil {
t.Fatal(err)
}
got := counts[job.ID]
if got["completed"] != 1 || got["failed"] != 1 || got["pending"] != 1 {
t.Errorf("task counts = %v, want completed=1 failed=1 pending=1", got)
}
}
func TestAdminJobCountsByDayAndWorkload(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
repo := NewAdminReadRepo(pool)
job, _ := seedJob(t, pool, 1)
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, job.ID) })
if _, err := pool.Exec(ctx, `UPDATE jobs SET created_at = now() - interval '2 days' WHERE id = $1`, job.ID); err != nil {
t.Fatal(err)
}
other, _ := seedJob(t, pool, 1)
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, other.ID) })
byDay, err := repo.JobCountsByDay(ctx, time.Now().UTC().Add(-6*24*time.Hour))
if err != nil {
t.Fatal(err)
}
today := time.Now().UTC().Format("2006-01-02")
twoDays := time.Now().UTC().Add(-48 * time.Hour).Format("2006-01-02")
if byDay[today] != 1 || byDay[twoDays] != 1 {
t.Errorf("by day = %v, want today=1 twoDays=1", byDay)
}
byWorkload, err := repo.JobCountsByWorkload(ctx)
if err != nil {
t.Fatal(err)
}
if byWorkload[job.Workload] < 2 {
t.Errorf("by workload = %v, want at least 2 for %s", byWorkload, job.Workload)
}
}
func TestAdminTaskStatsAndStorage(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
repo := NewAdminReadRepo(pool)
job, tasks := seedJob(t, pool, 2)
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM jobs WHERE id = $1`, job.ID) })
start := time.Now().UTC().Add(-2 * time.Minute)
done := time.Now().UTC().Add(-90 * time.Second)
resultArtifact := seedArtifact(t, pool, job.ID, &tasks[0].ID, domain.ArtifactPartialResult)
if _, err := pool.Exec(ctx, `UPDATE tasks SET status = 'completed', result_artifact_id = $1, started_at = $2, completed_at = $3 WHERE id = $4`,
resultArtifact.ID, start, done, tasks[0].ID); err != nil {
t.Fatal(err)
}
if _, err := pool.Exec(ctx, `UPDATE tasks SET status = 'failed' WHERE id = $1`, tasks[1].ID); err != nil {
t.Fatal(err)
}
completed, failed, avg, err := repo.TaskStats(ctx)
if err != nil {
t.Fatal(err)
}
if completed < 1 || failed < 1 {
t.Errorf("stats = completed %d failed %d, want >= 1/1", completed, failed)
}
if avg < 29 || avg > 31 {
t.Errorf("avg duration = %.1fs, want ~30s", avg)
}
for _, kind := range []domain.ArtifactKind{domain.ArtifactInput, domain.ArtifactShard} {
seedArtifact(t, pool, job.ID, nil, kind)
}
sizes, err := repo.ArtifactSizeByKind(ctx)
if err != nil {
t.Fatal(err)
}
if sizes["input"] < 1 || sizes["shard"] < 1 {
t.Errorf("sizes = %v, want input/shard > 0", sizes)
}
dbBytes, err := repo.DatabaseSizeBytes(ctx)
if err != nil {
t.Fatal(err)
}
if dbBytes <= 0 {
t.Errorf("database size = %d, want > 0", dbBytes)
}
}
func TestWorkloadSettingsRepoRoundTrip(t *testing.T) {
ensureMigrated(t)
pool := testPool(t)
ctx := context.Background()
repo := NewWorkloadSettingsRepo(pool)
enabled, err := repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if !enabled {
t.Error("workload without an override must be enabled")
}
now := time.Now().UTC()
if err := repo.SetEnabled(ctx, "similarity-search", false, now); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(context.Background(), `DELETE FROM workload_settings WHERE workload = 'similarity-search'`)
})
enabled, err = repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if enabled {
t.Error("workload must be disabled after the override")
}
if err := repo.SetEnabled(ctx, "similarity-search", true, now.Add(time.Hour)); err != nil {
t.Fatal(err)
}
enabled, err = repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if !enabled {
t.Error("workload must be re-enabled after the upsert")
}
list, err := repo.List(ctx)
if err != nil {
t.Fatal(err)
}
found := false
for _, s := range list {
if s.Workload == "similarity-search" {
found = true
if !s.Enabled {
t.Error("list must reflect the re-enabled state")
}
}
}
if !found {
t.Error("override missing from the settings list")
}
}
@@ -3,6 +3,7 @@ package postgres
import (
"context"
"errors"
"fmt"
"time"
sq "github.com/Masterminds/squirrel"
@@ -153,3 +154,46 @@ func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID,
}
return nil
}
// ListCompletedBefore returns jobs whose completion timestamp is older than
// the cutoff (completed and failed both count as finished).
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
sql, args, err := psql.Select(jobColumns...).From("jobs").
Where(sq.NotEq{"completed_at": nil}).
Where(sq.Lt{"completed_at": cutoff}).
OrderBy("completed_at 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 completed jobs: %w", err)
}
defer rows.Close()
var jobs []domain.Job
for rows.Next() {
var j domain.Job
var status string
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,
&j.OwnerID,
); err != nil {
return nil, err
}
j.Status = domain.JobStatus(status)
jobs = append(jobs, j)
}
return jobs, rows.Err()
}
// Delete removes the job row; tasks, artifacts and task_results cascade.
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
sql, args, err := psql.Delete("jobs").Where(sq.Eq{"id": id}).ToSql()
if err != nil {
return err
}
_, err = conn(ctx, r.pool).Exec(ctx, sql, args...)
return err
}
@@ -123,3 +123,19 @@ func scanWorker(row pgx.Row) (*domain.Worker, error) {
w.TrustLevel = domain.WorkerTrust(trust)
return &w, nil
}
// Delete removes a worker from the registry.
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
sql, args, err := psql.Delete("workers").Where(sq.Eq{"id": id}).ToSql()
if err != nil {
return err
}
tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...)
if err != nil {
return fmt.Errorf("delete worker: %w", err)
}
if tag.RowsAffected() == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
@@ -88,3 +88,47 @@ func TestWorkerSetTrust(t *testing.T) {
t.Errorf("unknown worker trust err = %v, want ErrWorkerNotFound", err)
}
}
func TestJobRepoListCompletedBeforeAndDelete(t *testing.T) {
db := newTestDB(t)
ctx := context.Background()
repo := NewJobRepo(db)
old := seedJob(t, db, 2)
oldTime := fixedTime().Add(-40 * 24 * time.Hour)
if err := repo.UpdateStatus(ctx, old.ID, domain.JobCompleted, &oldTime); err != nil {
t.Fatal(err)
}
fresh := seedJob(t, db, 2)
freshTime := fixedTime().Add(-2 * time.Hour)
if err := repo.UpdateStatus(ctx, fresh.ID, domain.JobCompleted, &freshTime); err != nil {
t.Fatal(err)
}
// The failing check constraint needs no result artifact for completed; the
// UpdateStatus path is fine, but tasks stay pending — irrelevant here.
list, err := repo.ListCompletedBefore(ctx, fixedTime().Add(-7*24*time.Hour))
if err != nil {
t.Fatal(err)
}
if len(list) != 1 || list[0].ID != old.ID {
t.Errorf("list = %d jobs, want only the old one", len(list))
}
if err := repo.Delete(ctx, old.ID); err != nil {
t.Fatal(err)
}
var n int
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM jobs WHERE id = ?", old.ID.String()).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Error("job row must be gone after Delete")
}
// Tasks cascaded away with the job.
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tasks WHERE job_id = ?", old.ID.String()).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Error("tasks must cascade with the job")
}
}
@@ -3,6 +3,7 @@ package sqlite
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/google/uuid"
@@ -166,3 +167,30 @@ func nullableUUID(id *uuid.UUID) any {
}
return id.String()
}
// ListCompletedBefore returns jobs whose completion timestamp is older than
// the cutoff (completed and failed both count as finished).
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT "+jobColumns+" FROM jobs WHERE completed_at IS NOT NULL AND completed_at < ? ORDER BY completed_at ASC",
encodeTime(cutoff))
if err != nil {
return nil, fmt.Errorf("list completed jobs: %w", err)
}
defer func() { _ = rows.Close() }()
var jobs []domain.Job
for rows.Next() {
job, err := scanJob(rows)
if err != nil {
return nil, err
}
jobs = append(jobs, *job)
}
return jobs, rows.Err()
}
// Delete removes the job row; tasks, artifacts and task_results cascade.
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
_, err := conn(ctx, r.db).ExecContext(ctx, "DELETE FROM jobs WHERE id = ?", id.String())
return err
}
@@ -107,3 +107,19 @@ func (r *WorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.Wo
}
return nil
}
// Delete removes a worker from the registry.
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
res, err := conn(ctx, r.db).ExecContext(ctx, "DELETE FROM workers WHERE id = ?", id.String())
if err != nil {
return err
}
affected, err := res.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
@@ -36,6 +36,7 @@ type UseCases struct {
Dashboard *usecase.Dashboard
PreviewArtifact *usecase.PreviewArtifact
Admin *usecase.Admin
PruneArtifacts *usecase.PruneArtifacts
}
type Server struct {
@@ -131,6 +132,11 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
mux.HandleFunc("GET /health", s.handleHealth)
// Unauthenticated like /health, so a Prometheus scraper needs no credential.
mux.Handle("GET /metrics", s.metrics.Handler())
// Worker-key exchange is fronted by the coordinator when the userservice
// is embedded (serve mode): the key itself is the credential.
if s.userserviceURL != "" {
mux.HandleFunc("POST /worker-tokens/exchange", s.handleWorkerTokenExchangeProxy)
}
hasBasicAuth := len(uiToken) > 0 && uiToken[0] != ""
if s.uc.Dashboard != nil && (s.uiSessionMode() || hasBasicAuth) {
@@ -184,6 +190,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
ui.Handle("GET /ui/admin/api/metrics", chain(http.HandlerFunc(s.handleUIAdminMetricsJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/workers", chain(http.HandlerFunc(s.handleUIAdminWorkersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workers/{id}/trust", chain(http.HandlerFunc(s.handleUIAdminSetTrustJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workers/{id}/remove", chain(http.HandlerFunc(s.handleUIAdminRemoveWorkerJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/users", chain(http.HandlerFunc(s.handleUIAdminUsersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/users/{id}/role", chain(http.HandlerFunc(s.handleUIAdminSetUserRoleJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/worker-keys", chain(http.HandlerFunc(s.handleUIAdminWorkerKeysJSON), gate, requireAdmin))
@@ -192,6 +199,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
ui.Handle("POST /ui/admin/api/workloads/{name}/enabled", chain(http.HandlerFunc(s.handleUIAdminSetWorkloadEnabledJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/settings", chain(http.HandlerFunc(s.handleUIAdminSettingsJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/token/reveal", chain(http.HandlerFunc(s.handleUIAdminRevealTokenJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/prune", chain(http.HandlerFunc(s.handleUIAdminPruneJSON), gate, requireAdmin))
} else {
for _, rt := range app {
ui.HandleFunc(rt.pattern, rt.handler)
@@ -5,9 +5,7 @@
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Add your machine · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:980px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:720px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}.card,.aside,.notice{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021}.card{padding:22px}.aside,.notice{padding:18px}.aside h2,.notice h2{margin:0;color:#f1f6ff;font-size:1rem}.aside p,.notice p{color:#9fb3cf}.aside ol{margin:13px 0 0;padding-left:20px;color:#aebfda}.aside li{margin:10px 0}label{display:block;margin:18px 0 5px;color:#eaf2ff;font-weight:750}input{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus{outline:2px solid #5d97f5;outline-offset:1px}.hint{margin:5px 0;color:#92a9c6;font-size:.87rem}.button{display:inline-flex;margin-top:16px;border:0;border-radius:10px;padding:11px 15px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.button.secondary{background:#23344d;color:#dce8ff}.button:disabled{opacity:.6;cursor:wait}.command{margin-top:18px;border:1px solid #2c8c70;border-radius:11px;padding:14px;background:#0c2b2a;color:#a8f1d0}.command strong{color:#e6fff4}.command pre{margin:10px 0 0;padding:12px;overflow-x:auto;border-radius:8px;background:#061a19;color:#c8ffe8;font:.82rem/1.5 ui-monospace,SFMono-Regular,monospace;white-space:pre;word-break:normal}.keys{margin-top:14px;display:grid;gap:9px}.key{display:flex;align-items:center;justify-content:space-between;gap:12px;border:1px solid #294662;border-radius:11px;padding:12px 14px;background:#0a1626}.key .kn{color:#f3f7ff;font-weight:700}.key .kp{margin-top:3px;color:#8196b3;font-family:ui-monospace,SFMono-Regular,monospace;font-size:.76rem}.key .kd{color:#8fa6c3;font-size:.8rem}.revoke{border:1px solid #6a2a3a;border-radius:8px;padding:7px 11px;background:#2a1420;color:#ff9bad;font:inherit;font-weight:700;cursor:pointer}.empty{padding:20px;border:1px dashed #35516f;border-radius:12px;color:#9ab0cb;text-align:center}.error{margin:12px 0 0;color:#ffacba}.warn{color:#ffd08a}.hidden{display:none}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 6px;color:#b5d3f5;font:.76rem ui-monospace,SFMono-Regular,monospace}@media(max-width:720px){.layout{grid-template-columns:1fr}.page{padding:22px 14px}}
</style>
{{template "ui-styles"}}
</head>
<body data-coordinator="{{.CoordinatorURL}}" data-userservice="{{.UserserviceURL}}">
<main class="page">
@@ -43,12 +41,12 @@
</main>
<script>
const coord=(document.body.dataset.coordinator||location.origin).replace(/\/+$/,'');
const users=(document.body.dataset.userservice||'').replace(/\/+$/,'');
const users=(document.body.dataset.userservice||coord).replace(/\/+$/,'');
const keysBox=document.querySelector('#keys'),cmdBox=document.querySelector('#command'),form=document.querySelector('#create'),nameInput=document.querySelector('#key-name'),createBtn=document.querySelector('#create-btn'),error=document.querySelector('#error');
const node=(tag,text,cls)=>{const n=document.createElement(tag);if(text!==undefined)n.textContent=text;if(cls)n.className=cls;return n};
const shq=s=>"'"+String(s).replace(/'/g,"'\\''")+"'";
const buildCommand=(key,name)=>['# install the worker binary (or download worker-agent from the release page)','curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/install.sh | bash -s worker','','export COORDINATOR_URL='+coord,'export USERSERVICE_URL='+(users||'<your userservice URL>')+' \\','export WORKER_KEY='+key,'export WORKER_NAME='+shq(name||'my-machine'),'export WORK_DIR=~/scimesh-worker','worker-agent'].join('\n');
const showCommand=(key,name)=>{cmdBox.replaceChildren();cmdBox.append(node('strong','Key created — copy it now, it is shown only once.'));const pre=node('pre',buildCommand(key,name));cmdBox.append(pre);const copy=node('button','Copy command','button secondary');copy.type='button';copy.addEventListener('click',()=>{navigator.clipboard&&navigator.clipboard.writeText(buildCommand(key,name)).then(()=>{copy.textContent='Copied ✓'},()=>{copy.textContent='Press Ctrl+C to copy'})});cmdBox.append(document.createElement('br'),copy);if(!users){cmdBox.append(node('p','Set USERSERVICE_URL to a userservice URL your machine can reach, or skip it and run the worker with WORKER_AUTH_TOKEN instead of a key.','warn'))}cmdBox.classList.remove('hidden')};
const showCommand=(key,name)=>{cmdBox.replaceChildren();cmdBox.append(node('strong','Key created — copy it now, it is shown only once.'));const pre=node('pre',buildCommand(key,name));cmdBox.append(pre);const copy=node('button','Copy command','button secondary');copy.type='button';copy.addEventListener('click',()=>{navigator.clipboard&&navigator.clipboard.writeText(buildCommand(key,name)).then(()=>{copy.textContent='Copied ✓'},()=>{copy.textContent='Press Ctrl+C to copy'})});cmdBox.append(document.createElement('br'),copy);cmdBox.classList.remove('hidden')};
const revoke=async id=>{const r=await fetch('/ui/api/worker-keys/'+encodeURIComponent(id)+'/revoke',{method:'POST'});if(r.status===204||r.ok){loadKeys()}else{error.textContent='Could not revoke the key.'}};
const renderKeys=keys=>{keysBox.replaceChildren();if(!keys.length){keysBox.append(node('div','No keys yet. Create one above to connect a machine.','empty'));return}for(const k of keys){const row=node('div',undefined,'key'),left=node('div');left.append(node('div',k.name||'unnamed','kn'),node('div',k.prefix+'…','kp'),node('div','Created '+new Date(k.created_at).toLocaleString()+(k.last_used_at?' · last used '+new Date(k.last_used_at).toLocaleString():' · never used'),'kd'));const btn=node('button','Revoke','revoke');btn.type='button';btn.addEventListener('click',()=>revoke(k.id));row.append(left,btn);keysBox.append(row)}};
const loadKeys=async()=>{try{const r=await fetch('/ui/api/worker-keys',{headers:{Accept:'application/json'}});if(!r.ok)throw Error();const data=await r.json();renderKeys(data.worker_keys||[])}catch(_){keysBox.replaceChildren(node('div','Could not load your keys.','empty'))}};
@@ -210,7 +210,7 @@ tbody tr:hover{background:var(--panel-2)}
<div class="section-note">Workers register themselves. Trust decides whether a machine's results are accepted directly or need quorum.</div>
<div class="card">
<table>
<thead><tr><th>Worker</th><th>Status</th><th>Capabilities</th><th>Trust</th><th>Owner</th><th>Last signal</th></tr></thead>
<thead><tr><th>Worker</th><th>Status</th><th>Capabilities</th><th>Trust</th><th>Owner</th><th>Last signal</th><th></th></tr></thead>
<tbody id="worker-rows"></tbody>
</table>
</div>
@@ -284,6 +284,12 @@ tbody tr:hover{background:var(--panel-2)}
<!-- ═══ SETTINGS ═══ -->
<section class="page" id="page-settings">
<div class="warn-strip" style="display:flex;gap:10px;align-items:flex-start;background:var(--amber-soft);border:1px solid #e5b64f33;border-radius:10px;padding:12px 14px;font-size:12.5px;color:#eecf8d"><span>The cluster token below authenticates <b>any</b> worker. Reveal it only on a trusted machine.</span></div>
<div class="section-title" style="color:var(--red)">Danger zone</div>
<div class="card" style="border-color:#f2647c33">
<dl class="kv">
<dt>Prune artifacts</dt><dd style="display:flex;justify-content:space-between;align-items:center;gap:14px"><span style="color:var(--text-2);font-size:12.5px">Delete finished jobs older than a cutoff and all their artifacts (blob files included).</span><span style="display:flex;gap:8px;align-items:center"><input id="prune-days" type="number" min="1" max="3650" value="30" style="width:80px"><button class="btn btn-danger btn-sm" id="prune">Prune…</button></span></dd>
</dl>
</div>
<div class="section-title">Cluster</div>
<div class="card">
<dl class="kv">
@@ -440,9 +446,15 @@ async function loadWorkers(){
'<td>'+(w.capabilities||[]).map(c=>'<span class="cap">'+esc(c)+'</span>').join('')+'</td>'+
'<td>'+trustSel+'</td>'+
'<td style="color:var(--text-2)">'+esc(w.owner)+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(w.last_heartbeat_at)+'</td>';
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(w.last_heartbeat_at)+'</td>'+
'<td>'+(w.status==='offline'?'<button class="btn btn-danger btn-sm worker-remove" data-id="'+w.id+'" data-name="'+esc(w.name)+'">Remove</button>':'')+'</td>';
rows.append(tr);
}
document.querySelectorAll('.worker-remove').forEach(btn=>btn.addEventListener('click',async()=>{
if(!confirm('Remove the offline worker "'+btn.dataset.name+'" from the registry? This cannot be undone.'))return;
await fetch('/ui/admin/api/workers/'+btn.dataset.id+'/remove',{method:'POST'});
loadWorkers();
}));
document.querySelectorAll('.trust-sel').forEach(sel=>sel.addEventListener('change',async()=>{
await fetch('/ui/admin/api/workers/'+sel.dataset.id+'/trust',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({trusted:sel.value==='trusted'})});
loadWorkers();
@@ -530,6 +542,17 @@ document.getElementById('reveal').addEventListener('click',async e=>{
}else{tok.textContent='••••••••••••••••••••••••';e.target.textContent='Reveal'}
});
document.querySelectorAll('.toggle').forEach(t=>t.addEventListener('click',()=>t.classList.toggle('on')));
document.getElementById('prune').addEventListener('click',async()=>{
const days=parseInt(document.getElementById('prune-days').value||'30',10);
if(!confirm('Delete finished jobs older than '+days+' days with all their artifacts? This cannot be undone.'))return;
const btn=document.getElementById('prune');btn.disabled=true;btn.textContent='Pruning…';
const r=await fetch('/ui/admin/api/prune',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({older_than_days:days})});
btn.disabled=false;btn.textContent='Prune…';
if(!r.ok){alert('Prune failed.');return}
const v=await r.json();
alert('Removed '+v.jobs+' jobs and '+v.artifacts+' artifacts, freed '+fmtBytes(v.freed_bytes)+'.');
if(current.page==='settings')loadSettings();
});
</script>
</body>
</html>
File diff suppressed because one or more lines are too long
@@ -5,9 +5,7 @@
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>New computation · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:980px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:720px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}.card,.aside,.notice{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021}.card{padding:22px}.aside,.notice{padding:18px}.aside h2,.notice h2{margin:0;color:#f1f6ff;font-size:1rem}.aside p,.notice p{color:#9fb3cf}.aside ol{margin:13px 0 0;padding-left:20px;color:#aebfda}.aside li{margin:10px 0}label{display:block;margin:18px 0 5px;color:#eaf2ff;font-weight:750}input,select{width:100%;border:1px solid #42617f;border-radius:9px;padding:10px 11px;background:#0a1626;color:#e5efff;font:inherit}input:focus,select:focus{outline:2px solid #5d97f5;outline-offset:1px}input[type=file]{padding:8px}.checkbox-row{display:flex;align-items:flex-start;gap:10px;margin-top:16px}.checkbox-row input[type=checkbox]{width:18px;height:18px;margin-top:4px;accent-color:#67e3b8}.checkbox-row label{margin:0}.checkbox-row .hint{margin:0}.hint{margin:5px 0;color:#92a9c6;font-size:.87rem}.split{display:grid;grid-template-columns:1fr 1fr;gap:12px}.run-preview{margin-top:18px;border:1px solid #2c8c70;border-radius:11px;padding:13px;background:#0c2b2a;color:#a8f1d0}.run-preview strong{color:#e6fff4}.button{display:inline-flex;margin-top:22px;border:0;border-radius:10px;padding:12px 16px;background:#67e3b8;color:#062018;font:inherit;font-weight:850;cursor:pointer}.button:disabled{opacity:.6;cursor:wait}.working{margin:14px 0 0;color:#9fc5ff}.error{margin:12px 0 0;color:#ffacba}.hidden{display:none}.cap{display:inline-block;margin:4px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 6px;color:#bcd2f0;font-size:.75rem;font-weight:700}.req{color:#ffb4c0}.workload-meta{margin:6px 0 0;color:#8fa7c8;font-size:.9rem}
</style>
{{template "ui-styles"}}
</head>
<body>
<main class="page">
@@ -5,8 +5,7 @@
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Profile · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 10% -8%,#183f77 0,transparent 32rem),#08111f}.page{max-width:720px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.top{display:flex;justify-content:space-between;align-items:center;gap:12px}.eyebrow{margin:0;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:6px 0 0;color:#f4f8ff;font-size:clamp(1.8rem,4vw,2.6rem);letter-spacing:-.04em}.btn{border:0;border-radius:9px;padding:10px 14px;font:inherit;font-weight:800;cursor:pointer;text-decoration:none}.btn-muted{background:#23344d;color:#dce8ff}.card{margin-top:24px;border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:6px 22px}.err{margin-top:24px;border-radius:10px;padding:12px 14px;background:#552334;color:#ff9bad;font-weight:700}.row{display:flex;justify-content:space-between;gap:16px;padding:15px 0;border-bottom:1px solid #1d3350}.row:last-child{border-bottom:0}.k{color:#9fb3cf}.v{color:#f2f7ff;font-weight:700;text-align:right;word-break:break-all}.mono{font-family:ui-monospace,SFMono-Regular,monospace;font-size:.9rem}.pill{display:inline-block;border-radius:999px;padding:3px 10px;font-size:.82rem;font-weight:800}.pill-yes{background:#123f34;color:#76efb5}.pill-no{background:#23344d;color:#b9cce9}.hint{margin-top:14px;color:#8ba2c2;font-size:.86rem}</style>
{{template "ui-styles"}}
</head>
<body>
<main class="page">
@@ -0,0 +1,141 @@
{{define "ui-styles"}}<style>
:root{--bg:#0b0e13;--panel:#11151d;--panel-2:#161c27;--border:#1f2634;--border-soft:#181f2b;--text:#e9ecf3;--text-2:#98a2b5;--text-3:#5d6879;--accent:#5b8cff;--accent-soft:#5b8cff1f;--accent-strong:#3f6fe0;--green:#3fce8a;--green-soft:#3fce8a1a;--amber:#e5b64f;--amber-soft:#e5b64f1a;--red:#f2647c;--red-soft:#f2647c1a;--mono:ui-monospace,"SF Mono",SFMono-Regular,Menlo,monospace;color-scheme:dark}
*{box-sizing:border-box;margin:0;padding:0}
body{background:var(--bg);color:var(--text);font:14px/1.55 Inter,-apple-system,"Segoe UI",Roboto,sans-serif;-webkit-font-smoothing:antialiased}
a{color:var(--accent);text-decoration:none}
a:hover{text-decoration:underline}
code{font-family:var(--mono);font-size:.86em}
.page{max-width:1100px;margin:auto;padding:30px 26px 70px}
.back{display:inline-flex;align-items:center;gap:6px;color:var(--text-2);font-size:13px;text-decoration:none}
.back:hover{color:var(--text)}
.eyebrow{margin:30px 0 4px;color:var(--accent);font-size:.72rem;font-weight:750;letter-spacing:.14em;text-transform:uppercase}
h1{margin:0;color:var(--text);font-size:clamp(1.7rem,4vw,2.6rem);letter-spacing:-.04em;line-height:1.1}
.lead{max-width:760px;margin:10px 0 0;color:var(--text-2);font-size:1rem}
.top{display:flex;justify-content:space-between;gap:24px;align-items:flex-start}
.title{margin:0;font-size:clamp(1.7rem,4vw,2.6rem);letter-spacing:-.04em}
.subtitle{margin:9px 0 0;color:var(--text-2)}
.live{color:var(--text-3);font-size:.85rem}
.hidden{display:none!important}
.card,.panel,.aside,.run-note,.notice,.artifact,.workload{background:var(--panel);border:1px solid var(--border-soft);border-radius:13px}
.card{padding:22px}
.panel{padding:20px}
.aside,.notice{padding:18px}
.aside h2,.notice h2{margin:0;color:var(--text);font-size:.95rem;font-weight:650}
.aside p,.notice p{color:var(--text-2)}
.aside ol{margin:13px 0 0;padding-left:20px;color:var(--text-2)}
.aside li{margin:10px 0}
.layout{display:grid;grid-template-columns:1.45fr .8fr;gap:15px;margin-top:28px}
label{display:block;margin:18px 0 5px;color:var(--text);font-weight:650}
input,select,textarea{width:100%;border:1px solid var(--border);border-radius:8px;padding:9px 11px;background:var(--panel-2);color:var(--text);font:inherit;outline:none}
input:focus,select:focus,textarea:focus{border-color:var(--accent);box-shadow:0 0 0 3px var(--accent-soft)}
input[type=file]{padding:8px}
.checkbox-row{display:flex;align-items:flex-start;gap:10px;margin-top:16px}
.checkbox-row input[type=checkbox]{width:17px;height:17px;margin-top:3px;accent-color:var(--accent)}
.checkbox-row label{margin:0}
.checkbox-row .hint{margin:0}
.hint{margin:5px 0;color:var(--text-3);font-size:.86rem}
.split{display:grid;grid-template-columns:1fr 1fr;gap:12px}
.run-preview{margin-top:18px;border:1px solid var(--green-soft);border-radius:10px;padding:13px 14px;background:var(--panel-2);color:var(--green)}
.run-preview strong{color:var(--text)}
.button,.btn{display:inline-flex;align-items:center;gap:7px;border:0;border-radius:8px;padding:10px 16px;background:var(--accent);color:#0a1222;font:inherit;font-weight:700;cursor:pointer;text-decoration:none}
.button:hover,.btn:hover{background:var(--accent-strong);color:#fff;text-decoration:none}
.button:disabled{opacity:.55;cursor:default}
.btn-muted{background:var(--panel-2);color:var(--text-2);border:1px solid var(--border)}
.btn-muted:hover{color:var(--text);border-color:#2a3446;background:var(--panel-2)}
.working{margin-top:14px;color:var(--text-2)}
.error{margin-top:12px;color:var(--red)}
.workload-meta{margin-top:8px;color:var(--text-2);font-size:.9rem}
.badge,.pill{display:inline-flex;align-items:center;gap:6px;border-radius:999px;padding:3px 10px;font-size:.78rem;font-weight:700;white-space:nowrap}
.badge i,.pill i{width:6px;height:6px;border-radius:50%;background:currentColor}
.badge-waiting,.pill-no{background:#ffffff12;color:var(--text-2)}
.badge-active,.pill-yes{background:var(--accent-soft);color:var(--accent)}
.badge-success{background:var(--green-soft);color:var(--green)}
.badge-danger{background:var(--red-soft);color:var(--red)}
.section-head{display:flex;align-items:baseline;justify-content:space-between;gap:12px;margin:34px 0 12px}
.section-head h2{margin:0;color:var(--text);font-size:1.05rem;font-weight:700}
.section-head p{margin:0;color:var(--text-3);font-size:.86rem}
.summary{margin-top:26px}
.summary-top{display:flex;justify-content:space-between;gap:20px;align-items:flex-start}
.bar{height:7px;margin:22px 0 9px;overflow:hidden;border-radius:999px;background:#ffffff10}
.bar span{display:block;height:100%;border-radius:inherit;background:linear-gradient(90deg,var(--accent),var(--green));transition:width .25s}
.progress-line{margin:0;color:var(--text-2)}
.metrics{display:grid;grid-template-columns:repeat(6,1fr);gap:10px;margin-top:17px}
.metric{padding:13px 15px}
.metric b{display:block;margin-top:5px;color:var(--text);font-size:1.35rem;line-height:1}
.metric small{color:var(--text-3)}
.stop{border:1px solid var(--red);border-radius:8px;padding:8px 12px;background:var(--red-soft);color:var(--red);font:inherit;font-weight:700;cursor:pointer}
.stop:disabled{opacity:.55}
.pipeline{display:grid;grid-template-columns:repeat(5,1fr);gap:10px}
.stage{position:relative;padding:13px 14px;min-height:104px;background:var(--panel);border:1px solid var(--border-soft);border-radius:11px}
.stage:not(:last-child):after{content:"";position:absolute;top:32px;right:-10px;width:10px;height:2px;background:#2a3446}
.stage .index{color:var(--text-3);font-size:.74rem;font-weight:700}
.stage b{display:block;margin-top:6px;color:var(--text);font-size:.92rem}
.stage p{margin-top:6px;color:var(--text-3);font-size:.82rem}
.stage-done{border-color:#3fce8a33;background:var(--green-soft)}
.stage-done .index{color:var(--green)}
.stage-active{border-color:var(--accent)}
.stage-active .index{color:var(--accent)}
.stage-waiting{opacity:.75}
.two-col{display:grid;grid-template-columns:1fr 1fr;gap:14px}
.parameter-list{display:grid;gap:8px;margin-top:12px}
.parameter{display:flex;justify-content:space-between;gap:14px;padding:8px 0;border-bottom:1px solid var(--border-soft)}
.parameter:last-child{border-bottom:0}
.parameter span{color:var(--text-2)}
.parameter code{color:var(--text)}
.result{border-color:var(--green)}
.result h3{color:var(--green)}
.alert{border-color:var(--red)}
.alert h3{color:var(--red)}
.download{display:inline-block;margin:10px 12px 0 0;color:var(--accent);font-weight:650}
.table-wrap{overflow-x:auto;background:var(--panel);border:1px solid var(--border-soft);border-radius:13px}
table{width:100%;border-collapse:collapse}
th{padding:10px 18px;text-align:left;font-size:11px;font-weight:650;letter-spacing:.07em;text-transform:uppercase;color:var(--text-3);border-bottom:1px solid var(--border-soft)}
td{padding:12px 18px;border-bottom:1px solid var(--border-soft);vertical-align:middle}
tr:last-child td{border-bottom:0}
tbody tr:hover{background:var(--panel-2)}
.muted{color:var(--text-3)}
.empty{padding:26px;text-align:center;color:var(--text-3)}
.artifact-grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(240px,1fr));gap:12px}
.artifact{padding:16px 18px}
.artifact strong{display:block;margin:7px 0 3px;color:var(--text)}
.artifact code{display:block;color:var(--text-3);font-size:.76rem;word-break:break-all}
.artifact a{margin-right:12px;font-weight:650}
.artifact-type{color:var(--text-3);font-size:.74rem;font-weight:700;letter-spacing:.06em;text-transform:uppercase}
.artifact-final{border-color:var(--green)}
.artifact-final .artifact-type{color:var(--green)}
.technical{margin-top:26px;color:var(--text-3);font-size:.86rem}
.technical summary{cursor:pointer}
.copy{border:1px solid var(--border);border-radius:6px;padding:2px 8px;margin-left:8px;background:var(--panel-2);color:var(--text-2);font:inherit;font-size:.8rem;cursor:pointer}
.speed-card{padding:18px}
.speed-stats{display:flex;gap:28px;flex-wrap:wrap}
.speed-stat b{display:block;color:var(--green);font-size:1.3rem}
.speed-stat small{color:var(--text-3)}
.speed-chart{display:block;width:100%;height:auto;margin-top:16px;border:1px solid var(--border);border-radius:10px;background:var(--panel-2)}
.speed-grid{stroke:#ffffff12;stroke-width:1}
.speed-line{fill:none;stroke:var(--green);stroke-width:2.5;stroke-linecap:round;stroke-linejoin:round}
.speed-point{fill:var(--green);stroke:var(--panel-2);stroke-width:2}
.speed-label{fill:var(--text-3);font-size:10px}
.library{display:grid;gap:14px;margin-top:30px}
.workload{padding:18px 22px}
.workload-head{display:flex;align-items:baseline;justify-content:space-between;gap:14px;flex-wrap:wrap}
.workload-head h2{margin:0;color:var(--text);font-size:1.05rem;letter-spacing:-.01em}
.version{margin:0;color:var(--text-3);font:.8rem var(--mono)}
.description{margin:8px 0 0;color:var(--text-2);max-width:860px}
.cap{display:inline-block;margin:12px 5px 0 0;border:1px solid var(--border);border-radius:6px;padding:2px 8px;color:var(--text-2);font:.75rem var(--mono)}
.schema-grid{display:grid;gap:10px;margin-top:14px}
.schema{margin:0;color:var(--text-2);font-size:.86rem}
.schema code{color:var(--accent)}
.schema-grid pre{background:var(--panel-2);border:1px solid var(--border);border-radius:8px;padding:12px;overflow-x:auto;color:var(--text-2);font:.8rem var(--mono)}
.params{margin-top:14px}
.keys{display:grid;gap:10px;margin-top:12px}
.keys .key{background:var(--panel-2);border:1px solid var(--border);border-radius:8px;padding:11px 13px;color:var(--text-2);font:.82rem var(--mono);word-break:break-all}
.command{background:var(--panel-2);border:1px solid var(--border);border-radius:8px;padding:12px 14px;color:var(--text-2);font:.82rem var(--mono);overflow-x:auto;white-space:pre-wrap}
.mono{font-family:var(--mono)}
.row{display:flex;justify-content:space-between;gap:16px;padding:14px 0;border-bottom:1px solid var(--border-soft)}
.row:last-child{border-bottom:0}
.k{color:var(--text-2)}
.v{color:var(--text);font-weight:650;text-align:right;word-break:break-all}
.err{margin-top:24px;border-radius:10px;padding:12px 14px;background:var(--red-soft);color:var(--red);font-weight:650}
.meta{color:var(--text-3)}
@media(max-width:900px){.layout,.two-col,.split{grid-template-columns:1fr}.pipeline{grid-template-columns:1fr 1fr}.metrics{grid-template-columns:repeat(3,1fr)}}
</style>{{end}}
@@ -5,9 +5,7 @@
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Workload library · SciMesh</title>
<style>
:root{color:#e5efff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 12% -8%,#183f77 0,transparent 32rem),radial-gradient(circle at 95% 5%,#123e39 0,transparent 29rem),#08111f}.page{max-width:1240px;margin:auto;padding:28px 22px 64px}a{color:#94bdff}.back{text-decoration:none}.eyebrow{margin:28px 0 4px;color:#7baaff;font-size:.78rem;font-weight:800;letter-spacing:.14em;text-transform:uppercase}h1{margin:0;color:#f4f8ff;font-size:clamp(2rem,5vw,3.25rem);letter-spacing:-.055em}.lead{max-width:760px;margin:10px 0 0;color:#aabed9;font-size:1.06rem}.library{display:grid;gap:14px;margin-top:30px}.workload{border:1px solid #294662;border-radius:15px;background:#0d1a2cdd;box-shadow:0 20px 45px #00000021;padding:20px 22px}.workload-head{display:flex;align-items:baseline;justify-content:space-between;gap:14px;flex-wrap:wrap}.workload-head h2{margin:0;color:#f2f7ff;font-size:1.22rem;letter-spacing:-.02em}.version{margin:0;color:#7d93b2;font:0.82rem ui-monospace,SFMono-Regular,monospace}.description{margin:8px 0 0;color:#b9c9e2;max-width:860px}.cap{display:inline-block;margin:12px 5px 0 0;border:1px solid #375978;border-radius:5px;padding:2px 7px;color:#b5d3f5;font:.76rem ui-monospace,SFMono-Regular,monospace}.badge{display:inline-flex;align-items:center;border-radius:999px;padding:4px 10px;font-size:.78rem;font-weight:800}.badge-success{background:#123f34;color:#76efb5}.badge-waiting{background:#23344d;color:#b9cce9}.meta{display:flex;gap:9px;flex-wrap:wrap;margin-top:14px}.meta span{border:1px solid #2b4a6b;border-radius:7px;padding:3px 8px;color:#a9c3e2;font-size:.8rem}.meta b{color:#dbe9fb;font-weight:750}.schema-grid{display:grid;grid-template-columns:1fr 1fr;gap:13px;margin-top:16px}.schema{border:1px solid #233e5c;border-radius:11px;background:#091627;padding:13px}.schema h3{margin:0 0 8px;color:#cfe1f7;font-size:.86rem;letter-spacing:.04em;text-transform:uppercase}.schema pre{margin:0;overflow:auto;max-height:300px;color:#9fc1e8;font:.76rem ui-monospace,SFMono-Regular,monospace;white-space:pre-wrap;word-break:break-word}.params{padding:13px}.params h3{margin:0 0 8px;color:#cfe1f7;font-size:.86rem;letter-spacing:.04em;text-transform:uppercase}.param{margin:0;padding:6px 0;border-bottom:1px dashed #223a56;color:#b9c9e2;font-size:.9rem}.param:last-child{border-bottom:0}.param b{color:#e8f2ff}.param small{display:block;margin-top:2px;color:#7f96b5}.empty{padding:30px;border:1px dashed #35516f;border-radius:14px;color:#9ab0cb;text-align:center}@media(max-width:820px){.schema-grid{grid-template-columns:1fr}}
</style>
{{template "ui-styles"}}
</head>
<body>
<main class="page">
@@ -8,6 +8,7 @@ import (
"net/url"
"strconv"
"strings"
"time"
"github.com/google/uuid"
@@ -390,3 +391,43 @@ func (s *Server) adminOwnerEmails(r *http.Request) map[uuid.UUID]string {
}
return out
}
// handleUIAdminPruneJSON deletes finished jobs older than the requested
// number of days (with all their artifacts) and reports what was freed.
func (s *Server) handleUIAdminPruneJSON(w http.ResponseWriter, r *http.Request) {
var body struct {
OlderThanDays int `json:"older_than_days"`
}
if err := decodeJSON(r, &body); err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
if body.OlderThanDays < 1 || body.OlderThanDays > 3650 {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
result, err := s.uc.PruneArtifacts.Execute(ctx, time.Duration(body.OlderThanDays)*24*time.Hour)
if err != nil {
s.writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, result)
}
// handleUIAdminRemoveWorkerJSON deletes an offline worker.
func (s *Server) handleUIAdminRemoveWorkerJSON(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(r.PathValue("id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
if err := s.uc.Admin.RemoveWorker(ctx, id); err != nil {
s.writeError(w, r, err)
return
}
w.WriteHeader(http.StatusNoContent)
}
@@ -9,6 +9,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
@@ -116,3 +117,38 @@ func (s *Server) callUserserviceAuthedBody(ctx context.Context, method, path, be
}
return resp.StatusCode, respBody, nil
}
// handleWorkerTokenExchangeProxy forwards a worker-key exchange to the
// userservice. The key itself is the credential, so this route is public —
// exactly like the userservice's own endpoint. In `serve` mode the embedded
// userservice binds loopback only, so workers need the coordinator to front
// the exchange for them.
func (s *Server) handleWorkerTokenExchangeProxy(w http.ResponseWriter, r *http.Request) {
if s.userserviceURL == "" {
writeJSON(w, http.StatusNotFound, map[string]string{"error": "not found"})
return
}
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
req, err := http.NewRequestWithContext(r.Context(), http.MethodPost, s.userserviceURL+"/worker-tokens/exchange", bytes.NewReader(body)) //nolint:gosec // G704: path is fixed, host is config
if err != nil {
s.writeError(w, r, err)
return
}
req.Header.Set("Content-Type", "application/json")
resp, err := s.httpClient.Do(req) //nolint:gosec // G704: see above
if err != nil {
s.writeError(w, r, err)
return
}
defer func() { _ = resp.Body.Close() }()
respBody, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
s.writeError(w, r, err)
return
}
proxyJSON(w, resp.StatusCode, respBody)
}
+17
View File
@@ -521,3 +521,20 @@ func (a *Admin) RevealWorkerToken(ctx context.Context, actor string) string {
}
return token
}
// RemoveWorker deletes an offline worker from the registry. Online or busy
// workers are refused: an admin console must never yank a live machine out
// from under a running task.
func (a *Admin) RemoveWorker(ctx context.Context, id uuid.UUID) error {
if a.workers == nil {
return domain.ErrWorkerNotFound
}
worker, err := a.workers.Get(ctx, id)
if err != nil {
return err
}
if worker.Status != domain.WorkerOffline {
return domain.ErrInvalidInput
}
return a.workers.Delete(ctx, id)
}
@@ -336,3 +336,30 @@ func TestAdminRevealToken(t *testing.T) {
t.Errorf("token = %q", got)
}
}
type removableWorkerRepo struct {
WorkerRepository
deleted uuid.UUID
}
func (f *removableWorkerRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Worker, error) {
return &domain.Worker{ID: id, Status: domain.WorkerOffline}, nil
}
func (f *removableWorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
f.deleted = id
return nil
}
func TestAdminRemoveWorker(t *testing.T) {
a := adminFixture()
repo := &removableWorkerRepo{}
a.workers = repo
id := uuid.New()
if err := a.RemoveWorker(context.Background(), id); err != nil {
t.Fatal(err)
}
if repo.deleted != id {
t.Error("offline worker must be deleted")
}
}
+9
View File
@@ -82,6 +82,12 @@ type JobRepository interface {
ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error)
CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error
FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error
// ListCompletedBefore returns jobs that finished (completed or failed)
// before the cutoff, for the admin artifact pruner.
ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error)
// Delete removes a job row; the engine cascades its tasks, artifacts and
// quorum votes. Blob files must be removed separately.
Delete(ctx context.Context, id uuid.UUID) error
}
// WorkerRepository persists the worker registry.
@@ -97,6 +103,9 @@ type WorkerRepository interface {
// SetTrust reclassifies a worker's trust level (trusted/untrusted). Returns
// ErrNotFound when the id is unknown.
SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error
// Delete removes a worker from the registry. Returns ErrNotFound when the
// id is unknown.
Delete(ctx context.Context, id uuid.UUID) error
}
// ArtifactRepository persists artifact metadata. The bytes live in a BlobStore;
+55
View File
@@ -0,0 +1,55 @@
package usecase
import (
"context"
"time"
)
// PruneArtifacts removes completed or failed jobs older than `olderThan` and
// every artifact they own: database rows cascade, blob files are deleted
// explicitly. It returns what was freed so the admin console can report it.
type PruneArtifacts struct {
jobs JobRepository
read UIReadRepository
blobs BlobStore
clk Clock
}
func NewPruneArtifacts(jobs JobRepository, read UIReadRepository, blobs BlobStore, clk Clock) *PruneArtifacts {
return &PruneArtifacts{jobs: jobs, read: read, blobs: blobs, clk: clk}
}
type PruneResult struct {
Jobs int `json:"jobs"`
Artifacts int `json:"artifacts"`
FreedBytes int64 `json:"freed_bytes"`
}
// Execute deletes finished jobs whose completion timestamp is older than the
// cutoff. Jobs that are still active are never touched.
func (uc *PruneArtifacts) Execute(ctx context.Context, olderThan time.Duration) (PruneResult, error) {
cutoff := uc.clk.Now().Add(-olderThan)
jobs, err := uc.jobs.ListCompletedBefore(ctx, cutoff)
if err != nil {
return PruneResult{}, err
}
out := PruneResult{}
for _, job := range jobs {
artifacts, err := uc.read.ListArtifactsByJob(ctx, job.ID)
if err != nil {
return out, err
}
for _, artifact := range artifacts {
if err := uc.blobs.Delete(ctx, artifact.StorageKey); err != nil {
return out, err
}
out.FreedBytes += artifact.SizeBytes
out.Artifacts++
}
if err := uc.jobs.Delete(ctx, job.ID); err != nil {
return out, err
}
out.Jobs++
}
return out, nil
}
@@ -980,3 +980,40 @@ func TestSubmitDatasetRejectsDisabledWorkload(t *testing.T) {
t.Fatalf("submit after re-enable: %v", err)
}
}
func TestPruneArtifactsRemovesOldFinishedJobs(t *testing.T) {
h := newHarness()
old := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: domain.JobCompleted, CreatedAt: old, CompletedAt: &old}
if err := h.jobs.Insert(context.Background(), job); err != nil {
t.Fatal(err)
}
art, err := domain.NewArtifact(job.ID, nil, domain.ArtifactFinalResult, "r.csv", "text/csv", old)
if err != nil {
t.Fatal(err)
}
art.SetContent("sha", 42)
if err := h.arts.Insert(context.Background(), art); err != nil {
t.Fatal(err)
}
// An active job must survive the prune.
active := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: domain.JobRunning, CreatedAt: old}
if err := h.jobs.Insert(context.Background(), active); err != nil {
t.Fatal(err)
}
prune := usecase.NewPruneArtifacts(h.jobs, memstore.NewUIReadRepo(h.jobs, h.tasks, h.work, h.arts), h.blobs, h.clk)
result, err := prune.Execute(context.Background(), 7*24*time.Hour)
if err != nil {
t.Fatal(err)
}
if result.Jobs != 1 || result.Artifacts != 1 || result.FreedBytes != 42 {
t.Errorf("prune = %+v, want 1 job / 1 artifact / 42 bytes", result)
}
if _, err := h.jobs.Get(context.Background(), job.ID); err == nil {
t.Error("finished job must be gone")
}
if _, err := h.jobs.Get(context.Background(), active.ID); err != nil {
t.Error("active job must survive the prune")
}
}
@@ -597,6 +597,206 @@
"upload_ready": true,
"verifier": "exact-artifact@1",
"version": "1.0.0"
},
{
"capabilities": [
"similarity-search-parallel"
],
"description": "Exact top-k Tanimoto molecular similarity search over deterministic TSV shards with a bounded merge; each shard is fingerprinted and scored across a thread pool. Output is byte-identical to similarity-search.",
"determinism": "byte_exact",
"enabled": true,
"inputs": {
"input": {
"allow_nested_collections": false,
"canonicalizer": "scimesh-tsv-v1",
"encoding": "utf-8",
"max_bytes": 10737418240,
"max_dimensions": [],
"max_records": 100000000,
"media_type": "text/tab-separated-values",
"privacy_class": "project",
"ref": "molecule-table@1",
"retention_class": "durable",
"streaming": false,
"validator": "delimited-table@1",
"validator_configuration": {
"required_columns": [
"canonical_smiles",
"chembl_id"
]
}
}
},
"name": "similarity-search-parallel",
"outputs": {
"result": {
"allow_nested_collections": false,
"canonicalizer": "scimesh-search-result-v1",
"encoding": "utf-8",
"max_bytes": 1073741824,
"max_dimensions": [],
"max_records": 100000,
"media_type": "text/csv",
"privacy_class": "project",
"ref": "similarity-search-result@1",
"retention_class": "durable",
"streaming": false,
"validator": "delimited-table@1",
"validator_configuration": {
"columns": [
"rank",
"chembl_id",
"canonical_smiles",
"similarity"
]
}
}
},
"parameters_schema": {
"additionalProperties": false,
"oneOf": [
{
"not": {
"required": [
"query_smiles"
]
},
"required": [
"query_id"
]
},
{
"not": {
"required": [
"query_id"
]
},
"required": [
"query_smiles"
]
}
],
"properties": {
"max_rows": {
"minimum": 1,
"type": "integer"
},
"progress_every": {
"minimum": 0,
"type": "integer"
},
"query_id": {
"maxLength": 200,
"minLength": 1,
"type": "string"
},
"query_smiles": {
"maxLength": 200,
"minLength": 1,
"type": "string"
},
"threads": {
"description": "Threads used to fingerprint and score one shard (default: CPU count).",
"minimum": 1,
"type": "integer"
},
"threshold": {
"maximum": 1,
"minimum": 0,
"type": "number"
},
"threshold_direction": {
"enum": [
"greater",
"less"
]
},
"top_k": {
"minimum": 1,
"type": "integer"
}
},
"type": "object"
},
"reduction": "top-k",
"trust_modes": [
"trusted",
"untrusted_quorum"
],
"ui_elements": [
{
"default": null,
"field": "query_id",
"group": "",
"help": "ChEMBL id of the query molecule. Provide exactly one of id or SMILES.",
"label": "Query molecule id",
"options": [],
"order": 1,
"placeholder": "",
"widget": "text"
},
{
"default": null,
"field": "query_smiles",
"group": "",
"help": "SMILES of the query molecule. Provide exactly one of id or SMILES.",
"label": "Query molecule SMILES",
"options": [],
"order": 2,
"placeholder": "",
"widget": "text"
},
{
"default": 20,
"field": "top_k",
"group": "",
"help": "Number of most similar molecules to keep per shard (global merge keeps the best of these).",
"label": "Top k",
"options": [],
"order": 3,
"placeholder": "",
"widget": "number"
},
{
"default": "greater",
"field": "threshold_direction",
"group": "",
"help": "Keep molecules with similarity greater or less than the threshold.",
"label": "Direction",
"options": [
"greater",
"less"
],
"order": 4,
"placeholder": "",
"widget": "select"
},
{
"default": null,
"field": "threshold",
"group": "",
"help": "Optional similarity bound: results are filtered to this direction.",
"label": "Similarity threshold",
"options": [],
"order": 5,
"placeholder": "e.g. 0.8",
"widget": "number"
},
{
"default": null,
"field": "threads",
"group": "",
"help": "Threads used to fingerprint and score one shard (default: CPU count).",
"label": "Threads per shard",
"options": [],
"order": 6,
"placeholder": "auto",
"widget": "number"
}
],
"upload_ready": true,
"verifier": "exact-artifact@1",
"version": "1.0.0"
}
]
}
+7
View File
@@ -59,6 +59,13 @@ curl -fsSL https://raw.githubusercontent.com/emil28092005/SciMesh/main/install.s
# the installer opens the local wizard at http://127.0.0.1:12700 automatically
```
On a headless server (no desktop environment), RDKit needs a few X11
libraries that desktops already ship — install them once with apt:
```bash
sudo apt-get install -y libxrender1 libxext6 libxcursor1 libxfixes3 libxi6 libxrandr2
```
Or configure by hand:
```bash
+5 -2
View File
@@ -1,10 +1,10 @@
[build-system]
requires = ["setuptools>=68"]
requires = ["setuptools>=68", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "scimesh"
version = "0.1.0"
dynamic = ["version"]
description = "Local scientific workloads for molecular similarity analysis"
readme = "README.md"
requires-python = ">=3.10"
@@ -23,6 +23,7 @@ scimesh = "scimesh.cli:main"
[project.entry-points."scimesh.workloads"]
"similarity-search@1.0.0" = "scimesh.workloads.search:workload_definition"
"similarity-search-parallel@1.0.0" = "scimesh.workloads.search_parallel:workload_definition"
"similarity-graph@1.0.0" = "scimesh.workloads.graph:workload_definition"
"descriptor-batch@1.0.0" = "scimesh.workloads.descriptors:workload_definition"
"molwt-filter@1.0.0" = "scimesh.workloads.molwt_filter:workload_definition"
@@ -40,3 +41,5 @@ venv = ".venv"
pythonVersion = "3.10"
typeCheckingMode = "basic"
exclude = ["SciMesh", "site", "coordinator"]
[tool.setuptools_scm]
+6
View File
@@ -21,6 +21,7 @@ from .environment import current_environment_digest
from .graph import similarity_graph_sdk_definition
from .molwt_filter import molwt_filter_sdk_definition
from .search import similarity_search_sdk_definition
from .search_parallel import similarity_search_parallel_sdk_definition
__all__ = [
"default_sdk_registry",
@@ -46,6 +47,10 @@ def default_sdk_registry(
similarity_search_sdk_definition(shard_rows=shard_rows).definition(),
enabled=True,
)
registry.register(
similarity_search_parallel_sdk_definition(shard_rows=shard_rows).definition(),
enabled=True,
)
registry.register(
similarity_graph_sdk_definition().definition(),
enabled=True,
@@ -82,6 +87,7 @@ def default_sdk_runtime(
workload_capabilities
or (
"similarity-search",
"similarity-search-parallel",
"similarity-graph",
"descriptor-batch",
"molwt-filter",
@@ -0,0 +1,28 @@
"""SDK-built ``similarity-search-parallel`` workload.
Same contract as ``similarity-search`` with a per-shard thread pool. See
``core.py`` for the parallel scoring core and ``definition.py`` for the
manifest-backed handlers.
"""
from .core import (
run_search_shard_parallel,
search_similar_parallel,
write_search_shards,
)
from .definition import (
MAP_ENTRY_POINT,
SimilaritySearchParallelSDKWorkload,
similarity_search_parallel_sdk_definition,
workload_definition,
)
__all__ = [
"MAP_ENTRY_POINT",
"SimilaritySearchParallelSDKWorkload",
"similarity_search_parallel_sdk_definition",
"workload_definition",
"run_search_shard_parallel",
"search_similar_parallel",
"write_search_shards",
]
+199
View File
@@ -0,0 +1,199 @@
"""Scientific core for the SDK-built ``similarity-search-parallel`` workload.
The exact same semantics as ``similarity-search`` identical partial format,
identical bounded merge, byte-identical output but the per-molecule
fingerprinting and Tanimoto scoring of one shard run across a thread pool
(``threads`` parameter, default = CPU count).
Parallelism is confined to the scoring phase: ``ThreadPoolExecutor.map`` keeps
the input row order, so the results are merged exactly like the sequential
reference (same ``_HeapEntry`` logic), which makes the output byte-identical
for every thread count by construction.
"""
from __future__ import annotations
import heapq
import os
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Mapping
from rdkit import Chem
from rdkit.Chem import DataStructs
from scimesh.chemistry.dataset import MoleculeRecord, parse_smiles
from scimesh.chemistry.fingerprints import fingerprint
from scimesh.workloads.search.core import (
run_search_shard,
write_search_partial,
write_search_shards,
)
from scimesh.workloads.similarity_search import (
DatasetStats,
SearchResult,
SimilarityMatch,
_HeapEntry,
iter_valid_molecules,
)
def search_similar_parallel(
tsv_path: Path,
query: MoleculeRecord,
top_k: int,
*,
threads: int = 0,
max_rows: int | None = None,
threshold: float | None = None,
threshold_direction: str = "greater",
) -> SearchResult:
"""Exact top-k matches with a bounded heap, scored by a thread pool.
Identical selection and ordering to ``search_similar`` for every thread
count: the merge runs in row order over the parallel-computed scores.
"""
if top_k < 1:
raise ValueError("--top-k must be a positive integer")
if threads < 0:
raise ValueError("threads must be a non-negative integer")
if threshold is not None and not 0.0 <= threshold <= 1.0:
raise ValueError("--threshold must be between 0 and 1")
if threshold_direction not in {"greater", "less"}:
raise ValueError("--threshold-direction must be 'greater' or 'less'")
workers = threads or (os.cpu_count() or 1)
query_fingerprint = fingerprint(query.molecule)
query_canonical_smiles = Chem.MolToSmiles(query.molecule, canonical=True)
stats = DatasetStats()
records = list(iter_valid_molecules(tsv_path, stats, max_rows=max_rows))
def score(record: MoleculeRecord):
candidate_smiles = Chem.MolToSmiles(record.molecule, canonical=True)
if (
record.molecule_id == query.molecule_id
or candidate_smiles == query_canonical_smiles
):
return None
similarity = DataStructs.TanimotoSimilarity(
query_fingerprint, fingerprint(record.molecule)
)
if threshold is not None and (
similarity < threshold
if threshold_direction == "greater"
else similarity > threshold
):
return None
return similarity
# map preserves the input order, so the merge below is exactly the
# sequential reference's merge, just over precomputed scores.
with ThreadPoolExecutor(max_workers=workers) as pool:
scored = pool.map(score, records)
heap: list[_HeapEntry] = []
for record, similarity in zip(records, scored):
if similarity is None:
continue
match = SimilarityMatch(similarity, record.molecule_id, record.smiles)
rank_key = match.sort_key(threshold_direction)
entry = _HeapEntry(match, rank_key)
if len(heap) < top_k:
heapq.heappush(heap, entry)
elif rank_key < heap[0].rank_key:
heapq.heapreplace(heap, entry)
matches = [entry.match for entry in sorted(heap, key=lambda e: e.rank_key)]
return SearchResult(matches=matches, stats=stats)
def run_search_shard_parallel(
input_path: Path,
parameters: Mapping[str, object],
output_path: Path,
) -> dict[str, int]:
"""Run one planned shard with the parallel scoring core.
Accepts the same parameters as ``similarity-search`` plus ``threads``.
"""
allowed = {
"query_id",
"query_smiles",
"top_k",
"threshold",
"threshold_direction",
"progress_every",
"threads",
}
unknown = set(parameters) - allowed
if unknown:
raise ValueError(
f"unsupported similarity-search-parallel parameters: {', '.join(sorted(unknown))}"
)
query_smiles = parameters.get("query_smiles")
query_id = parameters.get("query_id")
if isinstance(query_id, str) and not isinstance(query_smiles, str):
from rdkit import Chem
from scimesh.chemistry.dataset import find_molecule_by_id
record = find_molecule_by_id(input_path, query_id)
query_smiles = Chem.MolToSmiles(record.molecule, canonical=True)
if not isinstance(query_smiles, str) or not query_smiles.strip():
raise ValueError("query_smiles is required for a distributed shard")
molecule = parse_smiles(query_smiles)
if molecule is None:
raise ValueError("query_smiles is invalid")
top_k = _positive_int(parameters.get("top_k", 20), "top_k")
threads = _nonnegative_int(parameters.get("threads", 0), "threads")
threshold = None
if "threshold" in parameters:
threshold = _unit_interval(parameters["threshold"], "threshold")
direction = parameters.get("threshold_direction", "greater")
if direction not in {"greater", "less"}:
raise ValueError("threshold_direction must be 'greater' or 'less'")
assert isinstance(direction, str)
result = search_similar_parallel(
input_path,
MoleculeRecord("query", query_smiles, molecule),
top_k=top_k,
threads=threads,
threshold=threshold,
threshold_direction=direction,
)
write_search_partial(output_path, result.matches)
return {
"scanned_rows": result.stats.scanned,
"valid_molecules": result.stats.valid,
"invalid_smiles": result.stats.invalid,
"matches_emitted": len(result.matches),
}
def _positive_int(value: object, name: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 1:
raise ValueError(f"{name} must be a positive integer")
return value
def _nonnegative_int(value: object, name: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise ValueError(f"{name} must be a non-negative integer")
return value
def _unit_interval(value: object, name: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{name} must be a number between 0 and 1")
return float(value)
# The shared shard writer is re-exported so the workload definition can reuse
# the deterministic partitioning without importing search internals.
__all__ = [
"search_similar_parallel",
"run_search_shard_parallel",
"write_search_shards",
"run_search_shard",
]
@@ -0,0 +1,144 @@
"""SDK-built ``similarity-search-parallel`` workload definition and handlers.
A subclass of ``SimilaritySearchSDKWorkload``: identical contract (plan-time
query resolution, deterministic sharding, top-k reduction, byte-identical
partials), but each shard's fingerprinting and scoring runs across a thread
pool (``threads`` parameter, default = CPU count).
"""
from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from typing import Any
from scimesh.sdk.batch import MapReduceWorkload
from scimesh.sdk.identity import WorkloadId
from scimesh.sdk.plans import JobRequest
from scimesh.sdk.registry import WorkloadDefinition
from scimesh.sdk.ui import UIElement
from ..environment import current_environment_digest, current_scimesh_package_digest
from ..search.definition import SimilaritySearchSDKWorkload, _parameters_schema
from .core import run_search_shard_parallel
MAP_ENTRY_POINT = "scimesh.workloads.search_parallel.definition:map_search_parallel@v1"
# The parallel variant adds only the thread-count parameter on top of the
# search contract; everything else (schemas, entry points of the reduce stage,
# partitioning) is inherited.
_MAP_PARAMETERS = (
"query_id",
"query_smiles",
"top_k",
"threshold",
"threshold_direction",
"progress_every",
"threads",
)
def _parallel_parameters_schema() -> dict[str, Any]:
schema = dict(_parameters_schema())
properties = dict(schema["properties"])
properties["threads"] = {
"type": "integer",
"minimum": 1,
"description": "Threads used to fingerprint and score one shard (default: CPU count).",
}
schema["properties"] = properties
return schema
class SimilaritySearchParallelSDKWorkload(SimilaritySearchSDKWorkload):
"""Exact top-k Tanimoto search with a per-shard thread pool."""
workload_id = WorkloadId("similarity-search-parallel", "1.0.0")
description = (
"Exact top-k Tanimoto molecular similarity search over deterministic "
"TSV shards with a bounded merge; each shard is fingerprinted and "
"scored across a thread pool. Output is byte-identical to "
"similarity-search."
)
parameters_schema = _parallel_parameters_schema()
map_parameter_names = _MAP_PARAMETERS
map_entry_point = MAP_ENTRY_POINT
ui_elements = SimilaritySearchSDKWorkload.ui_elements + (
UIElement(
"threads",
"number",
"Threads per shard",
help="Threads used to fingerprint and score one shard (default: CPU count).",
placeholder="auto",
order=6,
),
)
def domain_validate(self, parameters: Mapping[str, Any]) -> None:
# The search base rejects unknown parameters; threads is our addition,
# so it is validated here and stripped before delegating.
rest = dict(parameters)
threads = rest.pop("threads", None)
if threads is not None and (
isinstance(threads, bool) or not isinstance(threads, int) or threads < 1
):
raise ValueError("threads must be a positive integer")
super().domain_validate(rest)
def resolved_parameters_for_plan(
self,
job,
input_path,
resolved,
):
# threads is a map-stage-only knob; strip it from the plan-level
# resolved parameters so the reduce stage projection stays clean.
resolved = super().resolved_parameters_for_plan(job, input_path, resolved)
stripped = dict(resolved)
stripped.pop("threads", None)
return stripped
def resolved_parameters(self, request: JobRequest) -> dict[str, Any]:
resolved = super().resolved_parameters(request)
if "threads" in request.parameters:
threads = request.parameters["threads"]
if isinstance(threads, bool) or not isinstance(threads, int) or threads < 1:
raise ValueError("threads must be a positive integer")
resolved["threads"] = threads
return resolved
def compute_shard(
self,
inputs: Mapping[str, Path],
parameters: Mapping[str, Any],
output_path: Path,
) -> Mapping[str, int | float]:
return run_search_shard_parallel(inputs["input"], parameters, output_path)
def similarity_search_parallel_sdk_definition(
*,
shard_rows: int = 10_000,
package_digest: str | None = None,
environment_digest: str | None = None,
) -> SimilaritySearchParallelSDKWorkload:
"""Build the SDK-built parallel similarity-search definition for tests."""
return SimilaritySearchParallelSDKWorkload(
shard_rows=shard_rows,
package_digest=package_digest or current_scimesh_package_digest(),
environment_digest=environment_digest or current_environment_digest(),
)
def workload_definition() -> WorkloadDefinition:
"""Installed entry-point factory for the SDK-built parallel search."""
return similarity_search_parallel_sdk_definition().definition()
def map_search_parallel(
input_path: Path,
parameters: Mapping[str, object],
output_path: Path,
) -> dict[str, int]:
"""Digest-pinned map entry point for the parallel search shard."""
return run_search_shard_parallel(input_path, parameters, output_path)
+7 -1
View File
@@ -244,7 +244,13 @@ def test_workload_cli_exports_the_library_as_json(tmp_path: Path) -> None:
assert payload["schema_version"] == 2
names = [item["name"] for item in payload["workloads"]]
assert names == sorted(
["descriptor-batch", "molwt-filter", "similarity-graph", "similarity-search"]
[
"descriptor-batch",
"molwt-filter",
"similarity-graph",
"similarity-search",
"similarity-search-parallel",
]
)
for item in payload["workloads"]:
assert item["version"] == "1.0.0"
+1 -1
View File
@@ -60,7 +60,7 @@ def _registered_similarity_search(shard_rows: int = 2):
registry = default_sdk_registry(shard_rows=shard_rows)
runtime = default_sdk_runtime()
descriptions = registry.descriptions()
assert len(descriptions) == 4
assert len(descriptions) == 5
description = next(
item for item in descriptions if item.workload.name == "similarity-search"
)
+185
View File
@@ -0,0 +1,185 @@
"""Tests for the SDK-built similarity-search-parallel workload."""
from __future__ import annotations
import csv
from pathlib import Path
import pytest
from scimesh.sdk import (
ArtifactCollection,
DeterminismProfile,
JobRequest,
LocalArtifactStore,
LocalCoreBatchExecutor,
LocalPlanningContext,
StageKind,
)
from scimesh.workloads.library import default_sdk_registry, default_sdk_runtime
from scimesh.workloads.search.core import run_search_shard, write_search_shards
from scimesh.workloads.search_parallel import (
run_search_shard_parallel,
search_similar_parallel,
)
from scimesh.workloads.similarity_search import (
find_molecule_by_id,
search_similar,
write_search_results,
)
def _write_dataset(path: Path, molecules: list[tuple[str, str]]) -> None:
path.write_text(
"chembl_id\tcanonical_smiles\n"
+ "".join(f"{mid}\t{smiles}\n" for mid, smiles in molecules),
encoding="utf-8",
)
def _tie_dataset(path: Path) -> None:
# Deliberate similarity ties: propanol isomers and duplicated rows, so the
# parallel merge must reproduce the sequential row-order preference.
_write_dataset(
path,
[
("QUERY", "CCO"),
("A1", "CCCO"),
("A2", "C(CC)O"),
("B", "CCN"),
("C1", "CCC"),
("C2", "CCC"),
("D", "CCCC"),
],
)
def test_parallel_matches_sequential_byte_exactly(tmp_path: Path) -> None:
dataset = tmp_path / "molecules.tsv"
_tie_dataset(dataset)
query = find_molecule_by_id(dataset, "QUERY")
reference = search_similar(dataset, query, top_k=5, progress_every=0)
reference_path = tmp_path / "reference.csv"
write_search_results(reference_path, reference.matches)
for threads in (1, 2, 4):
parallel = search_similar_parallel(dataset, query, top_k=5, threads=threads)
parallel_path = tmp_path / f"parallel-{threads}.csv"
write_search_results(parallel_path, parallel.matches)
assert parallel_path.read_bytes() == reference_path.read_bytes(), (
f"threads={threads} diverged from the reference"
)
def test_parallel_shard_matches_sequential_shard(tmp_path: Path) -> None:
dataset = tmp_path / "molecules.tsv"
_tie_dataset(dataset)
shard_dir = tmp_path / "shards"
shard_dir.mkdir()
shards = write_search_shards(dataset, shard_dir, shard_rows=2)
parameters = {"query_smiles": "CCO", "top_k": 3, "threads": 4}
sequential_out = tmp_path / "seq.tsv"
parallel_out = tmp_path / "par.tsv"
sequential_parameters = dict(parameters)
sequential_parameters.pop("threads")
run_search_shard(shards[0], sequential_parameters, sequential_out)
run_search_shard_parallel(shards[0], parameters, parallel_out)
assert parallel_out.read_bytes() == sequential_out.read_bytes()
with parallel_out.open(encoding="utf-8") as handle:
rows = list(csv.DictReader(handle))
assert rows[0]["rank"] == "1"
assert rows[0]["similarity"].startswith("0.5") # CCO vs CCCO
assert len(rows) <= 3
def test_parallel_rejects_bad_parameters(tmp_path: Path) -> None:
dataset = tmp_path / "molecules.tsv"
_tie_dataset(dataset)
output = tmp_path / "out.tsv"
with pytest.raises(ValueError, match="threads must be a non-negative integer"):
run_search_shard_parallel(
dataset, {"query_smiles": "CCO", "threads": -1}, output
)
with pytest.raises(ValueError, match="unsupported"):
run_search_shard_parallel(dataset, {"query_smiles": "CCO", "nope": 1}, output)
with pytest.raises(ValueError, match="query_smiles is invalid"):
run_search_shard_parallel(dataset, {"query_smiles": "СС"}, output)
def _registered_parallel_search(shard_rows: int = 2):
registry = default_sdk_registry(shard_rows=shard_rows)
runtime = default_sdk_runtime()
description = next(
item
for item in registry.descriptions()
if item.workload.name == "similarity-search-parallel"
)
definition, negotiated = registry.require(
description.workload.name,
description.workload.version,
description.package_digest,
runtime=runtime,
)
return registry, runtime, description, definition, negotiated
def test_parallel_manifest_is_registered_and_negotiable() -> None:
_, runtime, description, definition, negotiated = _registered_parallel_search()
manifest = definition.manifest
assert description.enabled is True
assert manifest.workload.name == "similarity-search-parallel"
assert manifest.workload.version == "1.0.0"
assert manifest.determinism is DeterminismProfile.BYTE_EXACT
assert manifest.verifier.verifier.canonical == "exact-artifact@1"
assert set(mode.value for mode in manifest.trust_modes) == {
"trusted",
"untrusted_quorum",
}
assert [stage.kind for stage in manifest.workflow.stages] == [
StageKind.MAP,
StageKind.REDUCE,
]
assert "threads" in manifest.parameters_schema["properties"]
assert negotiated is not None
assert runtime is not None
def test_parallel_executor_matches_reference(tmp_path: Path) -> None:
dataset = tmp_path / "molecules.tsv"
_tie_dataset(dataset)
registry, runtime, description, definition, _ = _registered_parallel_search()
artifact_store = LocalArtifactStore(tmp_path / "artifacts")
input_port = definition.manifest.inputs["input"]
dataset_artifact = artifact_store.import_file(
dataset,
declaration=input_port.schema,
)
request = JobRequest(
workload=definition.manifest.workload,
parameters={"query_id": "QUERY", "top_k": 3, "threads": 2},
inputs={"input": ArtifactCollection.single(dataset_artifact)},
)
result = LocalCoreBatchExecutor(
registry,
runtime,
artifact_store,
tmp_path / "sdk-work",
).execute(request, description.package_digest)
result_artifact = result.outputs["result"].items[0].artifact
reference_path = tmp_path / "reference.csv"
query = find_molecule_by_id(dataset, "QUERY")
reference = search_similar(dataset, query, top_k=3, progress_every=0)
write_search_results(reference_path, reference.matches)
assert (
artifact_store.materialize(result_artifact).read_bytes()
== reference_path.read_bytes()
)
assert result.task_key == "reduce/final"