Roll up qa-tools-fkst-staging into qa-tools

control-plane/src/domain/types.ts

@@ -69,6 +69,20 @@ export interface PendingInputRecord extends PendingInputIntent {
   consumed: boolean;
 }
 
+export interface PendingHandoffIntent {
+  schemaVersion: 'talos.task-handoff-intent/v1';
+  operationId: string;
+  id: string;
+  taskId: string;
+  userId: string;
+  claimId: string;
+  claimGeneration: number;
+  expiresInSeconds: number;
+  url: string;
+  expiresAt: string;
+  consumed: boolean;
+}
+
 export interface Artifact {
   id: string;
   name: string;
@@ -119,6 +133,7 @@ interface TaskBase {
   sessionActions?: readonly SessionActionRecord[];
   claimRecovery?: TaskClaimRecovery;
   pendingInputIntent?: PendingInputIntent;
+  pendingHandoffIntent?: PendingHandoffIntent;
 }
 
 export interface BrowserTask extends TaskBase {
@@ -251,12 +266,7 @@ export interface MachineLeaseReservation {
   expiresAt: string;
 }
 
-export interface HandoffLink {
-  id: string;
-  taskId: string;
-  userId: string;
-  url: string;
-  expiresAt: string;
+export interface HandoffRecord extends Omit<PendingHandoffIntent, 'consumed'> {
   used: boolean;
 }
 

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

@@ -8,7 +8,7 @@ import { MemoryRepository } from '../storage/memory-repository.js';
 import type { Repository } from '../storage/repository.js';
 import { createApiServer } from './server.js';
 import { loadOpenApiDocument } from '../openapi.js';
-import type { PendingInputIntent } from '../domain/types.js';
+import type { PendingHandoffIntent, PendingInputIntent, Task } from '../domain/types.js';
 
 class FirstMaterializationFailureRepository extends MemoryRepository {
   public readonly attemptedIntents: PendingInputIntent[] = [];
@@ -22,6 +22,26 @@ class FirstMaterializationFailureRepository extends MemoryRepository {
   }
 }
 
+class FirstCommittedInputAcknowledgementFailureRepository extends MemoryRepository {
+  public readonly attemptedIntents: PendingInputIntent[] = [];
+
+  public override async materializePendingInput(intent: PendingInputIntent): Promise<void> {
+    this.attemptedIntents.push(structuredClone(intent));
+    await super.materializePendingInput(intent);
+    if (this.attemptedIntents.length === 1) throw new Error('injected pending input acknowledgement failure');
+  }
+}
+
+class FirstCommittedHandoffAcknowledgementFailureRepository extends MemoryRepository {
+  public readonly attemptedIntents: PendingHandoffIntent[] = [];
+
+  public override async materializeHandoff(intent: PendingHandoffIntent): Promise<void> {
+    this.attemptedIntents.push(structuredClone(intent));
+    await super.materializeHandoff(intent);
+    if (this.attemptedIntents.length === 1) throw new Error('injected handoff acknowledgement failure');
+  }
+}
+
 describe('control-plane HTTP API', () => {
   it('serves cached OpenAPI JSON and YAML without authentication', async () => {
     const repository = new MemoryRepository();
@@ -292,11 +312,27 @@ describe('control-plane HTTP API', () => {
     expect((await fetch(`${base}/v1/admin/profiles`, { method: 'POST', headers, body: JSON.stringify({ id: 'profile', user_id: 'other-user' }) })).status).toBe(409);
     expect((await fetch(`${base}/v1/admin/machines/machine/rotate-token`, { method: 'POST', headers, body: JSON.stringify({ worker_token: 'rotated-worker-token-123456' }) })).status).toBe(200);
     expect((await repository.getMachine('machine'))?.workerTokenHash).toBe(hashWorkerToken('rotated-worker-token-123456'));
-    await repository.saveHandoff({ id: 'h', taskId: 't', userId: 'u', url: '/v1/handoffs/h', expiresAt: new Date(2000).toISOString(), used: false });
+    const handoffIntent: PendingHandoffIntent = {
+      schemaVersion: 'talos.task-handoff-intent/v1', operationId: 'handoff-operation-h', id: 'h', taskId: 't', userId: 'u',
+      claimId: 'claim-h', claimGeneration: 1, expiresInSeconds: 1, url: '/v1/handoffs/h', expiresAt: new Date(2000).toISOString(), consumed: false
+    };
+    await repository.saveTask({
+      id: 't', userId: 'u', kind: 'browse', goal: 'handoff', constraints: {}, mode: 'read_only', interaction: 'autonomous',
+      status: 'handoff', createdAt: new Date(0).toISOString(), updatedAt: new Date(0).toISOString(), findings: [], artifacts: [],
+      claimId: handoffIntent.claimId, claimGeneration: handoffIntent.claimGeneration, claimCommitted: true, taskVersion: 1,
+      handoff: { url: handoffIntent.url, expiresAt: handoffIntent.expiresAt }, pendingHandoffIntent: handoffIntent
+    } satisfies Task);
+    await repository.materializeHandoff(handoffIntent);
     const handoff = await fetch(`${base}/v1/handoffs/h`, { headers: { 'x-nyxid-identity-token': 'user:u' } });
     expect(handoff.status).toBe(501);
     expect((await fetch(`${base}/v1/handoffs/h`, { headers: { 'x-nyxid-identity-token': 'user:u' } })).status).toBe(409);
-    await repository.saveHandoff({ id: 'expired', taskId: 't', userId: 'u', url: '/v1/handoffs/expired', expiresAt: new Date(500).toISOString(), used: false });
+    await repository.materializeHandoff({
+      ...handoffIntent,
+      operationId: 'handoff-operation-expired',
+      id: 'expired',
+      url: '/v1/handoffs/expired',
+      expiresAt: new Date(500).toISOString()
+    });
     expect((await fetch(`${base}/v1/handoffs/expired`, { headers: { 'x-nyxid-identity-token': 'user:u' } })).status).toBe(409);
     server.close();
   });
@@ -615,4 +651,87 @@ describe('control-plane HTTP API', () => {
       server.close();
     }
   });
+
+  it('reconciles committed input materialization after a lost acknowledgement', async () => {
+    const now = Date.parse('2026-09-11T12:00:00.000Z');
+    const repository = new FirstCommittedInputAcknowledgementFailureRepository(() => now);
+    await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
+    await repository.saveMachine({ id: 'machine', poolId: 'pool', tags: {}, capacity: 1, activeLeases: 0, online: true, workerTokenHash: hashWorkerToken('worker-token-123456') });
+    const service = new TaskService(repository, new Scheduler(repository), new ProfileLockService(repository), new WebhookSigner('webhook-secret-1234'), { clock: () => now });
+    const server = createApiServer(service, repository, { clock: () => now });
+    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 publicHeaders = { 'content-type': 'application/json', 'x-nyxid-identity-token': 'user:user-a' };
+    const workerHeaders = { authorization: 'Bearer worker-token-123456', 'content-type': 'application/json', 'x-talos-worker-id': 'worker-a', 'x-talos-machine-id': 'machine' };
+    try {
+      const created = await (await fetch(`${base}/v1/tasks`, { method: 'POST', headers: publicHeaders, body: JSON.stringify({ kind: 'browse', goal: 'recover committed input' }) })).json() as { id: string };
+      const claim = await (await fetch(`${base}/v1/worker/claim`, { method: 'POST', headers: workerHeaders, body: JSON.stringify({ worker_id: 'worker-a', machine_id: 'machine' }) })).json() as { leaseToken: string };
+      await fetch(`${base}/v1/worker/tasks/${created.id}/needs-input`, { method: 'POST', headers: workerHeaders, body: JSON.stringify({ lease_token: claim.leaseToken }) });
+      const body = '{"kind":"text","value":" answer "}';
+      const first = await fetch(`${base}/v1/tasks/${created.id}/input`, { method: 'POST', headers: publicHeaders, body });
+      expect(first.status).toBe(500);
+      expect(await first.json()).toEqual({ error: { code: 'internal_error', message: 'internal server error', retryable: true } });
+      const retry = await fetch(`${base}/v1/tasks/${created.id}/input`, { method: 'POST', headers: publicHeaders, body });
+      expect(retry.status).toBe(200);
+      expect(repository.attemptedIntents).toHaveLength(2);
+      expect(repository.attemptedIntents[1]).toEqual(repository.attemptedIntents[0]);
+      const pollBody = JSON.stringify({ lease_token: claim.leaseToken, worker_token: 'worker-token-123456', worker_id: 'worker-a', machine_id: 'machine' });
+      const firstPoll = await fetch(`${base}/v1/worker/tasks/${created.id}/input/poll`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: pollBody });
+      expect(await firstPoll.json()).toEqual({ input: { kind: 'text', value: ' answer ' } });
+      const secondPoll = await fetch(`${base}/v1/worker/tasks/${created.id}/input/poll`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: pollBody });
+      expect(await secondPoll.json()).toEqual({});
+    } finally {
+      server.close();
+    }
+  });
+
+  it('reconciles one handoff after a committed materialization loses acknowledgement', async () => {
+    const now = Date.parse('2026-09-11T12:00:00.000Z');
+    const repository = new FirstCommittedHandoffAcknowledgementFailureRepository(() => now);
+    await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
+    await repository.saveMachine({ id: 'machine', poolId: 'pool', tags: {}, capacity: 1, activeLeases: 0, online: true, workerTokenHash: hashWorkerToken('worker-token-123456') });
+    const service = new TaskService(repository, new Scheduler(repository), new ProfileLockService(repository), new WebhookSigner('webhook-secret-1234'), { clock: () => now });
+    const server = createApiServer(service, repository, { clock: () => now });
+    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 publicHeaders = { 'content-type': 'application/json', 'x-nyxid-identity-token': 'user:user-a' };
+    const workerHeaders = { authorization: 'Bearer worker-token-123456', 'content-type': 'application/json', 'x-talos-worker-id': 'worker-a', 'x-talos-machine-id': 'machine' };
+    try {
+      const created = await (await fetch(`${base}/v1/tasks`, { method: 'POST', headers: publicHeaders, body: JSON.stringify({ kind: 'browse', goal: 'recover committed handoff' }) })).json() as { id: string };
+      const claim = await (await fetch(`${base}/v1/worker/claim`, { method: 'POST', headers: workerHeaders, body: JSON.stringify({ worker_id: 'worker-a', machine_id: 'machine' }) })).json() as { leaseToken: string };
+      await fetch(`${base}/v1/worker/tasks/${created.id}/heartbeat`, { method: 'POST', headers: workerHeaders, body: JSON.stringify({ lease_token: claim.leaseToken }) });
+      const first = await fetch(`${base}/v1/tasks/${created.id}/handoff`, { method: 'POST', headers: publicHeaders, body: '{"expires_in_seconds":900}' });
+      expect(first.status).toBe(500);
+      expect(await first.json()).toEqual({ error: { code: 'internal_error', message: 'internal server error', retryable: true } });
+      const retry = await fetch(`${base}/v1/tasks/${created.id}/handoff`, { method: 'POST', headers: publicHeaders, body: '{"expires_in_seconds":900}' });
+      expect(retry.status).toBe(200);
+      const response = await retry.json() as { handoff_url: string; expires: string };
+      expect(response).toEqual({ handoff_url: repository.attemptedIntents[0]?.url, expires: repository.attemptedIntents[0]?.expiresAt });
+      expect(repository.attemptedIntents).toHaveLength(2);
+      expect(repository.attemptedIntents[1]).toEqual(repository.attemptedIntents[0]);
+      const task = await (await fetch(`${base}/v1/tasks/${created.id}`, { headers: publicHeaders })).json() as Record<string, unknown>;
+      expect(task).not.toHaveProperty('pendingHandoffIntent');
+      expect(task).not.toHaveProperty('operationId');
+      const concurrent = await Promise.all([
+        fetch(`${base}${response.handoff_url}`, { headers: { 'x-nyxid-identity-token': 'user:user-a' } }),
+        fetch(`${base}${response.handoff_url}`, { headers: { 'x-nyxid-identity-token': 'user:user-a' } })
+      ]);
+      expect(concurrent.map((candidate) => candidate.status).sort()).toEqual([409, 501]);
+      for (const candidate of concurrent) {
+        expect(await candidate.json()).toEqual(candidate.status === 501
+          ? { error: { code: 'not_implemented', message: 'hosted handoff views are planned for Phase 3', retryable: false } }
+          : { error: { code: 'handoff_expired', message: 'handoff link is expired or already used', retryable: false } });
+      }
+      const repeated = await fetch(`${base}${response.handoff_url}`, { headers: { 'x-nyxid-identity-token': 'user:user-a' } });
+      expect(repeated.status).toBe(409);
+      expect(await repeated.json()).toEqual({ error: { code: 'handoff_expired', message: 'handoff link is expired or already used', retryable: false } });
+    } finally {
+      server.close();
+    }
+  });
+
 });

control-plane/src/http/server.ts

@@ -183,10 +183,9 @@ const route = async (
     const link = await repository.getHandoff(parts[2]);
     if (link === undefined) throw notFound('handoff not found');
     if (link.userId !== identity.userId) throw unauthorized('handoff belongs to another user');
-    if (link.used || Date.parse(link.expiresAt) <= (options.clock?.() ?? Date.now())) {
+    if (await repository.consumeHandoff(link, options.clock?.() ?? Date.now()) === undefined) {
       throw new TalosError('handoff_expired', 'handoff link is expired or already used', 409);
     }
-    await repository.saveHandoff({ ...link, used: true });
     throw notImplemented('hosted handoff views are planned for Phase 3');
   }
   if (parts[1] === 'admin') return adminRoute(request, response, repository, parts, options);

control-plane/src/services/task-service.test.ts

@@ -477,10 +477,10 @@ describe('task service', () => {
             return target.materializePendingInput(...args);
           };
         }
-        if (property === 'saveHandoff') {
-          return async (...args: Parameters<Repository['saveHandoff']>): Promise<void> => {
+        if (property === 'materializeHandoff') {
+          return async (...args: Parameters<Repository['materializeHandoff']>): Promise<void> => {
             handoffWrites += 1;
-            return target.saveHandoff(...args);
+            return target.materializeHandoff(...args);
           };
         }
         const value = Reflect.get(target, property);
@@ -527,10 +527,10 @@ describe('task service', () => {
             return claimed;
           };
         }
-        if (property === 'saveHandoff') {
-          return async (...args: Parameters<Repository['saveHandoff']>): Promise<void> => {
+        if (property === 'materializeHandoff') {
+          return async (...args: Parameters<Repository['materializeHandoff']>): Promise<void> => {
             handoffWrites += 1;
-            return target.saveHandoff(...args);
+            return target.materializeHandoff(...args);
           };
         }
         const value = Reflect.get(target, property);
@@ -610,10 +610,10 @@ describe('task service', () => {
             return target.materializePendingInput(...args);
           };
         }
-        if (property === 'saveHandoff') {
-          return async (...args: Parameters<Repository['saveHandoff']>): Promise<void> => {
+        if (property === 'materializeHandoff') {
+          return async (...args: Parameters<Repository['materializeHandoff']>): Promise<void> => {
             handoffWrites += 1;
-            return target.saveHandoff(...args);
+            return target.materializeHandoff(...args);
           };
         }
         const value = Reflect.get(target, property);

control-plane/src/services/task-service.ts

@@ -1,7 +1,7 @@
 import { concurrentUpdate, conflict, deadlineExceeded, forbidden, notFound, taskCancelled, unauthorized, TalosError } from '../domain/errors.js';
 import { timingSafeEqual } from 'node:crypto';
 import { taskCreateSchema } from '../domain/schemas.js';
-import type { Lease, MachineLeaseReservation, PendingInputIntent, PublicTask, SessionActionResult, Task, TaskClaimRecoveryReason, TaskClaimGuard, TaskFinding, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js';
+import type { Lease, MachineLeaseReservation, PendingHandoffIntent, PendingInputIntent, PublicTask, SessionActionResult, Task, TaskClaimRecoveryReason, TaskClaimGuard, TaskFinding, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js';
 import type { Repository, TaskMaintenanceCursor } from '../storage/repository.js';
 import { newId } from '../util/id.js';
 import type { ProfileLockService } from './profile-lock.js';
@@ -294,21 +294,39 @@ export class TaskService {
 
   public async requestHandoff(id: string, userId: string, expiresInSeconds: number): Promise<{ handoff_url: string; expires: string }> {
     const task = await this.authorizedTask(id, userId);
+    if (this.isPendingHandoffReconciliation(task, userId, expiresInSeconds)) {
+      await this.repository.materializeHandoff(task.pendingHandoffIntent);
+      return { handoff_url: task.pendingHandoffIntent.url, expires: task.pendingHandoffIntent.expiresAt };
+    }
     this.assertTaskCanRequestHandoff(task);
     const claimBinding = this.userClaimBinding(task);
     const expires = new Date(this.clock() + expiresInSeconds * 1000).toISOString();
     const linkId = newId('handoff');
     const url = `/v1/handoffs/${linkId}`;
+    const pendingHandoffIntent: PendingHandoffIntent = {
+      schemaVersion: 'talos.task-handoff-intent/v1',
+      operationId: newId('task-handoff'),
+      id: linkId,
+      taskId: id,
+      userId,
+      claimId: claimBinding.claimId,
+      claimGeneration: claimBinding.claimGeneration,
+      expiresInSeconds,
+      url,
+      expiresAt: expires,
+      consumed: false
+    };
     const { persisted } = await this.replaceAuthorizedTask(id, userId, task, (current) => {
       this.assertTaskCanRequestHandoff(current);
       return {
         ...current,
         status: 'handoff',
         updatedAt: new Date(this.clock()).toISOString(),
-        handoff: { url, expiresAt: expires }
+        handoff: { url, expiresAt: expires },
+        pendingHandoffIntent
       };
     }, { claimBinding, requireActiveClaim: true });
-    await this.repository.saveHandoff({ id: linkId, taskId: id, userId, url, expiresAt: expires, used: false });
+    await this.repository.materializeHandoff(pendingHandoffIntent);
     await this.emit(persisted, 'task.handoff_requested', { handoff_url: url, expires });
     return { handoff_url: url, expires };
   }
@@ -893,6 +911,30 @@ export class TaskService {
     if (task.claimCommitted !== true) throw conflict('task claim is not active');
   }
 
+  private isPendingHandoffReconciliation(
+    task: Task,
+    userId: string,
+    expiresInSeconds: number
+  ): task is Task & { pendingHandoffIntent: PendingHandoffIntent } {
+    const intent = task.pendingHandoffIntent;
+    return task.kind !== 'testing' &&
+      task.interaction !== 'interactive' &&
+      task.status === 'handoff' &&
+      task.userId === userId &&
+      task.claimCommitted === true &&
+      task.claimId !== undefined &&
+      task.claimGeneration !== undefined &&
+      task.claimGeneration > 0 &&
+      task.leaseExpiresAt !== undefined &&
+      Date.parse(task.leaseExpiresAt) > this.clock() &&
+      intent !== undefined &&
+      intent.taskId === task.id &&
+      intent.userId === userId &&
+      intent.claimId === task.claimId &&
+      intent.claimGeneration === task.claimGeneration &&
+      intent.expiresInSeconds === expiresInSeconds;
+  }
+
   private claimGuard(task: Task): TaskClaimGuard {
     if (task.claimId === undefined || task.claimGeneration === undefined || task.claimGeneration <= 0) {
       throw unauthorized('lease generation is no longer active');
@@ -1094,6 +1136,7 @@ export class TaskService {
       'sessionActions',
       'claimRecovery',
       'pendingInputIntent',
+      'pendingHandoffIntent',
       'testing'
     ]);
     return {