runa/vscode/src/protocol.ts

import { parse, stringify } from "lossless-json";

export const PROTOCOL_VERSION = 5;
export const MAX_MESSAGE_BYTES = 1024 * 1024;
export const WEBSOCKET_SUBPROTOCOL = "runa-debug";
const ACTOR_REF = /^[A-Za-z0-9_-]{21}$/;

export type Value =
  | { type: "null" }
  | { type: "bool"; data: boolean }
  | { type: "integer"; data: bigint }
  | { type: "string" | "actor_ref"; data: string }
  | { type: "list"; data: Value[] }
  | { type: "record"; data: Record<string, Value> };

export interface ActorSnapshot {
  behavior_path?: string | null;
  pending_behavior_path?: string | null;
  temporary_behavior?: boolean;
  log_level: LogLevel;
  actor_ref: string;
  mailbox_id: bigint;
  implementation_id: string;
  incarnation: bigint;
  state: Value;
  status: {
    status: "idle" | "running" | "awaiting" | "paused" | "stopped" | "failed";
    message?: string;
  };
  execution: {
    vm_steps: bigint;
    instruction_pointer: bigint;
    call_depth: bigint;
  } | null;
  queued: Value[];
  reserved: Value | null;
}

export type SystemKind = "stop" | "kill" | "restart" | "pause" | "resume" | "reload";
export type SystemMessage = { kind: SystemKind } | { kind: "become"; path: string };
export const LOG_LEVELS = ["trace", "debug", "info", "warn", "error", "fatal"] as const;
export type LogLevel = (typeof LOG_LEVELS)[number];
export interface LogEntry {
  sequence: bigint;
  actor_ref: string;
  level: LogLevel;
  message: string;
}

export interface SourceFile {
  path: string;
  source_id: string;
  source: string;
}

export interface SourceCommit {
  revision: string;
  source_id: string;
}

export type Command =
  | { operation: "actor_behavior"; parameters: { actor_ref: string } }
  | {
      operation: "write_actor_behavior";
      parameters: { actor_ref: string; expected_source_id: string; source: string };
    }
  | {
      operation: "update_behavior_source";
      parameters: {
        actor_ref: string;
        path: string;
        expected_source_id: string;
        temporary_source_id: string;
      };
    }
  | {
      operation: "rename_source";
      parameters: {
        path: string;
        target: string;
        expected_source_id: string;
        update_actors: boolean;
      };
    }
  | { operation: "delete_source"; parameters: { path: string; expected_source_id: string } }
  | { operation: "logs"; parameters: { actors: string[]; follow: false; after: bigint } }
  | { operation: "set_log_level"; parameters: { actor_ref: string; level: LogLevel } }
  | {
      operation: "edit_actor";
      parameters: {
        actor_ref: string;
        state: Value | null;
        messages: { reserved: Value | null; queued: Value[] } | null;
      };
    }
  | { operation: "list_actors" | "list_sources" }
  | { operation: "read_source" | "commit_source" | "source_history"; parameters: { path: string } }
  | {
      operation: "write_source";
      parameters: { path: string; expected_source_id: string | null; source: string };
    }
  | {
      operation: "revision_source" | "rollback_source";
      parameters: { path: string; revision: string };
    }
  | { operation: "system"; parameters: { actor_ref: string; message: SystemMessage } }
  | { operation: "inspect_actor"; parameters: { actor_ref: string } }
  | {
      operation: "spawn";
      parameters: { path: string; state: Value; mailbox_capacity: number };
    }
  | { operation: "send"; parameters: { actor_ref: string; message: Value } }
  | { operation: "tick"; parameters: { ticks: number; instruction_budget: number } };

export class ProtocolError extends Error {
  constructor(
    public readonly code: string,
    message: string,
  ) {
    super(`${code}: ${message}`);
  }
}

export function isActorRef(value: unknown): value is string {
  return typeof value === "string" && ACTOR_REF.test(value);
}

/** Canonical image source path, e.g. `actors/counter` (no `source/` root, no `.rn`). */
export function isSourcePath(value: unknown): value is string {
  return (
    typeof value === "string" &&
    value.length <= 128 &&
    !value.startsWith("source/") &&
    value.split("/").every((part) => /^[A-Za-z0-9_-]+$/.test(part))
  );
}

export function isSourceRevision(value: unknown): value is string {
  return typeof value === "string" && /^[A-Za-z0-9]{1,64}$/.test(value);
}

function isObject(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}

function object(value: unknown): Record<string, unknown> {
  if (!isObject(value)) {
    throw new Error("Expected a protocol object");
  }
  return value;
}

function text(value: unknown): string {
  if (typeof value !== "string") throw new Error("Expected a protocol string");
  return value;
}

function uint(value: unknown): bigint {
  if (typeof value !== "bigint" || value < 0n || value > 18446744073709551615n) {
    throw new Error("Expected an unsigned 64-bit protocol integer");
  }
  return value;
}

function array(value: unknown): unknown[] {
  if (!Array.isArray(value)) throw new Error("Expected a protocol array");
  return value;
}

export function decodeValue(input: unknown): Value {
  let nodes = 0;
  let bytes = 0;
  const visit = (input: unknown, depth: number): Value => {
    if (++nodes > 4096 || depth > 32) throw new Error("Value exceeds node/depth limits");
    const value = object(input);
    if (Object.keys(value).some((key) => key !== "type" && key !== "data")) {
      throw new Error("Unknown portable value field");
    }
    const type = value.type;
    switch (type) {
      case "null":
        if ("data" in value) throw new Error("Null must not carry data");
        return { type };
      case "bool":
        if (typeof value.data !== "boolean") throw new Error("Invalid boolean value");
        return { type, data: value.data };
      case "integer":
        if (
          typeof value.data !== "bigint" ||
          value.data < -9223372036854775808n ||
          value.data > 9223372036854775807n
        ) {
          throw new Error("Invalid signed 64-bit value");
        }
        return { type, data: value.data };
      case "string": {
        const data = text(value.data);
        bytes += Buffer.byteLength(data);
        if (bytes > MAX_MESSAGE_BYTES) throw new Error("Value exceeds string/key byte limit");
        return { type, data };
      }
      case "actor_ref":
        if (!isActorRef(value.data)) throw new Error("Invalid actor reference value");
        return { type, data: value.data };
      case "list":
        return { type, data: array(value.data).map((child) => visit(child, depth + 1)) };
      case "record": {
        const entries = Object.entries(object(value.data)).map(([key, child]) => {
          bytes += Buffer.byteLength(key);
          if (bytes > MAX_MESSAGE_BYTES) throw new Error("Value exceeds string/key byte limit");
          return [key, visit(child, depth + 1)];
        });
        return { type, data: Object.fromEntries(entries) };
      }
      default:
        throw new Error(`Unsupported portable value type: ${String(type)}`);
    }
  };
  return visit(input, 0);
}

function decodeActor(input: unknown): ActorSnapshot {
  const actor = object(input);
  if (!isActorRef(actor.actor_ref)) throw new Error("Invalid actor reference in snapshot");
  const status = object(actor.status);
  const name = status.status;
  if (
    name !== "idle" &&
    name !== "running" &&
    name !== "awaiting" &&
    name !== "paused" &&
    name !== "stopped" &&
    name !== "failed"
  ) {
    throw new Error("Unknown actor status");
  }
  const execution = actor.execution === null ? null : object(actor.execution);
  const level = LOG_LEVELS.find((level) => level === actor.log_level);
  if (level === undefined) throw new Error("Invalid actor log level");
  return {
    behavior_path:
      actor.behavior_path === undefined || actor.behavior_path === null
        ? null
        : text(actor.behavior_path),
    pending_behavior_path:
      actor.pending_behavior_path === undefined || actor.pending_behavior_path === null
        ? null
        : text(actor.pending_behavior_path),
    temporary_behavior:
      actor.temporary_behavior === undefined
        ? false
        : (() => {
            if (typeof actor.temporary_behavior !== "boolean")
              throw new Error("Invalid temporary behavior flag");
            return actor.temporary_behavior;
          })(),
    log_level: level,
    actor_ref: actor.actor_ref,
    mailbox_id: uint(actor.mailbox_id),
    implementation_id: text(actor.implementation_id),
    incarnation: uint(actor.incarnation),
    state: decodeValue(actor.state),
    status: name === "failed" ? { status: name, message: text(status.message) } : { status: name },
    execution: execution && {
      vm_steps: uint(execution.vm_steps),
      instruction_pointer: uint(execution.instruction_pointer),
      call_depth: uint(execution.call_depth),
    },
    queued: array(actor.queued).map(decodeValue),
    reserved: actor.reserved === null ? null : decodeValue(actor.reserved),
  };
}

function decodeSourceFile(input: unknown): SourceFile {
  const file = object(input);
  if (!isSourcePath(file.path)) throw new Error("Invalid source path");
  const source_id = text(file.source_id);
  if (!source_id) throw new Error("Invalid source identity");
  return { path: file.path, source_id, source: text(file.source) };
}

function decodeSourceCommit(input: unknown): SourceCommit {
  const commit = object(input);
  if (!isSourceRevision(commit.revision)) throw new Error("Invalid source revision");
  const source_id = text(commit.source_id);
  if (!source_id) throw new Error("Invalid source identity");
  return { revision: commit.revision, source_id };
}

export type Reply =
  | { kind: "actor_behavior"; data: { source: string; source_id: string } }
  | { kind: "logs"; data: { entries: LogEntry[]; next_sequence: bigint; has_more: boolean } }
  | { kind: "actors"; data: ActorSnapshot[] }
  | { kind: "actor"; data: ActorSnapshot }
  | { kind: "sources"; data: SourceFile[] }
  | { kind: "source"; data: SourceFile }
  | { kind: "source_committed"; data: SourceCommit }
  | { kind: "source_history"; data: SourceCommit[] }
  | { kind: "actor_spawned"; data: { actor_ref: string } }
  | { kind: "acknowledged" }
  | { kind: "behavior_source"; data: { source: string } }
  | { kind: "tick_completed"; data: { ticks: bigint } };

function decodeEnvelope(message: string): Record<string, unknown> {
  const envelope = object(
    parse(message, undefined, (literal) => {
      if (!/^-?(0|[1-9]\d*)$/.test(literal)) throw new Error("Non-integer protocol number");
      return BigInt(literal);
    }),
  );
  if (envelope.version !== BigInt(PROTOCOL_VERSION)) throw new Error("Protocol version mismatch");
  return envelope;
}

export function decodeNotification(message: string): bigint | undefined {
  const envelope = decodeEnvelope(message);
  if (!Object.hasOwn(envelope, "event")) return undefined;
  if (Object.keys(envelope).length !== 2) throw new Error("Invalid notification envelope");
  const event = object(envelope.event);
  if (event.event !== "image_changed" || Object.keys(event).length !== 2) {
    throw new Error("Unsupported server event");
  }
  const data = object(event.data);
  const revision = uint(data.revision);
  if (Object.keys(data).length !== 1 || revision === 0n) {
    throw new Error("Invalid image revision");
  }
  return revision;
}

export function decodeResponse(line: string, requestId: string, command: Command): Reply {
  const response = decodeEnvelope(line);
  if (response.request_id !== requestId) throw new Error("Response request ID mismatch");
  const result = object(response.result);
  if (Object.keys(result).length !== 1) throw new Error("Invalid response result");
  if (Object.hasOwn(result, "Err")) {
    const error = object(result.Err);
    const code = text(error.code);
    if (
      ![
        "invalid_request",
        "unsupported_version",
        "not_found",
        "conflict",
        "capacity_exceeded",
        "runtime_failure",
      ].includes(code)
    ) {
      throw new Error("Unknown protocol error code");
    }
    throw new ProtocolError(code, text(error.message));
  }
  const reply = object(result.Ok);
  if (
    (command.operation === "actor_behavior" || command.operation === "write_actor_behavior") &&
    reply.kind === "actor_behavior"
  ) {
    const data = object(reply.data);
    const source = text(data.source),
      source_id = text(data.source_id);
    if (!source_id) throw new Error("Invalid source identity");
    if (command.operation === "write_actor_behavior" && source !== command.parameters.source)
      throw new Error("Written actor behavior mismatch");
    return { kind: "actor_behavior", data: { source, source_id } };
  }
  if (command.operation === "logs" && reply.kind === "logs") {
    const data = object(reply.data);
    if (typeof data.has_more !== "boolean") throw new Error("Invalid log pagination flag");
    const entries = array(data.entries).map((input): LogEntry => {
      const entry = object(input);
      if (!isActorRef(entry.actor_ref)) throw new Error("Invalid log actor");
      const level = LOG_LEVELS.find((level) => level === entry.level);
      if (level === undefined) throw new Error("Invalid log level");
      return {
        sequence: uint(entry.sequence),
        actor_ref: entry.actor_ref,
        level,
        message: text(entry.message),
      };
    });
    let previous = command.parameters.after;
    for (const entry of entries) {
      if (entry.sequence <= previous || !command.parameters.actors.includes(entry.actor_ref))
        throw new Error("Invalid log ordering or actor filter");
      previous = entry.sequence;
    }
    const next_sequence = uint(data.next_sequence);
    if (
      next_sequence !== previous ||
      entries.length > 128 ||
      (data.has_more && entries.length === 0)
    )
      throw new Error("Invalid log pagination cursor");
    return { kind: "logs", data: { entries, next_sequence, has_more: data.has_more } };
  }
  if (
    (command.operation === "rename_source" || command.operation === "update_behavior_source") &&
    reply.kind === "source"
  ) {
    const data = decodeSourceFile(reply.data);
    if (command.operation === "rename_source" && data.path !== command.parameters.target)
      throw new Error("Renamed source mismatch");
    if (command.operation === "update_behavior_source" && data.path !== command.parameters.path)
      throw new Error("Updated source path mismatch");
    return { kind: "source", data };
  }
  if (command.operation === "revision_source" && reply.kind === "behavior_source") {
    return { kind: "behavior_source", data: { source: text(object(reply.data).source) } };
  }
  if (command.operation === "list_sources" && reply.kind === "sources") {
    const data = array(reply.data).map(decodeSourceFile);
    if (new Set(data.map((file) => file.path)).size !== data.length) {
      throw new Error("Duplicate source paths");
    }
    return { kind: "sources", data };
  }
  if (
    (command.operation === "read_source" ||
      command.operation === "write_source" ||
      command.operation === "rollback_source") &&
    reply.kind === "source"
  ) {
    const data = decodeSourceFile(reply.data);
    if (data.path !== command.parameters.path) throw new Error("Source path mismatch");
    if (command.operation === "write_source" && data.source !== command.parameters.source) {
      throw new Error("Written source mismatch");
    }
    return { kind: "source", data };
  }
  if (command.operation === "commit_source" && reply.kind === "source_committed") {
    return { kind: "source_committed", data: decodeSourceCommit(reply.data) };
  }
  if (command.operation === "source_history" && reply.kind === "source_history") {
    const data = array(reply.data).map(decodeSourceCommit);
    if (new Set(data.map((commit) => commit.revision)).size !== data.length) {
      throw new Error("Duplicate source revisions");
    }
    return { kind: "source_history", data };
  }
  if (command.operation === "spawn" && reply.kind === "actor_spawned") {
    const data = object(reply.data);
    if (!isActorRef(data.actor_ref)) throw new Error("Invalid spawned actor reference");
    return { kind: "actor_spawned", data: { actor_ref: data.actor_ref } };
  }
  if (command.operation === "tick" && reply.kind === "tick_completed") {
    const ticks = uint(object(reply.data).ticks);
    if (ticks !== BigInt(command.parameters.ticks)) throw new Error("Tick count mismatch");
    return { kind: "tick_completed", data: { ticks } };
  }
  if (
    (command.operation === "send" ||
      command.operation === "system" ||
      command.operation === "set_log_level" ||
      command.operation === "delete_source" ||
      command.operation === "edit_actor") &&
    reply.kind === "acknowledged"
  )
    return { kind: "acknowledged" };
  if (command.operation === "list_actors" && reply.kind === "actors") {
    const data = array(reply.data).map(decodeActor);
    if (new Set(data.map((actor) => actor.actor_ref)).size !== data.length) {
      throw new Error("Duplicate actor references in snapshot");
    }
    return { kind: "actors", data };
  }
  if (command.operation === "inspect_actor" && reply.kind === "actor") {
    const data = decodeActor(reply.data);
    if (data.actor_ref !== command.parameters.actor_ref) throw new Error("Snapshot actor mismatch");
    return { kind: "actor", data };
  }
  throw new Error("Unexpected reply kind");
}

export function formatSnapshot(
  actor: ActorSnapshot,
  endpoint: string,
  capturedAt: string,
  connected: boolean,
  removed = false,
): string {
  const result = stringify(
    {
      connection: { endpoint, connected },
      captured_at: capturedAt,
      ...(removed ? { removed: true } : {}),
      actor,
    },
    null,
    2,
  );
  if (result === undefined) throw new Error("Could not format actor snapshot");
  return result + "\n";
}