xagent

1// xagent: run Codex, Copilot and Antigravity as workers for a Claude Code session.
2//   dispatch runs one external agent; route picks models for a task by quota and complexity.
3import { atom, read, update } from 'claude-code'
4import type { EngineInterface, Register } from 'claude-code'
5
6import type { DispatchInput, DispatchResult, Parsed, ProviderQuota, RouteInput, XAgent } from '../types'
7import { buildArgv } from './backends'
8import { changedFiles, snapshot, treeOf } from './io'
9import { feed, firstLine, parse, sessionIdIn } from './parse'
10import { describe, describeOne, isBlocked } from './quota'
11import { clamp, defaults, parsePick, readConfig, ROUTE_SCHEMA, routerPrompt } from './routing'
12
13const agents = atom({ plugin: 'xagent', key: 'agents' } as const, [] as XAgent[])
14const blocked = atom({ plugin: 'xagent', key: 'blocked' } as const, {} as Record<string, ProviderQuota>)
15const effort = atom({ plugin: 'xagent', key: 'effort' } as const, 'medium')
16
17const DISPATCH_SCHEMA = {
18  type: 'object',
19  properties: {
20    backend: { type: 'string', enum: ['codex', 'copilot', 'agy'] },
21    model: { type: 'string' },
22    effort: { type: 'string', enum: ['low', 'medium', 'high', 'xhigh', 'max'] },
23    mode: { type: 'string', enum: ['read-only', 'write'] },
24    prompt: { type: 'string' },
25    dir: { type: 'string', description: 'working directory, absolute' },
26    schema: { type: 'string', description: 'path to a JSON schema file; the full answer comes back as `answer`' },
27    resume: { type: 'string', description: 'sessionId from an earlier dispatch, to continue it' },
28    timeoutMs: { type: 'number' },
29    override: { type: 'boolean', description: 'run on a backend at 90 % of its quota; only when the user said so' },
30  },
31  required: ['backend', 'mode', 'prompt'],
32}
33
34const ROUTE_SCHEMA_IN = {
35  type: 'object',
36  properties: {
37    task: { type: 'string', description: 'the task text from the plan' },
38    files: { type: 'array', items: { type: 'string' } },
39    instructions: { type: 'string', description: "the user's standing instructions for this run" },
40    override: { type: 'boolean', description: 'the user allows providers at 90 % of their quota' },
41  },
42  required: ['task'],
43}
44
45const DEFAULT_TIMEOUT_MS = 20 * 60_000
46const BACKSTOP_S = 5 // coreutils timeout sends TERM this long past the time limit, KILL 10 s later
47const STOP_SLACK_MS = 2_000 // how long past that TERM a stop may take to be confirmed
48// The headroom proxy wrapper exports these; inherited, they put copilot in BYOK mode and bill Claude, not the backend's own subscription login.
49const PROXY_ENV = ['COPILOT_PROVIDER_BASE_URL', 'COPILOT_PROVIDER_TYPE', 'COPILOT_PROVIDER_API_KEY', 'COPILOT_PROVIDER_BEARER_TOKEN', 'COPILOT_MODEL', 'OPENAI_BASE_URL', 'ANTHROPIC_BASE_URL']
50const COOLDOWN_MS = 30 * 60_000
51const LIMIT_WORDS = /usage limit|rate limit|quota|exceeded|\b429\b/i
52// `failed to run command`: coreutils timeout could not start the CLI (not installed, not on PATH).
53const AUTH_WORDS = /not logged in|log in|unauthori[sz]ed|\b401\b|ENOENT|command not found|failed to run command/i
54const RETRY_PROMPT = 'You were stopped at the time limit. Continue the same task from where you are, then give the final answer.'
55
56export function expandHome(path: string, home: string): string {
57  return path === '~' ? home : path.startsWith('~/') ? `${home}${path.slice(1)}` : path
58}
59
60// The OS backstop, still argv: coreutils timeout ends the child (and what it started) a few
61// seconds past the time limit, whether or not the engine's stream can interrupt a pending pull.
62export function withBackstop(argv: string[], timeoutMs: number): string[] {
63  return ['env', ...PROXY_ENV.flatMap(name => ['-u', name]), 'timeout', '--kill-after=10s', `${Math.ceil(timeoutMs / 1000) + BACKSTOP_S}s`, ...argv]
64}
65
66// How long after the time limit a stop may take to be confirmed: until just past the backstop's TERM.
67export function stopGraceMs(timeoutMs: number): number {
68  return Math.ceil(timeoutMs / 1000) * 1000 - timeoutMs + BACKSTOP_S * 1000 + STOP_SLACK_MS
69}
70
71// Ends the child through its stream (leaving the loop kills it) and says whether that is
72// confirmed within the grace period: stop() resolved, or the pull already under way ended the
73// stream (the child is gone) or failed. A pull that brings more output means it is still alive.
74// A rejected or hanging stop() alone is not confirmed.
75export async function stopStream(stop: () => Promise<unknown>, sleep: () => Promise<void>, pending?: Promise<IteratorResult<unknown, unknown>>): Promise<boolean> {
76  const never = new Promise<boolean>(() => {})
77  const ended = stop().then(() => true, () => false)
78  const pulled = pending ? pending.then(step => (step.done ? true : never), () => true) : never
79  return Promise.race([ended, pulled, sleep().then(() => false)])
80}
81
82async function mark($: EngineInterface, backend: string, q: ProviderQuota) {
83  await update($, blocked, all => ({ ...all, [backend]: q }))
84}
85
86async function track($: EngineInterface, id: string, patch: Partial<XAgent>) {
87  await update($, agents, list => list.map(a => (a.id === id ? { ...a, ...patch } : a)))
88}
89
90// One CLI run. `stopped` is false only when a child we had to stop was not confirmed gone.
91async function runOnce($: EngineInterface, input: DispatchInput, argv: string[], id: string, signal?: AbortSignal): Promise<Parsed & { stopped: boolean }> {
92  const out = feed(input.backend)
93  let stderr = ''
94  let failure: string | undefined
95  let end: { code: number | null; signal: string | null } = { code: null, signal: null }
96  let stopped = true
97  let pulling = false
98  const limit = input.timeoutMs ?? DEFAULT_TIMEOUT_MS
99  // spawn closes stdin from the start (no `input`): codex exec otherwise waits on it.
100  const stream = $.process.spawn({ argv: withBackstop(argv, limit), cwd: input.dir })
101  const stop = () => stream.return({ code: null, signal: null })
102  const grace = () => $.clock.sleep(stopGraceMs(limit))
103  // Wall clock, raced against each pull: a child that prints nothing must still stop.
104  let ring: () => void = () => {}
105  const expired = new Promise<'expired'>(resolve => (ring = () => resolve('expired')))
106  const timer = $.clock.after(limit, () => ring())
107  // The user interrupted (Esc): the engine kills the child; stop reading at once.
108  const aborted = new Promise<'aborted'>(resolve => {
109    if (signal?.aborted) resolve('aborted')
110    else signal?.addEventListener('abort', () => resolve('aborted'), { once: true })
111  })
112  try {
113    while (true) {
114      pulling = true
115      // The pull is kept, not dropped: if the stream cannot be stopped while it is pending, its
116      // end (the backstop killed the child) is what confirms the stop.
117      const pull = stream.next()
118      const step = await Promise.race([pull, expired, aborted])
119      pulling = false
120      if (step === 'aborted') {
121        void stop().catch(() => {})
122        void pull.catch(() => {}) // left unread: its end no longer matters
123        failure = 'interrupted'
124        break
125      }
126      if (step === 'expired') {
127        stopped = await stopStream(stop, grace, pull)
128        failure = stopped ? 'timed out' : 'timed out (child may still be running)'
129        break
130      }
131      if (step.done) {
132        end = step.value ?? end
133        break
134      }
135      const piece = step.value
136      if (piece.stream === 'stderr') {
137        stderr = (stderr + piece.text).slice(-2000)
138        continue
139      }
140      const status = out.push(piece.text)
141      if (status) await track($, id, { last: status })
142    }
143  } catch (err) {
144    failure ??= (err as Error).message
145    // A failed pull means the child never started or already ended. A failure in our own code
146    // between pulls leaves it running, so stop it.
147    if (!pulling) stopped = await stopStream(stop, grace)
148  } finally {
149    timer.cancel()
150  }
151  const stdout = out.text()
152  const sessionId = sessionIdIn(input.backend, stdout)
153  if (failure) return { ok: false, summary: '', sessionId, error: failure, stopped }
154  const result = parse(input.backend, stdout, Boolean(input.schema))
155  const why = stderr.trim().split('\n').slice(-3).join(' ').slice(0, 300)
156  // The CLI's own error (a usage limit, say) says more than stderr or the exit code.
157  const said = !result.ok && result.error !== 'no answer in output' ? result.error : undefined
158  // An answer only counts from a child that ended cleanly.
159  if (end.signal) return { ...result, ok: false, error: said ?? `killed by ${end.signal}`, stopped }
160  if (end.code !== 0) return { ...result, ok: false, error: said ?? (!result.ok && why ? why : `exited with code ${end.code}`), stopped }
161  if (!result.ok && result.error === 'no answer in output' && why) return { ...result, error: why, stopped }
162  return { ...result, stopped }
163}
164
165export async function runDispatch($: EngineInterface, raw: DispatchInput, task: string, signal?: AbortSignal): Promise<DispatchResult> {
166  const home = (await $.env.get('HOME')) ?? ''
167  const input: DispatchInput = { ...raw, dir: raw.dir && expandHome(raw.dir, home), schema: raw.schema && expandHome(raw.schema, home) }
168  let argv: string[]
169  try {
170    argv = buildArgv(input)
171  } catch (err) {
172    const q = (await snapshot($))[input.backend]
173    return { ok: false, summary: '', sessionId: null, error: (err as Error).message, attempts: 0, baseTree: null, tree: null, filesChanged: null, quota: q }
174  }
175  const startedAt = await $.clock.now()
176  const before = (await snapshot($))[input.backend]
177  const refuse = (error: string, extra: Partial<DispatchResult> = {}): DispatchResult => ({ ok: false, summary: '', sessionId: null, error, attempts: 0, baseTree: null, tree: null, filesChanged: null, quota: before, ...extra })
178  // The 90 % rule: no new work for a provider that close to a limit, unless the user said so.
179  if (!input.override && isBlocked(before, startedAt)) return refuse(`${input.backend} is at 90 % of a quota window: ask the user, or pass override if they allow it`)
180  // A writer may only start once its baseline is known: a lost baseline cannot be taken back after edits.
181  const base = input.mode === 'write' ? await treeOf($, input.dir) : { tree: null }
182  if (base.error) return refuse(`baseline snapshot failed: ${base.error}`, { treeError: base.error })
183  const id = `x${startedAt.toString(36)}${Math.random().toString(36).slice(2, 6)}`
184  const quotaLine = describeOne(input.backend, before, startedAt).split('\n')[0] ?? ''
185  const one: XAgent = { id, backend: input.backend, model: input.model ?? 'default', effort: input.effort ?? 'default', task, status: 'running', startedAt, last: 'starting', quota: quotaLine }
186  await update($, agents, list => [one, ...list].slice(0, 30))
187
188  // From here the agent is listed as running: whatever happens below, it must not stay that way.
189  let ended = false
190  let thrown: unknown
191  try {
192    let attempts = 1
193    let result = await runOnce($, input, argv, id, signal)
194    // One retry, owned here (the skill never retries): only once the first child is known to be
195    // gone, and never after the user interrupted.
196    if (result.error === 'timed out' && result.stopped && !signal?.aborted) {
197      attempts = 2
198      await track($, id, { last: 'timed out; retrying' })
199      const again: DispatchInput = result.sessionId ? { ...input, resume: result.sessionId, prompt: RETRY_PROMPT } : input
200      result = await runOnce($, again, buildArgv(again), id, signal)
201    }
202    // Interrupted: the engine has moved on and may refuse further calls; end the agent first and
203    // skip the tree, diff and quota work.
204    if (signal?.aborted) {
205      await track($, id, { status: 'failed', endedAt: await $.clock.now().catch(() => startedAt), last: 'stopped: interrupted' })
206      ended = true
207      const { stopped: _, ...parsed } = result
208      return { ...parsed, ok: false, error: 'interrupted', attempts, baseTree: base.tree, tree: null, filesChanged: null, quota: before }
209    }
210
211    const now = await $.clock.now()
212    // An interrupt says nothing about the backend: leave its quota marks alone.
213    if (!result.ok && !signal?.aborted) {
214      const why = result.error ?? ''
215      if (AUTH_WORDS.test(why)) await mark($, input.backend, { state: 'unavailable', reason: why.slice(0, 80) })
216      else if (LIMIT_WORDS.test(why)) await mark($, input.backend, { state: 'cooldown', until: now + COOLDOWN_MS })
217    }
218    const after = base.tree ? await treeOf($, input.dir) : { tree: null }
219    const diff = base.tree && after.tree && input.dir ? await changedFiles($, input.dir, base.tree, after.tree) : null
220    const treeError = after.error ?? (diff && 'error' in diff ? diff.error : undefined)
221    const quota = (await snapshot($))[input.backend]
222    await track($, id, { status: result.ok ? 'done' : 'failed', endedAt: now, last: result.ok ? 'finished' : `stopped: ${result.error}` })
223    ended = true
224    const { stopped: _, ...parsed } = result
225    return {
226      ...parsed,
227      attempts,
228      baseTree: base.tree,
229      tree: after.tree,
230      filesChanged: diff && 'files' in diff ? diff.files : null,
231      quota,
232      ...(treeError ? { treeError } : {}),
233    }
234  } catch (err) {
235    thrown = err
236    throw err
237  } finally {
238    if (!ended) {
239      const why = thrown instanceof Error ? thrown.message : signal?.aborted ? 'interrupted' : String(thrown ?? 'unknown')
240      const at = await $.clock.now().catch(() => startedAt)
241      await track($, id, { status: 'failed', endedAt: at, last: `stopped: ${why}` }).catch(() => {})
242    }
243  }
244}
245
246async function routingMarkdown($: EngineInterface): Promise<string> {
247  const path = `${(await $.env.get('HOME')) ?? ''}/.claude/skills/cross-model-dev/routing.md`
248  return (await $.fs.exists(path).catch(() => false)) ? await $.fs.read(path) : ''
249}
250
251async function route($: EngineInterface, input: RouteInput) {
252  const orchestrator = await read($, effort)
253  const markdown = await routingMarkdown($)
254  const { config, note } = readConfig(markdown)
255  const snap = await snapshot($)
256  const now = await $.clock.now()
257  const quota = describe(snap, now)
258  // The router gets no repository: an empty directory, and only the task text in its prompt.
259  const dir = `${(await $.env.get('HOME')) ?? ''}/.cache/xagent/router`
260  const schemaPath = `${dir}/route-schema.json`
261  await $.fs.write(schemaPath, JSON.stringify(ROUTE_SCHEMA))
262  const r = await runDispatch($, {
263    ...config.router,
264    mode: 'read-only',
265    ephemeral: true,
266    override: input.override ?? false,
267    dir,
268    schema: schemaPath,
269    prompt: routerPrompt(input.task, input.files ?? [], orchestrator, quota, markdown, input.instructions ?? ''),
270  }, `route: ${firstLine(input.task, 60)}`)
271  const proposed = r.ok && r.answer ? parsePick(r.answer, orchestrator, config) : { ...defaults(orchestrator, config), why: `router failed (${r.error ?? 'no answer'}); defaults` }
272  const pick = clamp(note ? { ...proposed, why: `${proposed.why}; ${note}` } : proposed, orchestrator, snap, now, config, input.override ?? false)
273  return { result: JSON.stringify({ pick, snapshot: quota, effort: orchestrator }) }
274}
275
276export const register: Register = on => {
277  on('session.start', async ($, e, next) => {
278    const r = await next(e)
279    await $.tool.register({ name: 'dispatch', description: 'Run one task on an external coding agent (codex, copilot, agy). Returns a short summary, the files it changed and a sessionId for resume.', inputSchema: DISPATCH_SCHEMA }).catch(() => {})
280    await $.tool.register({ name: 'route', description: 'Pick implementer and reviewer models and efforts for one plan task, from quota, complexity and the current effort.', inputSchema: ROUTE_SCHEMA_IN }).catch(() => {})
281    return r
282  })
283
284  // A gating hook answers even when something in it throws.
285  on('tool.call', { tool: 'mcp__xagent__dispatch' }, async ($, e, next) => {
286    const input: DispatchInput = e
287    return runDispatch($, input, firstLine(input.prompt, 80), next.signal)
288      .then(r => ({ result: JSON.stringify(r) }))
289      .catch((err: Error) => ({ deny: `xagent dispatch failed: ${err.message}` }))
290  })
291
292  // The transcript row echoes the whole input, prompt included; draw the prompt's first line only.
293  on('ui.render', { component: 'ToolUse' }, async ($, e, next) => {
294    if (e.props.tool !== 'mcp__xagent__dispatch') return next(e)
295    const input = (e.props.input ?? {}) as Record<string, unknown>
296    const prompt = typeof input.prompt === 'string' ? firstLine(input.prompt, 80) : input.prompt
297    return next({ ...e, props: { ...e.props, input: { ...input, prompt } } })
298  })
299
300  // The orchestrator's effort, from its own model requests (subagents carry an agentId).
301  on('turn.step', async function* ($, e, next) {
302    if (!e.agentId && typeof e.effort === 'string') await update($, effort, () => e.effort as string).catch(() => {})
303    return yield* next(e)
304  })
305
306  on('tool.call', { tool: 'mcp__xagent__route' }, async ($, e) => {
307    const input: RouteInput = e
308    return route($, input).catch((err: Error) => ({ deny: `xagent route failed: ${err.message}` }))
309  })
310}
311

1// The command line for one dispatch. No shell: each element is one argv entry.
2import type { DispatchInput } from '../types'
3
4const COPILOT_CREDIT_CAP = '30' // the CLI's minimum; one run cannot eat the month
5
6export function buildArgv(i: DispatchInput): string[] {
7  if (i.backend === 'codex') {
8    // `exec resume` takes no -s or -C, so the sandbox goes through -c and the directory through spawn's cwd.
9    const argv = ['codex', 'exec', ...(i.resume ? ['resume'] : []), '--json', '--skip-git-repo-check', '-c', `sandbox_mode="${i.mode === 'write' ? 'workspace-write' : 'read-only'}"`]
10    if (i.model) argv.push('-m', i.model)
11    if (i.effort) argv.push('-c', `model_reasoning_effort=${i.effort}`)
12    if (i.schema) argv.push('--output-schema', i.schema)
13    if (i.ephemeral) argv.push('--ephemeral')
14    // `--` ends the options: a prompt starting with `-` (a pasted "- [ ] Step") is not a flag.
15    if (i.resume) return [...argv, '--', i.resume, i.prompt]
16    if (i.dir) argv.push('-C', i.dir)
17    return [...argv, '--', i.prompt]
18  }
19  if (i.backend === 'copilot') {
20    if (i.mode === 'read-only') throw new Error('copilot has no read-only mode')
21    const argv = ['copilot', '-p', i.prompt, '--output-format', 'json', '--allow-all-tools']
22    if (i.effort) argv.push('--reasoning-effort', i.effort)
23    if (i.resume) argv.push('--resume', i.resume)
24    return [...argv, '--max-ai-credits', COPILOT_CREDIT_CAP]
25  }
26  const argv = ['agy', '-p', i.prompt, '--output-format', 'stream-json', '--mode', i.mode === 'write' ? 'accept-edits' : 'plan']
27  if (i.mode === 'write') argv.push('--sandbox', '--dangerously-skip-permissions')
28  if (i.model) argv.push('--model', i.model)
29  if (i.effort) argv.push('--effort', i.effort)
30  if (i.schema) argv.push('--json-schema', i.schema)
31  if (i.resume) argv.push('--conversation', i.resume)
32  return argv
33}
34

1// What xagent reads from the host: git trees of a working directory, and each provider's quota.
2import { atom, read } from 'claude-code'
3import type { EngineInterface } from 'claude-code'
4
5import type { ProviderQuota, Snapshot } from '../types'
6import { claudeFromUsage, codexFromRollout, copilotFromNanoAiu, copilotMonthStart, merge } from './quota'
7
8const blocked = atom({ plugin: 'xagent', key: 'blocked' } as const, {} as Record<string, ProviderQuota>)
9
10export type Tree = { tree: string | null; error?: string }
11
12async function home($: EngineInterface) {
13  return (await $.env.get('HOME')) ?? ''
14}
15
16const lastErr = (stderr: string) => stderr.trim().split('\n').slice(-1)[0] ?? ''
17
18// The working tree, untracked files included, as a git tree object. A scratch index keeps the
19// user's own index untouched. `tree: null` with no error: not a git repo.
20export async function treeOf($: EngineInterface, dir?: string): Promise<Tree> {
21  if (!dir) return { tree: null }
22  const inside = await $.process.run(['git', '-C', dir, 'rev-parse', '--is-inside-work-tree'], { timeoutMs: 10_000 }).catch(() => null)
23  if (!inside || inside.exitCode !== 0) return { tree: null }
24  const scratch = `${await home($)}/.cache/xagent`
25  const index = `${scratch}/index-${await $.clock.now()}-${Math.random().toString(36).slice(2, 8)}`
26  try {
27    const made = await $.process.run(['mkdir', '-p', scratch], { timeoutMs: 5_000 })
28    if (made.exitCode !== 0) return { tree: null, error: `mkdir ${scratch}: ${lastErr(made.stderr)}` }
29    const git = (args: string[]) => $.process.run(['git', '-C', dir, ...args], { env: { GIT_INDEX_FILE: index }, timeoutMs: 60_000 })
30    if ((await git(['read-tree', 'HEAD'])).exitCode !== 0) {
31      const empty = await git(['read-tree', '--empty']) // a repo with no commits yet
32      if (empty.exitCode !== 0) return { tree: null, error: `git read-tree: ${lastErr(empty.stderr)}` }
33    }
34    const added = await git(['add', '-A'])
35    if (added.exitCode !== 0) return { tree: null, error: `git add: ${lastErr(added.stderr)}` }
36    const tree = await git(['write-tree'])
37    if (tree.exitCode !== 0 || !tree.stdout.trim()) return { tree: null, error: `git write-tree: ${lastErr(tree.stderr)}` }
38    return { tree: tree.stdout.trim() }
39  } catch (err) {
40    return { tree: null, error: (err as Error).message }
41  } finally {
42    await $.process.run(['rm', '-f', index], { timeoutMs: 5_000 }).catch(() => null)
43  }
44}
45
46// NUL-separated names, so paths with spaces or non-ASCII come back as they are.
47export async function changedFiles($: EngineInterface, dir: string, a: string, b: string): Promise<{ files: string[] } | { error: string }> {
48  const r = await $.process.run(['git', '-C', dir, 'diff', '--name-only', '-z', a, b], { timeoutMs: 30_000 }).catch(() => null)
49  if (!r || r.exitCode !== 0) return { error: `git diff: ${lastErr(r?.stderr ?? 'did not run')}` }
50  if (r.isStdoutTruncated) return { error: 'git diff: file list too long' }
51  return { files: r.stdout.split('\0').filter(Boolean) }
52}
53
54// The rollout written most recently (a resumed old session counts), read from its end: the first
55// rate_limits sample found is the newest. Up to three files, for one with no sample yet.
56async function codexQuota($: EngineInterface) {
57  const sessions = `${await home($)}/.codex/sessions`
58  const found = await $.process.run(['find', sessions, '-name', 'rollout-*.jsonl', '-mtime', '-8', '-printf', '%T@ %p\\n'], { timeoutMs: 10_000 }).catch(() => null)
59  if (!found || found.exitCode !== 0) return codexFromRollout('')
60  const newest = found.stdout
61    .split('\n')
62    .map(l => /^(\S+) (.+)$/.exec(l))
63    .flatMap(m => (m?.[1] && m[2] ? [{ at: Number(m[1]), path: m[2] }] : []))
64    .sort((x, y) => y.at - x.at)
65    .slice(0, 3)
66  for (const f of newest) {
67    // tac puts the end first; process.run keeps the first 4 MB, so that is the file's last 4 MB.
68    const r = await $.process.run(['tac', f.path], { timeoutMs: 10_000 }).catch(() => null)
69    if (!r || r.exitCode !== 0) continue
70    const q = codexFromRollout(r.stdout.split('\n').reverse().join('\n'))
71    if (q.state !== 'unknown') return q
72  }
73  return codexFromRollout('')
74}
75
76// Copilot's own usage store: every CLI request on this machine, interactive ones included.
77async function copilotNanoAiu($: EngineInterface, now: number): Promise<number | null> {
78  const db = `${await home($)}/.copilot/session-store.db`
79  if (!(await $.fs.exists(db).catch(() => false))) return null
80  const sql = `select coalesce(sum(total_nano_aiu), 0) from assistant_usage_events where created_at >= '${copilotMonthStart(now)}'`
81  const r = await $.process.run(['sqlite3', '-readonly', db, sql], { timeoutMs: 5_000 }).catch(() => null)
82  if (!r || r.exitCode !== 0) return null
83  const n = Number(r.stdout.trim())
84  return Number.isFinite(n) ? n : null
85}
86
87export async function snapshot($: EngineInterface): Promise<Snapshot> {
88  const now = await $.clock.now()
89  const marks = await read($, blocked)
90  const usage = await $.session.usage().catch(() => null)
91  return {
92    claude: claudeFromUsage(usage?.rateLimits ?? []),
93    codex: merge(marks.codex, await codexQuota($), now),
94    copilot: merge(marks.copilot, copilotFromNanoAiu(await copilotNanoAiu($, now), now), now),
95    agy: merge(marks.agy, { state: 'unknown' }, now),
96  }
97}
98

1// Reads each CLI's JSON lines into one result, and one line of status for the radar.
2import type { Backend, Parsed } from '../types'
3
4export const SUMMARY_CAP = 4000
5export const STRUCTURED_CAP = 200_000
6export const KEEP_CAP = 2 * STRUCTURED_CAP // what feed() keeps of a run's stdout
7
8type J = Record<string, any>
9
10function lines(stdout: string): J[] {
11  const out: J[] = []
12  for (const l of stdout.split('\n')) {
13    try {
14      const j = JSON.parse(l)
15      if (j && typeof j === 'object') out.push(j)
16    } catch {}
17  }
18  return out
19}
20
21export function firstLine(text: string, n: number): string {
22  return (text.split('\n')[0] ?? '').slice(0, n)
23}
24
25function done(answer: string | null, sessionId: string | null, error: string | undefined, structured: boolean): Parsed {
26  const summary = (answer ?? '').trim().slice(0, SUMMARY_CAP)
27  if (error) return { ok: false, summary, sessionId, error }
28  if (answer === null) return { ok: false, summary: '', sessionId, error: 'no answer in output' }
29  if (!structured) return { ok: true, summary, sessionId }
30  // A schema answer goes back whole: a cut one is not JSON and would break the plan-review loop.
31  const whole = answer.trim()
32  if (whole.length > STRUCTURED_CAP) return { ok: false, summary, sessionId, error: 'answer too long' }
33  try {
34    JSON.parse(whole)
35  } catch {
36    return { ok: false, summary, sessionId, error: 'answer is not valid JSON' }
37  }
38  return { ok: true, summary, answer: whole, sessionId }
39}
40
41export function sessionIdIn(backend: Backend, stdout: string): string | null {
42  let id: string | null = null
43  for (const j of lines(stdout)) {
44    if (backend === 'codex' && j.type === 'thread.started') id = j.thread_id ?? null
45    if (backend === 'agy' && j.event === 'init') id = j.conversation_id ?? null
46    if (backend === 'copilot' && j.type === 'session.start') id = j.data?.sessionId ?? null
47  }
48  return id
49}
50
51export function parse(backend: Backend, stdout: string, structured: boolean): Parsed {
52  let answer: string | null = null
53  let error: string | undefined
54  const sessionId = sessionIdIn(backend, stdout)
55  if (backend === 'codex') {
56    for (const j of lines(stdout)) {
57      if (j.type === 'item.completed' && j.item?.type === 'agent_message') answer = String(j.item.text ?? '')
58      if (j.type === 'turn.failed' || j.type === 'error') error = j.error?.message ?? j.message ?? 'codex failed'
59    }
60    return done(answer, sessionId, error, structured)
61  }
62  if (backend === 'agy') {
63    let streamed = '' // agy's result.response can be empty after tool use (seen 2026-10-14)
64    for (const j of lines(stdout)) {
65      if (j.event === 'step_update' && j.step_update?.step_type === 'agent_response') streamed += j.step_update.text_delta ?? ''
66      if (j.event === 'result') {
67        answer = j.result?.response || streamed
68        if (j.result?.status !== 'SUCCESS') error = `agy ${j.result?.status ?? 'failed'}`
69        const denied = j.result?.denied_actions
70        if (Array.isArray(denied) && denied.length > 0) {
71          error = `agy denied: ${denied.map((d: J) => d?.display_name ?? d?.action ?? 'unknown').join(', ')}`
72        }
73        if (!String(answer ?? '').trim()) answer = null
74      }
75    }
76    return done(answer, sessionId, error, structured)
77  }
78  for (const j of lines(stdout)) {
79    if (j.type === 'assistant.message' && j.data?.content) answer = String(j.data.content)
80    if (j.type === 'session.error') error = j.data?.message ?? 'copilot failed'
81  }
82  return done(answer, sessionId, error, structured)
83}
84
85// The lines parse() and sessionIdIn() read. Command output and reasoning are dropped as they stream.
86function wanted(backend: Backend, j: J): boolean {
87  if (backend === 'codex') return j.type === 'thread.started' || j.type === 'turn.failed' || j.type === 'error' || (j.type === 'item.completed' && j.item?.type === 'agent_message')
88  if (backend === 'agy') return j.event === 'init' || j.event === 'result' || (j.event === 'step_update' && j.step_update?.step_type === 'agent_response')
89  return j.type === 'session.start' || j.type === 'assistant.message' || j.type === 'session.error'
90}
91
92// Takes stdout as it streams. An event can arrive split across chunks, so lines are put back
93// together first. Only wanted lines are kept, within KEEP_CAP: the oldest go first, except the
94// first, which carries the session id.
95export function feed(backend: Backend) {
96  let partial = ''
97  let size = 0
98  const kept: string[] = []
99  const take = (line: string): string | null => {
100    let j: J
101    try {
102      j = JSON.parse(line)
103    } catch {
104      return null
105    }
106    if (!j || typeof j !== 'object') return null
107    if (wanted(backend, j)) {
108      kept.push(line)
109      size += line.length + 1
110      while (size > KEEP_CAP && kept.length > 1) size -= (kept.splice(1, 1)[0] ?? '').length + 1
111    }
112    return lastLine(backend, line)
113  }
114  return {
115    // The newest radar status in this chunk, if any.
116    push(text: string): string | null {
117      const parts = (partial + text).split('\n')
118      partial = parts.pop() ?? ''
119      if (partial.length > KEEP_CAP) partial = '' // a line that long is never a usable answer
120      let status: string | null = null
121      for (const l of parts) status = take(l) ?? status
122      return status
123    },
124    // What parse() should read, including a last line that had no newline.
125    text(): string {
126      if (partial) take(partial)
127      partial = ''
128      return kept.join('\n')
129    },
130  }
131}
132
133export function lastLine(backend: Backend, line: string): string | null {
134  let j: J
135  try {
136    j = JSON.parse(line)
137  } catch {
138    return null
139  }
140  if (backend === 'codex') {
141    if (j.item?.type === 'command_execution') return `running ${String(j.item.command).slice(0, 60)}`
142    if (j.item?.type === 'file_change') return 'editing files'
143    if (j.item?.type === 'agent_message') return 'writing'
144    if (j.type === 'turn.started') return 'thinking'
145    return null
146  }
147  if (backend === 'agy') {
148    const s = j.step_update
149    if (!s) return null
150    return s.step_type === 'agent_response' ? 'writing' : String(s.step_type).replace(/_/g, ' ')
151  }
152  if (j.type === 'assistant.message') return 'writing'
153  if (typeof j.type === 'string' && j.type.startsWith('tool.')) return j.type.replace('tool.', 'tool: ')
154  return null
155}
156

1// Quota per provider: windows of used %, pace against elapsed time, and the 90 % stop.
2// A window past its reset and a cooldown past its end are old news: they never block.
3import type { ProviderQuota, Snapshot, Window } from '../types'
4
5export const BLOCK_AT = 90
6export const COPILOT_MONTHLY_CREDITS = 200
7export const NANO_AIU_PER_CREDIT = 1e9 // unverified; Task 11 checks it against copilot /usage
8const KIND_MINUTES: Record<string, number> = { five_hour: 300, seven_day: 10080 }
9
10const current = (w: Window, now: number) => w.resetsAt === null || w.resetsAt > now
11
12export function pace(w: Window, now: number): number | null {
13  if (w.resetsAt === null || w.resetsAt <= now) return null
14  const total = w.windowMinutes * 60_000
15  const elapsed = Math.min(Math.max(total - (w.resetsAt - now), 0), total)
16  return w.usedPercent - (elapsed / total) * 100
17}
18
19export function isBlocked(q: ProviderQuota, now: number): boolean {
20  if (q.state === 'unavailable') return true
21  if (q.state === 'cooldown') return q.until > now
22  if (q.state === 'windows') return q.windows.some(w => current(w, now) && w.usedPercent >= BLOCK_AT)
23  return false
24}
25
26export function merge(mark: ProviderQuota | undefined, live: ProviderQuota, now: number): ProviderQuota {
27  if (mark?.state === 'unavailable') return mark
28  if (mark?.state === 'cooldown' && mark.until > now) return mark
29  return live
30}
31
32export function codexFromRollout(text: string): ProviderQuota {
33  let last: any = null
34  for (const l of text.split('\n')) {
35    if (!l.includes('rate_limits')) continue
36    try {
37      const j = JSON.parse(l)
38      if (j.payload?.rate_limits) last = j.payload.rate_limits
39    } catch {}
40  }
41  if (!last) return { state: 'unknown' }
42  const windows: Window[] = [last.primary, last.secondary]
43    .filter((w: any) => w && typeof w.used_percent === 'number')
44    .map((w: any) => ({ usedPercent: w.used_percent, windowMinutes: w.window_minutes ?? 300, resetsAt: typeof w.resets_at === 'number' ? w.resets_at * 1000 : null }))
45  return windows.length ? { state: 'windows', windows } : { state: 'unknown' }
46}
47
48export function claudeFromUsage(limits: { kind: string; percentUsed: number; resetsAt?: string }[]): ProviderQuota {
49  if (!limits.length) return { state: 'unknown' }
50  return {
51    state: 'windows',
52    windows: limits.map(l => ({ usedPercent: l.percentUsed, windowMinutes: KIND_MINUTES[l.kind] ?? 300, resetsAt: l.resetsAt ? Date.parse(l.resetsAt) : null })),
53  }
54}
55
56export function copilotMonthStart(now: number): string {
57  return `${new Date(now).toISOString().slice(0, 7)}-01 00:00:00`
58}
59
60export function copilotFromNanoAiu(nano: number | null, now: number): ProviderQuota {
61  if (nano === null) return { state: 'unknown' }
62  const d = new Date(now)
63  const start = Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), 1)
64  const resetsAt = Date.UTC(d.getUTCFullYear(), d.getUTCMonth() + 1, 1)
65  const usedPercent = (nano / NANO_AIU_PER_CREDIT / COPILOT_MONTHLY_CREDITS) * 100
66  return { state: 'windows', windows: [{ usedPercent, windowMinutes: (resetsAt - start) / 60_000, resetsAt }] }
67}
68
69function windowLine(name: string, w: Window, now: number): string {
70  if (w.resetsAt !== null && w.resetsAt <= now) return `${name}: ${w.windowMinutes}m window reset passed`
71  const p = pace(w, now)
72  const lead = p === null ? 'reset time unknown' : p >= 0 ? `${Math.round(p)} ahead of pace` : `${Math.round(-p)} behind pace`
73  const resets = w.resetsAt === null ? '' : `, resets in ${Math.round((w.resetsAt - now) / 60_000)}m`
74  return `${name}: ${w.windowMinutes}m window ${Math.round(w.usedPercent)}% used, ${lead}${resets}`
75}
76
77export function describeOne(name: string, q: ProviderQuota, now: number): string {
78  if (q.state === 'unknown') return `${name}: unknown`
79  if (q.state === 'unavailable') return `${name}: unavailable (${q.reason})`
80  if (q.state === 'cooldown') return q.until > now ? `${name}: cooling down for ${Math.round((q.until - now) / 60_000)}m` : `${name}: unknown`
81  return q.windows.map(w => windowLine(name, w, now)).join('\n')
82}
83
84export function describe(s: Snapshot, now: number): string {
85  return (['claude', 'codex', 'copilot', 'agy'] as const).map(k => describeOne(k, s[k], now)).join('\n')
86}
87

1// The routing policy: orchestrator effort sets a band, complexity picks inside it, quota breaks
2// ties. The router model proposes; clamp() keeps the proposal inside the band. routing.md's
3// json block can change the router and the bands; its prose only informs the router.
4import type { Backend, Effort, Snapshot } from '../types'
5import { isBlocked } from './quota'
6
7export type Pick = {
8  complexity: 'trivial' | 'standard' | 'hard'
9  implementer: { backend: Backend; model: string; effort: Effort }
10  reviewer: { model: 'sonnet' | 'haiku'; effort: Effort }
11  reviewDepth: 'spec' | 'spec+quality' | 'adversarial'
12  secondSolReview: boolean
13  wait: string | null // set: the task must not start; the skill asks the user
14  why: string
15}
16
17export type Band = { impl: [Effort, Effort]; review: [Effort, Effort]; haiku: boolean }
18export type RoutingConfig = { router: { backend: Backend; model: string; effort: Effort }; bands: Record<Effort, Band> }
19
20const ORDER: Effort[] = ['low', 'medium', 'high', 'xhigh', 'max']
21const BACKENDS: Backend[] = ['codex', 'copilot', 'agy']
22const ROUTERS: Backend[] = ['codex', 'agy'] // the router runs read-only; copilot has no read-only mode
23const COMPLEXITY: Pick['complexity'][] = ['trivial', 'standard', 'hard']
24const DEPTHS: Pick['reviewDepth'][] = ['spec', 'spec+quality', 'adversarial']
25const REVIEWERS: Pick['reviewer']['model'][] = ['sonnet', 'haiku']
26
27const isEffort = (x: unknown): x is Effort => ORDER.includes(x as Effort)
28const rank = (e: string) => Math.max(0, ORDER.indexOf(e as Effort))
29const within = (e: Effort, [lo, hi]: [Effort, Effort]): Effort => ORDER[Math.min(Math.max(rank(e), rank(lo)), rank(hi))] ?? lo
30
31const LOW: Band = { impl: ['low', 'medium'], review: ['low', 'medium'], haiku: true }
32const HIGH: Band = { impl: ['medium', 'high'], review: ['medium', 'max'], haiku: false }
33const TOP: Band = { impl: ['high', 'xhigh'], review: ['high', 'max'], haiku: false }
34
35export const DEFAULT_CONFIG: RoutingConfig = {
36  router: { backend: 'codex', model: 'gpt-6-luna', effort: 'low' },
37  bands: { low: LOW, medium: LOW, high: HIGH, xhigh: TOP, max: TOP },
38}
39
40// A [low, high] effort range, low first.
41function isRange(r: unknown): r is [Effort, Effort] {
42  return Array.isArray(r) && r.length === 2 && isEffort(r[0]) && isEffort(r[1]) && rank(r[0]) <= rank(r[1])
43}
44
45function isBand(b: any): b is Band {
46  return b && isRange(b.impl) && isRange(b.review) && typeof b.haiku === 'boolean'
47}
48
49export function readConfig(markdown: string): { config: RoutingConfig; note: string | null } {
50  const block = /```json\n([\s\S]*?)\n```/.exec(markdown)?.[1]
51  if (!block) return { config: DEFAULT_CONFIG, note: null }
52  try {
53    const j = JSON.parse(block)
54    const router = j.router ?? DEFAULT_CONFIG.router
55    if (!ROUTERS.includes(router.backend) || typeof router.model !== 'string' || !router.model || !isEffort(router.effort)) throw new Error('router')
56    const bands = { ...DEFAULT_CONFIG.bands }
57    for (const [k, b] of Object.entries(j.bands ?? {})) {
58      if (!isEffort(k) || !isBand(b)) throw new Error(`band ${k}`)
59      bands[k] = b
60    }
61    return { config: { router, bands }, note: null }
62  } catch (err) {
63    return { config: DEFAULT_CONFIG, note: `routing.md config invalid (${(err as Error).message}); defaults` }
64  }
65}
66
67export function bandFor(orchestrator: string, c: RoutingConfig): Band {
68  return c.bands[isEffort(orchestrator) ? orchestrator : 'medium']
69}
70
71// Fallback implementers when the preferred backend is blocked, in order.
72const FALLBACK: { backend: Backend; model: string }[] = [
73  { backend: 'codex', model: 'gpt-6.1-sol' },
74  { backend: 'agy', model: 'gemini-3.8-flash-high' },
75  { backend: 'copilot', model: 'auto' },
76]
77
78export function defaults(orchestrator: string, c: RoutingConfig, complexity: Pick['complexity'] = 'standard'): Pick {
79  const b = bandFor(orchestrator, c)
80  return {
81    complexity,
82    implementer: { backend: 'codex', model: 'gpt-6.1-sol', effort: complexity === 'hard' ? b.impl[1] : b.impl[0] },
83    reviewer: { model: b.haiku && complexity === 'trivial' ? 'haiku' : 'sonnet', effort: b.review[0] },
84    reviewDepth: complexity === 'trivial' ? 'spec' : complexity === 'hard' ? 'adversarial' : 'spec+quality',
85    secondSolReview: false,
86    wait: null,
87    why: 'defaults',
88  }
89}
90
91// Keeps a pick inside the band and applies the 90 % rule. `override`: the user allows providers
92// at 90 %, so the pick stands as it is.
93export function clamp(p: Pick, orchestrator: string, snap: Snapshot, now: number, c: RoutingConfig, override = false): Pick {
94  const b = bandFor(orchestrator, c)
95  const notes: string[] = []
96  const waits: string[] = []
97  const blocked = (backend: keyof Snapshot) => !override && isBlocked(snap[backend], now)
98  let implementer = { ...p.implementer, effort: within(p.implementer.effort, b.impl) }
99  if (blocked(implementer.backend)) {
100    notes.push(`${implementer.backend} blocked`)
101    const open = FALLBACK.find(f => f.backend !== implementer.backend && !blocked(f.backend))
102    if (open) implementer = { ...open, effort: implementer.effort }
103    else waits.push(`every implementer is at 90 % or unavailable (${[implementer.backend, ...FALLBACK.map(f => f.backend).filter(x => x !== implementer.backend)].join(', ')})`)
104  }
105  if (implementer.backend === 'copilot') implementer = { ...implementer, model: 'auto' } // Student plan: auto only
106  const reviewer = { model: p.reviewer.model === 'haiku' && !b.haiku ? ('sonnet' as const) : p.reviewer.model, effort: within(p.reviewer.effort, b.review) }
107  if (blocked('claude')) waits.push('claude is at 90 %: the reviewer cannot run')
108  // A required second Sol review is never dropped: with codex at 90 % the task waits.
109  const secondSolReview = rank(orchestrator) >= 3 && p.complexity === 'hard'
110  if (secondSolReview && blocked('codex')) waits.push('codex is at 90 %: the second Sol review cannot run')
111  if (override) notes.push('user override: 90 % rule off')
112  return { ...p, implementer, reviewer, secondSolReview, wait: waits.length ? waits.join('; ') : null, why: [p.why, ...notes].join('; ') }
113}
114
115export const ROUTE_SCHEMA = {
116  type: 'object',
117  additionalProperties: false,
118  required: ['complexity', 'implementer', 'reviewer', 'reviewDepth', 'why'],
119  properties: {
120    complexity: { type: 'string', enum: COMPLEXITY },
121    implementer: {
122      type: 'object', additionalProperties: false, required: ['backend', 'model', 'effort'],
123      properties: { backend: { type: 'string', enum: BACKENDS }, model: { type: 'string' }, effort: { type: 'string', enum: ORDER } },
124    },
125    reviewer: {
126      type: 'object', additionalProperties: false, required: ['model', 'effort'],
127      properties: { model: { type: 'string', enum: REVIEWERS }, effort: { type: 'string', enum: ORDER } },
128    },
129    reviewDepth: { type: 'string', enum: DEPTHS },
130    why: { type: 'string' },
131  },
132}
133
134export function routerPrompt(task: string, files: string[], orchestrator: string, quota: string, routing: string, instructions: string): string {
135  return [
136    'You route one coding task to models. Answer only with JSON matching the schema. You have no repository; judge from the text below.',
137    'Rate complexity from the task alone. Prefer the cheapest pick that fits; move work away from providers ahead of pace and toward ones behind pace near reset.',
138    `Orchestrator effort: ${orchestrator}`,
139    `Quota now:\n${quota}`,
140    routing ? `Routing table:\n${routing}` : '',
141    instructions ? `User instructions (these win):\n${instructions}` : '',
142    `Files: ${files.join(', ') || 'not listed'}`,
143    `Task:\n${task}`,
144  ].filter(Boolean).join('\n\n')
145}
146
147// The router's answer, checked field by field; anything off falls back to the band defaults.
148export function parsePick(text: string, orchestrator: string, c: RoutingConfig): Pick {
149  try {
150    const j = JSON.parse(text)
151    const ok =
152      COMPLEXITY.includes(j.complexity) &&
153      BACKENDS.includes(j.implementer?.backend) &&
154      typeof j.implementer?.model === 'string' && j.implementer.model.length > 0 &&
155      isEffort(j.implementer?.effort) &&
156      REVIEWERS.includes(j.reviewer?.model) &&
157      isEffort(j.reviewer?.effort) &&
158      DEPTHS.includes(j.reviewDepth) &&
159      typeof j.why === 'string'
160    if (!ok) throw new Error('fields')
161    return {
162      complexity: j.complexity,
163      implementer: { backend: j.implementer.backend, model: j.implementer.model, effort: j.implementer.effort },
164      reviewer: { model: j.reviewer.model, effort: j.reviewer.effort },
165      reviewDepth: j.reviewDepth,
166      secondSolReview: false,
167      wait: null,
168      why: j.why,
169    }
170  } catch {
171    return { ...defaults(orchestrator, c), why: 'router output unusable; defaults' }
172  }
173}
174

1export type Backend = 'codex' | 'copilot' | 'agy'
2export type Mode = 'read-only' | 'write'
3export type Effort = 'low' | 'medium' | 'high' | 'xhigh' | 'max'
4
5export type DispatchInput = {
6  backend: Backend
7  model?: string
8  effort?: Effort
9  mode: Mode
10  prompt: string
11  dir?: string
12  schema?: string // path to a JSON schema file; the answer must parse as JSON
13  resume?: string // session id from an earlier result
14  timeoutMs?: number
15  ephemeral?: boolean // codex: keep no session (router calls)
16  override?: boolean // run even though the backend is at 90 % of a window: only when the user said so
17}
18
19export type RouteInput = {
20  task: string
21  files?: string[]
22  instructions?: string
23  override?: boolean // the user allows providers at 90 %
24}
25
26// What one CLI run's output says.
27export type Parsed = {
28  ok: boolean
29  summary: string // at most SUMMARY_CAP chars, for display
30  answer?: string // with a schema: the whole answer, valid JSON
31  sessionId: string | null
32  error?: string
33}
34
35export type DispatchResult = Parsed & {
36  attempts: number
37  baseTree: string | null // git tree of `dir` before the run; null outside a repo or read-only
38  tree: string | null // git tree after the run
39  filesChanged: string[] | null // paths that differ between baseTree and tree
40  treeError?: string // a git snapshot failed inside a repo (filesChanged is then null)
41  quota: ProviderQuota // this backend's quota, read after the run
42}
43
44export type XAgent = {
45  id: string
46  backend: Backend
47  model: string
48  effort: string
49  task: string
50  status: 'running' | 'done' | 'failed'
51  startedAt: number
52  endedAt?: number
53  last: string
54  quota: string // the backend's quota line when it started
55}
56
57export type Window = { usedPercent: number; windowMinutes: number; resetsAt: number | null } // resetsAt: epoch ms, null when unknown
58export type ProviderQuota =
59  | { state: 'unknown' }
60  | { state: 'windows'; windows: Window[] }
61  | { state: 'cooldown'; until: number }
62  | { state: 'unavailable'; reason: string }
63export type Snapshot = { claude: ProviderQuota; codex: ProviderQuota; copilot: ProviderQuota; agy: ProviderQuota }
64
65declare module 'claude-code' {
66  interface PluginState {
67    xagent: { agents: XAgent[]; effort: string; blocked: Record<string, ProviderQuota> }
68  }
69  // Written by hand: the generated MCP types only know servers that were connected when they were made.
70  interface McpToolInputs {
71    mcp__xagent__dispatch: DispatchInput
72    mcp__xagent__route: RouteInput
73  }
74}
75