From cbfd331f30c05f65f53b0ee96dd03ab1260bebf3 Mon Sep 17 00:00:00 2001 From: "chronoai-fkst[bot]" Date: Fri, 11 Sep 2026 05:36:05 +0000 Subject: [PATCH 1/2] auto-implement refs #50: Production slice: make Task input and handoff intents restart-convergent --- control-plane/src/domain/types.ts | 22 ++- control-plane/src/http/server.test.ts | 125 ++++++++++++- control-plane/src/http/server.ts | 3 +- .../src/services/task-service.test.ts | 18 +- control-plane/src/services/task-service.ts | 49 ++++- .../src/storage/memory-repository.ts | 68 ++++++- control-plane/src/storage/mongo-repository.ts | 104 ++++++++++- .../src/storage/repository-contract.test.ts | 176 +++++++++++++++++- control-plane/src/storage/repository.ts | 7 +- 9 files changed, 531 insertions(+), 41 deletions(-) diff --git a/control-plane/src/domain/types.ts b/control-plane/src/domain/types.ts index ddcb879..ffb4270 100644 --- a/control-plane/src/domain/types.ts +++ b/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 { used: boolean; } diff --git a/control-plane/src/http/server.test.ts b/control-plane/src/http/server.test.ts index 95cc515..6504a97 100644 --- a/control-plane/src/http/server.test.ts +++ b/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 { + 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 { + 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((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((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; + 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(); + } + }); + }); diff --git a/control-plane/src/http/server.ts b/control-plane/src/http/server.ts index 1d2156b..cf8869d 100644 --- a/control-plane/src/http/server.ts +++ b/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); diff --git a/control-plane/src/services/task-service.test.ts b/control-plane/src/services/task-service.test.ts index 420d566..6422dd8 100644 --- a/control-plane/src/services/task-service.test.ts +++ b/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): Promise => { + if (property === 'materializeHandoff') { + return async (...args: Parameters): Promise => { 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): Promise => { + if (property === 'materializeHandoff') { + return async (...args: Parameters): Promise => { 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): Promise => { + if (property === 'materializeHandoff') { + return async (...args: Parameters): Promise => { handoffWrites += 1; - return target.saveHandoff(...args); + return target.materializeHandoff(...args); }; } const value = Reflect.get(target, property); diff --git a/control-plane/src/services/task-service.ts b/control-plane/src/services/task-service.ts index 3a65210..7c18652 100644 --- a/control-plane/src/services/task-service.ts +++ b/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 { diff --git a/control-plane/src/storage/memory-repository.ts b/control-plane/src/storage/memory-repository.ts index 3f5031b..52d8516 100644 --- a/control-plane/src/storage/memory-repository.ts +++ b/control-plane/src/storage/memory-repository.ts @@ -1,4 +1,4 @@ -import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, PendingInputIntent, PendingInputRecord, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffRecord, Machine, MachineLeaseReservation, PendingHandoffIntent, PendingInputIntent, PendingInputRecord, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; import type { TestingMachineReservationRecord, TestingRunRecord } from '../domain/testing-types.js'; import type { Repository, SessionActionDispatchGuard, SessionActionResultGuard, TaskMaintenanceCursor, TestingAttemptDispatchGuard, TestingAttemptMutationGuard } from './repository.js'; @@ -38,12 +38,38 @@ const samePendingInput = (record: PendingInputRecord, intent: PendingInputIntent record.input.kind === intent.input.kind && record.input.value === intent.input.value; +const sameHandoff = (record: HandoffRecord, intent: Omit): boolean => + record.schemaVersion === intent.schemaVersion && + record.operationId === intent.operationId && + record.id === intent.id && + record.taskId === intent.taskId && + record.userId === intent.userId && + record.claimId === intent.claimId && + record.claimGeneration === intent.claimGeneration && + record.expiresInSeconds === intent.expiresInSeconds && + record.url === intent.url && + record.expiresAt === intent.expiresAt; + +const handoffRecord = (intent: PendingHandoffIntent): HandoffRecord => ({ + schemaVersion: intent.schemaVersion, + operationId: intent.operationId, + id: intent.id, + taskId: intent.taskId, + userId: intent.userId, + claimId: intent.claimId, + claimGeneration: intent.claimGeneration, + expiresInSeconds: intent.expiresInSeconds, + url: intent.url, + expiresAt: intent.expiresAt, + used: false +}); + export class MemoryRepository implements Repository { private readonly tasks = new Map(); private readonly pools = new Map(); private readonly machines = new Map(); private readonly profiles = new Map(); - private readonly handoffs = new Map(); + private readonly handoffs = new Map(); private readonly webhooks = new Map(); private readonly pendingInputs = new Map(); private readonly pendingActions = new Map(); @@ -390,12 +416,42 @@ export class MemoryRepository implements Repository { return [...this.profiles.values()].filter((profile) => profile.userId === userId); } - public async saveHandoff(link: HandoffLink): Promise { - this.handoffs.set(link.id, link); + public async getHandoff(id: string): Promise { + return this.handoffs.get(id); } - public async getHandoff(id: string): Promise { - return this.handoffs.get(id); + public async materializeHandoff(intent: PendingHandoffIntent): Promise { + const existing = this.handoffs.get(intent.id); + if (existing === undefined) { + this.handoffs.set(intent.id, handoffRecord(intent)); + return; + } + if (!sameHandoff(existing, intent)) throw new Error('handoff operation integrity failure'); + } + + public async consumeHandoff(link: HandoffRecord, now: number): Promise { + const existing = this.handoffs.get(link.id); + if (existing === undefined || existing.used || Date.parse(existing.expiresAt) <= now || !sameHandoff(existing, link)) return undefined; + const task = this.tasks.get(existing.taskId); + const intent = task?.pendingHandoffIntent; + if ( + task === undefined || + intent === undefined || + task.status !== 'handoff' || + task.userId !== existing.userId || + task.claimId !== existing.claimId || + task.claimGeneration !== existing.claimGeneration || + intent.consumed || + !sameHandoff(existing, intent) + ) return undefined; + this.tasks.set(task.id, { + ...task, + taskVersion: (task.taskVersion ?? 0) + 1, + pendingHandoffIntent: { ...intent, consumed: true } + }); + const consumed = { ...existing, used: true }; + this.handoffs.set(existing.id, consumed); + return consumed; } public async saveWebhook(event: WebhookEvent): Promise { diff --git a/control-plane/src/storage/mongo-repository.ts b/control-plane/src/storage/mongo-repository.ts index a051e8e..0de5314 100644 --- a/control-plane/src/storage/mongo-repository.ts +++ b/control-plane/src/storage/mongo-repository.ts @@ -1,5 +1,5 @@ import { MongoClient, type Collection, type Db, type Document as MongoDriverDocument, type Filter, type MongoClientOptions, type UpdateFilter } from 'mongodb'; -import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, PendingInputIntent, PendingInputRecord, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffRecord, Machine, MachineLeaseReservation, PendingHandoffIntent, PendingInputIntent, PendingInputRecord, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; import type { TestingMachineReservationRecord, TestingRunRecord } from '../domain/testing-types.js'; import type { Repository, SessionActionDispatchGuard, SessionActionResultGuard, TaskMaintenanceCursor, TestingAttemptDispatchGuard, TestingAttemptMutationGuard } from './repository.js'; @@ -45,6 +45,32 @@ const samePendingInput = (record: PendingInputRecord, intent: PendingInputIntent record.input.kind === intent.input.kind && record.input.value === intent.input.value; +const sameHandoff = (record: HandoffRecord, intent: Omit): boolean => + record.schemaVersion === intent.schemaVersion && + record.operationId === intent.operationId && + record.id === intent.id && + record.taskId === intent.taskId && + record.userId === intent.userId && + record.claimId === intent.claimId && + record.claimGeneration === intent.claimGeneration && + record.expiresInSeconds === intent.expiresInSeconds && + record.url === intent.url && + record.expiresAt === intent.expiresAt; + +const handoffRecord = (intent: PendingHandoffIntent): HandoffRecord => ({ + schemaVersion: intent.schemaVersion, + operationId: intent.operationId, + id: intent.id, + taskId: intent.taskId, + userId: intent.userId, + claimId: intent.claimId, + claimGeneration: intent.claimGeneration, + expiresInSeconds: intent.expiresInSeconds, + url: intent.url, + expiresAt: intent.expiresAt, + used: false +}); + const assertPositivePageLimit = (limit: number): void => { if (!Number.isSafeInteger(limit) || limit <= 0) throw new RangeError('page limit must be a positive safe integer'); }; @@ -101,6 +127,7 @@ export class MongoRepository implements Repository { await this.client.connect(); await Promise.all([ this.tasks.createIndex({ status: 1, queuePriority: 1, createdAt: 1 }), + this.handoffs.createIndex({ id: 1 }, { unique: true }), this.tasks.createIndex({ kind: 1, claimId: 1, status: 1, updatedAt: 1 }), this.tasks.createIndex( { kind: 1, status: 1, leaseExpiresAt: 1, _id: 1 }, @@ -533,13 +560,76 @@ export class MongoRepository implements Repository { return (await this.profiles.find({ userId }).toArray()).map(profileFromDocument); } - public async saveHandoff(link: HandoffLink): Promise { - await this.handoffs.replaceOne({ _id: link.id }, { ...link, _id: link.id }, { upsert: true }); + public async getHandoff(id: string): Promise { + const document = await this.handoffs.findOne({ id }); + return document === null ? undefined : handoffFromDocument(document); } - public async getHandoff(id: string): Promise { - const document = await this.handoffs.findOne({ _id: id }); - return document === null ? undefined : handoffFromDocument(document); + public async materializeHandoff(intent: PendingHandoffIntent): Promise { + const record = handoffRecord(intent); + let document: Document | null; + try { + document = await this.handoffs.findOneAndUpdate( + { _id: intent.operationId }, + { $setOnInsert: record }, + { upsert: true, returnDocument: 'after' } + ); + } catch (error) { + document = await this.handoffs.findOne({ $or: [{ _id: intent.operationId }, { id: intent.id }] }); + if (document === null) throw error; + } + const materialized = document === null ? undefined : handoffFromDocument(document); + if (materialized === undefined || !sameHandoff(materialized, intent)) { + throw new Error('handoff operation integrity failure'); + } + } + + public async consumeHandoff(link: HandoffRecord, now: number): Promise { + if (link.used || Date.parse(link.expiresAt) <= now) return undefined; + const materialized = await this.handoffs.findOne({ + _id: link.operationId, + schemaVersion: link.schemaVersion, + id: link.id, + taskId: link.taskId, + userId: link.userId, + claimId: link.claimId, + claimGeneration: link.claimGeneration, + expiresInSeconds: link.expiresInSeconds, + url: link.url, + expiresAt: link.expiresAt, + used: false + }); + if (materialized === null) return undefined; + const task = await this.tasks.findOneAndUpdate( + { + _id: link.taskId, + status: 'handoff', + userId: link.userId, + claimId: link.claimId, + claimGeneration: link.claimGeneration, + 'handoff.url': link.url, + 'handoff.expiresAt': link.expiresAt, + 'pendingHandoffIntent.schemaVersion': link.schemaVersion, + 'pendingHandoffIntent.operationId': link.operationId, + 'pendingHandoffIntent.id': link.id, + 'pendingHandoffIntent.taskId': link.taskId, + 'pendingHandoffIntent.userId': link.userId, + 'pendingHandoffIntent.claimId': link.claimId, + 'pendingHandoffIntent.claimGeneration': link.claimGeneration, + 'pendingHandoffIntent.expiresInSeconds': link.expiresInSeconds, + 'pendingHandoffIntent.url': link.url, + 'pendingHandoffIntent.expiresAt': link.expiresAt, + 'pendingHandoffIntent.consumed': false + }, + { $set: { 'pendingHandoffIntent.consumed': true }, $inc: { taskVersion: 1 } }, + { returnDocument: 'after' } + ); + if (task === null) return undefined; + await this.handoffs.updateOne( + { _id: link.operationId, used: false }, + { $set: { used: true } } + ); + return { ...link, used: true }; } public async saveWebhook(event: WebhookEvent): Promise { @@ -1024,7 +1114,7 @@ const machineFromDocument = ({ ? machine : { ...machine, leaseReservations }; const profileFromDocument = (document: Document): Profile => withoutId(document) as unknown as Profile; -const handoffFromDocument = (document: Document): HandoffLink => withoutId(document) as unknown as HandoffLink; +const handoffFromDocument = (document: Document): HandoffRecord => withoutId(document) as unknown as HandoffRecord; const webhookFromDocument = (document: Document): WebhookEvent => withoutId(document) as unknown as WebhookEvent; const sessionActionResultFromDocument = (document: Document): SessionActionResult => withoutId(document) as unknown as SessionActionResult; const completedSessionActionResultFromDocument = (document: Document): SessionActionResult => ({ diff --git a/control-plane/src/storage/repository-contract.test.ts b/control-plane/src/storage/repository-contract.test.ts index b6fec30..2a52299 100644 --- a/control-plane/src/storage/repository-contract.test.ts +++ b/control-plane/src/storage/repository-contract.test.ts @@ -4,7 +4,7 @@ import { MongoMemoryServer } from 'mongodb-memory-server'; import type { Repository, SessionActionDispatchGuard } from './repository.js'; import { MemoryRepository } from './memory-repository.js'; import { MongoRepository } from './mongo-repository.js'; -import type { BrowserTask, PendingInputIntent, PendingSessionAction, WebhookEvent } from '../domain/types.js'; +import type { BrowserTask, HandoffRecord, PendingHandoffIntent, PendingInputIntent, PendingSessionAction, WebhookEvent } from '../domain/types.js'; import { TaskService } from '../services/task-service.js'; import { SessionService } from '../services/session-service.js'; import { Scheduler } from '../services/scheduler.js'; @@ -397,6 +397,166 @@ const contractTests = (makeHarness: () => Promise): void => { } }, MONGODB_CONTRACT_TEST_TIMEOUT_MS); + it('materializes and atomically consumes one authoritative generation-bound handoff', async () => { + const { repository, close } = await makeHarness(); + const intent: PendingHandoffIntent = { + schemaVersion: 'talos.task-handoff-intent/v1', + operationId: 'handoff-operation-1', + id: 'handoff-link-1', + taskId: 'handoff-task-1', + userId: 'user-1', + claimId: 'handoff-claim-1', + claimGeneration: 1, + expiresInSeconds: 900, + url: '/v1/handoffs/handoff-link-1', + expiresAt: '2025-01-01T00:15:00.000Z', + consumed: false + }; + try { + await repository.saveTask(baseTask({ + id: intent.taskId, + status: 'handoff', + claimId: intent.claimId, + claimGeneration: intent.claimGeneration, + claimCommitted: true, + taskVersion: 1, + handoff: { url: intent.url, expiresAt: intent.expiresAt }, + pendingHandoffIntent: intent + })); + await repository.materializeHandoff(intent); + await repository.materializeHandoff(intent); + const link = await repository.getHandoff(intent.id); + if (link === undefined || !('operationId' in link)) throw new Error('handoff did not materialize'); + const results = await Promise.all([ + repository.consumeHandoff(link, Date.parse('2025-01-01T00:01:00.000Z')), + repository.consumeHandoff(link, Date.parse('2025-01-01T00:01:00.000Z')) + ]); + expect(results.filter((result): result is HandoffRecord => result !== undefined)).toHaveLength(1); + expect(await repository.consumeHandoff(link, Date.parse('2025-01-01T00:01:00.000Z'))).toBeUndefined(); + await repository.materializeHandoff(intent); + expect(await repository.consumeHandoff(link, Date.parse('2025-01-01T00:01:00.000Z'))).toBeUndefined(); + await expect(repository.materializeHandoff({ ...intent, url: '/v1/handoffs/different' })) + .rejects.toThrow('handoff operation integrity failure'); + await expect(repository.materializeHandoff({ ...intent, operationId: 'different-operation' })) + .rejects.toThrow('handoff operation integrity failure'); + } finally { + await close(); + } + }, MONGODB_CONTRACT_TEST_TIMEOUT_MS); + + it('reconciles input and handoff intents after repository restart', async () => { + const harness = await makeHarness(); + let repository = harness.repository; + const clock = { value: Date.parse('2026-09-11T12:00:00.000Z') }; + try { + await repository.savePool({ id: 'restart-pool', visibility: 'platform', tags: {} }); + await repository.saveMachine({ + id: 'restart-machine', poolId: 'restart-pool', tags: {}, capacity: 2, activeLeases: 0, online: true, workerTokenHash: 'hash' + }); + const inputService = taskService(repository, clock); + const inputTask = await inputService.createTask('user-1', { kind: 'browse', goal: 'restart input' }); + const inputClaim = await inputService.claim('restart-input-worker', 'restart-machine'); + await inputService.needsInput(inputTask.id, 'restart-input-worker', inputClaim.leaseToken); + let inputFaulted = false; + const inputFaultRepository = new Proxy(repository, { + get(target, property) { + if (property === 'materializePendingInput') { + return async (): Promise => { + if (!inputFaulted) { + inputFaulted = true; + throw new Error('injected input materialization failure'); + } + throw new Error('unexpected repeated input fault call'); + }; + } + const value = Reflect.get(target, property); + return typeof value === 'function' ? value.bind(target) : value; + } + }); + await expect(taskService(inputFaultRepository, clock).provideInput(inputTask.id, 'user-1', { kind: 'text', value: 'restart answer' })) + .rejects.toThrow('injected input materialization failure'); + repository = await harness.restart(); + const restartedInputService = taskService(repository, clock); + await restartedInputService.provideInput(inputTask.id, 'user-1', { kind: 'text', value: 'restart answer' }); + expect(await restartedInputService.getWorkerInput(inputTask.id, 'restart-input-worker', inputClaim.leaseToken)) + .toEqual({ kind: 'text', value: 'restart answer' }); + expect(await restartedInputService.getWorkerInput(inputTask.id, 'restart-input-worker', inputClaim.leaseToken)).toBeUndefined(); + + const handoffService = taskService(repository, clock); + const handoffTask = await handoffService.createTask('user-1', { kind: 'browse', goal: 'restart handoff' }); + const handoffClaim = await handoffService.claim('restart-handoff-worker', 'restart-machine'); + await handoffService.heartbeat(handoffTask.id, 'restart-handoff-worker', handoffClaim.leaseToken, 30); + let handoffFaulted = false; + const handoffFaultRepository = new Proxy(repository, { + get(target, property) { + if (property === 'materializeHandoff') { + return async (): Promise => { + if (!handoffFaulted) { + handoffFaulted = true; + throw new Error('injected handoff materialization failure'); + } + throw new Error('unexpected repeated handoff fault call'); + }; + } + const value = Reflect.get(target, property); + return typeof value === 'function' ? value.bind(target) : value; + } + }); + await expect(taskService(handoffFaultRepository, clock).requestHandoff(handoffTask.id, 'user-1', 900)) + .rejects.toThrow('injected handoff materialization failure'); + const storedIntent = (await repository.getTask(handoffTask.id))?.pendingHandoffIntent; + if (storedIntent === undefined) throw new Error('handoff intent was not persisted'); + repository = await harness.restart(); + expect(await taskService(repository, clock).requestHandoff(handoffTask.id, 'user-1', 900)).toEqual({ + handoff_url: storedIntent.url, + expires: storedIntent.expiresAt + }); + const link = await repository.getHandoff(storedIntent.id); + if (link === undefined || !('operationId' in link)) throw new Error('handoff did not materialize after restart'); + expect(await repository.consumeHandoff(link, clock.value)).toMatchObject({ id: storedIntent.id, used: true }); + expect(await repository.consumeHandoff(link, clock.value)).toBeUndefined(); + } finally { + await harness.close(); + } + }, MONGODB_CONTRACT_TEST_TIMEOUT_MS); + + it('rejects expired and stale-generation handoff consumption', async () => { + const { repository, close } = await makeHarness(); + const intent: PendingHandoffIntent = { + schemaVersion: 'talos.task-handoff-intent/v1', + operationId: 'handoff-operation-stale', + id: 'handoff-link-stale', + taskId: 'handoff-task-stale', + userId: 'user-1', + claimId: 'handoff-claim-1', + claimGeneration: 1, + expiresInSeconds: 60, + url: '/v1/handoffs/handoff-link-stale', + expiresAt: '2025-01-01T00:01:00.000Z', + consumed: false + }; + try { + await repository.saveTask(baseTask({ + id: intent.taskId, + status: 'handoff', + claimId: 'handoff-claim-2', + claimGeneration: 2, + claimCommitted: true, + taskVersion: 2, + handoff: { url: intent.url, expiresAt: intent.expiresAt }, + pendingHandoffIntent: intent + })); + await repository.materializeHandoff(intent); + const link = await repository.getHandoff(intent.id); + if (link === undefined || !('operationId' in link)) throw new Error('handoff did not materialize'); + expect(await repository.consumeHandoff(link, Date.parse('2025-01-01T00:00:30.000Z'))).toBeUndefined(); + expect(await repository.consumeHandoff(link, Date.parse('2025-01-01T00:01:00.000Z'))).toBeUndefined(); + expect((await repository.getHandoff(intent.id))?.used).toBe(false); + } finally { + await close(); + } + }, MONGODB_CONTRACT_TEST_TIMEOUT_MS); + it('returns one stable bounded page across deadline and lease expiry sources', async () => { const { repository, close } = await makeHarness(); try { @@ -2167,7 +2327,19 @@ const contractTests = (makeHarness: () => Promise): void => { await repository.createProfile({ id: 'profile-1', userId: 'user-1', machineId: 'machine-1' }); const task = baseTask({ profileId: 'profile-1', poolId: 'pool-1' }); await repository.saveTask(task); - await repository.saveHandoff({ id: 'handoff-1', taskId: task.id, userId: task.userId, url: '/v1/handoffs/handoff-1', expiresAt: '2025-01-01T00:10:00.000Z', used: false }); + await repository.materializeHandoff({ + schemaVersion: 'talos.task-handoff-intent/v1', + operationId: 'round-trip-handoff-operation', + id: 'handoff-1', + taskId: task.id, + userId: task.userId, + claimId: 'round-trip-claim', + claimGeneration: 1, + expiresInSeconds: 600, + url: '/v1/handoffs/handoff-1', + expiresAt: '2025-01-01T00:10:00.000Z', + consumed: false + }); const event: WebhookEvent = { id: 'event-1', type: 'task.state_changed', taskId: task.id, userId: task.userId, timestamp: task.createdAt, payload: { status: 'submitted' }, delivery: { status: 'pending', attempts: 0 } }; await repository.saveWebhook(event); const pendingInputIntent: PendingInputIntent = { diff --git a/control-plane/src/storage/repository.ts b/control-plane/src/storage/repository.ts index 3a5d938..6a4c99f 100644 --- a/control-plane/src/storage/repository.ts +++ b/control-plane/src/storage/repository.ts @@ -1,4 +1,4 @@ -import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, PendingInputIntent, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffRecord, Machine, MachineLeaseReservation, PendingHandoffIntent, PendingInputIntent, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; import type { TestingAttemptStatus, TestingMachineReservationRecord, TestingRunRecord } from '../domain/testing-types.js'; export interface TestingAttemptMutationGuard { @@ -82,8 +82,9 @@ export interface Repository { releaseLegacyProfileLease(profileId: string, taskId: string): Promise; listProfiles(): Promise; listProfilesByUser(userId: string): Promise; - saveHandoff(link: HandoffLink): Promise; - getHandoff(id: string): Promise; + getHandoff(id: string): Promise; + materializeHandoff(intent: PendingHandoffIntent): Promise; + consumeHandoff(link: HandoffRecord, now: number): Promise; saveWebhook(event: WebhookEvent): Promise; getWebhook(id: string): Promise; listWebhooks(): Promise; From 1eeda8e57aab0d3351efbda04143954fff956a6b Mon Sep 17 00:00:00 2001 From: "chronoai-fkst[bot]" Date: Fri, 11 Sep 2026 06:02:08 +0000 Subject: [PATCH 2/2] auto-fix refs #50: Production slice: make Task input and handoff intents restart-convergent --- control-plane/src/storage/memory-repository.ts | 3 ++- control-plane/src/storage/repository-contract.test.ts | 2 ++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/control-plane/src/storage/memory-repository.ts b/control-plane/src/storage/memory-repository.ts index 52d8516..7f1fa42 100644 --- a/control-plane/src/storage/memory-repository.ts +++ b/control-plane/src/storage/memory-repository.ts @@ -421,7 +421,8 @@ export class MemoryRepository implements Repository { } public async materializeHandoff(intent: PendingHandoffIntent): Promise { - const existing = this.handoffs.get(intent.id); + const existing = [...this.handoffs.values()].find((handoff) => handoff.operationId === intent.operationId) + ?? this.handoffs.get(intent.id); if (existing === undefined) { this.handoffs.set(intent.id, handoffRecord(intent)); return; diff --git a/control-plane/src/storage/repository-contract.test.ts b/control-plane/src/storage/repository-contract.test.ts index 2a52299..ae7fa2c 100644 --- a/control-plane/src/storage/repository-contract.test.ts +++ b/control-plane/src/storage/repository-contract.test.ts @@ -425,6 +425,8 @@ const contractTests = (makeHarness: () => Promise): void => { })); await repository.materializeHandoff(intent); await repository.materializeHandoff(intent); + await expect(repository.materializeHandoff({ ...intent, id: 'different-handoff-link' })) + .rejects.toThrow('handoff operation integrity failure'); const link = await repository.getHandoff(intent.id); if (link === undefined || !('operationId' in link)) throw new Error('handoff did not materialize'); const results = await Promise.all([