Class InMemoryRunEventLog

packages/ai-sandbox/src/run-log.ts:112–224

 * backend mirrors.
 */
export class InMemoryRunEventLog implements RunEventLog {
  private readonly runs = new Map<string, RunState>()

  private now(): number {
    return Date.now()
  }

  private require(runId: string): RunState {
    const state = this.runs.get(runId)
    if (!state) throw new Error(`run-log: unknown runId "${runId}"`)
    return state
  }

  private wake(state: RunState): void {
    const waiters = [...state.waiters]
    state.waiters.clear()
    for (const resolve of waiters) resolve()
  }

  // Mutators return a Promise without `async` so contract violations REJECT
  // (rather than throwing synchronously from a Promise-typed method — a
  // `.catch()` footgun) without an `await`-less async body.
  open(input: { runId: string; threadId?: string }): Promise<RunRecord> {
    const existing = this.runs.get(input.runId)
    if (existing) return Promise.resolve({ ...existing.record })
    const now = this.now()
    const record: RunRecord = {
      runId: input.runId,
      ...(input.threadId !== undefined ? { threadId: input.threadId } : {}),
      status: 'running',
      lastSeq: -1,
      createdAt: now,
      updatedAt: now,
    }
    this.runs.set(input.runId, { record, chunks: [], waiters: new Set() })
    return Promise.resolve({ ...record })
  }

  append(runId: string, chunk: StreamChunk): Promise<number> {
    const state = this.runs.get(runId)
    if (!state) {
      return Promise.reject(new Error(`run-log: unknown runId "${runId}"`))
    }
    if (isTerminalRunStatus(state.record.status)) {
      return Promise.reject(
        new Error(
          `run-log: cannot append to terminal run "${runId}" (status=${state.record.status})`,
        ),
      )
    }
    // Derive seq from the record's cursor (not `chunks.length`) so the gap-free
    // invariant holds the same way the durable backend computes it, even if the
    // backlog is ever trimmed/compacted.
    const seq = state.record.lastSeq + 1
    state.chunks.push(chunk)
    state.record.lastSeq = seq
    state.record.updatedAt = this.now()
    this.wake(state)