fix(agent-harness): preserve ordered Cloud Agent dispatch identity

apps/web/src/lib/agent-harness/cloud-agent-context.test.ts

@@ -1,8 +1,14 @@
 import { beforeEach, describe, expect, it, jest } from '@jest/globals';
-import { TRPCError } from '@trpc/server';
+import { TRPCClientError } from '@trpc/client';
+import { getTRPCErrorFromUnknown, TRPCError } from '@trpc/server';
+import { getHTTPStatusCodeFromError } from '@trpc/server/http';
+import { TRPC_ERROR_CODES_BY_KEY } from '@trpc/server/rpc';
+import { generateMessageId } from '@kilocode/cloud-agent-sdk/message-id';
+import { insertSorted } from '@kilocode/cloud-agent-sdk/storage/helpers';
 import { z } from 'zod';
 import {
   caller,
+  dispatchStartedAt,
   fixture,
   invocation,
   message,
@@ -21,7 +27,7 @@
 jest.mock('@sentry/nextjs', () => ({ captureException: jest.fn() }));
 
 // The repository transformer does not hoist mocks.
-const { createHarnessCloudAgentContext } =
+const { createHarnessCloudAgentContext, normalizeCloudAgentAdmissionError } =
   jest.requireActual<typeof Context>('./cloud-agent-context');
 const { fetchSessionMessagesPage } = jest.requireActual<typeof SessionIngest>(
   '@/lib/session-ingest-client'
@@ -106,7 +112,7 @@
     const decode = jest.spyOn(TextDecoder.prototype, 'decode');
     await expect(invoke('attach', reference)).rejects.toThrow('size limit');
     expect(decode).not.toHaveBeenCalled();
     expect(read).toHaveBeenNthCalledWith(2, { session_id: sessionId, limit: 20, bounded: true });
   });
   it.each(cases)('denies removed access for %s', async (name, args) => {
     fixture.revoked = true;
@@ -179,12 +185,105 @@
 });
 it('keeps replay identity stable without colliding across dispatches or inputs', () => {
   const input = invocation('continue', { ...reference, message: 'one' });
-  const identity = (value: unknown) => createHarnessCloudAgentContext('token', value).messageId;
-  expect(identity(input)).toBe(identity({ ...input, arguments: { message: 'one', ...reference } }));
+  const identity = (value: unknown) => createHarnessCloudAgentContext('token', value).messageId!;
+  const harness = identity(input);
+  const clock = jest.spyOn(Date, 'now').mockReturnValue(dispatchStartedAt - 1);
+  const legacy = generateMessageId();
+  clock.mockReturnValue(dispatchStartedAt);
+  expect(harness.slice(4, 16)).toBe(generateMessageId().slice(4, 16));
+  clock.mockReturnValue(dispatchStartedAt + 1);
+  const assistant = generateMessageId();
+  clock.mockReturnValue(dispatchStartedAt + 60_000);
+  expect(identity({ ...input, arguments: { message: 'one', ...reference } })).toBe(harness);
+  expect(harness).toMatch(/^msg_[a-f0-9]{12}[A-Za-z0-9]{14}$/);
+  expect([assistant, harness, legacy].reduce(insertSorted, [])).toEqual([
+    legacy,
+    harness,
+    assistant,
+  ]);
   for (const change of [
     { operationId: org },
     { conversationId: org },
+    { dispatchStartedAt: dispatchStartedAt + 1 },
     { arguments: { ...reference, message: 'two' } },
   ])
-    expect(identity({ ...input, ...change })).not.toBe(identity(input));
+    expect(identity({ ...input, ...change })).not.toBe(harness);
+});
+
+const mutations = [
+  ['start', { prompt: 'Fix', modelId: 'model' }, 'prompt'],
+  ['continue', { ...reference, message: 'Continue' }, 'message'],
+  ['stop', reference, 'sessionId'],
+] as const;
+it.each(mutations)('authenticates %s identity in both scopes', async (name, args, field) => {
+  for (const scope of [null, org]) {
+    fixture.organizationId = fixture.sessionScope = scope;
+    const input = invocation(name, args);
+    const fresh = (value: unknown) => createHarnessCloudAgentContext(input.name, value).fresh();
+    expect((await fresh(input)).authority).toEqual({ userId, organizationId: scope });
+    for (const change of [
+      { dispatchStartedAt: dispatchStartedAt + 1 },
+      { arguments: { ...args, [field]: 'changed' } },
+    ])
+      await expect(fresh({ ...input, ...change })).rejects.toMatchObject({ code: 'FORBIDDEN' });
+  }
+});
+it.each(mutations)('rejects missing or unusable %s dispatch identity', (name, args) => {
+  const input = invocation(name, args);
+  for (const time of [undefined, null, -1, 1.5, NaN, Infinity, 2 ** 53, '0'])
+    expect(() =>
+      createHarnessCloudAgentContext(input.name, { ...input, dispatchStartedAt: time })
+    ).toThrow(z.ZodError);
+  expect(() =>
+    createHarnessCloudAgentContext(input.name, { ...input, messageId: 'unsigned-override' })
+  ).toThrow(z.ZodError);
+});
+
+const remoteError = (code: TRPCError['code']) =>
+  TRPCClientError.from({
+    error: {
+      message: 'provider-private-text',
+      code: TRPC_ERROR_CODES_BY_KEY[code],
+      data: { code, httpStatus: getHTTPStatusCodeFromError(new TRPCError({ code })) },
+    },
+  });
+it.each([
+  'UNAUTHORIZED',
+  'FORBIDDEN',
+  'BAD_REQUEST',
+  'PRECONDITION_FAILED',
+  'PAYMENT_REQUIRED',
+] as const)('preserves sanitized %s rejection through the server caller wrapper', code => {
+  const remote = remoteError(code);
+  const wrapped = getTRPCErrorFromUnknown(remote);
+  expect(wrapped).toMatchObject({ code: 'INTERNAL_SERVER_ERROR', cause: remote });
+  for (const error of [
+    remote,
+    wrapped,
+    new TRPCError({ code, message: 'provider-private-text' }),
+  ]) {
+    const normalized = normalizeCloudAgentAdmissionError(error);
+    expect(normalized).toMatchObject({ code, message: 'Cloud Agent rejected this operation.' });
+    expect(normalized?.cause).toBeUndefined();
+  }
+});
+it.each([
+  new Error('response lost'),
+  TRPCClientError.from(new Error('response lost')),
+  remoteError('SERVICE_UNAVAILABLE'),
+  remoteError('TIMEOUT'),
+  remoteError('CONFLICT'),
+  { data: { code: 'FORBIDDEN', httpStatus: 403 } },
+  ...[
+    undefined,
+    { code: 'BAD_REQUEST' },
+    { code: 'BAD_REQUEST', httpStatus: 500 },
+    { code: 'FORBIDDEN', httpStatus: 403 },
+    { code: 'BAD_REQUEST', httpStatus: '400' },
+  ].map(data =>
+    TRPCClientError.from({ error: { message: 'provider-private-text', code: -32600, data } })
+  ),
+])('keeps transport, ambiguous, or malformed errors uncertain: %s', error => {
+  expect(normalizeCloudAgentAdmissionError(error)).toBeUndefined();
+  expect(normalizeCloudAgentAdmissionError(getTRPCErrorFromUnknown(error))).toBeUndefined();
 });

apps/web/src/lib/agent-harness/cloud-agent-context.ts

@@ -1,19 +1,63 @@
 import 'server-only';
+import { TRPCClientError } from '@trpc/client';
 import { TRPCError } from '@trpc/server';
+import { getHTTPStatusCodeFromError } from '@trpc/server/http';
+import { TRPC_ERROR_CODES_BY_KEY } from '@trpc/server/rpc';
 import { z } from 'zod';
 import { type ToolOutcome } from '@kilocode/agent-harness/contracts';
 import { ToolRequestSchema, toolDefinitions } from '@kilocode/agent-harness/tools';
 import { rootRouter } from '@/routers/root-router';
 import { authorizeHarnessCapability, harnessInputDigest } from './authorization';
 
 const Id = z.uuid().transform(value => value.toLowerCase());
+const DispatchTime = z.int().nonnegative();
 const definitions = toolDefinitions.filter(tool => tool.name.startsWith('kilo.sessions.'));
 const Invocation = z.strictObject({
   conversationId: Id,
   operationId: Id,
   name: z.enum(definitions.map(tool => tool.name)),
   arguments: z.unknown(),
+  dispatchStartedAt: DispatchTime.optional(),
 });
+const AdmissionCode = z.enum([
+  'UNAUTHORIZED',
+  'FORBIDDEN',
+  'BAD_REQUEST',
+  'PRECONDITION_FAILED',
+  'PAYMENT_REQUIRED',
+]);
+const RemoteAdmission = z.object({
+  message: z.string(),
+  code: z.int(),
+  data: z.object({ code: AdmissionCode, httpStatus: z.int() }),
+});
+
+export function normalizeCloudAgentAdmissionError(error: unknown): TRPCError | undefined {
+  // Only unwrap server caller wrappers, never infer rejection from transport text or arbitrary causes.
+  for (
+    let depth = 0;
+    depth < 4 && error instanceof TRPCError && error.code === 'INTERNAL_SERVER_ERROR';
+    depth++
+  )
+    error = error.cause;
+  const remote = error instanceof TRPCClientError ? RemoteAdmission.safeParse(error.shape) : null;
+  const code = AdmissionCode.safeParse(
+    error instanceof TRPCError ? error.code : remote?.success ? remote.data.data.code : undefined
+  );
+  if (!code.success) return undefined;
+  const normalized = new TRPCError({
+    code: code.data,
+    message: 'Cloud Agent rejected this operation.',
+  });
+  if (
+    remote?.success &&
+    (remote.data.code !== TRPC_ERROR_CODES_BY_KEY[code.data] ||
+      remote.data.data.httpStatus !== getHTTPStatusCodeFromError(normalized))
+  )
+    return undefined;
+  return normalized;
+}
+
 function bounded<T>(value: T): T {
   if (Buffer.byteLength(JSON.stringify(value), 'utf8') > 64 * 1024)
     throw new TRPCError({ code: 'PAYLOAD_TOO_LARGE' });
@@ -29,17 +73,29 @@
   bounded(request);
   const definition = definitions.find(tool => tool.name === request.name);
   if (!definition) throw new TRPCError({ code: 'BAD_REQUEST' });
+  // Reads retain their deployed argument-only digest. Mutations cannot invent legacy dispatch times.
+  const dispatchStartedAt =
+    definition.effect === 'read' ? undefined : DispatchTime.parse(invocation.dispatchStartedAt);
   const scope = {
     audience: 'agent-harness:operations',
     conversationId: invocation.conversationId,
     operation: request.name,
     definitionVersion: definition.version,
-    inputDigest: harnessInputDigest(request.arguments),
+    inputDigest: harnessInputDigest(
+      dispatchStartedAt === undefined
+        ? request.arguments
+        : { arguments: request.arguments, dispatchStartedAt }
+    ),
     dispatchId: invocation.operationId,
     target: { kind: 'backend' } as const,
   };
-  // Stable, schema-valid message identity; include the immutable input and conversation in its digest.
-  const messageId = `msg_${harnessInputDigest(scope).slice(0, 26)}`;
+  // Match the deployed SDK's six-byte millisecond << 12 prefix; only the suffix is a scoped digest.
+  const messageId =
+    dispatchStartedAt === undefined
+      ? undefined
+      : `msg_${BigInt.asUintN(48, BigInt(dispatchStartedAt) << 12n)
+          .toString(16)
+          .padStart(12, '0')}${harnessInputDigest(scope).slice(0, 14)}`;
   const fresh = async () => {
     const { ctx, authority } = await authorizeHarnessCapability(token, scope);
     return { caller: rootRouter.createCaller(ctx), authority };

apps/web/src/lib/agent-harness/cloud-agent-test-fixture.ts

@@ -1,6 +1,7 @@
 import { beforeEach, jest } from '@jest/globals';
 import { createHash } from 'node:crypto';
 import { TRPCError } from '@trpc/server';
+import { canonicalizeValidatedInput } from '@kilocode/agent-harness/commands';
 import type { HarnessCapabilityScope } from './authorization';
 
 const conversationId = '11111111-1111-4111-8111-111111111111';
@@ -10,7 +11,9 @@ export const userId = 'oauth/github:owner';
 export const sessionId = 'ses_12345678901234567890123456';
 export const cloudId = 'agent_real_reference';
 export const reference = { sessionId };
-const digest = (value: unknown) => createHash('sha256').update(JSON.stringify(value)).digest('hex');
+export const dispatchStartedAt = 1717986919400;
+const digest = (value: unknown) =>
+  createHash('sha256').update(canonicalizeValidatedInput(value)).digest('hex');
 export const message = (id: string, content: string, role = 'user') => ({
   info: { id, sessionID: sessionId, role },
   parts: [
@@ -23,6 +26,7 @@ const initialFixture = () => ({
   sessionScope: org as string | null,
   revoked: false,
   grantRevoked: false,
+  inputDigest: '',
   hideEvidence: false,
   unavailable: false,
   historyKind: undefined as string | undefined,
@@ -46,6 +50,8 @@ jest.mock('./authorization', () => ({
     if (
       fixture.grantRevoked ||
       token !== scope.operation ||
+      scope.inputDigest !== fixture.inputDigest ||
+      scope.definitionVersion !== '1' ||
       scope.conversationId !== conversationId ||
       scope.dispatchId !== operationId ||
       scope.target.kind !== 'backend' ||
@@ -113,12 +119,19 @@ export const caller = {
   },
 };
 jest.mock('@/routers/root-router', () => ({ rootRouter: { createCaller: () => caller } }));
-export const invocation = (name: string, args: unknown) => ({
-  conversationId,
-  operationId,
-  name: `kilo.sessions.${name}`,
-  arguments: args,
-});
+export const invocation = (name: string, args: unknown) => {
+  const identity = ['start', 'continue', 'stop'].includes(name) ? { dispatchStartedAt } : {};
+  fixture.inputDigest = digest(
+    'dispatchStartedAt' in identity ? { arguments: args, ...identity } : args
+  );
+  return {
+    conversationId,
+    operationId,
+    name: `kilo.sessions.${name}`,
+    arguments: args,
+    ...identity,
+  };
+};
 beforeEach(() => {
   jest.restoreAllMocks();
   Object.assign(fixture, initialFixture());