Secure user worker operations

PLAN.md

@@ -66,7 +66,8 @@ Coordinator reducer -> final artifact -> download/status API
 
 - cloud object storage, Kubernetes, autoscaling, and multi-region operation;
 - arbitrary shell commands sent by coordinator to workers;
-- user accounts, multi-tenancy, billing, or sophisticated authorization;
+- user accounts, multi-tenancy, billing, or sophisticated authorization
+  (planned after the first release in CTX-15);
 - GPU scheduling and multiprocessing inside a worker;
 - Docker as a required runtime dependency;
 - video/CV processing implementation;
@@ -820,6 +821,71 @@ CTX-09 enables final result downloads.
 - failure/retry scenarios have automated coverage;
 - README contains architecture diagram, security caveat, and troubleshooting.
 
+### CTX-13 — In-worker CPU parallelism
+
+**Goal:** Allow a worker to use a bounded, configured number of CPU threads or
+processes while preserving the existing one-task-per-lease coordinator model.
+
+**Depends on:** CTX-12.
+
+**Acceptance criteria:**
+
+- worker concurrency is an explicit configuration value with a safe default of
+  one;
+- a task's internal parallel execution has bounded memory and does not build a
+  dense N×N similarity matrix;
+- CPU-parallel `similarity-search` and `similarity-graph` outputs match the
+  single-threaded local reference byte-for-byte where ordering is observable;
+- result ordering is deterministic across worker counts and block sizes;
+- cancellation, lease loss, and worker failure stop child work safely and do
+  not report a successful result;
+- benchmarks and tests cover one-worker and multi-worker configurations.
+
+### CTX-14 — GPU-accelerated workload execution
+
+**Goal:** Add an optional GPU execution backend for supported molecular
+workloads, while retaining the validated CPU implementation as the reference
+and fallback.
+
+**Depends on:** CTX-13.
+
+**Acceptance criteria:**
+
+- GPU capability and backend version are advertised explicitly by a worker;
+- the coordinator schedules GPU work only to compatible workers and CPU-only
+  workers continue to claim CPU tasks;
+- unsupported hardware, unavailable drivers, and GPU execution errors produce
+  sanitized failures or a documented CPU fallback;
+- GPU results match the CPU reference within a documented, tested numerical
+  tolerance and preserve deterministic output ordering;
+- GPU memory use is bounded and no dense N×N similarity matrix is created;
+- CPU-only CI verifies backend selection and contract behavior, with GPU
+  integration tests documented for compatible runners.
+
+### CTX-15 — User Service and access control
+
+**Goal:** Introduce a dedicated User Service for user identity and access
+control, without coupling workers to user credentials or moving scientific
+workload logic into the service.
+
+**Depends on:** CTX-12.
+
+**Acceptance criteria:**
+
+- the service has a versioned, documented API in
+  [`docs/user-service-api-contract.md`](docs/user-service-api-contract.md) and
+  owns user identity data;
+- credentials and authentication tokens are stored and handled securely; they
+  are never logged or exposed to workers;
+- authenticated identity is propagated to coordinator requests through an
+  explicit, validated boundary;
+- authorization restricts access to jobs and artifacts to the intended user or
+  project;
+- unauthenticated, expired-token, and cross-user access attempts have
+  automated failure tests;
+- the existing single-operator demo remains usable through a documented local
+  development configuration.
+
 ---
 
 ## 10. Suggested assignment bundles
@@ -948,12 +1014,15 @@ Before merging a task, reviewer checks:
 Do not start these before CTX-12 is accepted.
 
 - Replace local artifact storage with S3/MinIO behind an `ArtifactStore` API.
-- Add worker labels/capacity-aware scheduling and concurrency > 1.
+- Add worker labels and capacity-aware scheduling.
+- Implement CTX-13 for bounded in-worker CPU parallelism.
+- Implement CTX-14 for optional GPU-accelerated workload execution.
+- Implement CTX-15 for the User Service and authenticated user/project access.
 - Add cancellation propagation to workers.
 - Add image outputs and final PDF reporting to job artifacts.
 - Add CV/video workloads using the same planner/runner/reducer contract.
 - Add observability export (Prometheus/OpenTelemetry).
-- Add per-user/project authorization and signed artifact URLs.
+- Add signed artifact URLs.
 - Add shard caching and content-addressed input deduplication.
 - Add job priority and fair scheduling.
 - Add a CLI for submitting and monitoring remote jobs.

README.md

@@ -137,5 +137,6 @@ The package separates common dataset parsing and fingerprints from independent w
 
 - [Emil](https://github.com/emil28092005) — Project Lead
 - [Kristina](https://github.com/kristtma) — Tech Lead
+- [Veniamin](https://t.me/Veniamin_Kt) — Scientific Lead
 - [Arkhip](https://github.com/hIpa-ussr) — Programmer
-- [Makar](https://github.com/RERAN4K) — Programmer
+- [Reranchik](https://github.com/RERAN4K) — Programmer

STATUS.md

@@ -1,7 +1,7 @@
 # SciMesh Status
 
-**Updated:** 2026-07-24
-**Branch baseline:** `main` at `6e67daa` (distributed similarity-search)
+**Updated:** 2026-07-27
+**Branch baseline:** `main` at `f5ead0a` (team and scaling-roadmap documentation)
 
 ## Current state
 
@@ -23,6 +23,9 @@ are reduced once into a checksum-protected final CSV, which is downloadable
 through the coordinator. The full Go checks (including a fresh migration and
 real PostgreSQL smoke test) passed on 2026-07-24.
 
+A User Service is being developed on a separate programmer branch. It is not
+yet merged, reviewed, or integrated with the coordinator/worker contract.
+
 ## Milestone tracker
 
 | CTX | Status | Notes |
@@ -40,6 +43,7 @@ real PostgreSQL smoke test) passed on 2026-07-24.
 | CTX-10 Distributed similarity-graph | Not started | Local reference exists. |
 | CTX-11 Dashboard/operator view | Implemented | Protected live control room: recent-run/worker overview, real pipeline-stage visualization, shard attempts and safe failures, validated similarity-search upload, coordinator artifacts, final-result download, and bounded polling. |
 | CTX-12 Reliability, security, CI | In progress | Unit, race, PostgreSQL integration, and smoke checks exist; CI hardening remains. |
+| CTX-15 User Service and access control | In progress (separate branch) | Proposed implementation is under development; API, security review, tests, and integration are pending. |
 
 ## Next recommended assignment
 

coordinator/cmd/coordinator/main.go

@@ -78,10 +78,10 @@ func run() error {
 		RenewLease:       usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration),
 		CompleteTask:     usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize),
 		ReduceJob:        usecase.NewReduceJob(jobRepo, taskRepo, artifactRepo, blobStore, tx, clk),
-		FailTask:         usecase.NewFailTask(taskRepo, jobRepo, tx, clk),
+		FailTask:         usecase.NewFailTask(taskRepo, jobRepo, workerRepo, tx, clk),
 		GetJobStatus:     usecase.NewGetJobStatus(jobRepo, taskRepo),
 		CancelJob:        usecase.NewCancelJob(jobRepo, taskRepo, tx, clk),
-		UploadArtifact:   usecase.NewUploadArtifact(taskRepo, artifactRepo, blobStore, tx, clk),
+		UploadArtifact:   usecase.NewUploadArtifact(taskRepo, workerRepo, artifactRepo, blobStore, tx, clk),
 		DownloadArtifact: usecase.NewDownloadArtifact(artifactRepo, blobStore),
 		GetJobResult:     usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)),
 		GetTaskInput:     usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),

coordinator/internal/transport/http/server_test.go

@@ -52,11 +52,11 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
 		RenewLease:       usecase.NewRenewLease(tasks, work, tx, clk, lease),
 		CompleteTask:     usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2),
 		ReduceJob:        usecase.NewReduceJob(jobs, tasks, arts, blobs, tx, clk),
-		FailTask:         usecase.NewFailTask(tasks, jobs, tx, clk),
+		FailTask:         usecase.NewFailTask(tasks, jobs, work, tx, clk),
 		GetJobStatus:     usecase.NewGetJobStatus(jobs, tasks),
 		GetJobResult:     usecase.NewGetJobResult(jobs, downloadArtifact),
 		CancelJob:        usecase.NewCancelJob(jobs, tasks, tx, clk),
-		UploadArtifact:   usecase.NewUploadArtifact(tasks, arts, blobs, tx, clk),
+		UploadArtifact:   usecase.NewUploadArtifact(tasks, work, arts, blobs, tx, clk),
 		DownloadArtifact: downloadArtifact,
 		GetTaskInput:     usecase.NewGetTaskInput(tasks, arts, blobs),
 		Dashboard:        usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),

coordinator/internal/usecase/artifact.go

@@ -12,18 +12,22 @@ import (
 // UploadArtifact stores a worker's partial-result bytes and records the metadata.
 type UploadArtifact struct {
 	tasks     TaskRepository
+	workers   WorkerRepository
 	artifacts ArtifactRepository
 	blobs     BlobStore
 	tx        TxManager
 	clk       Clock
 }
 
-func NewUploadArtifact(tasks TaskRepository, artifacts ArtifactRepository,
+func NewUploadArtifact(tasks TaskRepository, workers WorkerRepository, artifacts ArtifactRepository,
 	blobs BlobStore, tx TxManager, clk Clock) *UploadArtifact {
-	return &UploadArtifact{tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clk: clk}
+	return &UploadArtifact{tasks: tasks, workers: workers, artifacts: artifacts, blobs: blobs, tx: tx, clk: clk}
 }
 
 func (uc *UploadArtifact) Execute(ctx context.Context, in UploadArtifactInput) (*domain.Artifact, error) {
+	if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil {
+		return nil, err
+	}
 	task, err := uc.tasks.Get(ctx, in.TaskID)
 	if err != nil {
 		return nil, err

coordinator/internal/usecase/task.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/google/uuid"
 
-	"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
 	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
 )
 
@@ -59,11 +58,8 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl
 		// tier would be read off a caller-supplied worker_id, letting anyone who
 		// knows a trusted worker's id claim as it. A shared-token caller (no
 		// requester) is a lab operator and may act as any worker.
-		if r, ok := authctx.From(ctx); ok {
-			if worker.OwnerID == nil || *worker.OwnerID != r.UserID {
-				// Don't disclose that another user's worker exists.
-				return nil, domain.ErrWorkerNotFound
-			}
+		if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil {
+			return nil, err
 		}
 		// An untrusted volunteer may claim, but never a chunk its owner has
 		// already voted on — so quorum needs genuinely independent computations.
@@ -126,6 +122,9 @@ func NewRenewLease(tasks TaskRepository, workers WorkerRepository, tx TxManager,
 // locked: two concurrent heartbeats must not interleave into a lost update.
 // Whether the caller may renew at all is decided by the entity, not here.
 func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain.ClaimedTask, error) {
+	if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil {
+		return nil, err
+	}
 	var claimed domain.ClaimedTask
 
 	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
@@ -186,6 +185,9 @@ func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts Artifac
 // Lease ownership, staleness, and idempotent replays are all decided by
 // Task.CompleteWith; this use case only orchestrates.
 func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*domain.Task, error) {
+	if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil {
+		return nil, err
+	}
 	var out *domain.Task
 
 	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {
@@ -317,19 +319,23 @@ func (uc *CompleteTask) verifyResultArtifact(ctx context.Context, taskID uuid.UU
 // --- FailTask ------------------------------------------------------------
 
 type FailTask struct {
-	tasks TaskRepository
-	jobs  JobRepository
-	tx    TxManager
-	clock Clock
+	tasks   TaskRepository
+	jobs    JobRepository
+	workers WorkerRepository
+	tx      TxManager
+	clock   Clock
 }
 
-func NewFailTask(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *FailTask {
-	return &FailTask{tasks: tasks, jobs: jobs, tx: tx, clock: clock}
+func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock) *FailTask {
+	return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock}
 }
 
 // Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps
 // the parent job's status consistent in the same transaction.
 func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task, error) {
+	if err := authorizeWorkerOwner(ctx, uc.workers, in.WorkerID); err != nil {
+		return nil, err
+	}
 	var out *domain.Task
 
 	err := uc.tx.WithinTx(ctx, func(ctx context.Context) error {

coordinator/internal/usecase/usecase_test.go

@@ -77,11 +77,11 @@ func newHarness() *harness {
 	h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease)
 	h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease)
 	h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2)
-	h.fail = usecase.NewFailTask(h.tasks, h.jobs, tx, h.clk)
+	h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk)
 	h.status = usecase.NewGetJobStatus(h.jobs, h.tasks)
 	h.results = usecase.NewListResults(h.tasks)
 	h.register = usecase.NewRegisterWorker(h.work, h.clk)
-	h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.arts, h.blobs, tx, h.clk)
+	h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.work, h.arts, h.blobs, tx, h.clk)
 	h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs)
 	h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs)
 	h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk)
@@ -318,6 +318,36 @@ func TestJWTCallerCannotClaimAsAnotherUsersWorker(t *testing.T) {
 	}
 }
 
+func TestJWTCallerCannotMutateAnotherUsersWorkerLease(t *testing.T) {
+	h := newHarness()
+	h.seedJob(t, "w", 1)
+	victimOwner := uuid.New()
+	victim, err := h.register.Execute(ctx, usecase.RegisterWorkerInput{
+		Name: "victim", Capabilities: []string{"w"}, OwnerID: &victimOwner, TrustLevel: domain.WorkerTrusted,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := h.claim.Execute(ctx, usecase.ClaimTaskInput{WorkerID: victim.ID.String()})
+	if err != nil || claimed == nil {
+		t.Fatalf("claim = (%v, %v)", claimed, err)
+	}
+	attacker := authctx.With(ctx, authctx.Requester{UserID: uuid.New(), Role: "user"})
+
+	if _, err := h.renew.Execute(attacker, usecase.RenewLeaseInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt}); !errors.Is(err, domain.ErrWorkerNotFound) {
+		t.Errorf("foreign heartbeat err = %v, want ErrWorkerNotFound", err)
+	}
+	if _, err := h.fail.Execute(attacker, usecase.FailTaskInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, ErrorCode: "x"}); !errors.Is(err, domain.ErrWorkerNotFound) {
+		t.Errorf("foreign failure err = %v, want ErrWorkerNotFound", err)
+	}
+	if _, err := h.uploadArt.Execute(attacker, usecase.UploadArtifactInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, Filename: "x.csv", ContentType: "text/csv", Body: strings.NewReader("x")}); !errors.Is(err, domain.ErrWorkerNotFound) {
+		t.Errorf("foreign upload err = %v, want ErrWorkerNotFound", err)
+	}
+	if _, err := h.complete.Execute(attacker, usecase.CompleteTaskInput{TaskID: claimed.TaskID, WorkerID: victim.ID.String(), Attempt: claimed.Attempt, ResultArtifactID: uuid.New()}); !errors.Is(err, domain.ErrWorkerNotFound) {
+		t.Errorf("foreign result err = %v, want ErrWorkerNotFound", err)
+	}
+}
+
 func TestJWTCallerClaimsAsOwnTrustedWorker(t *testing.T) {
 	h := newHarness()
 	h.seedJob(t, "w", 1)
@@ -551,7 +581,7 @@ func TestUploadRejectsLeaseThatExpiresDuringStreaming(t *testing.T) {
 	h.seedJob(t, "w", 1)
 	taskID, attempt := h.leaseOne(t, "w1", "w")
 	h.uploadArt = usecase.NewUploadArtifact(
-		h.tasks, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, memstore.Tx{}, h.clk,
+		h.tasks, h.work, h.arts, expiringBlobStore{BlobStore: h.blobs, clock: h.clk}, memstore.Tx{}, h.clk,
 	)
 
 	_, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{

coordinator/internal/usecase/worker_authorization.go

@@ -0,0 +1,33 @@
+package usecase
+
+import (
+	"context"
+
+	"github.com/google/uuid"
+
+	"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
+	"github.com/emil28092005/SciMesh/coordinator/internal/domain"
+)
+
+// authorizeWorkerOwner binds a JWT-authenticated requester to a worker. The
+// shared coordinator token intentionally has no requester and retains its
+// existing operator privileges.
+func authorizeWorkerOwner(ctx context.Context, workers WorkerRepository, workerID string) error {
+	requester, ok := authctx.From(ctx)
+	if !ok {
+		return nil
+	}
+	id, err := uuid.Parse(workerID)
+	if err != nil {
+		return domain.ErrWorkerNotFound
+	}
+	worker, err := workers.Get(ctx, id)
+	if err != nil {
+		return err
+	}
+	if worker.OwnerID == nil || *worker.OwnerID != requester.UserID {
+		// Mask ownership and existence from another user.
+		return domain.ErrWorkerNotFound
+	}
+	return nil
+}

docs/user-service-api-contract.md

@@ -0,0 +1,47 @@
+# SciMesh User Service API contract (v1)
+
+**Status:** `v1`. The User Service owns user accounts and issues access tokens.
+The coordinator never receives user passwords and never accesses the User
+Service database.
+
+## Authentication boundary
+
+- User Service signs access tokens; coordinator verifies them before accepting
+  user-scoped requests.
+- Tokens contain a UUID `sub`, `role` (`user` or `admin`), `verified`, `iat`,
+  and `exp` claims.
+- A user-authenticated caller may operate only workers whose `owner_id` equals
+  `sub`. This applies to claim, heartbeat, result, failure, and artifact upload.
+- Worker traffic authenticated with the coordinator's shared worker token has
+  no user identity and remains an operator-only compatibility path.
+- Role or verification changes take effect when the access token is renewed.
+  Deployments needing immediate revocation must use a short token lifetime or a
+  revocation mechanism before enabling volunteer-worker trust.
+
+## Endpoints
+
+All JSON request bodies reject unknown fields and are size-limited. Error
+responses are JSON with a stable `error` value and request ID.
+
+| Method | Path | Auth | Success |
+| --- | --- | --- | --- |
+| `GET` | `/health` | none | `200 {"status":"ok"}` |
+| `POST` | `/register` | none | `201` user object |
+| `POST` | `/login` | none | `200` user object and access token |
+| `GET` | `/me` | Bearer access token | `200` current user |
+| `POST` | `/users/{id}/verify` | Bearer admin token | `204` |
+| `POST` | `/users/{id}/unverify` | Bearer admin token | `204` |
+| `POST` | `/users/{id}/promote` | Bearer admin token | `204` |
+| `POST` | `/users/{id}/demote` | Bearer admin token | `204` |
+
+`POST /register` accepts `{ "email": string, "password": string }` and
+always creates role `user` with `verified: false`. `POST /login` accepts the
+same shape and returns `{ "token": string, "user": User }`. Password hashes,
+JWT signing material, and raw tokens must never be logged.
+
+## Coordinator integration tests
+
+The coordinator must test that a JWT user cannot claim or mutate another
+user's worker lease, including heartbeat, failure, result, and artifact upload.
+Job and artifact access is restricted to the job owner unless the caller has
+the admin role.

users/README.md

@@ -5,6 +5,9 @@ and issues the JWTs the coordinator trusts. It is a **separate bounded context**
 from the coordinator: its own database, its own binary. The only thing shared
 between the two services is the JWT signing secret.
 
+The versioned external contract is
+[`docs/user-service-api-contract.md`](../docs/user-service-api-contract.md).
+
 Built as a modular monolith following Clean Architecture — one binary, four
 layers, dependencies pointing strictly inward: