packages/runtime-host/src/protocol/message.ts

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import { invalidProtocolFrame } from './errors.js';
import {
  assertExactKeys,
  requireCount,
  requireEntityId,
  requireExactRecord,
  requireId,
  requireRecord,
  requireShapedRecord,
  requireUtf8String,
} from './codec.js';
import { defineOperation } from './operation-spec.js';
import {
  decodeMessageContent,
  decodeMessageAdmissionContent,
  decodeSkillIds,
  decodeTurnOrchestration,
  decodeTurnSnapshot,
  type MessageContent,
  TURN_MESSAGE_TEXT_MAX_BYTES,
  type TurnSnapshot,
} from './turn.js';
import { decodeSkillInvocationResult } from '@maka/core/skill-invocation';
import type { SkillInvocationResult } from '@maka/core/skill-invocation';
import type { TurnOrchestration } from '@maka/core/runtime-inputs';

export const MESSAGE_QUEUE_MAX_ENTRIES = 64;
export const MESSAGE_QUEUE_PROJECTION_MAX_BYTES = 52 * 1024;
export const MESSAGE_OPERATION_RESULT_MAX_BYTES = 56 * 1024;

export type MessagePlacement = 'current_turn' | 'next_turn';

interface MessageQueueEntrySnapshotBase {
  readonly entryId: string;
  readonly messageId: string;
  readonly content: MessageContent;
  readonly placement: MessagePlacement;
}

export interface QueuedMessageSnapshot extends MessageQueueEntrySnapshotBase {
  readonly state: 'queued';
}

export interface InFlightMessageSnapshot extends MessageQueueEntrySnapshotBase {
  readonly placement: 'current_turn';
  readonly state: 'in_flight';
}

export interface RetractedMessageSnapshot extends MessageQueueEntrySnapshotBase {
  readonly state: 'retracted';
}

export type MessageQueueEntrySnapshot =
  | QueuedMessageSnapshot
  | InFlightMessageSnapshot
  | RetractedMessageSnapshot;

export type SteeringMessageSnapshot =
  | (QueuedMessageSnapshot & { readonly placement: 'current_turn' })
  | InFlightMessageSnapshot;

export interface SessionMessageQueueProjection {
  readonly hostEpoch: string;
  readonly queueRevision: number;
  readonly steering: readonly SteeringMessageSnapshot[];
  readonly followup: readonly QueuedMessageSnapshot[];
}

/**
 * The sole client admission input for a user Message. `skillIds` and
 * `turnOrchestration` carry exact-Turn intent: the Host, not the client,
 * decides that such a Message can only open its own Turn.
 */
export interface TurnMessageSubmitInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly messageId: string;
  readonly content: MessageContent;
  readonly placement: MessagePlacement;
  readonly skillIds?: readonly string[];
  readonly turnOrchestration?: TurnOrchestration;
}

export type TurnMessageSubmitResult = {
  readonly skillInvocation: SkillInvocationResult;
} & (
  | {
      readonly disposition: 'steering' | 'followup';
      /** Absent when an older Host Epoch can prove admission but not its transient revision. */
      readonly queueRevision?: number;
    }
  | { readonly disposition: 'turn_started'; readonly turnId: string }
  | { readonly disposition: 'blocked' }
);

export interface TurnMessageQueryInput {
  readonly sessionId: string;
  readonly messageIds: readonly string[];
}

/**
 * Durable cancellation proof for the queried identities. Only a cancelled
 * Message retires a client's transient row; every other identity stays visible
 * until canonical transcript replaces it, so absence needs no status of its own.
 */
export interface TurnMessageQueryResult {
  readonly cancelledMessageIds: readonly string[];
}

export interface TurnMessageExecutionQueryInput {
  readonly sessionId: string;
  readonly messageIds: readonly string[];
}

export interface TurnMessageExecutionQueryResult {
  readonly resolutions: readonly TurnMessageExecutionResolution[];
}

export type TurnMessageExecutionResolution =
  | { readonly messageId: string; readonly state: 'pending' }
  | { readonly messageId: string; readonly state: 'cancelled' }
  | {
      readonly messageId: string;
      readonly state: 'owned';
      readonly turnId: string;
      readonly runId: string;
    };

export interface QueueRetractInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly retractId: string;
}

export interface QueueRetractResult {
  readonly queueRevision: number;
  readonly retracted: readonly RetractedMessageSnapshot[];
}

export interface QueueEntryRetractInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly entryId: string;
  readonly retractId: string;
}

export interface QueueMutationResult {
  readonly queueRevision: number;
}

export interface QueueEntryPromoteInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly entryId: string;
  readonly promoteId: string;
}

export interface QueueEntryUpdateInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly entryId: string;
  readonly updateId: string;
  readonly expectedQueueRevision: number;
  readonly text: string;
}

export interface QueueEntriesReorderInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly reorderId: string;
  readonly entryIds: readonly string[];
}

export interface TurnInterruptInput {
  readonly originHostEpoch: string;
  readonly sessionId: string;
  readonly interruptId: string;
  readonly turnId: string;
  readonly runId: string;
}

export interface TurnInterruptResult {
  readonly queueRevision: number;
  readonly retracted: readonly RetractedMessageSnapshot[];
  readonly turn: TurnSnapshot;
}

const MESSAGE_OPERATION_ERRORS = [
  'host_not_ready',
  'host_draining',
  'operation_unavailable',
  'not_found',
  'session_archived',
  'session_busy',
  'operation_conflict',
  'outcome_unknown',
  'internal_failure',
] as const;

export const MESSAGE_OPERATION_SPECS = {
  'turn.message.query': defineOperation({
    mode: 'query',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeTurnMessageQueryInput,
    decodeOutput: decodeTurnMessageQueryResult,
  }),
  'turn.message.execution.query': defineOperation({
    mode: 'query',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeTurnMessageExecutionQueryInput,
    decodeOutput: decodeTurnMessageExecutionQueryResult,
  }),
  'turn.message.submit': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeTurnMessageSubmitInput,
    decodeOutput: decodeTurnMessageSubmitResult,
  }),
  'queue.retract': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeQueueRetractInput,
    decodeOutput: decodeQueueRetractResult,
  }),
  'queue.entry.retract': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeQueueEntryRetractInput,
    decodeOutput: decodeQueueMutationResult,
  }),
  'queue.entry.promote': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeQueueEntryPromoteInput,
    decodeOutput: decodeQueueMutationResult,
  }),
  'queue.entry.update': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeQueueEntryUpdateInput,
    decodeOutput: decodeQueueMutationResult,
  }),
  'queue.entries.reorder': defineOperation({
    mode: 'command',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeQueueEntriesReorderInput,
    decodeOutput: decodeQueueMutationResult,
  }),
  'turn.interrupt': defineOperation({
    mode: 'control',
    availability: 'ready',
    errors: MESSAGE_OPERATION_ERRORS,
    decodeInput: decodeTurnInterruptInput,
    decodeOutput: decodeTurnInterruptResult,
  }),
} as const;

export function decodeSessionMessageQueueProjection(value: unknown): SessionMessageQueueProjection {
  const record = requireExactRecord(value, 'Session message queue projection', [
    'hostEpoch',
    'queueRevision',
    'steering',
    'followup',
  ]);
  const steering = decodeSteeringMessages(record.steering);
  const followup = decodeFollowupMessages(record.followup);
  if (steering.length + followup.length > MESSAGE_QUEUE_MAX_ENTRIES) {
    throw invalidProtocolFrame('Invalid Session message queue projection');
  }
  assertUniqueQueueEntries([...steering, ...followup], 'Session message queue projection');
  const projection = {
    hostEpoch: requireId(record.hostEpoch, 'queue hostEpoch'),
    queueRevision: requireCount(record.queueRevision, 'queueRevision'),
    steering,
    followup,
  };
  requireEncodedByteLimit(
    projection,
    'Session message queue projection',
    MESSAGE_QUEUE_PROJECTION_MAX_BYTES,
  );
  return projection;
}

function decodeTurnMessageSubmitInput(value: unknown): TurnMessageSubmitInput {
  const record = requireShapedRecord(
    value,
    'turn.message.submit input',
    ['originHostEpoch', 'sessionId', 'messageId', 'content', 'placement'],
    ['skillIds', 'turnOrchestration'],
  );
  const skillIds = decodeSkillIds(record.skillIds);
  const placement = requireMessagePlacement(record.placement);
  const turnOrchestration =
    record.turnOrchestration !== undefined
      ? decodeTurnOrchestration(record.turnOrchestration)
      : undefined;
  // Exact-Turn intent has no queued form: a Skill or orchestration Message
  // opens its own Turn or fails closed, so `next_turn` cannot describe it.
  if ((skillIds.length > 0 || turnOrchestration !== undefined) && placement !== 'current_turn') {
    throw invalidProtocolFrame('Invalid turn.message.submit placement for an exact Turn');
  }
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    messageId: requireEntityId(record.messageId, 'messageId'),
    content: decodeMessageAdmissionContent(record.content, skillIds.length > 0),
    placement,
    ...(skillIds.length > 0 ? { skillIds } : {}),
    ...(turnOrchestration !== undefined ? { turnOrchestration } : {}),
  };
}

function decodeTurnMessageQueryInput(value: unknown): TurnMessageQueryInput {
  const record = requireExactRecord(value, 'turn.message.query input', ['sessionId', 'messageIds']);
  if (!Array.isArray(record.messageIds) || record.messageIds.length > MESSAGE_QUEUE_MAX_ENTRIES) {
    throw invalidProtocolFrame('Invalid turn.message.query messageIds');
  }
  const messageIds = record.messageIds.map((messageId) => requireEntityId(messageId, 'messageId'));
  if (new Set(messageIds).size !== messageIds.length) {
    throw invalidProtocolFrame('Duplicate turn.message.query messageId');
  }
  return {
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    messageIds,
  };
}

function decodeTurnMessageQueryResult(value: unknown): TurnMessageQueryResult {
  const record = requireExactRecord(value, 'turn.message.query result', ['cancelledMessageIds']);
  if (
    !Array.isArray(record.cancelledMessageIds) ||
    record.cancelledMessageIds.length > MESSAGE_QUEUE_MAX_ENTRIES
  ) {
    throw invalidProtocolFrame('Invalid turn.message.query cancelledMessageIds');
  }
  const cancelledMessageIds = record.cancelledMessageIds.map((messageId) =>
    requireEntityId(messageId, 'messageId'),
  );
  if (new Set(cancelledMessageIds).size !== cancelledMessageIds.length) {
    throw invalidProtocolFrame('Duplicate turn.message.query cancelledMessageId');
  }
  return { cancelledMessageIds };
}

function decodeTurnMessageExecutionQueryInput(value: unknown): TurnMessageExecutionQueryInput {
  return decodeTurnMessageQueryInput(value);
}

function decodeTurnMessageExecutionQueryResult(value: unknown): TurnMessageExecutionQueryResult {
  const record = requireExactRecord(value, 'turn.message.execution.query result', ['resolutions']);
  if (!Array.isArray(record.resolutions) || record.resolutions.length > MESSAGE_QUEUE_MAX_ENTRIES) {
    throw invalidProtocolFrame('Invalid turn.message.execution.query resolutions');
  }
  const resolutions = record.resolutions.map((value): TurnMessageExecutionResolution => {
    const resolution = requireRecord(value, 'turn.message.execution.query resolution');
    if (resolution.state === 'pending') {
      assertExactKeys(resolution, 'turn.message.execution.query pending resolution', [
        'messageId',
        'state',
      ]);
      return {
        messageId: requireEntityId(resolution.messageId, 'messageId'),
        state: 'pending',
      };
    }
    if (resolution.state === 'cancelled') {
      assertExactKeys(resolution, 'turn.message.execution.query cancelled resolution', [
        'messageId',
        'state',
      ]);
      return {
        messageId: requireEntityId(resolution.messageId, 'messageId'),
        state: 'cancelled',
      };
    }
    if (resolution.state === 'owned') {
      assertExactKeys(resolution, 'turn.message.execution.query owned resolution', [
        'messageId',
        'state',
        'turnId',
        'runId',
      ]);
      return {
        messageId: requireEntityId(resolution.messageId, 'messageId'),
        state: 'owned',
        turnId: requireEntityId(resolution.turnId, 'turnId'),
        runId: requireEntityId(resolution.runId, 'runId'),
      };
    }
    throw invalidProtocolFrame('Invalid turn.message.execution.query resolution state');
  });
  if (new Set(resolutions.map(({ messageId }) => messageId)).size !== resolutions.length) {
    throw invalidProtocolFrame('Duplicate turn.message.execution.query messageId');
  }
  return { resolutions };
}

function decodeTurnMessageSubmitResult(value: unknown): TurnMessageSubmitResult {
  const record = requireRecord(value, 'turn.message.submit result');
  if (record.disposition === 'turn_started') {
    assertExactKeys(record, 'turn.message.submit turn_started result', [
      'disposition',
      'turnId',
      'skillInvocation',
    ]);
    return {
      disposition: 'turn_started',
      turnId: requireEntityId(record.turnId, 'turnId'),
      skillInvocation: decodeSubmitSkillInvocation(record.skillInvocation),
    };
  }
  if (record.disposition === 'blocked') {
    assertExactKeys(record, 'turn.message.submit blocked result', [
      'disposition',
      'skillInvocation',
    ]);
    const skillInvocation = decodeSubmitSkillInvocation(record.skillInvocation);
    if (skillInvocation.loaded.length !== 0 || skillInvocation.failed.length === 0) {
      throw invalidProtocolFrame('Invalid blocked turn.message.submit Skill invocation');
    }
    return { disposition: 'blocked', skillInvocation };
  }
  if (record.disposition === 'steering' || record.disposition === 'followup') {
    const shaped = requireShapedRecord(
      record,
      'turn.message.submit queued result',
      ['disposition', 'skillInvocation'],
      ['queueRevision'],
    );
    return {
      disposition: record.disposition,
      ...(shaped.queueRevision !== undefined
        ? { queueRevision: requireCount(shaped.queueRevision, 'queueRevision') }
        : {}),
      skillInvocation: decodeSubmitSkillInvocation(shaped.skillInvocation),
    };
  }
  throw invalidProtocolFrame('Invalid turn.message.submit disposition');
}

function decodeSubmitSkillInvocation(value: unknown): SkillInvocationResult {
  try {
    return decodeSkillInvocationResult(value);
  } catch {
    throw invalidProtocolFrame('Invalid turn.message.submit Skill invocation');
  }
}

function decodeQueueRetractInput(value: unknown): QueueRetractInput {
  const record = requireExactRecord(value, 'queue.retract input', [
    'originHostEpoch',
    'sessionId',
    'retractId',
  ]);
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    retractId: requireEntityId(record.retractId, 'retractId'),
  };
}

function decodeQueueRetractResult(value: unknown): QueueRetractResult {
  const record = requireExactRecord(value, 'queue.retract result', ['queueRevision', 'retracted']);
  const result = {
    queueRevision: requireCount(record.queueRevision, 'queueRevision'),
    retracted: decodeRetractedMessages(record.retracted),
  };
  requireEncodedByteLimit(result, 'queue.retract result', MESSAGE_OPERATION_RESULT_MAX_BYTES);
  return result;
}

function decodeQueueEntryRetractInput(value: unknown): QueueEntryRetractInput {
  const record = requireExactRecord(value, 'queue.entry.retract input', [
    'originHostEpoch',
    'sessionId',
    'entryId',
    'retractId',
  ]);
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    entryId: requireEntityId(record.entryId, 'entryId'),
    retractId: requireEntityId(record.retractId, 'retractId'),
  };
}

function decodeQueueMutationResult(value: unknown): QueueMutationResult {
  const record = requireExactRecord(value, 'queue mutation result', ['queueRevision']);
  return { queueRevision: requireCount(record.queueRevision, 'queueRevision') };
}

function decodeQueueEntryPromoteInput(value: unknown): QueueEntryPromoteInput {
  const record = requireExactRecord(value, 'queue.entry.promote input', [
    'originHostEpoch',
    'sessionId',
    'entryId',
    'promoteId',
  ]);
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    entryId: requireEntityId(record.entryId, 'entryId'),
    promoteId: requireEntityId(record.promoteId, 'promoteId'),
  };
}

function decodeQueueEntryUpdateInput(value: unknown): QueueEntryUpdateInput {
  const record = requireExactRecord(value, 'queue.entry.update input', [
    'originHostEpoch',
    'sessionId',
    'entryId',
    'updateId',
    'expectedQueueRevision',
    'text',
  ]);
  const text = requireUtf8String(record.text, 'Message text', TURN_MESSAGE_TEXT_MAX_BYTES);
  if (text.trim().length === 0) throw invalidProtocolFrame('Invalid Message text');
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    entryId: requireEntityId(record.entryId, 'entryId'),
    updateId: requireEntityId(record.updateId, 'updateId'),
    expectedQueueRevision: requireCount(record.expectedQueueRevision, 'expectedQueueRevision'),
    text,
  };
}

function decodeQueueEntriesReorderInput(value: unknown): QueueEntriesReorderInput {
  const record = requireExactRecord(value, 'queue.entries.reorder input', [
    'originHostEpoch',
    'sessionId',
    'reorderId',
    'entryIds',
  ]);
  const entryIds = requireBoundedArray(record.entryIds, 'reorder entry identities').map((entryId) =>
    requireEntityId(entryId, 'entryId'),
  );
  if (new Set(entryIds).size !== entryIds.length) {
    throw invalidProtocolFrame('Invalid reorder entry identities');
  }
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    reorderId: requireEntityId(record.reorderId, 'reorderId'),
    entryIds,
  };
}

function decodeTurnInterruptInput(value: unknown): TurnInterruptInput {
  const record = requireExactRecord(value, 'turn.interrupt input', [
    'originHostEpoch',
    'sessionId',
    'interruptId',
    'turnId',
    'runId',
  ]);
  return {
    originHostEpoch: requireId(record.originHostEpoch, 'originHostEpoch'),
    sessionId: requireEntityId(record.sessionId, 'sessionId'),
    interruptId: requireEntityId(record.interruptId, 'interruptId'),
    turnId: requireEntityId(record.turnId, 'turnId'),
    runId: requireEntityId(record.runId, 'runId'),
  };
}

function decodeTurnInterruptResult(value: unknown): TurnInterruptResult {
  const record = requireExactRecord(value, 'turn.interrupt result', [
    'queueRevision',
    'retracted',
    'turn',
  ]);
  const result = {
    queueRevision: requireCount(record.queueRevision, 'queueRevision'),
    retracted: decodeRetractedMessages(record.retracted),
    turn: decodeTurnSnapshot(record.turn),
  };
  requireEncodedByteLimit(result, 'turn.interrupt result', MESSAGE_OPERATION_RESULT_MAX_BYTES);
  return result;
}

function decodeSteeringMessages(value: unknown): SteeringMessageSnapshot[] {
  return requireBoundedArray(value, 'steering queue').map((entry) => {
    const decoded = decodeMessageQueueEntrySnapshot(entry);
    if (decoded.placement !== 'current_turn') {
      throw invalidProtocolFrame('Invalid steering queue entry');
    }
    if (decoded.state === 'queued') return { ...decoded, placement: 'current_turn' };
    if (decoded.state === 'in_flight') return decoded;
    throw invalidProtocolFrame('Invalid steering queue entry');
  });
}

function decodeFollowupMessages(value: unknown): QueuedMessageSnapshot[] {
  return requireBoundedArray(value, 'followup queue').map((entry) => {
    const decoded = decodeMessageQueueEntrySnapshot(entry);
    if (decoded.state !== 'queued' || decoded.placement !== 'next_turn') {
      throw invalidProtocolFrame('Invalid followup queue entry');
    }
    return { ...decoded, placement: 'next_turn' };
  });
}

function decodeRetractedMessages(value: unknown): RetractedMessageSnapshot[] {
  const entries = requireBoundedArray(value, 'retracted messages').map((entry) => {
    const decoded = decodeMessageQueueEntrySnapshot(entry);
    if (decoded.state !== 'retracted') {
      throw invalidProtocolFrame('Invalid retracted message state');
    }
    return decoded;
  });
  assertUniqueQueueEntries(entries, 'retracted messages');
  return entries;
}

function decodeMessageQueueEntrySnapshot(value: unknown): MessageQueueEntrySnapshot {
  const record = requireExactRecord(value, 'message queue entry snapshot', [
    'entryId',
    'messageId',
    'content',
    'placement',
    'state',
  ]);
  const base = {
    entryId: requireEntityId(record.entryId, 'entryId'),
    messageId: requireEntityId(record.messageId, 'messageId'),
    content: decodeMessageAdmissionContent(record.content),
    placement: requireMessagePlacement(record.placement),
  };
  if (record.state === 'queued' || record.state === 'retracted') {
    return { ...base, state: record.state };
  }
  if (record.state === 'in_flight' && base.placement === 'current_turn') {
    return { ...base, placement: 'current_turn', state: record.state };
  }
  throw invalidProtocolFrame('Invalid message queue entry state');
}

function requireMessagePlacement(value: unknown): MessagePlacement {
  if (value === 'current_turn' || value === 'next_turn') return value;
  throw invalidProtocolFrame('Invalid message placement');
}

function requireBoundedArray(value: unknown, label: string): unknown[] {
  if (!Array.isArray(value) || value.length > MESSAGE_QUEUE_MAX_ENTRIES) {
    throw invalidProtocolFrame(`Invalid ${label}`);
  }
  return value;
}

function assertUniqueQueueEntries(
  entries: readonly MessageQueueEntrySnapshot[],
  label: string,
): void {
  const entryIds = new Set<string>();
  const messageIds = new Set<string>();
  for (const entry of entries) {
    if (entryIds.has(entry.entryId) || messageIds.has(entry.messageId)) {
      throw invalidProtocolFrame(`${label} repeats a message identity`);
    }
    entryIds.add(entry.entryId);
    messageIds.add(entry.messageId);
  }
}

function requireEncodedByteLimit(value: unknown, label: string, maxBytes: number): void {
  let encoded: string | undefined;
  try {
    encoded = JSON.stringify(value);
  } catch {
    throw invalidProtocolFrame(`Invalid ${label}`);
  }
  if (encoded === undefined || Buffer.byteLength(encoded, 'utf8') > maxBytes) {
    throw invalidProtocolFrame(`Invalid ${label}`);
  }
}