github-devloop implementation for #42

control-plane/src/domain/schemas.ts

@@ -167,18 +167,28 @@ export const workerInputPollSchema = workerNeedsInputSchema;
 
 export const workerActionPollSchema = workerNeedsInputSchema;
 
+export const workerActionResultPayloadSchema = z.object({
+  screenshot: z.object({
+    mimeType: z.enum(['image/jpeg', 'image/png']),
+    data: z.string(),
+    width: z.number().int().nonnegative(),
+    height: z.number().int().nonnegative()
+  }).optional(),
+  value: z.unknown().optional(),
+  error: z.object({ code: z.string().min(1), message: z.string().min(1) }).optional()
+}).strict();
+
+export const workerActionResultEnvelopeSchema = z.object({
+  worker_token: z.unknown().optional(),
+  worker_id: z.unknown().optional(),
+  machine_id: z.unknown().optional(),
+  lease_token: z.unknown().optional(),
+  result: z.unknown().optional()
+}).passthrough();
+
 export const workerActionResultSchema = workerBodyCredentialsSchema.extend({
   lease_token: z.string().min(1),
-  result: z.object({
-    screenshot: z.object({
-      mimeType: z.enum(['image/jpeg', 'image/png']),
-      data: z.string(),
-      width: z.number().int().nonnegative(),
-      height: z.number().int().nonnegative()
-    }).optional(),
-    value: z.unknown().optional(),
-    error: z.object({ code: z.string().min(1), message: z.string().min(1) }).optional()
-  }).strict()
+  result: workerActionResultPayloadSchema
 });
 
 export const resultSchema = workerBodyCredentialsSchema.extend({

control-plane/src/domain/types.ts

@@ -103,6 +103,7 @@ interface TaskBase {
   handoff?: { url: string; expiresAt: string };
   pendingActionId?: string;
   lastActionId?: string;
+  sessionActions?: readonly SessionActionRecord[];
   claimRecovery?: TaskClaimRecovery;
 }
 
@@ -143,11 +144,25 @@ export interface PublicTask {
 
 export type SessionAction = BrowserAction;
 
+export interface ActionDispatchBinding {
+  schemaVersion: 'talos.internal-action-dispatch-binding/v1';
+  dispatchId: string;
+  dispatchGeneration: number;
+  workerId: string;
+  machineId: string;
+  leaseTokenDigest: string;
+}
+
 export interface PendingSessionAction {
+  schemaVersion: 'talos.internal-session-action/v1';
   id: string;
   taskId: string;
   action: SessionAction;
   state: 'pending' | 'dispatched';
+  dispatchGeneration: number;
+  dispatchBinding?: ActionDispatchBinding;
+  dispatchClaimId?: string;
+  dispatchClaimGeneration?: number;
   createdAt: string;
 }
 
@@ -156,8 +171,15 @@ export interface SessionActionResult {
   taskId: string;
   result: unknown;
   completedAt: string;
+  dispatchBinding?: ActionDispatchBinding;
+  unbound?: true;
 }
 
+export type SessionActionRecord = PendingSessionAction | (Omit<PendingSessionAction, 'state'> & {
+  state: 'completed';
+  completion: SessionActionResult;
+});
+
 export interface Pool {
   id: string;
   visibility: 'private' | 'org' | 'platform';

control-plane/src/http/server.ts

@@ -32,7 +32,7 @@ import {
   workerBodyCredentialsSchema,
   workerClaimSchema,
   workerActionPollSchema,
-  workerActionResultSchema,
+  workerActionResultEnvelopeSchema,
   workerInputPollSchema,
   workerNeedsInputSchema
 } from '../domain/schemas.js';
@@ -144,8 +144,11 @@ export const createApiServer = (
   });
 };
 
-const isWorkerActionResultPath = (method: string | undefined, path: string): boolean =>
-  method === 'POST' && /^\/v1\/worker\/tasks\/[^/]+\/actions\/[^/]+\/result$/.test(path);
+const isWorkerActionResultPath = (method: string | undefined, path: string): boolean => {
+  const parts = path.split('/').filter(Boolean);
+  return method === 'POST' && parts.length === 7 && parts[0] === 'v1' && parts[1] === 'worker' &&
+    parts[2] === 'tasks' && parts[4] === 'actions' && parts[6] === 'result';
+};
 
 const route = async (
   request: IncomingMessage,
@@ -422,19 +425,11 @@ const workerRoute = async (
   options: ServerOptions
 ): Promise<void> => {
   const method = request.method ?? 'GET';
-  const isActionResult = parts[2] === 'tasks' && parts[4] === 'actions' && parts[6] === 'result';
+  const isActionResult = method === 'POST' && parts.length === 7 && parts[2] === 'tasks' && parts[4] === 'actions' && parts[5] !== undefined && parts[6] === 'result';
   const maxBodyBytes = isActionResult ? 8 * 1024 * 1024 : options.maxBodyBytes;
   const body = method === 'POST' ? await readBody(request, maxBodyBytes) : undefined;
-  const workerIdentity = await requireWorker(request, repository, body);
-  if (isActionResult) {
-    const bodyCredentials = workerBodyCredentialsSchema.safeParse(body);
-    if (bodyCredentials.success && (
-      (bodyCredentials.data.machine_id !== undefined && bodyCredentials.data.machine_id !== workerIdentity.machineId) ||
-      (bodyCredentials.data.worker_id !== undefined && bodyCredentials.data.worker_id !== workerIdentity.workerId)
-    )) {
-      throw unauthorized('action result credentials do not match authenticated worker');
-    }
-  }
+  const workerIdentity = await requireWorker(request, repository, body, isActionResult);
+
   if (parts[2] === 'testing') {
     return testingWorkerRoute(response, testingAttempts, parts, method, body, workerIdentity);
   }
@@ -449,8 +444,10 @@ const workerRoute = async (
     return send(response, 404, publicErrorEnvelope('not_found', 'route not found', 404));
   }
   const taskId = parts[3];
-  const task = await repository.getTask(taskId);
-  if (task?.machineId !== workerIdentity.machineId) throw unauthorized('task is assigned to another machine');
+  if (!isActionResult) {
+    const task = await repository.getTask(taskId);
+    if (task?.machineId !== workerIdentity.machineId) throw unauthorized('task is assigned to another machine');
+  }
   const worker = workerIdentity.workerId;
   if (method === 'POST' && parts[4] === 'heartbeat') {
     const input = heartbeatSchema.parse(body);
@@ -468,9 +465,10 @@ const workerRoute = async (
     const input = workerActionPollSchema.parse(body);
     return send(response, 200, await sessions.pollWorkerAction(taskId, worker, input.lease_token));
   }
-  if (method === 'POST' && parts[4] === 'actions' && parts[5] !== undefined && parts[6] === 'result') {
-    const input = workerActionResultSchema.parse(body);
-    await sessions.saveWorkerResult(taskId, parts[5], worker, input.lease_token, input.result, workerIdentity.machineId);
+  if (isActionResult && parts[5] !== undefined) {
+    const envelope = workerActionResultEnvelopeSchema.parse(body);
+    const leaseToken = boundedCredential(envelope.lease_token);
+    await sessions.saveWorkerResult(taskId, parts[5], worker, leaseToken, envelope.result, workerIdentity.machineId);
     return send(response, 200, { stored: true });
   }
   if (method === 'GET' && parts[4] === 'input') {
@@ -662,7 +660,8 @@ interface WorkerIdentity {
 const requireWorker = async (
   request: IncomingMessage,
   repository: Repository,
-  body: unknown
+  body: unknown,
+  requireMatchingCarriers = false
 ): Promise<WorkerIdentity> => {
   const auth = request.headers.authorization;
   const headerToken = request.headers['x-talos-worker-token']?.toString();
@@ -672,12 +671,22 @@ const requireWorker = async (
   const bodyCredentials = workerBodyCredentialsSchema.safeParse(body);
   const bodyData = bodyCredentials.success ? bodyCredentials.data : {};
   const headersSelected = headerToken !== undefined || bearerToken !== undefined;
+  const bodySelected = bodyData.worker_token !== undefined || bodyData.machine_id !== undefined || bodyData.worker_id !== undefined;
+  if (requireMatchingCarriers && headersSelected && bodySelected && (
+    headerMachineId === undefined || headerWorkerId === undefined ||
+    bodyData.worker_token === undefined || bodyData.machine_id === undefined || bodyData.worker_id === undefined ||
+    (headerToken ?? bearerToken) !== bodyData.worker_token || headerMachineId !== bodyData.machine_id || headerWorkerId !== bodyData.worker_id
+  )) throw unauthorized('unauthorized');
   const token = headerToken ?? bearerToken ?? bodyData.worker_token;
   const machineId = headersSelected ? headerMachineId : bodyData.machine_id;
   const workerId = headersSelected ? headerWorkerId : bodyData.worker_id;
   if (token === undefined || machineId === undefined || workerId === undefined) {
     throw unauthorized('worker token, machine id, and worker id are required');
   }
+  if (requireMatchingCarriers) {
+    boundedCredential(token);
+    if (Buffer.byteLength(workerId, 'utf8') > 255 || Buffer.byteLength(machineId, 'utf8') > 255) throw unauthorized('unauthorized');
+  }
   const machine = await repository.getMachine(machineId);
   if (machine === undefined) throw unauthorized('invalid worker token');
   const expected = Buffer.from(machine.workerTokenHash);
@@ -726,3 +735,10 @@ const boundedPublicErrorMessage = (message: string): string => message.slice(0,
 export const parseError = z.object({
   error: z.object({ code: z.string(), message: z.string(), retryable: z.boolean() }).passthrough()
 });
+
+const boundedCredential = (value: unknown): string => {
+  if (typeof value !== 'string' || value.length === 0 || Buffer.byteLength(value, 'utf8') > 4096) {
+    throw unauthorized('unauthorized');
+  }
+  return value;
+};

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

@@ -164,7 +164,7 @@ describe('interactive session HTTP API', () => {
 
   it('returns one opaque denial for invalid terminal action-result bindings', async () => {
     const clock = { value: 1_000 };
-    const repository = new MemoryRepository();
+    const repository = new MemoryRepository(() => clock.value);
     await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
     await repository.saveMachine({
       id: 'machine',
@@ -332,7 +332,7 @@ describe('interactive session HTTP API', () => {
       headers: originalHeaders,
       body: JSON.stringify({ lease_token: claim.leaseToken })
     });
-    expect(authorizedMalformed.status).toBe(400);
-    expect(await authorizedMalformed.json()).toMatchObject({ error: { code: 'validation_error' } });
+    expect(authorizedMalformed.status).toBe(409);
+    expect(await authorizedMalformed.json()).toMatchObject({ error: { code: 'action_already_completed' } });
   });
 });

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

@@ -8,7 +8,7 @@ import { WebhookSigner } from './webhook-signer.js';
 
 const setup = async () => {
   const clock = { value: 1_000 };
-  const repository = new MemoryRepository();
+  const repository = new MemoryRepository(() => clock.value);
   await repository.savePool({ id: 'pool', visibility: 'platform', tags: {} });
   await repository.saveMachine({
     id: 'machine',
@@ -97,7 +97,8 @@ describe('session service', () => {
             pending.action_id,
             'worker',
             claim.leaseToken,
-            { value: 'loaded' }
+            { value: 'loaded' },
+            'machine'
           );
         }
       }
@@ -123,7 +124,8 @@ describe('session service', () => {
       first.action_id,
       'worker',
       claim.leaseToken,
-      { value: 'done' }
+      { value: 'done' },
+      'machine'
     );
 
     await expect(sessions.getAction(created.id, first.action_id, 'user-a', 0)).resolves.toEqual({
@@ -148,15 +150,17 @@ describe('session service', () => {
       pending.action_id,
       'worker',
       claim.leaseToken,
-      { value: 'done' }
+      { value: 'done' },
+      'machine'
     );
 
     await expect(sessions.saveWorkerResult(
       created.id,
       pending.action_id,
       'worker',
       claim.leaseToken,
-      { value: 'done' }
+      { value: 'done' },
+      'machine'
     )).rejects.toMatchObject({ code: 'action_already_completed', status: 409 });
   });
 
@@ -166,7 +170,7 @@ describe('session service', () => {
     const claim = await tasks.claim('worker', 'machine');
     const sent = await sessions.sendAction(created.id, 'user-a', { type: 'wait', milliseconds: 1 }, 0);
     await sessions.pollWorkerAction(created.id, 'worker', claim.leaseToken);
-    await sessions.saveWorkerResult(created.id, sent.action_id, 'worker', claim.leaseToken, { value: 'winner' });
+    await sessions.saveWorkerResult(created.id, sent.action_id, 'worker', claim.leaseToken, { value: 'winner' }, 'machine');
     await sessions.close(created.id, 'user-a');
     clock.value = 12_000;
     await tasks.expireLeases();
@@ -215,7 +219,7 @@ describe('session service', () => {
   });
 
   it('returns a dispatched action to pending when an interactive lease is requeued', async () => {
-    const { clock, sessions, tasks } = await setup();
+    const { clock, repository, sessions, tasks } = await setup();
     const created = await sessions.create('user-a', { mode: 'act', constraints: {} });
     const claim = await tasks.claim('worker-one', 'machine');
     const sent = await sessions.sendAction(created.id, 'user-a', { type: 'wait', milliseconds: 1 }, 0);
@@ -225,5 +229,12 @@ describe('session service', () => {
     await tasks.expireLeases();
     const replacement = await tasks.claim('worker-two', 'machine');
     expect((await sessions.pollWorkerAction(created.id, 'worker-two', replacement.leaseToken)).action?.id).toBe(sent.action_id);
+    await expect(sessions.saveWorkerResult(
+      created.id, sent.action_id, 'worker-one', claim.leaseToken, { value: 'stale' }, 'machine'
+    )).rejects.toMatchObject({ code: 'unauthorized', status: 401 });
+    await sessions.saveWorkerResult(
+      created.id, sent.action_id, 'worker-two', replacement.leaseToken, { value: 'winner' }, 'machine'
+    );
+    expect((await repository.getSessionActionResult(sent.action_id))?.result).toEqual({ value: 'winner' });
   });
 });