github-devloop implementation for #34

.github/workflows/ci.yml

@@ -37,7 +37,7 @@ jobs:
         run: git diff --exit-code -- specs/testing-contract-fixtures.json
 
   test:
-    name: Test
+    name: test
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

control-plane/src/domain/errors.ts

@@ -74,6 +74,8 @@ export const unauthorized = (message: string): TalosError =>
 export const forbidden = (message: string): TalosError => new TalosError('forbidden', message, 403);
 export const modeForbidden = (message: string): TalosError => new TalosError('mode_forbidden', message, 403);
 export const conflict = (message: string): TalosError => new TalosError('conflict', message, 409);
+export const concurrentUpdate = (message = 'task state changed concurrently'): TalosError =>
+  new TalosError('concurrent_update', message, 409);
 export const actionAlreadyCompleted = (): TalosError =>
   new TalosError('action_already_completed', 'session action result was already stored', 409);
 export const taskCancelled = (message = 'task was cancelled'): TalosError => new TalosError('task_cancelled', message, 409);

control-plane/src/domain/types.ts

@@ -14,6 +14,29 @@ export type TaskStatus =
   | 'failed'
   | 'cancelled';
 
+export type TaskClaimRecoveryReason =
+  | 'partial_claim_identity'
+  | 'invalid_claim_generation'
+  | 'invalid_claim_credentials'
+  | 'invalid_lease_expiry'
+  | 'active_claim_missing_credentials'
+  | 'active_claim_marked_released'
+  | 'legacy_profile_identity_conflict';
+
+export interface TaskClaimRecovery {
+  schemaVersion: 'talos.task-claim-recovery/v1';
+  recoveryId: string;
+  kind: 'legacy' | 'malformed';
+  phase: 'draining' | 'finalizing' | 'quarantined';
+  sourceStatus: TaskStatus;
+  sourceMachineId?: string;
+  sourceProfileId?: string;
+  restoredQueuePriority: number;
+  reasonCode?: TaskClaimRecoveryReason;
+  startedAt: string;
+  updatedAt: string;
+}
+
 export type CapabilityTag =
   | 'os'
   | 'region'
@@ -65,6 +88,12 @@ interface TaskBase {
   claimedAt?: string;
   leaseExpiresAt?: string;
   leaseToken?: string;
+  claimId?: string;
+  claimGeneration?: number;
+  taskVersion?: number;
+  claimCommitted?: boolean;
+  claimReleased?: boolean;
+  claimQueuePriority?: number;
   workerId?: string;
   machineId?: string;
   findings: readonly TaskFinding[];
@@ -74,6 +103,7 @@ interface TaskBase {
   handoff?: { url: string; expiresAt: string };
   pendingActionId?: string;
   lastActionId?: string;
+  claimRecovery?: TaskClaimRecovery;
 }
 
 export interface BrowserTask extends TaskBase {
@@ -142,6 +172,7 @@ export interface Machine {
   tags: Readonly<Record<string, string | boolean>>;
   capacity: number;
   activeLeases: number;
+  leaseReservations?: readonly MachineLeaseReservation[];
   online: boolean;
   workerTokenHash: string;
 }
@@ -151,9 +182,39 @@ export interface Profile {
   userId: string;
   machineId?: string;
   lockedByTaskId?: string;
+  lockedByClaimId?: string;
+  lockedByClaimGeneration?: number;
   lockExpiresAt?: string;
 }
 
+export interface TaskClaimGuard {
+  claimId: string;
+  claimGeneration: number;
+  taskVersion: number;
+  status: TaskStatus;
+}
+
+export interface TaskActiveClaimGuard extends TaskClaimGuard {
+  leaseExpiresAt: string;
+}
+
+export interface TaskRecoveryGuard {
+  status: TaskStatus;
+  taskVersion: number;
+  updatedAt: string;
+  claimId?: string;
+  claimGeneration?: number;
+  recoveryId?: string;
+  recoveryPhase?: TaskClaimRecovery['phase'];
+}
+
+export interface MachineLeaseReservation {
+  claimId: string;
+  claimGeneration: number;
+  taskId: string;
+  expiresAt: string;
+}
+
 export interface HandoffLink {
   id: string;
   taskId: string;

control-plane/src/http/server.test.ts

@@ -5,6 +5,7 @@ import { Scheduler } from '../services/scheduler.js';
 import { TaskService } from '../services/task-service.js';
 import { WebhookSigner } from '../services/webhook-signer.js';
 import { MemoryRepository } from '../storage/memory-repository.js';
+import type { Repository } from '../storage/repository.js';
 import { createApiServer } from './server.js';
 import { loadOpenApiDocument } from '../openapi.js';
 
@@ -58,6 +59,114 @@ describe('control-plane HTTP API', () => {
     server.close();
   });
 
+  it('maps repository failures to an opaque public error', async () => {
+    const repository = new MemoryRepository();
+    repository.getTask = async () => { throw new Error('claim-secret-sentinel lease-token-sentinel'); };
+    const service = new TaskService(repository, new Scheduler(repository), new ProfileLockService(repository), new WebhookSigner('webhook-secret-1234'));
+    const server = createApiServer(service, repository);
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const address = server.address();
+    if (address === null || typeof address === 'string') throw new Error('server did not bind');
+
+    const response = await fetch(`http://127.0.0.1:${address.port}/v1/tasks/task`, {
+      headers: { 'x-nyxid-identity-token': 'user:user-a' }
+    });
+    expect(response.status).toBe(500);
+    const body = JSON.stringify(await response.json());
+    expect(body).toContain('internal_error');
+    expect(body).not.toContain('claim-secret-sentinel');
+    expect(body).not.toContain('lease-token-sentinel');
+    server.close();
+  });
+
+  it('returns a retryable public conflict after authorized task CAS exhaustion', async () => {
+    const storage = new MemoryRepository();
+    let attempts = 0;
+    const repository = new Proxy<Repository>(storage, {
+      get(target, property) {
+        if (property === 'replaceSubmittedTask') {
+          return async (...args: Parameters<Repository['replaceSubmittedTask']>): Promise<boolean> => {
+            if (args[0].status === 'cancelled') {
+              attempts += 1;
+              return false;
+            }
+            return target.replaceSubmittedTask(...args);
+          };
+        }
+        const value = Reflect.get(target, property);
+        return typeof value === 'function' ? value.bind(target) : value;
+      }
+    });
+    const service = new TaskService(
+      repository,
+      new Scheduler(repository),
+      new ProfileLockService(repository),
+      new WebhookSigner('webhook-secret-1234')
+    );
+    const server = createApiServer(service, repository);
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const address = server.address();
+    if (address === null || typeof address === 'string') throw new Error('server did not bind');
+    const base = `http://127.0.0.1:${address.port}`;
+    const headers = { 'content-type': 'application/json', 'x-nyxid-identity-token': 'user:user-a' };
+    const created = await fetch(`${base}/v1/tasks`, {
+      method: 'POST',
+      headers,
+      body: JSON.stringify({ kind: 'browse', goal: 'public CAS exhaustion' })
+    });
+    const task = await created.json() as { id: string };
+
+    const response = await fetch(`${base}/v1/tasks/${task.id}/cancel`, { method: 'POST', headers });
+    expect(response.status).toBe(409);
+    expect(await response.json()).toEqual({
+      error: {
+        code: 'concurrent_update',
+        message: 'task state changed concurrently',
+        retryable: true
+      }
+    });
+    expect(attempts).toBe(3);
+    server.close();
+  });
+
+  it('reports an expired handoff claim as a non-retryable public conflict', async () => {
+    const clock = { value: Date.now() };
+    const repository = new MemoryRepository(() => clock.value);
+    await repository.savePool({ id: 'expired-handoff-pool', visibility: 'platform', tags: {} });
+    await repository.saveMachine({ id: 'expired-handoff-machine', poolId: 'expired-handoff-pool', tags: {}, capacity: 1, activeLeases: 0, online: true, workerTokenHash: 'hash' });
+    const service = new TaskService(
+      repository,
+      new Scheduler(repository),
+      new ProfileLockService(repository),
+      new WebhookSigner('webhook-secret-1234'),
+      { clock: () => clock.value, leaseSeconds: 10 }
+    );
+    const task = await service.createTask('user-a', { kind: 'browse', goal: 'expired handoff' });
+    const claim = await service.claim('worker-a', 'expired-handoff-machine', clock.value);
+    clock.value = Date.parse(claim.task.leaseExpiresAt!);
+    const server = createApiServer(service, repository, { clock: () => clock.value });
+    await new Promise<void>((resolve) => server.listen(0, resolve));
+    const address = server.address();
+    if (address === null || typeof address === 'string') throw new Error('server did not bind');
+    const base = `http://127.0.0.1:${address.port}`;
+    const response = await fetch(`${base}/v1/tasks/${task.id}/handoff`, {
+      method: 'POST',
+      headers: { 'content-type': 'application/json', 'x-nyxid-identity-token': 'user:user-a' },
+      body: '{}'
+    });
+
+    expect(response.status).toBe(409);
+    expect(await response.json()).toEqual({
+      error: {
+        code: 'conflict',
+        message: 'task claim is not active',
+        retryable: false
+      }
+    });
+    expect(await repository.getTask(task.id)).not.toHaveProperty('handoff');
+    server.close();
+  });
+
   it('enforces NyxID and worker authentication across lifecycle routes', async () => {
     const repository = new MemoryRepository();
     await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
@@ -77,8 +186,30 @@ describe('control-plane HTTP API', () => {
     expect(badWorker.status).toBe(401);
     const claimResponse = await fetch(`${base}/v1/worker/claim`, { method: 'POST', headers: { authorization: 'Bearer worker-token-123456', 'x-talos-worker-id': 'w', 'x-talos-machine-id': 'machine', 'content-type': 'application/json' }, body: JSON.stringify({ worker_id: 'w', machine_id: 'machine' }) });
     expect(claimResponse.status).toBe(200);
-    const claim = await claimResponse.json() as { task: { id: string }; leaseToken: string };
+    const claim = await claimResponse.json() as { task: Record<string, unknown> & { id: string }; leaseToken: string };
     expect(claim.task.id).toBe(created.id);
+    const internalAuthorityFields = [
+      'claimId',
+      'claimGeneration',
+      'taskVersion',
+      'claimCommitted',
+      'claimReleased',
+      'claimQueuePriority',
+      'queuePriority',
+      'workerId',
+      'machineId',
+      'leaseExpiresAt',
+      'leaseToken',
+      'claimRecovery'
+    ];
+    for (const field of internalAuthorityFields) expect(claim.task).not.toHaveProperty(field);
+    const publicTaskResponse = await fetch(`${base}/v1/tasks/${created.id}`, {
+      headers: { 'x-nyxid-identity-token': 'user:user-a' }
+    });
+    expect(publicTaskResponse.status).toBe(200);
+    const publicTask = await publicTaskResponse.json() as Record<string, unknown>;
+    for (const field of internalAuthorityFields) expect(publicTask).not.toHaveProperty(field);
+    expect(JSON.stringify(publicTask)).not.toContain(claim.leaseToken);
     const heartbeat = await fetch(`${base}/v1/worker/tasks/${created.id}/heartbeat`, { method: 'POST', headers: { authorization: 'Bearer worker-token-123456', 'x-talos-worker-id': 'w', 'x-talos-machine-id': 'machine', 'content-type': 'application/json' }, body: JSON.stringify({ lease_token: claim.leaseToken }) });
     expect(heartbeat.status).toBe(200);
     server.close();
@@ -218,6 +349,7 @@ describe('control-plane HTTP API', () => {
     expect((await fetch(`${base}/v1/pools`, { method: 'POST', headers: user('bob'), body: JSON.stringify({ id: 'bob-pool' }) })).status).toBe(201);
     expect((await fetch(`${base}/v1/pools/bob-pool/machines`, { method: 'POST', headers: user('bob'), body: JSON.stringify({ id: 'bob-machine' }) })).status).toBe(201);
     expect((await fetch(`${base}/v1/profiles`, { method: 'POST', headers: user('bob'), body: JSON.stringify({ id: 'bob-profile', machine_id: 'bob-machine' }) })).status).toBe(201);
+    expect((await fetch(`${base}/v1/profiles`, { method: 'POST', headers: user('bob'), body: JSON.stringify({ id: 'bob-profile' }) })).status).toBe(409);
     const bobPools = await fetch(`${base}/v1/pools`, { headers: user('bob') });
     expect(bobPools.status).toBe(200);
     expect((await bobPools.json() as Array<{ id: string }>).map((pool) => pool.id)).toEqual(['bob-pool']);
@@ -290,7 +422,7 @@ describe('control-plane HTTP API', () => {
     const repository = new MemoryRepository();
     await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
     await repository.saveMachine({ id: 'machine', poolId: 'pool', tags: {}, capacity: 2, activeLeases: 0, online: true, workerTokenHash: hashWorkerToken('worker-token-123456') });
-    await repository.saveProfile({ id: 'p', userId: 'u' });
+    await repository.createProfile({ id: 'p', userId: 'u' });
     const service = new TaskService(repository, new Scheduler(repository), new ProfileLockService(repository), new WebhookSigner('webhook-secret-1234'));
     const server = createApiServer(service, repository);
     await new Promise<void>((resolve) => server.listen(0, resolve));

control-plane/src/http/server.ts

@@ -217,7 +217,9 @@ const route = async (
     await assertPoolOwner(repository, machine.poolId, userId);
     selfRotateMachineSchema.parse(await readBody(request, options.maxBodyBytes));
     const workerToken = issueWorkerToken();
-    await repository.saveMachine({ ...machine, workerTokenHash: hashWorkerToken(workerToken) });
+    if (!await repository.rotateMachineToken(machine.id, machine.workerTokenHash, hashWorkerToken(workerToken))) {
+      throw conflict('machine token changed concurrently');
+    }
     return send(response, 200, { id: machine.id, rotated: true, worker_token: workerToken });
   }
   if (parts[1] === 'profiles' && parts.length === 2 && method === 'POST') {
@@ -228,8 +230,9 @@ const route = async (
       await assertPoolOwner(repository, machine.poolId, userId);
     }
     const id = input.id ?? newId('profile');
-    if (await repository.getProfile(id) !== undefined) throw conflict('profile already exists');
-    await repository.saveProfile({ id, userId, ...(input.machine_id === undefined ? {} : { machineId: input.machine_id }) });
+    if (!await repository.createProfile({ id, userId, ...(input.machine_id === undefined ? {} : { machineId: input.machine_id }) })) {
+      throw conflict('profile already exists');
+    }
     return send(response, 201, {
       id,
       userId,
@@ -385,7 +388,9 @@ const adminRoute = async (
     const machine = await repository.getMachine(parts[3]);
     if (machine === undefined) throw notFound('machine not found');
     const workerToken = input.worker_token ?? issueWorkerToken();
-    await repository.saveMachine({ ...machine, workerTokenHash: hashWorkerToken(workerToken) });
+    if (!await repository.rotateMachineToken(machine.id, machine.workerTokenHash, hashWorkerToken(workerToken))) {
+      throw conflict('machine token changed concurrently');
+    }
     return send(response, 200, { id: machine.id, rotated: true, worker_token: workerToken });
   }
   if (parts[2] === 'machines') {
@@ -398,8 +403,9 @@ const adminRoute = async (
   }
   if (parts[2] === 'profiles') {
     const input = adminProfileSchema.parse(await readBody(request, options.maxBodyBytes));
-    if (await repository.getProfile(input.id) !== undefined) throw conflict('profile already exists');
-    await repository.saveProfile({ id: input.id, userId: input.user_id, ...(input.machine_id === undefined ? {} : { machineId: input.machine_id }) });
+    if (!await repository.createProfile({ id: input.id, userId: input.user_id, ...(input.machine_id === undefined ? {} : { machineId: input.machine_id }) })) {
+      throw conflict('profile already exists');
+    }
     return send(response, 201, { id: input.id });
   }
   return send(response, 404, publicErrorEnvelope('not_found', 'route not found', 404));

control-plane/src/http/session-routes.integration.test.ts

@@ -87,6 +87,22 @@ describe('interactive session HTTP API', () => {
       })
     });
     const claim = await claimResponse.json() as { leaseToken: string };
+    const publicSession = await (await fetch(`${base}/v1/sessions/${session.id}`, { headers: alice })).json() as Record<string, unknown>;
+    for (const field of [
+      'claimId',
+      'claimGeneration',
+      'taskVersion',
+      'claimCommitted',
+      'claimReleased',
+      'claimQueuePriority',
+      'queuePriority',
+      'workerId',
+      'machineId',
+      'leaseExpiresAt',
+      'leaseToken',
+      'claimRecovery'
+    ]) expect(publicSession).not.toHaveProperty(field);
+    expect(JSON.stringify(publicSession)).not.toContain(claim.leaseToken);
     const forbiddenAction = await fetch(`${base}/v1/sessions/${session.id}/actions?wait_seconds=0`, {
       method: 'POST',
       headers: alice,