From 7a0c3099dd4cc5e086008d523ed6376a8c52a46e Mon Sep 17 00:00:00 2001 From: "chronoai-fkst[bot]" Date: Fri, 11 Sep 2026 04:05:45 +0000 Subject: [PATCH 1/3] auto-implement refs #47: Walking skeleton: recover one generation-bound Task input after post-CAS write failure --- control-plane/src/domain/types.ts | 14 ++ control-plane/src/http/server.test.ts | 154 +++++++++++++++++- .../http/session-routes.integration.test.ts | 3 +- control-plane/src/openapi.test.ts | 3 +- .../src/services/task-service.test.ts | 38 ++--- control-plane/src/services/task-service.ts | 55 ++++++- .../src/services/webhook-dispatcher.test.ts | 3 +- .../src/storage/memory-repository.ts | 31 +++- control-plane/src/storage/mongo-repository.ts | 39 ++++- .../src/storage/repository-contract.test.ts | 43 ++++- control-plane/src/storage/repository.ts | 6 +- 11 files changed, 337 insertions(+), 52 deletions(-) diff --git a/control-plane/src/domain/types.ts b/control-plane/src/domain/types.ts index 151988c..ddcb879 100644 --- a/control-plane/src/domain/types.ts +++ b/control-plane/src/domain/types.ts @@ -56,6 +56,19 @@ export interface TaskInput { value: string; } +export interface PendingInputIntent { + schemaVersion: 'talos.task-input-intent/v1'; + operationId: string; + taskId: string; + claimId: string; + claimGeneration: number; + input: TaskInput; +} + +export interface PendingInputRecord extends PendingInputIntent { + consumed: boolean; +} + export interface Artifact { id: string; name: string; @@ -105,6 +118,7 @@ interface TaskBase { lastActionId?: string; sessionActions?: readonly SessionActionRecord[]; claimRecovery?: TaskClaimRecovery; + pendingInputIntent?: PendingInputIntent; } export interface BrowserTask extends TaskBase { diff --git a/control-plane/src/http/server.test.ts b/control-plane/src/http/server.test.ts index 4a4df40..95cc515 100644 --- a/control-plane/src/http/server.test.ts +++ b/control-plane/src/http/server.test.ts @@ -8,6 +8,19 @@ 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'; + +class FirstMaterializationFailureRepository extends MemoryRepository { + public readonly attemptedIntents: PendingInputIntent[] = []; + public successfulMaterializations = 0; + + public override async materializePendingInput(intent: PendingInputIntent): Promise { + this.attemptedIntents.push(structuredClone(intent)); + if (this.attemptedIntents.length === 1) throw new Error('injected pending input materialization failure'); + await super.materializePendingInput(intent); + this.successfulMaterializations += 1; + } +} describe('control-plane HTTP API', () => { it('serves cached OpenAPI JSON and YAML without authentication', async () => { @@ -200,7 +213,8 @@ describe('control-plane HTTP API', () => { 'machineId', 'leaseExpiresAt', 'leaseToken', - 'claimRecovery' + 'claimRecovery', + 'pendingInputIntent' ]; for (const field of internalAuthorityFields) expect(claim.task).not.toHaveProperty(field); const publicTaskResponse = await fetch(`${base}/v1/tasks/${created.id}`, { @@ -463,4 +477,142 @@ describe('control-plane HTTP API', () => { expect((await fetch(`${base}/v1/profiles/p/login-link`, { method: 'POST', headers: publicHeaders })).status).toBe(501); server.close(); }); + + it('reconciles one generation-bound input after post-CAS materialization failure', async () => { + const now = Date.parse('2026-09-11T12:00:00.000Z'); + const repository = new FirstMaterializationFailureRepository(() => now); + await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} }); + await repository.saveMachine({ + id: 'machine-a', + 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-a' + }; + const internalFields = [ + 'pendingInputIntent', + 'operationId', + 'claimId', + 'claimGeneration', + 'leaseToken', + 'workerId', + 'machineId' + ]; + const assertPublicTask = (task: Record): void => { + for (const field of internalFields) expect(task).not.toHaveProperty(field); + }; + + try { + const createdResponse = await fetch(`${base}/v1/tasks`, { + method: 'POST', + headers: publicHeaders, + body: JSON.stringify({ kind: 'browse', goal: 'recover input' }) + }); + const created = await createdResponse.json() as Record & { id: string }; + expect(createdResponse.status).toBe(201); + assertPublicTask(created); + + const claimResponse = await fetch(`${base}/v1/worker/claim`, { + method: 'POST', + headers: workerHeaders, + body: JSON.stringify({ worker_id: 'worker-a', machine_id: 'machine-a' }) + }); + const claim = await claimResponse.json() as { task: Record; leaseToken: string }; + expect(claimResponse.status).toBe(200); + assertPublicTask(claim.task); + + const needsInputResponse = await fetch(`${base}/v1/worker/tasks/${created.id}/needs-input`, { + method: 'POST', + headers: workerHeaders, + body: JSON.stringify({ lease_token: claim.leaseToken }) + }); + expect(needsInputResponse.status).toBe(200); + assertPublicTask(await needsInputResponse.json() as Record); + + const firstInputResponse = await fetch(`${base}/v1/tasks/${created.id}/input`, { + method: 'POST', + headers: publicHeaders, + body: '{"kind":"text","value":"answer"}' + }); + expect(firstInputResponse.status).toBe(500); + expect(await firstInputResponse.json()).toEqual({ + error: { code: 'internal_error', message: 'internal server error', retryable: true } + }); + const failedTask = await repository.getTask(created.id); + expect(failedTask).toMatchObject({ + status: 'running', + pendingInputIntent: { + taskId: created.id, + claimId: failedTask?.claimId, + claimGeneration: failedTask?.claimGeneration, + input: { kind: 'text', value: 'answer' } + } + }); + expect(failedTask?.pendingInputIntent?.claimGeneration).toBeGreaterThan(0); + expect(repository.attemptedIntents).toHaveLength(1); + expect(repository.successfulMaterializations).toBe(0); + + const retryResponse = await fetch(`${base}/v1/tasks/${created.id}/input`, { + method: 'POST', + headers: publicHeaders, + body: '{"kind":"text","value":"answer"}' + }); + expect(retryResponse.status).toBe(200); + const retriedTask = await retryResponse.json() as Record & { status: string }; + expect(retriedTask.status).toBe('running'); + assertPublicTask(retriedTask); + expect(repository.attemptedIntents).toHaveLength(2); + expect(repository.attemptedIntents[1]?.operationId).toBe(repository.attemptedIntents[0]?.operationId); + expect(repository.successfulMaterializations).toBe(1); + + const pollBody = JSON.stringify({ + lease_token: claim.leaseToken, + worker_token: 'worker-token-123456', + worker_id: 'worker-a', + machine_id: 'machine-a' + }); + const firstPoll = await fetch(`${base}/v1/worker/tasks/${created.id}/input/poll`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: pollBody + }); + expect(firstPoll.status).toBe(200); + 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(secondPoll.status).toBe(200); + expect(await secondPoll.json()).toEqual({}); + expect(await repository.getTask(created.id)).not.toHaveProperty('pendingInputIntent'); + } finally { + server.close(); + } + }); }); diff --git a/control-plane/src/http/session-routes.integration.test.ts b/control-plane/src/http/session-routes.integration.test.ts index 0bd146f..fd05348 100644 --- a/control-plane/src/http/session-routes.integration.test.ts +++ b/control-plane/src/http/session-routes.integration.test.ts @@ -100,7 +100,8 @@ describe('interactive session HTTP API', () => { 'machineId', 'leaseExpiresAt', 'leaseToken', - 'claimRecovery' + 'claimRecovery', + 'pendingInputIntent' ]) 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`, { diff --git a/control-plane/src/openapi.test.ts b/control-plane/src/openapi.test.ts index 8a46c1f..c3772a4 100644 --- a/control-plane/src/openapi.test.ts +++ b/control-plane/src/openapi.test.ts @@ -82,7 +82,8 @@ describe('OpenAPI loader', () => { 'machineId', 'leaseExpiresAt', 'leaseToken', - 'claimRecovery' + 'claimRecovery', + 'pendingInputIntent' ]; for (const publicSchema of ['Task', 'Session']) { const publicProperties = properties(publicSchema); diff --git a/control-plane/src/services/task-service.test.ts b/control-plane/src/services/task-service.test.ts index 0c1c11a..128e84f 100644 --- a/control-plane/src/services/task-service.test.ts +++ b/control-plane/src/services/task-service.test.ts @@ -253,7 +253,8 @@ describe('task service', () => { 'machineId', 'leaseExpiresAt', 'leaseToken', - 'claimRecovery' + 'claimRecovery', + 'pendingInputIntent' ]) expect(serialized).not.toContain(`\"${field}\"`); expect(serialized).not.toContain(claim.leaseToken); expect(serialized).not.toContain(stored?.claimId); @@ -450,17 +451,13 @@ describe('task service', () => { let handoffWrites = 0; const repository = new Proxy(storage, { get(target, property) { - if (property === 'replaceTaskForClaim') { - return async (...args: Parameters): Promise => { + if (property === 'replaceTaskForActiveClaim') { + return async (...args: Parameters): Promise => { if (args[0].id === inputTask.id && args[0].status === 'running' && args[1].status === 'needs_input') { inputAttempts += 1; + await authority.heartbeat(inputTask.id, 'worker-input', inputClaim.leaseToken, 30 + inputAttempts); return false; } - return target.replaceTaskForClaim(...args); - }; - } - if (property === 'replaceTaskForActiveClaim') { - return async (...args: Parameters): Promise => { if (args[0].id === handoffTask.id && args[0].status === 'handoff') { handoffAttempts += 1; await authority.heartbeat( @@ -474,10 +471,10 @@ describe('task service', () => { return target.replaceTaskForActiveClaim(...args); }; } - if (property === 'savePendingInput') { - return async (...args: Parameters): Promise => { + if (property === 'materializePendingInput') { + return async (...args: Parameters): Promise => { pendingInputWrites += 1; - return target.savePendingInput(...args); + return target.materializePendingInput(...args); }; } if (property === 'saveHandoff') { @@ -506,7 +503,7 @@ describe('task service', () => { expect(handoffAttempts).toBe(3); expect(pendingInputWrites).toBe(0); expect(handoffWrites).toBe(0); - expect(await storage.takePendingInput(inputTask.id)).toBeUndefined(); + expect(pendingInputWrites).toBe(0); expect(await storage.getTask(handoffTask.id)).not.toHaveProperty('handoff'); }); @@ -592,18 +589,13 @@ describe('task service', () => { let handoffWrites = 0; const repository = new Proxy(storage, { get(target, property) { - if (property === 'replaceTaskForClaim') { - return async (...args: Parameters): Promise => { + if (property === 'replaceTaskForActiveClaim') { + return async (...args: Parameters): Promise => { if (!inputReclaimed && args[0].id === inputTask.id && args[0].status === 'running' && args[1].status === 'needs_input') { inputReclaimed = true; await advanceClaimGeneration(target, inputTask.id, 'needs_input'); return false; } - return target.replaceTaskForClaim(...args); - }; - } - if (property === 'replaceTaskForActiveClaim') { - return async (...args: Parameters): Promise => { if (!handoffReclaimed && args[0].id === handoffTask.id && args[0].status === 'handoff') { handoffReclaimed = true; await advanceClaimGeneration(target, handoffTask.id, 'running'); @@ -612,10 +604,10 @@ describe('task service', () => { return target.replaceTaskForActiveClaim(...args); }; } - if (property === 'savePendingInput') { - return async (...args: Parameters): Promise => { + if (property === 'materializePendingInput') { + return async (...args: Parameters): Promise => { pendingInputWrites += 1; - return target.savePendingInput(...args); + return target.materializePendingInput(...args); }; } if (property === 'saveHandoff') { @@ -649,7 +641,7 @@ describe('task service', () => { expect((await storage.getTask(handoffTask.id))?.claimGeneration).toBe((handoffClaim.task.claimGeneration ?? 0) + 1); expect(pendingInputWrites).toBe(0); expect(handoffWrites).toBe(0); - expect(await storage.takePendingInput(inputTask.id)).toBeUndefined(); + expect(pendingInputWrites).toBe(0); expect(await storage.getTask(handoffTask.id)).not.toHaveProperty('handoff'); }); diff --git a/control-plane/src/services/task-service.ts b/control-plane/src/services/task-service.ts index f7ffca8..0920d2b 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, PublicTask, SessionActionResult, Task, TaskClaimRecoveryReason, TaskClaimGuard, TaskFinding, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { Lease, MachineLeaseReservation, 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'; @@ -220,8 +220,21 @@ export class TaskService { public async provideInput(id: string, userId: string, input: NonNullable): Promise { const task = await this.authorizedTask(id, userId); + if (this.isPendingInputReconciliation(task, input)) { + await this.repository.materializePendingInput(task.pendingInputIntent); + if (!await this.ensureClaimProjections(task)) throw conflict('lease accounting could not be renewed'); + return task; + } this.assertTaskAcceptsInput(task); const claimBinding = this.userClaimBinding(task); + const pendingInputIntent: PendingInputIntent = { + schemaVersion: 'talos.task-input-intent/v1', + operationId: newId('task-input'), + taskId: task.id, + claimId: claimBinding.claimId, + claimGeneration: claimBinding.claimGeneration, + input + }; const { persisted } = await this.replaceAuthorizedTask(id, userId, task, (current) => { this.assertTaskAcceptsInput(current); const now = this.clock(); @@ -230,14 +243,15 @@ export class TaskService { return { ...current, status: 'running', + pendingInputIntent, updatedAt: new Date(now).toISOString(), leaseExpiresAt: new Date(Math.max( Number.isFinite(currentLeaseExpiry) ? currentLeaseExpiry : 0, extendedLeaseExpiry )).toISOString() }; - }, { claimBinding }); - await this.repository.savePendingInput(id, input); + }, { claimBinding, requireActiveClaim: true }); + await this.repository.materializePendingInput(pendingInputIntent); if (!await this.ensureClaimProjections(persisted)) throw conflict('lease accounting could not be renewed'); await this.emit(persisted, 'task.state_changed', { status: persisted.status }); return persisted; @@ -259,7 +273,18 @@ export class TaskService { public async getWorkerInput(taskId: string, workerId: string, leaseToken: string): Promise { const task = await this.getWorkerTask(taskId, workerId, leaseToken); if (task.interaction === 'interactive') throw conflict('interactive sessions do not accept task input'); - return this.repository.takePendingInput(taskId); + const intent = task.pendingInputIntent; + if (intent === undefined) return undefined; + const input = await this.repository.consumePendingInput(intent); + if (input !== undefined && task.leaseExpiresAt !== undefined) { + const cleared: Task = { ...task }; + delete cleared.pendingInputIntent; + await this.repository.replaceTaskForActiveClaim(cleared, { + ...this.claimGuard(task), + leaseExpiresAt: task.leaseExpiresAt + }); + } + return input; } public async requestHandoff(id: string, userId: string, expiresInSeconds: number): Promise<{ handoff_url: string; expires: string }> { @@ -832,8 +857,29 @@ export class TaskService { } private assertTaskAcceptsInput(task: Task): void { + if (task.kind === 'testing') throw conflict('testing tasks do not accept task input'); if (task.interaction === 'interactive') throw conflict('interactive sessions do not accept task input'); if (task.status !== 'needs_input') throw conflict('task is not waiting for input'); + if (task.claimCommitted !== true) throw conflict('task claim is not active'); + } + + private isPendingInputReconciliation(task: Task, input: TaskInput): task is Task & { pendingInputIntent: PendingInputIntent } { + const intent = task.pendingInputIntent; + return task.kind !== 'testing' && + task.interaction !== 'interactive' && + task.status === 'running' && + 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.claimId === task.claimId && + intent.claimGeneration === task.claimGeneration && + intent.input.kind === input.kind && + intent.input.value === input.value; } private assertTaskCanRequestHandoff(task: Task): void { @@ -1042,6 +1088,7 @@ export class TaskService { 'lastActionId', 'sessionActions', 'claimRecovery', + 'pendingInputIntent', 'testing' ]); return { diff --git a/control-plane/src/services/webhook-dispatcher.test.ts b/control-plane/src/services/webhook-dispatcher.test.ts index 2dc8a7f..b9ce07d 100644 --- a/control-plane/src/services/webhook-dispatcher.test.ts +++ b/control-plane/src/services/webhook-dispatcher.test.ts @@ -131,7 +131,8 @@ describe('WebhookDispatcher', () => { 'machineId', 'leaseExpiresAt', 'leaseToken', - 'claimRecovery' + 'claimRecovery', + 'pendingInputIntent' ]) expect(body).not.toContain(`\"${field}\"`); expect(body).not.toContain(claim.leaseToken); expect(body).not.toContain(stored?.claimId); diff --git a/control-plane/src/storage/memory-repository.ts b/control-plane/src/storage/memory-repository.ts index a1edc42..3f5031b 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, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, 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'; @@ -29,6 +29,15 @@ const assertPositivePageLimit = (limit: number): void => { if (!Number.isSafeInteger(limit) || limit <= 0) throw new RangeError('page limit must be a positive safe integer'); }; +const samePendingInput = (record: PendingInputRecord, intent: PendingInputIntent): boolean => + record.schemaVersion === intent.schemaVersion && + record.operationId === intent.operationId && + record.taskId === intent.taskId && + record.claimId === intent.claimId && + record.claimGeneration === intent.claimGeneration && + record.input.kind === intent.input.kind && + record.input.value === intent.input.value; + export class MemoryRepository implements Repository { private readonly tasks = new Map(); private readonly pools = new Map(); @@ -36,7 +45,7 @@ export class MemoryRepository implements Repository { private readonly profiles = new Map(); private readonly handoffs = new Map(); private readonly webhooks = new Map(); - private readonly pendingInputs = new Map(); + private readonly pendingInputs = new Map(); private readonly pendingActions = new Map(); private readonly actionResults = new Map(); private readonly testingRuns = new Map(); @@ -401,14 +410,20 @@ export class MemoryRepository implements Repository { return [...this.webhooks.values()]; } - public async savePendingInput(taskId: string, input: TaskInput): Promise { - this.pendingInputs.set(taskId, input); + public async materializePendingInput(intent: PendingInputIntent): Promise { + const existing = this.pendingInputs.get(intent.operationId); + if (existing === undefined) { + this.pendingInputs.set(intent.operationId, { ...intent, consumed: false }); + return; + } + if (!samePendingInput(existing, intent)) throw new Error('pending input operation integrity failure'); } - public async takePendingInput(taskId: string): Promise { - const input = this.pendingInputs.get(taskId); - this.pendingInputs.delete(taskId); - return input; + public async consumePendingInput(intent: PendingInputIntent): Promise { + const existing = this.pendingInputs.get(intent.operationId); + if (existing === undefined || existing.consumed || !samePendingInput(existing, intent)) return undefined; + this.pendingInputs.set(intent.operationId, { ...existing, consumed: true }); + return existing.input; } public async enqueueSessionAction(action: PendingSessionAction): Promise { diff --git a/control-plane/src/storage/mongo-repository.ts b/control-plane/src/storage/mongo-repository.ts index 57543c9..879753e 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, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, 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'; @@ -36,6 +36,15 @@ const LEGACY_ACTION_RECOVERY_ERROR = { } } as const; +const samePendingInput = (record: PendingInputRecord, intent: PendingInputIntent): boolean => + record.schemaVersion === intent.schemaVersion && + record.operationId === intent.operationId && + record.taskId === intent.taskId && + record.claimId === intent.claimId && + record.claimGeneration === intent.claimGeneration && + record.input.kind === intent.input.kind && + record.input.value === intent.input.value; + const assertPositivePageLimit = (limit: number): void => { if (!Number.isSafeInteger(limit) || limit <= 0) throw new RangeError('page limit must be a positive safe integer'); }; @@ -544,13 +553,31 @@ export class MongoRepository implements Repository { return (await this.webhooks.find({}).toArray()).map(webhookFromDocument); } - public async savePendingInput(taskId: string, input: TaskInput): Promise { - await this.pendingInputs.replaceOne({ _id: taskId }, { _id: taskId, input }, { upsert: true }); + public async materializePendingInput(intent: PendingInputIntent): Promise { + const document = await this.pendingInputs.findOneAndUpdate( + { _id: intent.operationId }, + { $setOnInsert: { ...intent, consumed: false } }, + { upsert: true, returnDocument: 'after' } + ); + if (document === null || !samePendingInput(document as PendingInputRecord, intent)) { + throw new Error('pending input operation integrity failure'); + } } - public async takePendingInput(taskId: string): Promise { - const result = await this.pendingInputs.findOneAndDelete({ _id: taskId }); - const document = result ?? null; + public async consumePendingInput(intent: PendingInputIntent): Promise { + const document = await this.pendingInputs.findOneAndUpdate( + { + _id: intent.operationId, + taskId: intent.taskId, + claimId: intent.claimId, + claimGeneration: intent.claimGeneration, + 'input.kind': intent.input.kind, + 'input.value': intent.input.value, + consumed: false + }, + { $set: { consumed: true } }, + { returnDocument: 'before' } + ); return document === null ? undefined : document.input as TaskInput; } diff --git a/control-plane/src/storage/repository-contract.test.ts b/control-plane/src/storage/repository-contract.test.ts index a34a5c8..b6fec30 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, PendingSessionAction, WebhookEvent } from '../domain/types.js'; +import type { BrowserTask, 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'; @@ -370,6 +370,33 @@ const normalizedMongoSort = (value: unknown): Readonly> }; const contractTests = (makeHarness: () => Promise): void => { + it('materializes and consumes one generation-bound input without resurrection', async () => { + const { repository, close } = await makeHarness(); + const intent: PendingInputIntent = { + schemaVersion: 'talos.task-input-intent/v1', + operationId: 'input-operation-1', + taskId: 'input-task-1', + claimId: 'input-claim-1', + claimGeneration: 1, + input: { kind: 'text', value: 'answer' } + }; + try { + await repository.materializePendingInput(intent); + await repository.materializePendingInput(intent); + expect(await repository.consumePendingInput(intent)).toEqual(intent.input); + expect(await repository.consumePendingInput(intent)).toBeUndefined(); + await repository.materializePendingInput(intent); + expect(await repository.consumePendingInput(intent)).toBeUndefined(); + await expect(repository.materializePendingInput({ + ...intent, + input: { kind: 'text', value: 'different' } + })).rejects.toThrow('pending input operation integrity failure'); + expect(await repository.consumePendingInput({ ...intent, claimGeneration: 2 })).toBeUndefined(); + } 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 { @@ -2143,7 +2170,15 @@ const contractTests = (makeHarness: () => Promise): void => { 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 }); 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); - await repository.savePendingInput(task.id, { kind: 'text', value: 'secret' }); + const pendingInputIntent: PendingInputIntent = { + schemaVersion: 'talos.task-input-intent/v1', + operationId: 'round-trip-input-operation', + taskId: task.id, + claimId: 'round-trip-claim', + claimGeneration: 1, + input: { kind: 'text', value: 'secret' } + }; + await repository.materializePendingInput(pendingInputIntent); expect(await repository.getPool('pool-1')).toMatchObject({ ownerUserId: 'user-1', sharedWithGroups: ['eng'] }); expect(await repository.listPoolsByOwner('user-1')).toHaveLength(1); expect(await repository.getMachine('machine-1')).toMatchObject({ activeLeases: 0 }); @@ -2155,8 +2190,8 @@ const contractTests = (makeHarness: () => Promise): void => { expect(await repository.getHandoff('handoff-1')).toMatchObject({ used: false }); expect(await repository.getWebhook('event-1')).toMatchObject({ delivery: { status: 'pending', attempts: 0 } }); expect(await repository.listWebhooks()).toHaveLength(1); - expect(await repository.takePendingInput(task.id)).toEqual({ kind: 'text', value: 'secret' }); - expect(await repository.takePendingInput(task.id)).toBeUndefined(); + expect(await repository.consumePendingInput(pendingInputIntent)).toEqual({ kind: 'text', value: 'secret' }); + expect(await repository.consumePendingInput(pendingInputIntent)).toBeUndefined(); await repository.ping(); } finally { await close(); diff --git a/control-plane/src/storage/repository.ts b/control-plane/src/storage/repository.ts index 78ba513..3a5d938 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, PendingSessionAction, Pool, Profile, SessionActionResult, Task, TaskActiveClaimGuard, TaskClaimGuard, TaskInput, TaskRecoveryGuard, WebhookEvent } from '../domain/types.js'; +import type { ActionDispatchBinding, HandoffLink, Machine, MachineLeaseReservation, 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 { @@ -87,8 +87,8 @@ export interface Repository { saveWebhook(event: WebhookEvent): Promise; getWebhook(id: string): Promise; listWebhooks(): Promise; - savePendingInput(taskId: string, input: TaskInput): Promise; - takePendingInput(taskId: string): Promise; + materializePendingInput(intent: PendingInputIntent): Promise; + consumePendingInput(intent: PendingInputIntent): Promise; enqueueSessionAction(action: PendingSessionAction): Promise; getPendingSessionAction(taskId: string): Promise; takePendingSessionAction(taskId: string, guard: SessionActionDispatchGuard): Promise; From 74be4f87fe30637b62ea6a43d83135953501fda2 Mon Sep 17 00:00:00 2001 From: "chronoai-fkst[bot]" Date: Fri, 11 Sep 2026 04:21:21 +0000 Subject: [PATCH 2/3] auto-fix refs #47: Walking skeleton: recover one generation-bound Task input after post-CAS write failure --- control-plane/src/services/task-service.test.ts | 17 +++++++++++++++++ control-plane/src/services/task-service.ts | 7 ++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/control-plane/src/services/task-service.test.ts b/control-plane/src/services/task-service.test.ts index 128e84f..420d566 100644 --- a/control-plane/src/services/task-service.test.ts +++ b/control-plane/src/services/task-service.test.ts @@ -645,6 +645,23 @@ describe('task service', () => { expect(await storage.getTask(handoffTask.id)).not.toHaveProperty('handoff'); }); + it('does not consume pending input retained from a previous claim generation', async () => { + const { repository, service } = setup({ value: 1_000 }); + await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} }); + await repository.saveMachine({ id: 'machine', poolId: 'pool', tags: {}, capacity: 1, activeLeases: 0, online: true, workerTokenHash: 'x' }); + const task = await service.createTask('user-a', { kind: 'browse', goal: 'generation-bound input consumption' }); + const claim = await service.claim('worker-a', 'machine'); + await service.needsInput(task.id, 'worker-a', claim.leaseToken); + await service.provideInput(task.id, 'user-a', { kind: 'text', value: 'stale input' }); + const intent = (await repository.getTask(task.id))?.pendingInputIntent; + if (intent === undefined) throw new Error('test task does not have a pending input intent'); + + await advanceClaimGeneration(repository, task.id, 'running'); + + await expect(service.getWorkerInput(task.id, 'worker-next', 'lease-next')).resolves.toBeUndefined(); + await expect(repository.consumePendingInput(intent)).resolves.toEqual(intent.input); + }); + it('lets either eligible machine claim queued work', async () => { const { repository, service } = setup(); await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} }); diff --git a/control-plane/src/services/task-service.ts b/control-plane/src/services/task-service.ts index 0920d2b..3a65210 100644 --- a/control-plane/src/services/task-service.ts +++ b/control-plane/src/services/task-service.ts @@ -274,7 +274,12 @@ export class TaskService { const task = await this.getWorkerTask(taskId, workerId, leaseToken); if (task.interaction === 'interactive') throw conflict('interactive sessions do not accept task input'); const intent = task.pendingInputIntent; - if (intent === undefined) return undefined; + if ( + intent === undefined || + intent.taskId !== task.id || + intent.claimId !== task.claimId || + intent.claimGeneration !== task.claimGeneration + ) return undefined; const input = await this.repository.consumePendingInput(intent); if (input !== undefined && task.leaseExpiresAt !== undefined) { const cleared: Task = { ...task }; From 58ae387fb199f4037a6e4af18b028b7e349c3921 Mon Sep 17 00:00:00 2001 From: "chronoai-fkst[bot]" Date: Fri, 11 Sep 2026 04:40:16 +0000 Subject: [PATCH 3/3] auto-fix refs #47: Walking skeleton: recover one generation-bound Task input after post-CAS write failure --- control-plane/src/storage/mongo-repository.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/control-plane/src/storage/mongo-repository.ts b/control-plane/src/storage/mongo-repository.ts index 879753e..a051e8e 100644 --- a/control-plane/src/storage/mongo-repository.ts +++ b/control-plane/src/storage/mongo-repository.ts @@ -62,6 +62,8 @@ export interface MongoRepositoryOptions { clientOptions?: MongoClientOptions; } +type PendingInputDocument = PendingInputRecord & { _id: string }; + export class MongoRepository implements Repository { private readonly client: MongoClient; private readonly database: Db; @@ -71,7 +73,7 @@ export class MongoRepository implements Repository { private readonly profiles: Collection; private readonly handoffs: Collection; private readonly webhooks: Collection; - private readonly pendingInputs: Collection; + private readonly pendingInputs: Collection; private readonly pendingActions: Collection; private readonly actionResults: Collection; private readonly testingRuns: Collection; @@ -559,7 +561,7 @@ export class MongoRepository implements Repository { { $setOnInsert: { ...intent, consumed: false } }, { upsert: true, returnDocument: 'after' } ); - if (document === null || !samePendingInput(document as PendingInputRecord, intent)) { + if (document === null || !samePendingInput(document, intent)) { throw new Error('pending input operation integrity failure'); } }