Agent Run State

// Normalized agent run state and the event contract that updates it.
// Framework- and provider-independent: map your backend's stream to these events,
// or replace a run wholesale with a snapshot from your API.

export type RunStatus = "queued" | "running" | "waiting-for-approval" | "succeeded" | "failed" | "cancelled";
export type StepStatus = "queued" | "running" | "succeeded" | "failed" | "skipped";
export type ApprovalStatus = "pending" | "approved" | "rejected" | "expired";
export type MessageStatus = "streaming" | "complete" | "failed";
export type ArtifactStatus = "generating" | "ready" | "failed";

export interface RunStep {
  id: string;
  /** Tool name for tool calls, or a short label for other steps. */
  name: string;
  kind: "tool" | "step";
  status: StepStatus;
  summary?: string;
  input?: unknown;
  output?: string;
  error?: string;
  startedAt?: string;
  finishedAt?: string;
}

export interface RunApproval {
  id: string;
  title: string;
  description?: string;
  /** What will happen if approved, e.g. a command or diff. Rendered as text, never as HTML. */
  detail?: string;
  status: ApprovalStatus;
  expiresAt?: string;
  decidedAt?: string;
}

export interface RunMessage {
  id: string;
  role: "user" | "assistant";
  content: string;
  status: MessageStatus;
  error?: string;
}

export interface RunArtifact {
  id: string;
  name: string;
  /** MIME type, e.g. text/markdown, text/csv, image/png. */
  type: string;
  status: ArtifactStatus;
  /** Where the artifact can be opened or downloaded. */
  path?: string;
  /** Inline text content for previews of text artifacts. */
  content?: string;
  sizeBytes?: number;
  error?: string;
}

export interface AgentRun {
  id: string;
  title: string;
  status: RunStatus;
  steps: RunStep[];
  approvals: RunApproval[];
  messages: RunMessage[];
  artifacts: RunArtifact[];
  createdAt?: string;
  finishedAt?: string;
  error?: string;
}

interface EventBase<Type extends string, Payload> {
  /** Unique event id; repeated deliveries are ignored. */
  id: string;
  runId: string;
  /** Increases by one per event within a run. */
  sequence: number;
  /** ISO 8601 time the backend produced the event. */
  timestamp: string;
  type: Type;
  payload: Payload;
}

export type AgentRunEvent =
  | EventBase<"run.status", { status: RunStatus; error?: string }>
  | EventBase<"step.started", { step: Omit<RunStep, "status" | "startedAt"> }>
  | EventBase<"step.updated", { stepId: string; status: StepStatus; output?: string; error?: string }>
  | EventBase<"approval.requested", { approval: Omit<RunApproval, "status"> }>
  | EventBase<"approval.resolved", { approvalId: string; status: Exclude<ApprovalStatus, "pending"> }>
  | EventBase<"message.started", { message: Omit<RunMessage, "status" | "content"> & { content?: string } }>
  | EventBase<"message.delta", { messageId: string; delta: string }>
  | EventBase<"message.completed", { messageId: string; status: Exclude<MessageStatus, "streaming">; error?: string }>
  | EventBase<"artifact.updated", { artifact: RunArtifact }>;

/** A run plus the bookkeeping needed to apply events safely. */
export interface RunStore {
  run: AgentRun;
  /** Highest sequence applied without a gap. */
  sequence: number;
  /** Event ids already applied or buffered, for de-duplication. */
  seen: string[];
  /** Events that arrived ahead of a missing sequence, waiting for the gap to fill. */
  pending: AgentRunEvent[];
  /** True when events are missing; refresh from a snapshot (see `fromSnapshot`). */
  stale: boolean;
}

/** How many out-of-order events to hold before declaring the store stale. */
const MAX_PENDING = 50;

/** Starts a store from a run snapshot and the sequence it reflects. */
export function fromSnapshot(run: AgentRun, sequence = 0): RunStore {
  return { run, sequence, seen: [], pending: [], stale: false };
}

/**
 * Applies events: ignores duplicates (by id) and other runs, buffers events that
 * arrive early, applies them in sequence order, and marks the store stale if a
 * gap does not fill. Pure: returns a new store and never mutates its input.
 */
export function applyEvents(store: RunStore, events: AgentRunEvent[]): RunStore {
  const seen = new Set(store.seen);
  let pending = [...store.pending];
  for (const event of events) {
    if (event.runId !== store.run.id || seen.has(event.id) || event.sequence <= store.sequence) continue;
    seen.add(event.id);
    pending.push(event);
  }
  pending.sort((a, b) => a.sequence - b.sequence);

  let { run, sequence } = store;
  while (pending.length && pending[0]!.sequence === sequence + 1) {
    run = reduce(run, pending.shift()!);
    sequence += 1;
  }
  // Drop stale duplicates of already applied sequences that slipped in under another id.
  pending = pending.filter((event) => event.sequence > sequence);

  return { run, sequence, seen: [...seen].slice(-500), pending, stale: store.stale || pending.length > MAX_PENDING };
}

/** True when events after `sequence` are buffered behind a missing one. */
export function hasGap(store: RunStore): boolean {
  return store.pending.length > 0;
}

function reduce(run: AgentRun, event: AgentRunEvent): AgentRun {
  const at = event.timestamp;
  switch (event.type) {
    case "run.status":
      return {
        ...run,
        status: event.payload.status,
        error: event.payload.error ?? run.error,
        finishedAt: ["succeeded", "failed", "cancelled"].includes(event.payload.status) ? at : run.finishedAt,
      };
    case "step.started":
      return upsert(run, "steps", { ...event.payload.step, status: "running", startedAt: at });
    case "step.updated":
      return patch(run, "steps", event.payload.stepId, (step) => ({
        ...step,
        status: event.payload.status,
        output: event.payload.output ?? step.output,
        error: event.payload.error ?? step.error,
        finishedAt: event.payload.status === "running" ? step.finishedAt : at,
      }));
    case "approval.requested":
      return upsert(run, "approvals", { ...event.payload.approval, status: "pending" });
    case "approval.resolved":
      return patch(run, "approvals", event.payload.approvalId, (approval) => ({ ...approval, status: event.payload.status, decidedAt: at }));
    case "message.started":
      return upsert(run, "messages", { content: "", ...event.payload.message, status: "streaming" });
    case "message.delta":
      return patch(run, "messages", event.payload.messageId, (message) => ({ ...message, content: message.content + event.payload.delta }));
    case "message.completed":
      return patch(run, "messages", event.payload.messageId, (message) => ({ ...message, status: event.payload.status, error: event.payload.error }));
    case "artifact.updated":
      return upsert(run, "artifacts", event.payload.artifact);
  }
}

type Collection = "steps" | "approvals" | "messages" | "artifacts";
type Entry<K extends Collection> = AgentRun[K][number];

function upsert<K extends Collection>(run: AgentRun, key: K, entry: Entry<K>): AgentRun {
  const list = run[key] as Entry<K>[];
  const index = list.findIndex((item) => item.id === entry.id);
  const next = index === -1 ? [...list, entry] : list.map((item, i) => (i === index ? { ...item, ...entry } : item));
  return { ...run, [key]: next };
}

function patch<K extends Collection>(run: AgentRun, key: K, id: string, update: (entry: Entry<K>) => Entry<K>): AgentRun {
  const list = run[key] as Entry<K>[];
  return { ...run, [key]: list.map((item) => (item.id === id ? update(item) : item)) };
}