ChronoAIProject · chronoai-fkst · Sep 11, 2026 · Sep 11, 2026 · Sep 11, 2026 · Sep 11, 2026
diff --git 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
@@ -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<void> {
+    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<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-a'
+    };
+    const internalFields = [
+      'pendingInputIntent',
+      'operationId',
+      'claimId',
+      'claimGeneration',
+      'leaseToken',
+      'workerId',
+      'machineId'
+    ];
+    const assertPublicTask = (task: Record<string, unknown>): 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<string, unknown> & { 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<string, unknown>; 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<string, unknown>);
+
+      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<string, unknown> & { 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
@@ -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
@@ -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
@@ -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<Repository>(storage, {
       get(target, property) {
-        if (property === 'replaceTaskForClaim') {
-          return async (...args: Parameters<Repository['replaceTaskForClaim']>): Promise<boolean> => {
+        if (property === 'replaceTaskForActiveClaim') {
+          return async (...args: Parameters<Repository['replaceTaskForActiveClaim']>): Promise<boolean> => {
             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<Repository['replaceTaskForActiveClaim']>): Promise<boolean> => {
             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<Repository['savePendingInput']>): Promise<void> => {
+        if (property === 'materializePendingInput') {
+          return async (...args: Parameters<Repository['materializePendingInput']>): Promise<void> => {
             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<Repository>(storage, {
       get(target, property) {
-        if (property === 'replaceTaskForClaim') {
-          return async (...args: Parameters<Repository['replaceTaskForClaim']>): Promise<boolean> => {
+        if (property === 'replaceTaskForActiveClaim') {
+          return async (...args: Parameters<Repository['replaceTaskForActiveClaim']>): Promise<boolean> => {
             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<Repository['replaceTaskForActiveClaim']>): Promise<boolean> => {
             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<Repository['savePendingInput']>): Promise<void> => {
+        if (property === 'materializePendingInput') {
+          return async (...args: Parameters<Repository['materializePendingInput']>): Promise<void> => {
             pendingInputWrites += 1;
-            return target.savePendingInput(...args);
+            return target.materializePendingInput(...args);
           };
         }
         if (property === 'saveHandoff') {
@@ -649,10 +641,27 @@ 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');
   });
 
+  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: {} });