omp-advisor/index.js

/**
 * @local/omp-advisor — oh-my-pi's "advisor": a second model watches every step the main agent
 * takes and injects notes inline (an aside, a concern, or a hard blocker).
 *
 * After each completed step (tool results in) the advisor model reviews a compact digest of the
 * transcript on its own context and answers with JSON {verdict: ok|note|concern|blocker, message}.
 * Anything but "ok" is delivered to the main agent before its next request:
 *   - note / concern: appended to the next step as a system reminder;
 *   - blocker: the reminder demands the agent stop and address it, and at the turn boundary a
 *     still-unacknowledged blocker steers the agent into one more step.
 * Reviews run in the background and never delay the main loop unless `blocking: true`.
 */
import { createUserMessage } from '@deepseek-ai/dsh-llm'

export const name = 'omp-advisor'
export const inject = ['agents', 'llm']

const SOURCE = { kind: 'omp-advisor' }

const SYSTEM = `You are the ADVISOR: a senior engineer silently pairing with a coding agent. You see a digest of the conversation and the agent's latest step (its message, tool calls and results). Intervene only when it matters:
- "blocker": the agent is about to cause or has caused real damage (data loss, destructive command, secret leak, pushing broken code, ignoring an explicit user instruction, going in circles), or is clearly heading the wrong way.
- "concern": a likely bug, wrong assumption, missed requirement, unverified claim, or a much better approach.
- "note": a brief useful aside the agent would want.
- "ok": nothing worth saying (this should be the common case).
Be specific and short (max 3 sentences), reference files/commands concretely, and never repeat advice you already gave (listed under PREVIOUS ADVICE). Reply with ONLY one JSON object: {"verdict":"ok|note|concern|blocker","message":"..."}`

export function apply(ctx, config = {}) {
  const cfg = {
    enabled: true,
    provider: '',          // empty = same provider as the agent
    model: '',             // empty = same model as the agent
    reasoningEffort: undefined,
    minLevel: 'note',      // lowest verdict delivered: note | concern | blocker
    everyNSteps: 1,
    maxDigestChars: 24000,
    maxToolResultChars: 1500,
    blocking: false,       // true: wait for the review before the next step
    timeoutMs: 60000,
    skipSubagents: true,
    ...config,
  }
  const LEVEL = { ok: 0, note: 1, concern: 2, blocker: 3 }
  const minLevel = LEVEL[cfg.minLevel] ?? 1
  const installed = new Map()

  function install(agent) {
    if (installed.has(agent) || !agent?.ctx || !cfg.enabled) return
    if (cfg.skipSubagents && agent.session?.header?.parentSession) return
    const st = { pending: null, queue: [], history: [], step: 0, openBlocker: null, lastReviewedSeq: -1, failures: 0 }
    const disposers = []

    // Review after every step that ran tools: the tool results are the newest surface nodes.
    disposers.push(agent.ctx.on('agent/pre-step', async (payload, next) => {
      const decision = await next()
      if (decision.kind !== 'enter') return decision
      st.step++
      if (payload.step > 1 && st.step % cfg.everyNSteps === 0 && !st.pending) {
        st.pending = review(agent, st, payload.signal).finally(() => { st.pending = null })
      }
      if (cfg.blocking && st.pending) await st.pending
      const notes = st.queue.splice(0)
      if (!notes.length) return decision
      const text = notes.map(formatAdvice).join('\n\n')
      return { ...decision, messages: [...decision.messages, createUserMessage({ content: [{ type: 'text', text }], source: SOURCE })] }
    }))

    // At the end of a turn: deliver late advice; a blocker forces one more step.
    disposers.push(agent.ctx.on('agent/turn-stopping', async ({ signal }) => {
      if (st.pending) await Promise.race([st.pending, new Promise((r) => setTimeout(r, cfg.blocking ? cfg.timeoutMs : 4000))])
      const notes = st.queue.splice(0)
      const block = notes.find((n) => n.verdict === 'blocker')
      if (block && !signal?.aborted) {
        agent.steer(createUserMessage({ content: [{ type: 'text', text: notes.map(formatAdvice).join('\n\n') }], source: SOURCE }))
      } else if (notes.length) {
        st.queue.push(...notes) // deliver with the next user turn
      }
    }))

    installed.set(agent, () => { for (const d of disposers) { try { d() } catch {} } })
  }

  function formatAdvice(n) {
    const tag = n.verdict === 'blocker' ? 'ADVISOR BLOCKER — stop and address this before continuing' : `Advisor ${n.verdict}`
    return `<system-reminder>\n[${tag}] ${n.message}\n(Second-model review of your last step. Acknowledge briefly and act on it if it is right; say why if you disagree.)\n</system-reminder>`
  }

  async function review(agent, st, signal) {
    try {
      const header = agent.session.requestHeader?.()
      const provider = cfg.provider || header?.config?.provider
      const model = cfg.model || header?.config?.model
      if (!provider || !model) return
      const digest = buildDigest(agent.session, st)
      if (!digest) return
      const prev = st.history.slice(-8).map((h) => `- [${h.verdict}] ${h.message}`).join('\n') || '(none)'
      const ac = new AbortController()
      const timer = setTimeout(() => ac.abort(), cfg.timeoutMs)
      signal?.addEventListener?.('abort', () => ac.abort(), { once: true })
      let text = ''
      try {
        for await (const ch of ctx.llm.stream({
          provider, model,
          ...(cfg.reasoningEffort ? { reasoningEffort: cfg.reasoningEffort } : {}),
          system: SYSTEM,
          messages: [createUserMessage({ content: [{ type: 'text', text: `PREVIOUS ADVICE:\n${prev}\n\nTRANSCRIPT DIGEST (oldest first, latest step last):\n${digest}\n\nYour JSON verdict:` }], source: SOURCE })],
          maxTokens: 600,
          signal: ac.signal,
        })) {
          if (ch.type === 'text-delta') text += ch.text
          if (ch.type === 'finish' && ch.reason?.kind === 'error') throw new Error(ch.reason.error?.message ?? 'advisor model error')
        }
      } finally { clearTimeout(timer) }
      const v = parseVerdict(text)
      if (!v) return
      st.history.push(v)
      if ((LEVEL[v.verdict] ?? 0) >= minLevel && LEVEL[v.verdict] > 0) {
        if (st.history.slice(0, -1).some((h) => h.message === v.message)) return
        st.queue.push(v)
      }
      st.failures = 0
    } catch (e) {
      if (++st.failures <= 3) console.warn(`[omp-advisor] review failed: ${e?.message ?? e}`)
    }
  }

  function buildDigest(session, st) {
    const nodes = [...session.surface.nodes]
    if (!nodes.length) return ''
    const parts = []
    for (const seq of nodes) {
      const ev = session.eventAt(seq)
      if (!ev) continue
      let msg
      try { msg = session.deriveEventMessage(ev) } catch { continue }
      if (!msg || msg.source?.kind === 'omp-advisor') continue
      parts.push(renderMessage(msg))
    }
    st.lastReviewedSeq = nodes.at(-1)
    // Keep the head (task) and as much of the tail (latest steps) as fits.
    let out = parts.join('\n')
    if (out.length > cfg.maxDigestChars) {
      const head = parts.slice(0, 2).join('\n').slice(0, Math.floor(cfg.maxDigestChars * 0.25))
      const tail = out.slice(-Math.floor(cfg.maxDigestChars * 0.75))
      out = `${head}\n…[earlier steps omitted]…\n${tail}`
    }
    return out
  }

  function renderMessage(msg) {
    const role = msg.role === 'assistant' ? 'AGENT' : msg.role === 'tool' ? 'TOOL RESULT' : msg.role.toUpperCase()
    const bits = []
    for (const b of msg.content ?? []) {
      if (b.type === 'text') bits.push(clip(b.text, msg.role === 'tool' ? cfg.maxToolResultChars : 4000))
      else if (b.type === 'tool-call') bits.push(`→ ${b.name}(${clip(typeof b.arguments === 'string' ? b.arguments : JSON.stringify(b.arguments ?? b.input ?? {}), 1200)})`)
      else if (b.type === 'tool-result') bits.push(clip(contentText(b.content), cfg.maxToolResultChars))
    }
    return `### ${role}${msg.toolName ? ` (${msg.toolName})` : ''}\n${bits.join('\n')}`
  }

  const contentText = (c) => (Array.isArray(c) ? c.map((x) => (x.type === 'text' ? x.text : `[${x.type}]`)).join('\n') : String(c ?? ''))
  const clip = (s, n) => { const t = String(s ?? ''); return t.length > n ? `${t.slice(0, n)}…[+${t.length - n} chars]` : t }

  for (const agent of ctx.agents.list()) install(agent)
  ctx.on('agent/created', ({ agent }) => { install(agent) })
  ctx.on('agent/disposed', ({ agent }) => { installed.get(agent)?.(); installed.delete(agent) })
  ctx.effect(() => () => { for (const d of installed.values()) d(); installed.clear() }, 'omp-advisor: agents')
}

export function parseVerdict(text) {
  const m = /\{[\s\S]*\}/.exec(String(text ?? ''))
  if (!m) return null
  try {
    const v = JSON.parse(m[0])
    const verdict = String(v.verdict ?? '').toLowerCase()
    if (!['ok', 'note', 'concern', 'blocker'].includes(verdict)) return null
    return { verdict, message: String(v.message ?? '').trim().slice(0, 800) }
  } catch { return null }
}