HologramAgent-DSH-Memory-Port/lib/loop/inbox.js

/**
 * Durable agent-inbox projection and its command facade.
 *
 * ## Why this file exists at all
 *
 * `dsh-agent-loop` keeps `inboxProjectionDefinition` module-private: its
 * package entry exports only `AgentLoop`, `DEFAULT_MAX_PARALLEL_TOOL_CALLS`,
 * `CONFIGURED_AGENT_IDENTITIES_KEY` and `turnBoundaryProjectionDefinition`
 * (dsh-agent-loop/lib/index.js:2023). A replacement loop therefore cannot
 * import the fold, and re-deriving pending input in memory instead of in a
 * projection would lose queued messages on `resume` — the pending splices live
 * in the durable log, so the state has to be rebuilt from it.
 *
 * So the fold is reproduced here, field for field, from
 * `dsh-agent-loop/lib/index.js:26-60` (0.2.1-rc.2, MIT). The projection KEY
 * must stay `"inbox"` and `stateVersion` must stay 1: a different key would
 * silently read as "no pending input" for sessions written by the stock loop,
 * and a different version would make `stateSchema.parse` refuse those rows.
 *
 * ★ The one deliberate deviation: the host validates with zod
 * (`z.array(z.custom())`), which is a dependency this plugin does not carry.
 * {@link inboxProjectionSchema} provides the same contract the projection
 * registry actually uses — `parse()` that throws on a malformed persisted row —
 * without pulling zod in. Nothing else about the fold changed.
 *
 * @module dsh-memory/lib/loop/inbox
 */

/**
 * Validate one persisted inbox state.
 *
 * The registry calls `stateSchema.parse` on durable checkpoint rows and on the
 * value a fold returns, so this must reject anything that is not the two-array
 * shape. Message contents stay opaque (`z.custom()` in the host): they are
 * already validated by the session layer when they were added to the surface,
 * and re-validating them here would make the fold able to reject a log the
 * session accepted.
 */
export const inboxProjectionSchema = {
  /**
   * @param {unknown} value candidate state
   * @returns {{'next-turn': Array<object>, 'next-step': Array<object>}} the same object
   * @throws {Error} when the shape is not two message arrays
   */
  parse(value) {
    if (value === null || typeof value !== 'object' || Array.isArray(value)) {
      throw new Error('inbox projection state must be an object')
    }
    const nextTurn = value['next-turn']
    const nextStep = value['next-step']
    if (!Array.isArray(nextTurn) || !Array.isArray(nextStep)) {
      throw new Error('inbox projection state must hold "next-turn" and "next-step" message arrays')
    }
    return value
  },
}

/** Standard fold that reconstructs pending input and rejects invalid splice history. */
export const inboxProjectionDefinition = {
  key: 'inbox',
  stateSchema: inboxProjectionSchema,
  init: () => ({
    'next-turn': [],
    'next-step': [],
  }),
  apply(state, event) {
    if (event.type !== 'agent/inbox/spliced') return state
    const splice = event.data
    try {
      const inbox = state[splice.target]
      const removedCount = splice.removedCount ?? 0
      if (
        !Number.isSafeInteger(splice.start) ||
        splice.start < 0 ||
        splice.start > inbox.length ||
        !Number.isSafeInteger(removedCount) ||
        removedCount < 0 ||
        splice.start + removedCount > inbox.length
      ) {
        throw new Error('invalid inbox splice')
      }
      const next = inbox.toSpliced(splice.start, removedCount, ...splice.inserted)
      const ids = new Set()
      for (const message of splice.target === 'next-turn' ? [...next, ...state['next-step']] : [...state['next-turn'], ...next]) {
        if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
        ids.add(message.id)
      }
      return splice.target === 'next-turn'
        ? { 'next-turn': next, 'next-step': state['next-step'] }
        : { 'next-turn': state['next-turn'], 'next-step': next }
    } catch (error) {
      throw new Error(`invalid persisted inbox splice at session seq ${event.seq}`, { cause: error })
    }
  },
  wire: {
    viewSchema: inboxProjectionSchema,
    view: (state) => state,
  },
  stateVersion: 1,
}

/**
 * Driver-owned durable Inbox implementation.
 *
 * Every mutation is a durable `agent/inbox/spliced` append; the projection is
 * the only state holder, so a resumed session sees exactly the pending input
 * the previous process left behind.
 */
export class ReactLoopInbox {
  /**
   * @param {object} projections `ctx.sessionProjections`
   * @param {object} session the owning session
   * @param {object} dispatch agent-scoped dispatcher
   */
  constructor(projections, session, dispatch) {
    this.projections = projections
    this.session = session
    this.dispatch = dispatch
  }

  /** Prompts awaiting individual turns. */
  get nextTurn() {
    return this.current()['next-turn']
  }

  /** Input awaiting the next step boundary. */
  get nextStep() {
    return this.current()['next-step']
  }

  /** Whether either pending-message list contains work. */
  get hasPending() {
    const state = this.current()
    return state['next-turn'].length > 0 || state['next-step'].length > 0
  }

  /** Durably cancel all pending input, clearing next-step before next-turn. */
  clear() {
    this.splice('next-step', 0, this.nextStep.length, [])
    this.splice('next-turn', 0, this.nextTurn.length, [])
  }

  /**
   * Remove and return the complete batch proposed for one step.
   * @param {'next-turn'|'next-step'} target whether this boundary also consumes one queued turn
   * @param {number} turn turn that will own the claimed batch
   */
  claim(target, turn) {
    const claimed = this.mutate('next-step', 0, this.nextStep.length, [], false)
    if (target === 'next-turn') claimed.push(...this.mutate('next-turn', 0, 1, [], false))
    for (const message of claimed) this.dispatch.emit('agent/inbox/claimed', { message, turn })
    return claimed
  }

  /**
   * Append one message to a pending list.
   * @param {'next-turn'|'next-step'} target pending list to extend
   * @param {object} message message to append
   */
  append(target, message) {
    this.splice(target, this.current()[target].length, 0, [message])
  }

  /**
   * Prepend one message to a pending list.
   * @param {'next-turn'|'next-step'} target pending list to extend
   * @param {object} message message to prepend
   */
  prepend(target, message) {
    this.splice(target, 0, 0, [message])
  }

  /**
   * Replace one pending message in place.
   * @param {string} messageId identity of the pending message to replace
   * @param {object} newMessage replacement message
   * @returns {boolean} whether the message was still pending
   */
  replace(messageId, newMessage) {
    const location = this.locate(messageId)
    if (location === undefined) return false
    this.splice(location.target, location.index, 1, [newMessage])
    return true
  }

  /**
   * Remove one pending message.
   * @param {string} messageId identity of the pending message to remove
   * @returns {boolean} whether the message was still pending
   */
  remove(messageId) {
    const location = this.locate(messageId)
    if (location === undefined) return false
    this.splice(location.target, location.index, 1, [])
    return true
  }

  /**
   * Apply standard splice semantics and durably record the normalized result.
   * @param {'next-turn'|'next-step'} target pending list to mutate
   * @param {number} start splice position
   * @param {number} deleteCount maximum number of messages to remove
   * @param {Array<object>} inserted messages to insert at the resolved position
   */
  splice(target, start, deleteCount, inserted) {
    return this.mutate(target, start, deleteCount, inserted, true)
  }

  /** Locate one pending identity across both owned lists. */
  locate(messageId) {
    const state = this.current()
    for (const target of ['next-turn', 'next-step']) {
      const index = state[target].findIndex((message) => message.id === messageId)
      if (index >= 0) return { target, index }
    }
  }

  /** Read the current durable projection state. */
  current() {
    const state = this.projections.stateOf(this.session, 'inbox')
    if (state === undefined) throw new Error(`agent "${this.session.id}" cannot read inbox state: its projection registration is not active`)
    return state
  }

  /** Commit one normalized mutation and publish its live events. */
  mutate(target, start, deleteCount, inserted, discardRemoved) {
    const state = this.current()
    const inbox = state[target]
    const truncatedStart = Math.trunc(start)
    const offset = Number.isNaN(truncatedStart) ? 0 : truncatedStart
    const actualStart = offset < 0 ? Math.max(inbox.length + offset, 0) : Math.min(offset, inbox.length)
    const truncatedDeleteCount = Math.trunc(deleteCount)
    const actualDeleteCount = Math.min(Math.max(Number.isNaN(truncatedDeleteCount) ? 0 : truncatedDeleteCount, 0), inbox.length - actualStart)
    if (actualDeleteCount === 0 && inserted.length === 0) return []
    const candidate = inbox.toSpliced(actualStart, actualDeleteCount, ...inserted)
    const ids = new Set()
    for (const message of target === 'next-turn' ? [...candidate, ...state['next-step']] : [...state['next-turn'], ...candidate]) {
      if (ids.has(message.id)) throw new Error(`message "${message.id}" is already pending`)
      ids.add(message.id)
    }
    const outcome = discardRemoved && actualDeleteCount > 0 ? 'canceled' : undefined
    const splice = {
      target,
      start: actualStart,
      ...(actualDeleteCount === 0 ? {} : { removedCount: actualDeleteCount }),
      inserted,
      ...(outcome === undefined ? {} : { outcome }),
    }
    const removed = inbox.slice(actualStart, actualStart + actualDeleteCount)
    const event = this.session.append('agent/inbox/spliced', splice)
    if (discardRemoved) for (const message of removed) this.dispatch.emit('agent/inbox/discarded', { message })
    for (const message of event.data.inserted) this.dispatch.emit('agent/inbox/inserted', { message })
    return removed
  }
}