claw-code-parity/ts-worker/turnManager.ts

/**
 * TurnManager — owns the lifecycle of a single user turn.
 *
 * Each turn gets a unique ID. The gateway:
 *   1. POSTs /v1/turns  →  gets back { turn_id }
 *   2. GETs  /v1/turns/:turn_id/events  →  SSE stream of WorkerTurnEvent
 *   3. POSTs /v1/turns/:turn_id/approval  →  resolves a pending tool approval
 *   4. POSTs /v1/turns/:turn_id/cancel   →  aborts the turn
 *
 * Internally we hold:
 *   - a queue of WorkerTurnEvent waiting to be flushed to SSE clients
 *   - an optional pending approval (blocks the engine until resolved)
 *   - an AbortController so cancel works
 */

import { randomUUID } from 'crypto'
import type {
  ApprovalDecision,
  WorkerTurnEvent,
  WorkerTurnRequest,
} from './protocol.js'

export type PendingApproval = {
  approval_id: string
  resolve: (decision: ApprovalDecision) => void
}

export type Turn = {
  id: string
  request: WorkerTurnRequest
  events: WorkerTurnEvent[]
  /** SSE clients listening for events on this turn */
  listeners: Set<(event: WorkerTurnEvent) => void>
  pendingApproval: PendingApproval | null
  abortController: AbortController
  done: boolean
}

export class TurnManager {
  private turns = new Map<string, Turn>()

  create(request: WorkerTurnRequest): Turn {
    const id = randomUUID()
    const turn: Turn = {
      id,
      request,
      events: [],
      listeners: new Set(),
      pendingApproval: null,
      abortController: new AbortController(),
      done: false,
    }
    this.turns.set(id, turn)
    return turn
  }

  get(turnId: string): Turn | undefined {
    return this.turns.get(turnId)
  }

  /** Push an event to the turn's log and broadcast to all SSE listeners. */
  emit(turnId: string, event: WorkerTurnEvent): void {
    const turn = this.turns.get(turnId)
    if (!turn) return
    turn.events.push(event)
    for (const listener of turn.listeners) {
      listener(event)
    }
    if (event.type === 'completed' || event.type === 'failed') {
      turn.done = true
    }
  }

  /** Register an SSE listener. Returns unsubscribe function. */
  subscribe(
    turnId: string,
    listener: (event: WorkerTurnEvent) => void,
  ): (() => void) | null {
    const turn = this.turns.get(turnId)
    if (!turn) return null

    // Replay buffered events first
    for (const event of turn.events) {
      listener(event)
    }

    if (turn.done) return null

    turn.listeners.add(listener)
    return () => {
      turn.listeners.delete(listener)
    }
  }

  /** Set a pending approval on the turn. Returns a promise that resolves
   *  when the gateway sends the decision. */
  requestApproval(turnId: string, approvalId: string): Promise<ApprovalDecision> {
    const turn = this.turns.get(turnId)
    if (!turn) return Promise.reject(new Error(`unknown turn ${turnId}`))

    return new Promise<ApprovalDecision>((resolve) => {
      turn.pendingApproval = { approval_id: approvalId, resolve }
    })
  }

  resolveApproval(turnId: string, approvalId: string, decision: ApprovalDecision): boolean {
    const turn = this.turns.get(turnId)
    if (!turn?.pendingApproval) return false
    if (turn.pendingApproval.approval_id !== approvalId) return false
    turn.pendingApproval.resolve(decision)
    turn.pendingApproval = null
    return true
  }

  cancel(turnId: string): boolean {
    const turn = this.turns.get(turnId)
    if (!turn) return false
    turn.abortController.abort()
    return true
  }

  /** Prune completed turns older than `maxAgeMs`. */
  prune(maxAgeMs = 5 * 60 * 1000): void {
    const now = Date.now()
    for (const [id, turn] of this.turns) {
      if (turn.done) {
        const lastEvent = turn.events[turn.events.length - 1]
        // Simple heuristic — if turn is done and old, drop it
        if (lastEvent && turn.events.length > 0) {
          this.turns.delete(id)
        }
      }
    }
  }

  get activeTurnId(): string | null {
    for (const [id, turn] of this.turns) {
      if (!turn.done) return id
    }
    return null
  }

  get isBusy(): boolean {
    return this.activeTurnId !== null
  }
}