270 lines
14 KiB
JavaScript
270 lines
14 KiB
JavaScript
/**
|
|
* @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 fs from 'node:fs'
|
|
import os from 'node:os'
|
|
import path from 'node:path'
|
|
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
|
|
|
export const name = 'omp-advisor'
|
|
export const inject = ['agents', 'llm']
|
|
|
|
/**
|
|
* Per-session advisor choices made in the UI: { [sessionId]: { enabled, provider, model,
|
|
* reasoningEffort, minLevel, at } }. A session without an entry uses the config defaults, which
|
|
* are OFF. The pre-1.2 global file (`omp-advisor.json`) is intentionally ignored so that no
|
|
* existing session silently stays on.
|
|
*/
|
|
function settingsFile() {
|
|
const dir = process.env.DSH_PROFILE_DIR || path.join(process.env.DSH_HOME || path.join(os.homedir(), '.dsh'), 'profiles', 'web')
|
|
return path.join(dir, 'omp-advisor', 'sessions.json')
|
|
}
|
|
const MAX_SESSIONS = 2000
|
|
function loadSessions() {
|
|
try {
|
|
const v = JSON.parse(fs.readFileSync(settingsFile(), 'utf8'))
|
|
return v && typeof v === 'object' && !Array.isArray(v) ? v : {}
|
|
} catch { return {} }
|
|
}
|
|
function saveSessions(all) {
|
|
const f = settingsFile()
|
|
fs.mkdirSync(path.dirname(f), { recursive: true })
|
|
// Keep the most recently changed sessions only.
|
|
const entries = Object.entries(all)
|
|
const kept = entries.length > MAX_SESSIONS
|
|
? Object.fromEntries(entries.sort((a, b) => (b[1]?.at ?? 0) - (a[1]?.at ?? 0)).slice(0, MAX_SESSIONS))
|
|
: all
|
|
const tmp = `${f}.${process.pid}.tmp`
|
|
fs.writeFileSync(tmp, JSON.stringify(kept, null, 1))
|
|
fs.renameSync(tmp, f) // atomic replace: a crash never leaves half a file
|
|
}
|
|
const validSessionId = (s) => typeof s === 'string' && s.length > 0 && s.length <= 200 && /^[\w.:-]+$/.test(s)
|
|
|
|
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: false, // default for every session until it is turned on in the composer
|
|
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 }
|
|
let sessions = loadSessions()
|
|
const sessionIdOf = (agent) => agent?.id ?? agent?.session?.header?.id
|
|
/** Effective settings for one session: its own UI choice over the config defaults. */
|
|
const eff = (sessionId) => {
|
|
const live = (sessionId && sessions[sessionId]) || {}
|
|
return {
|
|
enabled: live.enabled ?? cfg.enabled,
|
|
provider: live.provider ?? cfg.provider,
|
|
model: live.model ?? cfg.model,
|
|
reasoningEffort: live.reasoningEffort === null ? undefined : (live.reasoningEffort ?? cfg.reasoningEffort),
|
|
minLevel: live.minLevel ?? cfg.minLevel,
|
|
}
|
|
}
|
|
const minLevel = (sessionId) => LEVEL[eff(sessionId).minLevel] ?? 1
|
|
const installed = new Map()
|
|
|
|
function install(agent) {
|
|
if (installed.has(agent) || !agent?.ctx) 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
|
|
if (!eff(sessionIdOf(agent)).enabled) { st.queue.length = 0; 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 (!eff(sessionIdOf(agent)).enabled) return
|
|
if (st.pending) {
|
|
let t
|
|
await Promise.race([st.pending, new Promise((r) => { t = setTimeout(r, cfg.blocking ? cfg.timeoutMs : 4000) })]).finally(() => clearTimeout(t))
|
|
}
|
|
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 e = eff(sessionIdOf(agent))
|
|
// A half-set route (provider without model) falls back to the agent's model as a whole.
|
|
const own = e.provider && e.model
|
|
const provider = own ? e.provider : header?.config?.provider
|
|
const model = own ? e.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,
|
|
...(own && e.reasoningEffort ? { reasoningEffort: e.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(sessionIdOf(agent)) && 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 }
|
|
|
|
// ---------------------------------------------------------------- Web UI settings routes
|
|
// Mounted only while the web server and its trust fence exist (a headless profile has neither).
|
|
ctx.inject?.(['webServer', 'connection'], (wctx) => {
|
|
const webServer = wctx.webServer
|
|
const connection = wctx.connection
|
|
const send = (res, code, body) => { res.statusCode = code; res.setHeader('content-type', 'application/json; charset=utf-8'); res.setHeader('cache-control', 'no-store'); res.end(JSON.stringify(body)) }
|
|
const readBody = async (req) => { let b = ''; for await (const c of req) { b += c; if (b.length > 16384) throw new Error('body too large') } return b ? JSON.parse(b) : {} }
|
|
const state = (sessionId) => ({ sessionId, ...eff(sessionId), configured: { provider: cfg.provider || null, model: cfg.model || null, enabled: !!cfg.enabled } })
|
|
wctx.effect(() => webServer.register({
|
|
kind: 'exact',
|
|
path: '/omp-advisor/settings',
|
|
handler: async (req, res) => {
|
|
const rej = connection.requestRejection(req)
|
|
if (rej !== undefined) return send(res, rej, { code: 'rejected' })
|
|
try {
|
|
// Settings are per session: ?session=<id> on every call.
|
|
const sessionId = new URL(req.url ?? '/', 'http://x').searchParams.get('session')
|
|
if (!validSessionId(sessionId)) return send(res, 400, { code: 'error', message: 'session query parameter is required' })
|
|
if (req.method === 'GET') return send(res, 200, state(sessionId))
|
|
if (req.method !== 'POST') { res.statusCode = 405; return res.end() }
|
|
const b = await readBody(req)
|
|
const next = { ...(sessions[sessionId] ?? {}) }
|
|
if (typeof b.enabled === 'boolean') next.enabled = b.enabled
|
|
if ('provider' in b) next.provider = typeof b.provider === 'string' && b.provider ? b.provider : ''
|
|
if ('model' in b) next.model = typeof b.model === 'string' && b.model ? b.model : ''
|
|
if ('reasoningEffort' in b) next.reasoningEffort = typeof b.reasoningEffort === 'string' && b.reasoningEffort ? b.reasoningEffort : null
|
|
if (typeof b.minLevel === 'string' && b.minLevel in LEVEL && b.minLevel !== 'ok') next.minLevel = b.minLevel
|
|
next.at = Date.now()
|
|
sessions = { ...sessions, [sessionId]: next }
|
|
saveSessions(sessions)
|
|
return send(res, 200, state(sessionId))
|
|
} catch (e) { return send(res, 400, { code: 'error', message: String(e?.message ?? e) }) }
|
|
},
|
|
}), 'omp-advisor: settings route')
|
|
})
|
|
|
|
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 }
|
|
}
|