cc-orq

1import type { Register } from 'claude-code'
2import { registerCard, startCard, stopCard } from './card.ts'
3import { registerEngine, registerOrqCommand } from './commands.ts'
4import { loadConfig } from './config.ts'
5import { registerPanel, registerPanelCommand, stopPanel } from './panel.ts'
6import { registerTask, registerTaskSurface } from './task.ts'
7import { logUnfinished } from './tools.ts'
8
9export const register: Register = (on, options) => {
10  registerEngine(on)
11  registerPanel(on)
12  registerCard(on, options)
13  registerTask(on)
14  on('session.start', async ($, e, next) => {
15    try {
16      const { config, source, warning } = await loadConfig($)
17      if (warning) await $.ui.log(`orq: ${warning}`, { to: 'debug' })
18      await $.ui.log(`orq loaded: config ${source}, jobsDir ${config.jobsDir}, ${config.instances.length} instances`, { to: 'debug' })
19    } catch (err) {
20      await $.ui.log(`orq: session.start failed: ${String(err)}`, { to: 'debug' })
21    }
22    try {
23      await registerOrqCommand($)
24    } catch (err) {
25      await $.ui.log(`orq: could not register /orq: ${String(err)}`, { to: 'debug' })
26    }
27    try {
28      await registerPanelCommand($)
29    } catch (err) {
30      await $.ui.log(`orq: could not register /orq-panel: ${String(err)}`, { to: 'debug' })
31    }
32    try {
33      startCard($)
34    } catch (err) {
35      await $.ui.log(`orq: could not start the task card: ${String(err)}`, { to: 'debug' })
36    }
37    try {
38      await registerTaskSurface($)
39      await logUnfinished($, (await loadConfig($)).config)
40    } catch (err) {
41      await $.ui.log(`orq: could not register cc-orq:task and log the unfinished jobs: ${String(err)}`, { to: 'debug' })
42    }
43    return next(e)
44  })
45  on('session.end', async ($, e, next) => {
46    await stopCard($, e)
47    stopPanel(e)
48    return next(e)
49  })
50}
51

1import type { EngineInterface, On, PluginOptions, RenderElement, RenderInput, SessionEndInput, Timer } from 'claude-code'
2
3import { loadConfig } from './config.ts'
4import { listJobs, type Engine } from './engine.ts'
5import { renderTail } from './event-line.ts'
6import { cachedJobView, tailEvents } from './panel.ts'
7import type { OrqCardRow } from '../types/index'
8import { formatDuration, type JobRecord } from './types.ts'
9
10export const REFRESH_MS = 2000
11const ENDED_SHOWN_MS = 60000
12const STUCK_REFRESH_MS = 60000
13
14export type CardMode = 'band' | 'status' | 'off'
15
16export type CardRow = OrqCardRow
17
18type CardEngine = Engine & Pick<EngineInterface, 'state'>
19
20/** The rows, newest first. Session state survives a hot reload, and a render that reads it is drawn again when it changes. */
21const CARD = { plugin: 'cc-orq', key: 'card' } as const
22
23let mode: CardMode = 'band'
24/** Rows of jobs that ended, kept so an ended job is read once rather than on every tick. */
25let endedRowsById = new Map<string, CardRow>()
26let timer: Timer | null = null
27/** The refresh now running, and the clock time it began; null when none runs. */
28let refreshing: { startedAt: number } | null = null
29let statusShown = false
30/** True until the status line is cleared: a hot reload starts this module while the host may still show an old line. */
31let staleStatus = true
32/** Bumped when the card stops, so a refresh that began before the stop writes nothing. */
33let generation = 0
34
35export function displayMode(value: unknown): CardMode {
36  return value === 'status' || value === 'off' ? value : 'band'
37}
38
39export function cardLine(row: CardRow): string {
40  return `${row.instance} · ${row.model} · ${row.status} · ${row.lastAction} · ${formatDuration(row.durationMs)}`
41}
42
43async function debugLog($: Engine, what: string, err: unknown): Promise<void> {
44  try {
45    await $.ui.log(`orq-card: ${what}: ${String(err)}`, { to: 'debug' })
46  } catch {}
47}
48
49async function exitFileMtimeMs($: Engine, job: JobRecord): Promise<number | null> {
50  try {
51    return (await $.fs.stat(`${job.dir}/exit`)).mtimeMs
52  } catch {
53    return null
54  }
55}
56
57async function rowOf($: Engine, job: JobRecord, fallbackEndAt: number): Promise<CardRow> {
58  const view = await cachedJobView($, job)
59  const [lastAction] = renderTail(await tailEvents($, job.dir), 1)
60  return {
61    id: job.id,
62    instance: job.instanceId ?? job.runner ?? job.kind,
63    model: job.model ?? '?',
64    status: view.status,
65    lastAction: lastAction ?? '—',
66    durationMs: (view.endedAt ?? fallbackEndAt) - job.startedAt,
67  }
68}
69
70async function refresh($: CardEngine): Promise<void> {
71  let run: { startedAt: number } | null = null
72  try {
73    const now = await $.clock.now()
74    // A refresh still running after STUCK_REFRESH_MS is taken as stuck, so this tick starts another one.
75    if (refreshing && now - refreshing.startedAt < STUCK_REFRESH_MS) return
76    run = { startedAt: now }
77    refreshing = run
78    const started = generation
79    const { config } = await loadConfig($)
80    const nextRows: CardRow[] = []
81    const nextEnded = new Map<string, CardRow>()
82    for (const job of await listJobs($, config.jobsDir)) {
83      const exitAt = await exitFileMtimeMs($, job)
84      if (exitAt === null) {
85        nextRows.push(await rowOf($, job, now))
86      } else if (now - exitAt < ENDED_SHOWN_MS) {
87        const row = endedRowsById.get(job.id) ?? (await rowOf($, job, exitAt))
88        nextEnded.set(job.id, row)
89        nextRows.push(row)
90      }
91    }
92    if (started !== generation) return
93    endedRowsById = nextEnded
94    await show($, nextRows)
95  } catch (err) {
96    await debugLog($, 'refresh failed', err)
97  } finally {
98    // A stuck refresh that a newer one has replaced must not clear the newer one's mark.
99    if (run && refreshing === run) refreshing = null
100  }
101}
102
103/** Clear the status line. A failure goes to the debug log. */
104function clearStatus($: Engine): void {
105  try {
106    $.ui.status(undefined)
107  } catch (err) {
108    void debugLog($, 'clearing the status line failed', err)
109  }
110}
111
112async function show($: CardEngine, rows: CardRow[]): Promise<void> {
113  try {
114    if (mode === 'band') {
115      // An empty band before and after has nothing to write.
116      if (rows.length === 0 && ((await $.state.get(CARD)).value ?? []).length === 0) return
117      await $.state.set(CARD, rows)
118      return
119    }
120    if (mode !== 'status') return
121    const [newest] = rows
122    if (newest) {
123      const more = rows.length > 1 ? ` (+${rows.length - 1} more)` : ''
124      $.ui.status(`${cardLine(newest)}${more}`)
125      statusShown = true
126      staleStatus = false
127    } else if (statusShown || staleStatus) {
128      clearStatus($)
129      statusShown = false
130      staleStatus = false
131    }
132  } catch (err) {
133    void debugLog($, 'redraw failed', err)
134  }
135}
136
137function shouldDraw(e: RenderInput<'AbovePrompt'>): boolean {
138  if (mode !== 'band') return false
139  if (e.props.hasSurvey || e.props.maxRows < 1) return false
140  return e.surface === 'terminal' || e.surface === 'desktop'
141}
142
143/** The band's lines for `max` rows. With one row the count leads, since truncate-end cuts a long line at its end. */
144function bandLines(rows: CardRow[], max: number): { key: string; text: string; dim?: boolean }[] {
145  if (rows.length <= max) return rows.map((row) => ({ key: row.id, text: cardLine(row) }))
146  const [newest] = rows
147  if (max === 1 && newest) return [{ key: newest.id, text: `(+${rows.length - 1} more) ${cardLine(newest)}` }]
148  const shown = rows.slice(0, max - 1)
149  return [
150    ...shown.map((row) => ({ key: row.id, text: cardLine(row) })),
151    { key: 'card:more', text: `+${rows.length - shown.length} more`, dim: true },
152  ]
153}
154
155function drawBand($: Engine, e: RenderInput<'AbovePrompt'>, rows: CardRow[], rest: RenderElement): RenderElement {
156  const { Box, Text } = $.ui.resolve(e)
157  const children: RenderElement[] = bandLines(rows, e.props.maxRows).map((line) =>
158    Box({ key: line.key, children: [Text({ children: line.text, wrap: 'truncate-end', dimColor: line.dim })] }),
159  )
160  return Box({ flexDirection: 'column', children: [...children, rest] })
161}
162
163/** Empty the band's rows, which a session that drew the band may have left in state. */
164async function emptyBand($: CardEngine): Promise<void> {
165  try {
166    if (((await $.state.get(CARD)).value ?? []).length > 0) await $.state.set(CARD, [])
167  } catch (err) {
168    await debugLog($, 'clearing the card failed', err)
169  }
170}
171
172export function startCard($: CardEngine): void {
173  if (mode !== 'band') void emptyBand($)
174  if (mode === 'off' || timer) return
175  timer = $.clock.every(REFRESH_MS, () => {
176    void refresh($)
177  })
178}
179
180/** A /clear or a resume keeps the card: the engine sends no session.start after either. */
181export async function stopCard($: CardEngine, e: Pick<SessionEndInput, 'reason'>): Promise<void> {
182  if (e.reason === 'clear' || e.reason === 'resume') return
183  timer?.cancel()
184  timer = null
185  generation += 1
186  endedRowsById = new Map()
187  if (statusShown) clearStatus($)
188  statusShown = false
189  if (mode === 'band') await emptyBand($)
190}
191
192export function registerCard(on: On, options: PluginOptions): void {
193  mode = displayMode(options.cardDisplay)
194  on('ui.render', { component: 'AbovePrompt' }, async ($, e, next) => {
195    if (!shouldDraw(e)) return next(e)
196    let rows: CardRow[]
197    try {
198      rows = (await $.state.get(CARD)).value ?? []
199    } catch {
200      return next(e)
201    }
202    if (rows.length === 0) return next(e)
203    return drawBand($, e, rows, await next(e))
204  })
205}
206
207export function clearCardState(): void {
208  timer?.cancel()
209  timer = null
210  endedRowsById = new Map()
211  refreshing = null
212  statusShown = false
213  staleStatus = true
214}
215

1import type { EngineInterface, On } from 'claude-code'
2
3import { loadConfig } from './config.ts'
4import type { OrqConfig } from './config.ts'
5import {
6  composeAnswer,
7  jobView,
8  launchJob,
9  listJobs,
10  parseLaunch,
11  readMeta,
12  writeCancel,
13  type Engine,
14  type JobView,
15} from './engine.ts'
16import { QUESTION_PROTOCOL } from './question.ts'
17import { DRIVER_RUNNER, formatDuration, renderTable, type JobRecord, type TableRow } from './types.ts'
18
19const USAGE = 'orq: usage: /orq list | /orq result JOB_ID | /orq cancel JOB_ID | /orq launch [--worktree] INSTANCE MODEL -- PROMPT'
20
21/** Declare /orq in the typeahead. Called from session.start. */
22export async function registerOrqCommand($: Pick<EngineInterface, 'command'>): Promise<void> {
23  await $.command.register({
24    name: 'orq',
25    description: 'List, show, cancel or launch detached jobs',
26    argumentHint: 'list | result JOB_ID | cancel JOB_ID | launch [--worktree] INSTANCE MODEL -- PROMPT',
27  })
28}
29
30async function resultText($: Engine, config: OrqConfig, id: string): Promise<string> {
31  const job = await readMeta($, config.jobsDir, id)
32  if (!job) return `orq: unknown job ${id || '(missing)'}`
33  return composeAnswer($, config, job)
34}
35
36function elapsedText(job: JobRecord, view: JobView, now: number): string {
37  if (view.endedAt !== null) return formatDuration(view.endedAt - job.startedAt)
38  if (view.status === 'running' || view.status === 'cancelling' || view.status === 'waiting') {
39    return formatDuration(now - job.startedAt)
40  }
41  return '—'
42}
43
44function originText(job: JobRecord): string {
45  if (job.parent !== undefined) return `↳ ${job.parent}`
46  switch (job.kind) {
47    case 'mcp':
48      return 'delegate'
49    case 'task':
50      return 'agent'
51    default:
52      return 'manual'
53  }
54}
55
56async function listText($: Engine, config: OrqConfig): Promise<string> {
57  const jobs = await listJobs($, config.jobsDir)
58  if (jobs.length === 0) return 'orq: no jobs yet'
59  const now = await $.clock.now()
60  const rows: TableRow[] = await Promise.all(
61    jobs.map(async (job) => {
62      const view = await jobView($, job)
63      return {
64        id: job.id,
65        kind: job.kind,
66        status: view.status,
67        elapsed: elapsedText(job, view, now),
68        origin: originText(job),
69        cwd: job.cwd,
70      }
71    }),
72  )
73  return renderTable(rows)
74}
75
76async function cancelText($: Engine, config: OrqConfig, id: string): Promise<string> {
77  const job = await readMeta($, config.jobsDir, id)
78  if (!job) return `orq: unknown job ${id || '(missing)'}`
79  const { status } = await jobView($, job)
80  if (status === 'interrupted') return `orq: ${job.id} has no live supervisor; nothing to cancel`
81  if (status === 'done' || status === 'failed' || status === 'cancelled') {
82    return `orq: ${job.id} already finished (${status})`
83  }
84  await writeCancel($, job)
85  return `orq: cancel requested for ${job.id}`
86}
87
88async function launchText($: Engine, config: OrqConfig, rest: string): Promise<string> {
89  const parsed = parseLaunch(rest)
90  if ('error' in parsed) return `orq: ${parsed.error}`
91  const instance = config.instances.find((i) => i.id === parsed.instance)
92  if (!instance) {
93    const known = config.instances.map((i) => i.id).join(', ')
94    return `orq: unknown instance ${parsed.instance} (configured: ${known})`
95  }
96  const result = await launchJob($, config, {
97    kind: 'manual',
98    agentId: 'manual',
99    runner: DRIVER_RUNNER[instance.driver],
100    model: parsed.model,
101    cwd: await $.session.cwd(),
102    prompt: `${parsed.prompt}\n\n${QUESTION_PROTOCOL}`,
103    env: instance.env ?? {},
104    instanceId: instance.id,
105    driver: instance.driver,
106    ...(parsed.worktree ? { isolation: 'worktree' as const } : {}),
107  })
108  return 'refusal' in result ? result.refusal : `orq: launched ${result.job.id}`
109}
110
111/** Dispatch the text after `/orq`. Every answer is one line or a table. */
112export async function runOrq($: Engine, config: OrqConfig, args: string): Promise<string> {
113  const m = /^(\S*)\s*([\s\S]*)$/.exec(args.trim())
114  const sub = m?.[1] ?? ''
115  const rest = m?.[2] ?? ''
116  switch (sub) {
117    case 'list':
118      return listText($, config)
119    case 'result':
120      return resultText($, config, rest.trim())
121    case 'cancel':
122      return cancelText($, config, rest.trim())
123    case 'launch':
124      return launchText($, config, rest)
125    default:
126      return USAGE
127  }
128}
129
130export function registerEngine(on: On): void {
131  on('command.run', { command: 'orq' }, async ($, e) => {
132    try {
133      const { config, warning } = await loadConfig($)
134      if (warning) await $.ui.log(`orq: ${warning}`, { to: 'debug' })
135      return { text: await runOrq($, config, e.args) }
136    } catch (err) {
137      await $.ui.log(`orq: /orq failed: ${String(err)}`, { to: 'debug' })
138      return { text: `orq: /orq failed: ${String(err)}` }
139    }
140  })
141}
142

1import { defaultCeilingConfig, parseCeilingConfig, type CeilingConfig } from './ceiling.ts'
2import type { CoreEngine } from './engine.ts'
3
4export type DriverKind = 'codex' | 'opencode' | 'antigravity' | 'claudeAgent'
5
6export type InstanceConfig = {
7  id: string
8  driver: DriverKind
9  env?: Record<string, string>
10}
11
12/** The longest a wait call blocks: the engine aborts a subagent silent for 600 s, so each wait stays well under it. */
13export const MAX_WAIT_SECONDS = 240
14
15/** `run` has `integrate` run the checkout's git hooks as they are; `off`, and anything the config says that is not `run`, runs none. */
16export type IntegrateGitHooks = 'off' | 'run'
17
18export type OrqConfig = {
19  jobsDir: string
20  instances: InstanceConfig[]
21  waitSeconds: number
22  /** Absolute directories, besides the session's own, that a cc-orq:task may run in. */
23  allowedRoots: string[]
24  integrate: { gitHooks: IntegrateGitHooks }
25  /** The most a task may run as. The session's permission mode can lower it, never raise it. */
26  ceiling: CeilingConfig
27}
28
29export const CONFIG_PATH = '.config/orq/config.json'
30
31export const DEFAULT_INSTANCES: InstanceConfig[] = [
32  { id: 'codex', driver: 'codex' },
33  { id: 'opencode', driver: 'opencode' },
34  { id: 'claudeAgent', driver: 'claudeAgent' },
35]
36
37const DRIVERS: DriverKind[] = ['codex', 'opencode', 'antigravity', 'claudeAgent']
38
39export function defaultConfig(home: string): OrqConfig {
40  return {
41    jobsDir: `${home}/.cache/orq/jobs`,
42    instances: DEFAULT_INSTANCES.map((i) => ({ ...i })),
43    waitSeconds: 120,
44    allowedRoots: [],
45    integrate: { gitHooks: 'off' },
46    ceiling: defaultCeilingConfig(),
47  }
48}
49
50export function expandHome(path: string, home: string): string {
51  return path === '~' ? home : path.startsWith('~/') ? `${home}/${path.slice(2)}` : path
52}
53
54export type ParsedConfig = { config: OrqConfig; source: 'defaults' | 'file'; warning?: string }
55
56export function parseConfig(text: string | null, home: string): ParsedConfig {
57  const base = defaultConfig(home)
58  if (text === null) return { config: base, source: 'defaults' }
59  let raw: unknown
60  try {
61    raw = JSON.parse(text)
62  } catch {
63    return { config: base, source: 'defaults', warning: 'config: invalid JSON, using defaults' }
64  }
65  if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) {
66    return { config: base, source: 'defaults', warning: 'config: root is not an object, using defaults' }
67  }
68  const obj = raw as Record<string, unknown>
69  const config: OrqConfig = { ...base }
70  const warnings: string[] = []
71  if (obj.jobsDir !== undefined) {
72    const dir = typeof obj.jobsDir === 'string' ? expandHome(obj.jobsDir.trim(), home) : ''
73    if (dir.startsWith('/')) config.jobsDir = dir
74    else warnings.push('jobsDir must be an absolute path or start with ~/, keeping the default')
75  }
76  if (obj.allowedRoots !== undefined) {
77    if (!Array.isArray(obj.allowedRoots)) {
78      warnings.push('allowedRoots must be an array of absolute paths, keeping none')
79    } else {
80      const roots: string[] = []
81      obj.allowedRoots.forEach((item: unknown, index) => {
82        const dir = typeof item === 'string' ? expandHome(item.trim(), home) : ''
83        if (dir.startsWith('/')) roots.push(dir)
84        else warnings.push(`allowedRoots[${index}] must be an absolute path or start with ~/, dropped`)
85      })
86      config.allowedRoots = roots
87    }
88  }
89  if (obj.waitSeconds !== undefined) {
90    const w = typeof obj.waitSeconds === 'number' && Number.isFinite(obj.waitSeconds) ? Math.floor(obj.waitSeconds) : Number.NaN
91    if (w >= 10 && w <= MAX_WAIT_SECONDS) config.waitSeconds = w
92    else warnings.push(`waitSeconds must be a number between 10 and ${MAX_WAIT_SECONDS}, keeping the default`)
93  }
94  if (obj.integrate !== undefined) {
95    if (typeof obj.integrate !== 'object' || obj.integrate === null || Array.isArray(obj.integrate)) {
96      warnings.push('integrate must be an object, keeping git hooks off')
97    } else {
98      const gitHooks = (obj.integrate as Record<string, unknown>).gitHooks
99      if (gitHooks === 'run') config.integrate = { gitHooks: 'run' }
100      else if (gitHooks !== undefined && gitHooks !== 'off') warnings.push('integrate.gitHooks must be "run" or "off", keeping git hooks off')
101    }
102  }
103  if (obj.ceiling !== undefined) {
104    const parsed = parseCeilingConfig(obj.ceiling)
105    config.ceiling = parsed.ceiling
106    warnings.push(...parsed.warnings)
107  }
108  if (obj.instances !== undefined) {
109    if (!Array.isArray(obj.instances)) {
110      warnings.push('instances must be an array, keeping the defaults')
111    } else {
112      const instances: InstanceConfig[] = []
113      const seen = new Set<string>()
114      obj.instances.forEach((item, index) => {
115        const label = `instances[${index}]`
116        if (typeof item !== 'object' || item === null || Array.isArray(item)) {
117          warnings.push(`${label} is not an object, dropped`)
118          return
119        }
120        const inst = item as Record<string, unknown>
121        if (typeof inst.id !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$/.test(inst.id)) {
122          warnings.push(`${label} has no valid id (letters, digits, _ . -, up to 64 chars), dropped`)
123          return
124        }
125        if (typeof inst.driver !== 'string' || !DRIVERS.includes(inst.driver as DriverKind)) {
126          warnings.push(`${label} (${inst.id}) has an unknown driver, expected one of ${DRIVERS.join(', ')}, dropped`)
127          return
128        }
129        if (seen.has(inst.id)) {
130          warnings.push(`${label} repeats the id ${inst.id}, dropped`)
131          return
132        }
133        seen.add(inst.id)
134        const env: Record<string, string> = {}
135        if (inst.env !== undefined) {
136          if (typeof inst.env !== 'object' || inst.env === null || Array.isArray(inst.env)) {
137            warnings.push(`${label} (${inst.id}) env is not an object, ignored`)
138          } else {
139            for (const [k, v] of Object.entries(inst.env as Record<string, unknown>)) {
140              if (!/^[A-Z_][A-Z0-9_]*$/.test(k)) warnings.push(`${label} (${inst.id}) env key ${k} must be UPPER_CASE, dropped`)
141              else if (typeof v !== 'string') warnings.push(`${label} (${inst.id}) env ${k} must be a string, dropped`)
142              else env[k] = expandHome(v, home)
143            }
144          }
145        }
146        instances.push(Object.keys(env).length > 0 ? { id: inst.id, driver: inst.driver as DriverKind, env } : { id: inst.id, driver: inst.driver as DriverKind })
147      })
148      if (instances.length > 0) config.instances = instances
149      else warnings.push('instances has no valid entry, keeping the defaults')
150    }
151  }
152  const result: ParsedConfig = { config, source: 'file' }
153  if (warnings.length > 0) result.warning = `config: ${warnings.join('; ')}`
154  return result
155}
156
157export async function loadConfig($: Pick<CoreEngine, 'fs' | 'env'>): Promise<ParsedConfig> {
158  const home = ((await $.env.get('HOME')) ?? '').trim()
159  if (home === '') {
160    return { ...parseConfig(null, '/root'), warning: 'config: HOME is not set, using defaults under /root' }
161  }
162  const path = `${home}/${CONFIG_PATH}`
163  let text: string | null = null
164  try {
165    if (await $.fs.exists(path)) text = String(await $.fs.read(path))
166  } catch (err) {
167    return { ...parseConfig(null, home), warning: `config: cannot read ${path} (${String(err)})` }
168  }
169  return parseConfig(text, home)
170}
171

1import type {
2  CommandRunInput,
3  EngineInterface,
4  On,
5  RenderElement,
6  RenderInput,
7  SessionEndInput,
8  Timer,
9  UiOpenResult,
10} from 'claude-code'
11
12import { loadConfig } from './config.ts'
13import { jobView, listJobs, readMeta, writeCancel, type Engine, type JobView } from './engine.ts'
14import { renderTail } from './event-line.ts'
15import { formatDuration, isSafeJobId, shellWord, type JobRecord, type JobStatus } from './types.ts'
16
17export const PANEL_ID = 'orq-panel'
18export const PANEL_COMMAND = 'orq-panel'
19export const PANEL_TITLE = 'orq'
20
21const TAIL_BYTES = 32768
22const REFRESH_MS = 2000
23const PROCESS_TIMEOUT_MS = 15000
24const FINAL_VIEWS_MAX = 200
25const USAGE = 'orq-panel: usage: /orq-panel [show <id> | cancel <id>]'
26
27/** A job row: its meta.json, the status jobView reads for it, and how long it has run or ran until it ended. */
28export type PanelJob = JobRecord & { status: JobStatus; elapsedMs: number }
29
30export type PanelSnapshot = {
31  jobs: PanelJob[]
32  selectedId: string | null
33  tail: string[]
34  now: number
35  jobsDir: string
36  note: string
37}
38
39let selectedId: string | null = null
40let refreshTimer: Timer | null = null
41/** The views of ended jobs, by job id. See cachedJobView. */
42const finalViews = new Map<string, JobView>()
43
44function statusMark(status: JobStatus): string {
45  switch (status) {
46    case 'running':
47      return '●'
48    case 'cancelling':
49      return '◌'
50    case 'waiting':
51      return '◐'
52    case 'done':
53      return '✓'
54    case 'cancelled':
55      return '✕'
56    case 'failed':
57      return '!'
58    case 'interrupted':
59      return '?'
60  }
61}
62
63/** Log a failed panel call at debug level. The pane never throws into the host. */
64async function debugLog($: Engine, what: string, err: unknown): Promise<void> {
65  try {
66    await $.ui.log(`orq-panel: ${what}: ${String(err)}`, { to: 'debug' })
67  } catch {
68    /* nothing left to report to */
69  }
70}
71
72/** Ask the host to redraw the pane. */
73async function redraw($: Engine): Promise<void> {
74  try {
75    $.ui.invalidate('ui.render')
76  } catch (err) {
77    await debugLog($, 'redraw failed', err)
78  }
79}
80
81/** The last TAIL_BYTES of a job's events file. A full read would be too big on a long run. */
82export async function tailEvents($: Engine, dir: string): Promise<string> {
83  const script = `# orq-panel:tail\ntail -c ${TAIL_BYTES} ${shellWord(`${dir}/events.jsonl`)} 2>/dev/null; exit 0`
84  try {
85    const r = await $.process.run(['bash', '-c', script], { timeoutMs: PROCESS_TIMEOUT_MS })
86    return r.stdout
87  } catch (err) {
88    await debugLog($, 'tail failed', err)
89    return ''
90  }
91}
92
93/** Cancel one job, the way mcp__plugin_cc-orq_orq__cancel does: only a running or waiting job gets a cancel file. */
94export async function cancelJob($: Engine, id: string): Promise<string> {
95  const { config } = await loadConfig($)
96  const job = await readMeta($, config.jobsDir, id)
97  if (!job) return `orq-panel: unknown job ${id || '(missing)'}`
98  const { status } = await jobView($, job)
99  if (status === 'running' || status === 'waiting') {
100    await writeCancel($, job)
101    return `orq-panel: cancel requested for ${job.id}`
102  }
103  if (status === 'done' || status === 'failed' || status === 'cancelled') {
104    return `orq-panel: ${job.id} already finished; nothing to cancel`
105  }
106  return `orq-panel: ${job.id} is ${status}; nothing to cancel`
107}
108
109/**
110 * jobView, with the view of an ended job kept. An exit file never changes, so an ended job is read once, and a
111 * finished job costs one extract.py run, not one per render. Running and interrupted jobs are read every time.
112 */
113export async function cachedJobView($: Engine, job: JobRecord): Promise<JobView> {
114  const known = finalViews.get(job.id)
115  if (known) return known
116  const view = await jobView($, job)
117  if (view.endedAt !== null) {
118    // The cap is well above the listing, so the map stays small and a job still listed is never dropped.
119    if (finalViews.size >= FINAL_VIEWS_MAX) {
120      const oldest = finalViews.keys().next().value
121      if (oldest !== undefined) finalViews.delete(oldest)
122    }
123    finalViews.set(job.id, view)
124  }
125  return view
126}
127
128/**
129 * The panel's state: the newest jobs with their status and elapsed time, and the selected job's tail. A job whose
130 * view throws is logged at debug level with its id and left out; the other jobs are still listed.
131 */
132export async function collectSnapshot($: Engine): Promise<PanelSnapshot> {
133  const now = await $.clock.now()
134  const { config } = await loadConfig($)
135  const listed = await Promise.all(
136    (await listJobs($, config.jobsDir)).map(async (job): Promise<PanelJob | null> => {
137      try {
138        const view = await cachedJobView($, job)
139        return { ...job, status: view.status, elapsedMs: Math.max(0, (view.endedAt ?? now) - job.startedAt) }
140      } catch (err) {
141        await debugLog($, `job ${job.id} could not be read`, err)
142        return null
143      }
144    }),
145  )
146  const jobs = listed.filter((job): job is PanelJob => job !== null)
147  const selected = jobs.find((job) => job.id === selectedId)
148  let tail: string[] = []
149  let note = ''
150  if (selected) {
151    tail = renderTail(await tailEvents($, selected.dir))
152  } else if (selectedId) {
153    note = `selected job ${selectedId} is gone`
154  }
155  return { jobs, selectedId: selected?.id ?? null, tail, now, jobsDir: config.jobsDir, note }
156}
157
158function jobLine(job: PanelJob, active: boolean): string {
159  const model = job.model ? ` · ${job.model}` : ''
160  return `${statusMark(job.status)} ${job.status} ${formatDuration(job.elapsedMs)} ${job.kind}${model}${active ? ' ←' : ''}`
161}
162
163async function cancelAndRedraw($: Engine, id: string): Promise<void> {
164  try {
165    await cancelJob($, id)
166  } catch (err) {
167    await debugLog($, 'cancel failed', err)
168  }
169  await redraw($)
170}
171
172function drawSnapshot($: Engine, e: RenderInput<'Pane'>, snap: PanelSnapshot): RenderElement {
173  const { Box, Button, Text } = $.ui.resolve(e)
174  const count = `${snap.jobs.length} job${snap.jobs.length === 1 ? '' : 's'}`
175  const rows: RenderElement[] = [Text({ dimColor: true, children: `orq · ${count} · ${snap.jobsDir}` })]
176  if (snap.jobs.length === 0) rows.push(Text({ children: 'no jobs yet' }))
177  for (const job of snap.jobs) {
178    const active = job.id === snap.selectedId
179    rows.push(
180      Box({
181        key: `job:${job.id}`,
182        flexDirection: 'row',
183        children: [
184          Button({
185            key: `sel:${job.id}`,
186            plain: true,
187            label: jobLine(job, active),
188            onPress: () => {
189              selectedId = job.id
190              void redraw($)
191            },
192          }),
193          job.status === 'running' || job.status === 'waiting'
194            ? Button({
195                key: `cancel:${job.id}`,
196                label: 'cancel',
197                onPress: () => {
198                  void cancelAndRedraw($, job.id)
199                },
200              })
201            : Text({ dimColor: true, children: ` ${job.id.slice(-6)}` }),
202        ],
203      }),
204    )
205  }
206  if (snap.selectedId) {
207    rows.push(Text({ bold: true, children: `— ${snap.selectedId}` }))
208    const parent = snap.jobs.find((job) => job.id === snap.selectedId)?.parent
209    if (parent !== undefined) rows.push(Text({ dimColor: true, children: `↳ continues ${parent}` }))
210    if (snap.tail.length === 0) rows.push(Text({ dimColor: true, children: '(no events yet)' }))
211    for (const [i, line] of snap.tail.entries()) {
212      rows.push(
213        Box({
214          key: `tail:${i}`,
215          children: [Text({ children: line, wrap: 'truncate-end' })],
216        }),
217      )
218    }
219  } else {
220    rows.push(Text({ dimColor: true, children: 'select a job to watch its live events' }))
221  }
222  if (snap.note) rows.push(Text({ dimColor: true, children: snap.note }))
223  rows.push(
224    Box({
225      flexDirection: 'row',
226      children: [
227        Button({
228          key: 'panel:refresh',
229          label: 'refresh',
230          onPress: () => {
231            void redraw($)
232          },
233        }),
234        Button({
235          key: 'panel:close',
236          label: 'close',
237          onPress: () => {
238            // Once the close settles, one tick finds the pane gone and stops the timer. A close a hook
239            // answered without next leaves the pane up, and the same tick keeps the timer.
240            void $.ui.close({ id: PANEL_ID }).then(
241              () => refreshTick($),
242              (err: unknown) => debugLog($, 'close failed', err),
243            )
244          },
245        }),
246      ],
247    }),
248  )
249  return Box({ flexDirection: 'column', children: rows })
250}
251
252function cancelRefresh(): void {
253  refreshTimer?.cancel()
254  refreshTimer = null
255}
256
257/** One tick of the refresh timer: redraw while the host lists the pane, and stop the timer once it does not. */
258async function refreshTick($: Engine): Promise<void> {
259  try {
260    const open = (await $.ui.panes()).some((pane) => pane.id === PANEL_ID)
261    if (!open) {
262      cancelRefresh()
263      return
264    }
265  } catch (err) {
266    await debugLog($, 'pane check failed', err)
267  }
268  await redraw($)
269}
270
271/** Redraw the pane every REFRESH_MS while it is open. Started once, when the pane opens. */
272function ensureRefresh($: Engine): void {
273  if (refreshTimer) return
274  try {
275    refreshTimer = $.clock.every(REFRESH_MS, () => {
276      void refreshTick($)
277    })
278  } catch (err) {
279    void debugLog($, 'refresh timer failed', err)
280  }
281}
282
283/** Stop the refresh timer when the session ends. A /clear or a resume keeps it, as it keeps the card. */
284export function stopPanel(e: Pick<SessionEndInput, 'reason'>): void {
285  if (e.reason === 'clear' || e.reason === 'resume') return
286  cancelRefresh()
287}
288
289async function openPane($: Engine): Promise<{ text: string }> {
290  let opened: UiOpenResult
291  try {
292    opened = await $.ui.open({ id: PANEL_ID, title: PANEL_TITLE, focus: true })
293  } catch (err) {
294    return { text: `orq-panel: could not open the pane: ${String(err)}` }
295  }
296  ensureRefresh($)
297  if (!opened.isPlaced) return { text: `orq-panel: pane waits undrawn (${opened.reason})` }
298  return { text: 'orq-panel: pane opened' }
299}
300
301/** The text after /orq-panel: nothing opens the pane, `show <id>` selects a job, `cancel <id>` cancels one. */
302export async function runPanelCommand($: Engine, args: string): Promise<{ text: string }> {
303  const m = /^(\S*)\s*([\s\S]*)$/.exec(args.trim())
304  const sub = m?.[1] ?? ''
305  const rest = (m?.[2] ?? '').trim()
306  if (sub === 'cancel') return { text: await cancelJob($, rest) }
307  if (sub === 'show') {
308    if (!isSafeJobId(rest)) return { text: `orq-panel: unknown job ${rest || '(missing)'}` }
309    selectedId = rest
310  } else if (sub !== '') {
311    return { text: USAGE }
312  }
313  return openPane($)
314}
315
316/** Declare /orq-panel in the typeahead. Called from session.start, which owns the one unmatched hook. */
317export async function registerPanelCommand($: Pick<EngineInterface, 'command'>): Promise<void> {
318  await $.command.register({
319    name: PANEL_COMMAND,
320    description: 'Watch orq background jobs live: list, tail one job, cancel it.',
321    argumentHint: '[show <id> | cancel <id>]',
322  })
323}
324
325/** The command.run hook for /orq-panel. A failure becomes the command's text. */
326export async function onPanelCommand($: Engine, e: CommandRunInput): Promise<{ text: string }> {
327  try {
328    return await runPanelCommand($, e.args)
329  } catch (err) {
330    await debugLog($, 'command failed', err)
331    return { text: `orq-panel: ${String(err)}` }
332  }
333}
334
335/** Draw the panel's pane: the jobs, the selected job's tail, and the refresh and close buttons. */
336export async function onPaneRender($: Engine, e: RenderInput<'Pane'>): Promise<RenderElement> {
337  try {
338    return drawSnapshot($, e, await collectSnapshot($))
339  } catch (err) {
340    const { Box, Text } = $.ui.resolve(e)
341    return Box({
342      flexDirection: 'column',
343      children: [Text({ children: `orq-panel: could not read jobs: ${String(err)}` })],
344    })
345  }
346}
347
348/** Test seam: drop the selected job, the refresh timer and the ended jobs' views between isolated runs. */
349export function clearPanelSelection(): void {
350  cancelRefresh()
351  selectedId = null
352  finalViews.clear()
353}
354
355/** The panel's own hooks. The command itself is declared from session.start, through registerPanelCommand. */
356export function registerPanel(on: On): void {
357  on('command.run', { command: PANEL_COMMAND }, onPanelCommand)
358  on('ui.render', { component: 'Pane' }, async ($, e, next) => {
359    if (e.requestId !== PANEL_ID) return next(e)
360    return onPaneRender($, e)
361  })
362}
363

1// The cc-orq:task subagent. Its prompt names a target; the mod runs the task as a detached job and serves
2// each step of the subagent itself, so no model call happens for it. The subagent waits on its job
3// through mcp__cc-orq__wait, and its answer goes back through SubagentHandback in auto mode, as text otherwise.
4
5import type { EngineInterface, On, TurnStepChunk, TurnStepInput, TurnStepResult } from 'claude-code'
6
7import { resolveCeiling } from './ceiling.ts'
8import { loadConfig, MAX_WAIT_SECONDS, type OrqConfig } from './config.ts'
9import { isFailure } from './contract.ts'
10import { composeAnswer, jobLiveness, launchTaskFromHeader, listJobs, readMeta, writeCancel, type Engine } from './engine.ts'
11import { parseTaskHeader } from './task-header.ts'
12import { parentRuntimeModeFromPermissionMode } from './target.ts'
13import { formatDuration, isRecord, refusalText, shellWord, TASK_REFUSAL, type JobRecord } from './types.ts'
14
15type TaskEngine = Engine & Pick<EngineInterface, 'agent' | 'settings' | 'store' | 'tool' | 'ui'>
16
17/** The subagent type a parent spawns with `Agent({ subagent_type: "cc-orq:task" })`. */
18export const TASK_SUBAGENT = 'cc-orq:task'
19const WAIT_TOOL = 'mcp__cc-orq__wait'
20const HANDBACK_TOOL = 'SubagentHandback'
21
22const WAIT_DESCRIPTION =
23  'Wait for an orq background job. Input is {"job": "<job-id>"}. Answers finished, supervisor gone, or still running with how long it has run; call again until it finishes.'
24const WAIT_SCHEMA = {
25  type: 'object',
26  properties: { job: { type: 'string' } },
27  required: ['job'],
28  additionalProperties: false,
29}
30// Everything the parent needs to start a task and handle its answer. The guidance for the orq tools is the MCP
31// server's instructions (server/instructions.md), so it is not repeated here.
32const TASK_AGENT_DESCRIPTION =
33  'Runs one self-contained task on a configured orq instance (Codex, OpenCode, Antigravity or Claude) in the background. ' +
34  'Spawn it with `run_in_background: true`. The child gets no history, so the prompt carries the whole brief: the goal, the files it may touch, and the check that shows it worked. ' +
35  'The prompt starts with header lines, in any order. Required: `target: <instance> <model> [effort=<value>] [runtime=<mode>] [interaction=<mode>]` ' +
36  '(runtime and interaction can narrow the session, never widen it) and `cwd: /absolute/path` (inside the session directory or an allowed root). ' +
37  'Optional: `isolation: worktree | shared` (default worktree: a new branch in its own git worktree, so cwd must be in a git checkout; use shared only for a task that edits nothing) ' +
38  'and `role: implementation | research | review | design | test | general`. ' +
39  "Never pass the Agent tool's own `isolation` option. " +
40  'After spawning, end the turn and do not poll: the answer arrives when the job ends, and its footer carries `taskId <id>`. ' +
41  'A handback that starts `BLOCKED on a question` is answered with the `answer` tool, which resumes the same worker in a new job: settle the question yourself when the brief and the conversation do, and ask the user when they do not. ' +
42  'A review round is a new task too, run by a different model family than the implementer.'
43const ONE_LINE_PROMPT =
44  'An orq runner: the orq plugin serves your steps and runs your task on the external CLI its target line names. Call mcp__cc-orq__wait with the job id until the job finishes.'
45
46/** A launched job's id, or the refusal a step answers with. Stored per agent so a repeated step reuses it. */
47type TaskRun = { jobId: string } | { refusal: string }
48
49/** The longest refusal kept and answered: a refusal can echo a long path from the header. */
50const MAX_REFUSAL_CHARS = 2000
51
52function stringValue(value: unknown): string | undefined {
53  return typeof value === 'string' && value !== '' ? value : undefined
54}
55
56async function storeRead($: TaskEngine, key: string): Promise<unknown> {
57  try {
58    return await $.store.get(key)
59  } catch {
60    return undefined
61  }
62}
63
64async function sessionId($: TaskEngine): Promise<string | undefined> {
65  try {
66    return stringValue(await $.session.id())
67  } catch {
68    return undefined
69  }
70}
71
72/** The agent id from an agent.spawn result, bare or wrapped in `value`; a denied spawn has none. */
73function spawnedAgentId(result: unknown): string | undefined {
74  if (!isRecord(result)) return undefined
75  const outcome = typeof result.agentId === 'string' || typeof result.deny === 'string' || !isRecord(result.value) ? result : result.value
76  if (typeof outcome.deny === 'string') return undefined
77  return stringValue(outcome.agentId)
78}
79
80function parseTaskRun(value: unknown): TaskRun | null {
81  if (!isRecord(value)) return null
82  if (typeof value.jobId === 'string') return { jobId: value.jobId }
83  if (typeof value.refusal === 'string') return { refusal: value.refusal }
84  return null
85}
86
87/** The mode this task's own spawn captured. It is the only input to the task's privilege ceiling. */
88async function spawnPermissionMode($: TaskEngine, agentId: string): Promise<string | undefined> {
89  return stringValue(await storeRead($, `perm:${agentId}`))
90}
91
92/**
93 * The mode that picks how a finished task is delivered: the spawn's, then the session's last cc-orq:task spawn,
94 * then the settings default. It chooses a delivery format, never a privilege.
95 */
96async function handbackPermissionMode($: TaskEngine, agentId: string): Promise<string | undefined> {
97  const own = await spawnPermissionMode($, agentId)
98  if (own !== undefined) return own
99  const sid = await sessionId($)
100  const bySession = sid === undefined ? undefined : stringValue(await storeRead($, `perm:session:${sid}`))
101  if (bySession !== undefined) return bySession
102  try {
103    const settings = await $.settings.read()
104    return isRecord(settings.permissions) ? stringValue(settings.permissions.defaultMode) : undefined
105  } catch {
106    return undefined
107  }
108}
109
110async function sessionCwd($: TaskEngine): Promise<string> {
111  try {
112    return await $.session.cwd()
113  } catch {
114    return ''
115  }
116}
117
118function firstUserText(messages: { role: string; text: string }[]): string | undefined {
119  return messages.find((message) => message.role === 'user')?.text
120}
121
122/** A task agent is one whose spawn the mod saw as cc-orq:task. Every other agent passes through untouched. */
123async function isTaskAgent($: TaskEngine, agentId: string): Promise<boolean> {
124  return (await storeRead($, `task:${agentId}`)) === TASK_SUBAGENT
125}
126
127async function readMessages($: TaskEngine, agentId: string): Promise<{ role: string; text: string }[]> {
128  try {
129    const messages = await $.session.messages({ agentId })
130    return Array.isArray(messages) ? messages : []
131  } catch {
132    return []
133  }
134}
135
136/** Read the header, hold it to the ceiling that the config and the spawn's permission mode set, then launch. Every refusal is text. */
137async function launchForAgent($: TaskEngine, config: OrqConfig, agentId: string): Promise<TaskRun> {
138  const text = firstUserText(await readMessages($, agentId))
139  if (text === undefined) return { refusal: `${TASK_REFUSAL}: its prompt could not be read.` }
140  const header = parseTaskHeader(text)
141  if (isFailure(header)) return { refusal: `${TASK_REFUSAL}: ${header.error.message}` }
142  const spawnMode = await spawnPermissionMode($, agentId)
143  if (spawnMode === undefined) {
144    await $.ui.log(`orq: ceiling for ${agentId}: no permission mode captured at spawn, using approval-required`, { to: 'debug' })
145  }
146  const { effective, refusal } = resolveCeiling({
147    config: config.ceiling,
148    session: parentRuntimeModeFromPermissionMode(spawnMode),
149    isolation: header.isolation,
150    request: { runtimeMode: header.runtimeMode, interactionMode: header.interactionMode },
151  })
152  if (refusal !== null) return { refusal: refusalText(refusal) }
153  const launched = await launchTaskFromHeader($, config, { agentId, header, ceiling: effective, sessionDir: await sessionCwd($) })
154  return 'refusal' in launched ? { refusal: launched.refusal } : { jobId: launched.job.id }
155}
156
157/** The task job this agent launched, found in the jobs directory when its run record was never stored. */
158async function jobForAgent($: TaskEngine, config: OrqConfig, agentId: string): Promise<JobRecord | null> {
159  const jobs = await listJobs($, config.jobsDir)
160  return jobs.find((job) => job.kind === 'task' && job.agentId === agentId) ?? null
161}
162
163/** The run recorded for this agent, or a new one. A repeated step finds the record and does not relaunch. */
164async function runFor($: TaskEngine, config: OrqConfig, agentId: string): Promise<TaskRun> {
165  const key = `agent:${agentId}`
166  const stored = parseTaskRun(await storeRead($, key))
167  if (stored !== null) return stored
168  const existing = await jobForAgent($, config, agentId)
169  const launched = existing === null ? await launchForAgent($, config, agentId) : { jobId: existing.id }
170  const run = 'refusal' in launched ? { refusal: launched.refusal.slice(0, MAX_REFUSAL_CHARS) } : launched
171  try {
172    await $.store.set(key, run)
173  } catch (err) {
174    await $.ui.log(`orq: could not record the run of ${agentId}: ${String(err)}`, { to: 'debug' })
175  }
176  return run
177}
178
179async function cancelRun($: TaskEngine, config: OrqConfig, agentId: string, reason: string): Promise<void> {
180  try {
181    const run = parseTaskRun(await storeRead($, `agent:${agentId}`))
182    if (run === null || !('jobId' in run)) return
183    const job = await readMeta($, config.jobsDir, run.jobId)
184    if (job !== null) await writeCancel($, job, reason)
185  } catch (err) {
186    await $.ui.log(`orq: could not cancel the run of ${agentId}: ${String(err)}`, { to: 'debug' })
187  }
188}
189
190async function handbackDue($: TaskEngine, agentId: string): Promise<boolean> {
191  if ((await storeRead($, `handedback:${agentId}`)) === true) return false
192  const mode = await handbackPermissionMode($, agentId)
193  if (mode === 'auto') return true
194  const why = mode === undefined ? 'no permission mode' : `mode is ${mode}, not auto`
195  await $.ui.log(`orq: handback skipped for ${agentId}: ${why}`, { to: 'debug' })
196  return false
197}
198
199/** The step's final answer: SubagentHandback under auto mode, the text otherwise. */
200async function* answerStep($: TaskEngine, e: TurnStepInput, agentId: string, answer: string): AsyncGenerator<TurnStepChunk, TurnStepResult> {
201  if (await handbackDue($, agentId)) {
202    await $.store.set(`handedback:${agentId}`, true)
203    const input = { message: answer }
204    yield { kind: 'tool', index: 0, id: `handback-${agentId}-${e.index}`, name: HANDBACK_TOOL }
205    yield { kind: 'input', index: 0, json: JSON.stringify(input) }
206    yield { kind: 'stop', stopReason: 'tool_use', usage: null }
207    return { turnId: e.turnId, index: e.index, answer: '', toolUses: [{ name: HANDBACK_TOOL, input }], stopReason: 'tool_use', usage: null }
208  }
209  yield { kind: 'text', index: 0, text: answer }
210  yield { kind: 'stop', stopReason: 'end_turn', usage: null }
211  return { turnId: e.turnId, index: e.index, answer, toolUses: [], stopReason: 'end_turn', usage: null }
212}
213
214/** The step that waits on a running job: one mcp__cc-orq__wait call, with no model call. */
215async function* waitStep(e: TurnStepInput, jobId: string): AsyncGenerator<TurnStepChunk, TurnStepResult> {
216  yield { kind: 'tool', index: 0, id: `wait-${jobId}-${e.index}`, name: WAIT_TOOL }
217  yield { kind: 'input', index: 0, json: JSON.stringify({ job: jobId }) }
218  yield { kind: 'stop', stopReason: 'tool_use', usage: null }
219  return { turnId: e.turnId, index: e.index, answer: '', toolUses: [{ name: WAIT_TOOL, input: { job: jobId } }], stopReason: 'tool_use', usage: null }
220}
221
222/** Sleeps in 2-second steps until the job exits (0), its supervisor is gone (12), or the budget ends (11). */
223function waitScript(dir: string, iterations: number): string {
224  const exit = shellWord(`${dir}/exit`)
225  const pgid = shellWord(`${dir}/pgid`)
226  return (
227    `# orq:wait\nfor i in $(seq 1 ${iterations}); do\n` +
228    `  [ -e ${exit} ] && exit 0\n` +
229    `  pgid=$(cat ${pgid} 2>/dev/null | tr -d ' \\t\\n')\n` +
230    `  case "$pgid" in ''|*[!0-9]*) exit 12 ;; esac\n` +
231    `  kill -0 -"$pgid" 2>/dev/null || { [ -e ${exit} ] && exit 0; exit 12; }\n` +
232    `  sleep 2\n` +
233    `done\nexit 11`
234  )
235}
236
237/** Declares the wait tool and the cc-orq:task agent. Called from the session.start hook, which owns that event. */
238export async function registerTaskSurface($: Pick<EngineInterface, 'agent' | 'tool'>): Promise<void> {
239  await $.tool.register({ name: 'wait', description: WAIT_DESCRIPTION, inputSchema: WAIT_SCHEMA })
240  await $.agent.register({
241    name: 'task',
242    description: TASK_AGENT_DESCRIPTION,
243    prompt: ONE_LINE_PROMPT,
244    tools: [WAIT_TOOL],
245    background: true,
246  })
247}
248
249export function registerTask(on: On): void {
250  on('agent.spawn', async ($, e, next) => {
251    const r = await next(e)
252    try {
253      const agentId = spawnedAgentId(r)
254      if (agentId !== undefined && e.subagentType === TASK_SUBAGENT) {
255        await $.store.set(`task:${agentId}`, TASK_SUBAGENT)
256        const mode = stringValue(e.permissionMode)
257        if (mode !== undefined) {
258          await $.store.set(`perm:${agentId}`, mode)
259          const sid = await sessionId($)
260          if (sid !== undefined) await $.store.set(`perm:session:${sid}`, mode)
261        }
262      }
263    } catch (err) {
264      await $.ui.log(`orq: could not record the spawn: ${String(err)}`, { to: 'debug' })
265    }
266    return r
267  })
268
269  on('turn.step', async function* ($, e, next) {
270    const agentId = e.agentId
271    if (agentId === undefined || !(await isTaskAgent($, agentId))) return yield* next(e)
272    try {
273      const { config } = await loadConfig($)
274      next.signal.addEventListener('abort', () => void cancelRun($, config, agentId, 'turn-abort'), { once: true })
275      const run = await runFor($, config, agentId)
276      if (next.signal.aborted) await cancelRun($, config, agentId, 'turn-abort')
277      if ('refusal' in run) return yield* answerStep($, e, agentId, run.refusal)
278      const job = await readMeta($, config.jobsDir, run.jobId)
279      if (job === null) return yield* answerStep($, e, agentId, `orq: task job ${run.jobId} has no readable metadata.`)
280      if ((await jobLiveness($, job)) === 'running') return yield* waitStep(e, job.id)
281      return yield* answerStep($, e, agentId, await composeAnswer($, config, job))
282    } catch (err) {
283      // Nothing is yielded before the body's last step, so this text is the whole answer. The model never sees a throw.
284      const answer = `orq: task failed: ${err instanceof Error ? err.message : String(err)}`
285      yield { kind: 'text', index: 0, text: answer }
286      yield { kind: 'stop', stopReason: 'end_turn', usage: null }
287      return { turnId: e.turnId, index: e.index, answer, toolUses: [], stopReason: 'end_turn', usage: null }
288    }
289  })
290
291  on('tool.call', { tool: WAIT_TOOL }, async ($, e, next) => {
292    const jobId = isRecord(e) && typeof e.job === 'string' ? e.job : ''
293    const { config } = await loadConfig($)
294    const job = await readMeta($, config.jobsDir, jobId)
295    if (job === null) return { result: `unknown job ${jobId || '(missing)'}` }
296    next.signal.addEventListener('abort', () => void writeCancel($, job, 'wait-abort').catch(() => undefined), { once: true })
297    try {
298      const slice = Math.min(config.waitSeconds, MAX_WAIT_SECONDS)
299      const w = await $.process.run(['bash', '-c', waitScript(job.dir, Math.floor(slice / 2))], {
300        timeoutMs: slice * 1000 + 5000,
301      })
302      if (w.exitCode === 0) return { result: 'finished' }
303      if (w.exitCode === 12) return { result: 'supervisor gone' }
304      if (w.exitCode !== 11) return { result: `orq: wait failed (exit ${w.exitCode})` }
305      const now = await $.clock.now()
306      return { result: `still running, ${formatDuration(now - job.startedAt)}` }
307    } catch {
308      // The engine rejects the wait's process run when the dispatch aborts, so a throw here is a stop.
309      await writeCancel($, job, 'wait-throw')
310      return { result: 'cancelled' }
311    }
312  })
313}
314

1// What the orq tools capabilities, status and cancel answer, and the session-start log of unfinished jobs.
2// The stdio server in server/ serves the tools; the mod no longer registers them. Every function takes the
3// CoreEngine part of `$`, so the server runs the same code on its Node shim. A refusal is a typed failure,
4// `{ error: { code, message } }`, so the model can branch on the code.
5
6import { buildCapabilities, parseCatalogInstances } from './catalog.ts'
7import type { OrqConfig } from './config.ts'
8import {
9  type CapabilitiesResult,
10  type Failure,
11  type InteractionMode,
12  type ProviderInstance,
13  type RuntimeMode,
14  type TaskCancelResult,
15  type TaskStatusResult,
16  fail,
17} from './contract.ts'
18import { jobView, listJobs, readJobStatus, readMeta, taskStatusOf, writeCancel, type CoreEngine } from './engine.ts'
19import { formatDuration, isRecord } from './types.ts'
20
21/** The names the model calls the tools by. A refusal that quotes the expected input names the tool it answers. */
22export const STATUS_TOOL = 'mcp__plugin_cc-orq_orq__status'
23export const CANCEL_TOOL = 'mcp__plugin_cc-orq_orq__cancel'
24
25export const CANCEL_REASON_MAX = 200
26
27const CAPABILITY_FEATURES: CapabilitiesResult['features'] = {
28  appOwnedSubagents: true,
29  asyncPolling: true,
30  cancellation: true,
31  batchThreadCreation: false,
32  threadManagement: false,
33  incrementalThreadRead: false,
34  scheduledTasks: false,
35  maxBatchThreads: 0,
36}
37
38const CATALOG_CACHE_MS = 10 * 60 * 1000
39const CATALOG_TIMEOUT_MS = 90000
40
41/** The last probe result. One per server, so a process, and a test, starts with none. */
42export type CatalogCache = { entry: { at: number; key: string; instances: ProviderInstance[] } | null }
43
44export function newCatalogCache(): CatalogCache {
45  return { entry: null }
46}
47
48/** The taskId a tool call carries; null when the input has none. */
49function taskIdOf(input: unknown): string | null {
50  return isRecord(input) && typeof input.taskId === 'string' && input.taskId !== '' ? input.taskId : null
51}
52
53/** The provider instances the probe reports for the configured ones, cached for ten minutes. A failed probe is not cached. */
54async function probeInstances($: CoreEngine, config: OrqConfig, cache: CatalogCache): Promise<ProviderInstance[]> {
55  const probeConfig = config.instances.map((i) => ({ instanceId: i.id, driver: i.driver, ...(i.env ? { env: i.env } : {}) }))
56  const key = JSON.stringify(probeConfig)
57  const now = await $.clock.now()
58  const cached = cache.entry
59  if (cached && cached.key === key && now >= cached.at && now - cached.at < CATALOG_CACHE_MS) return cached.instances
60  try {
61    const run = await $.process.run(['python3', `${$.plugin.root}/scripts/catalog.py`], {
62      timeoutMs: CATALOG_TIMEOUT_MS,
63      stdin: key,
64    })
65    if (run.exitCode !== 0) throw new Error(`catalog.py exited ${run.exitCode}`)
66    const instances = parseCatalogInstances(run.stdout)
67    cache.entry = { at: now, key, instances }
68    return instances
69  } catch (err) {
70    await $.ui.log(`orq: catalog probe failed: ${String(err)}`, { to: 'debug' })
71    return []
72  }
73}
74
75/** The providers the probe reports, with the modes a task may run in. The caller knows where those modes come from. */
76export async function capabilitiesAnswer(
77  $: CoreEngine,
78  config: OrqConfig,
79  modes: { runtimeMode: RuntimeMode; interactionMode: InteractionMode },
80  cache: CatalogCache,
81): Promise<CapabilitiesResult> {
82  const instances = await probeInstances($, config, cache)
83  return {
84    parentThreadId: null,
85    inheritedProviderInstanceId: null,
86    inheritedModel: null,
87    ...modes,
88    providers: buildCapabilities(instances),
89    features: CAPABILITY_FEATURES,
90  }
91}
92
93export async function statusAnswer($: CoreEngine, config: OrqConfig, input: unknown): Promise<TaskStatusResult | Failure> {
94  const taskId = taskIdOf(input)
95  if (taskId === null) return fail('invalid_request', `${STATUS_TOOL} needs {"taskId": "<id>"}`)
96  return (await readJobStatus($, config, taskId)) ?? fail('task_not_found', `no task ${taskId}`)
97}
98
99export async function cancelAnswer($: CoreEngine, config: OrqConfig, input: unknown): Promise<TaskCancelResult | Failure> {
100  const taskId = taskIdOf(input)
101  if (taskId === null) return fail('invalid_request', `${CANCEL_TOOL} needs {"taskId": "<id>"}`)
102  const reason = isRecord(input) ? input.reason : undefined
103  if (reason !== undefined && (typeof reason !== 'string' || [...reason].length > CANCEL_REASON_MAX)) {
104    return fail('invalid_request', `reason must be text of at most ${CANCEL_REASON_MAX} characters`)
105  }
106  const job = await readMeta($, config.jobsDir, taskId)
107  if (!job) return fail('task_not_found', `no task ${taskId}`)
108  const status = taskStatusOf(await jobView($, job))
109  if (status !== 'running' && status !== 'waiting') {
110    return fail('task_not_cancellable', `task ${taskId} is ${status}, so it has nothing left to cancel`)
111  }
112  await writeCancel($, job, reason)
113  return { taskId: job.id, status: 'cancel_requested' }
114}
115
116/** One debug line per job in the jobs directory that has no exit file, with its kind and age. */
117export async function logUnfinished($: CoreEngine, config: OrqConfig): Promise<void> {
118  const jobs = await listJobs($, config.jobsDir)
119  if (jobs.length === 0) return
120  const now = await $.clock.now()
121  for (const job of jobs) {
122    if (await $.fs.exists(`${job.dir}/exit`)) continue
123    await $.ui.log(`orq: unfinished job ${job.id} (${job.kind}), started ${formatDuration(now - job.startedAt)} ago`, {
124      to: 'debug',
125    })
126  }
127}
128

1import type { EngineInterface } from 'claude-code'
2
3import type { DriverKind, InstanceConfig, OrqConfig } from './config.ts'
4import { EFFORT_OPTION, fail, isFailure, type Failure, type GitFailure, type InteractionMode, type RuntimeMode, type TaskIsolation, type TaskStatus, type TaskStatusResult, type WorkState } from './contract.ts'
5import { appliedEffort, effortFlag, hasControlCharacters, plainArgument, resolveTarget, runnerPolicy, taskPrompt } from './target.ts'
6import { QUESTION_FILE } from './question.ts'
7import { LAUNCH_GRACE_MS, claimCheckout, rearmLock, releaseLock, type Checkout, type HeldLock, type OwnerProbe } from './lock.ts'
8import { scopedCwd } from './scope.ts'
9import type { TaskHeader } from './task-header.ts'
10import {
11  DRIVER_RUNNER,
12  buildFooter,
13  failureSnippet,
14  isRecord,
15  isWithin,
16  isSafeJobId,
17  parseJobMeta,
18  parseRolloutModelEffort,
19  refusalText,
20  TASK_REFUSAL,
21  rand36,
22  shellWord,
23  type JobRecord,
24  type Runner,
25} from './types.ts'
26import {
27  GIT_TIMEOUT_MS,
28  branchDeleteArgv,
29  branchFor,
30  gitArgv,
31  gitDetail,
32  NO_HOOKS,
33  revParseHeadArgv,
34  revParseToplevelArgv,
35  WORKTREE_INCLUDE_DEADLINE_S,
36  WORKTREE_INCLUDE_TIMEOUT_MS,
37  worktreeAddArgv,
38  worktreePathFor,
39  worktreeRemoveArgv,
40} from './worktree.ts'
41
42export type Engine = Pick<EngineInterface, 'clock' | 'env' | 'fs' | 'plugin' | 'process' | 'session' | 'ui'>
43
44/** The part of `$` the job core calls: the mod passes its `$`, and the MCP server passes `server/node-engine.ts`. */
45export type CoreEngine = {
46  clock: Pick<EngineInterface['clock'], 'now'>
47  env: Pick<EngineInterface['env'], 'get'>
48  fs: Pick<EngineInterface['fs'], 'write' | 'list' | 'exists' | 'stat'> & { read: (path: string) => Promise<string> }
49  plugin: Pick<EngineInterface['plugin'], 'root'>
50  process: Pick<EngineInterface['process'], 'run'>
51  ui: Pick<EngineInterface['ui'], 'log'>
52}
53
54export const LIST_LIMIT = 50
55const PROCESS_TIMEOUT_MS = 30000
56
57/** Who answers a question the worker asks: the parent session, or the human through a form. */
58export type QuestionMode = 'parent' | 'human'
59
60export type LaunchSpec = {
61  kind: string
62  reservedId?: string
63  agentId: string
64  runner: Runner
65  model: string
66  cwd: string
67  prompt: string
68  env: Record<string, string>
69  instanceId: string
70  driver: DriverKind
71  /** Extra runner arguments, one argv item each, written to the job's flags file. */
72  flags?: string[]
73  runtimeMode?: RuntimeMode
74  interactionMode?: InteractionMode
75  /** The reasoning effort `flags` give the CLI, recorded for the answer's footer. */
76  effort?: string
77  /** Recorded in meta.json for a job the server launches. */
78  questions?: QuestionMode
79  /** `worktree` runs the task in its own git worktree on a new branch; absent runs in `cwd` as it is. */
80  isolation?: 'worktree'
81  /** When set, the checkout that holds `cwd` must lie inside one of these resolved roots to get a worktree. */
82  worktreeScope?: string[]
83  /** Set for a job that resumes the worker of another: `cwd` is that job's, and no worktree is made or removed for it. */
84  continuation?: Continuation
85}
86
87/** What a job that answers a question takes over from the job that asked it. */
88export type Continuation = {
89  parent: string
90  /** The session id of the worker, which the supervisor resumes. */
91  session: string
92  /** The worktree of the task, which the job keeps working in; absent for a task that ran in the checkout itself. */
93  worktree?: Pick<WorktreePlan, 'repo' | 'baseSha' | 'branch' | 'worktree'>
94}
95
96export type TaskLaunchSpec = {
97  /** The job kind: `task` for the cc-orq:task agent, which is the default, and `mcp` for the server's delegate. */
98  kind?: string
99  reservedId?: string
100  questions?: QuestionMode
101  agentId: string
102  instance: InstanceConfig
103  model: string
104  cwd: string
105  prompt: string
106  flags: string[]
107  runtimeMode: RuntimeMode
108  interactionMode: InteractionMode
109  effort?: string
110  isolation: TaskIsolation
111  /** The session directory and the allowed roots that scoped `cwd`. */
112  scopeRoots: string[]
113}
114
115export type LaunchResult = { job: JobRecord } | { refusal: string }
116
117/** `workdir` is where the job runs: the worktree, or the same subdirectory of it that `repo` is of `toplevel`. */
118export type WorktreePlan = { repo: string; toplevel: string; baseSha: string; branch: string; worktree: string; workdir: string }
119
120export type WorktreeResult = { plan: WorktreePlan } | { failure: GitFailure }
121
122async function readText($: CoreEngine, path: string): Promise<string | null> {
123  try {
124    return await $.fs.read(path)
125  } catch {
126    return null
127  }
128}
129
130async function isDirectory($: CoreEngine, path: string): Promise<boolean> {
131  try {
132    return (await $.fs.stat(path)).kind === 'dir'
133  } catch {
134    return false
135  }
136}
137
138/** The question a job wrote to question.txt, trimmed; null when the file is missing or blank. */
139export async function readQuestion($: CoreEngine, job: JobRecord): Promise<string | null> {
140  const text = ((await readText($, `${job.dir}/${QUESTION_FILE}`)) ?? '').trim()
141  return text === '' ? null : text
142}
143
144async function exists($: CoreEngine, path: string): Promise<boolean> {
145  try {
146    return await $.fs.exists(path)
147  } catch {
148    return false
149  }
150}
151
152/** Cancel-file body; the supervisor only checks that the file exists. */
153export function cancelText(reason?: string): string {
154  const why = reason?.trim()
155  return `cancelled at ${new Date().toISOString()}${why ? ` why=${JSON.stringify(why)}` : ''}\n`
156}
157
158export async function writeCancel($: CoreEngine, job: JobRecord, reason?: string): Promise<void> {
159  await $.fs.write(`${job.dir}/cancel`, cancelText(reason))
160}
161
162/** Read one job's meta.json. Unsafe ids and unreadable or incomplete metadata give null. */
163export async function readMeta($: CoreEngine, jobsDir: string, id: string): Promise<JobRecord | null> {
164  if (!isSafeJobId(id)) return null
165  const raw = await readText($, `${jobsDir}/${id}/meta.json`)
166  if (raw === null) return null
167  try {
168    return parseJobMeta(JSON.parse(raw), jobsDir, id)
169  } catch {
170    return null
171  }
172}
173
174/** The jobs in the directory, newest first, at most LIST_LIMIT. */
175export async function listJobs($: CoreEngine, jobsDir: string): Promise<JobRecord[]> {
176  let entries: { name: string; kind: string }[]
177  try {
178    entries = await $.fs.list(jobsDir)
179  } catch {
180    return []
181  }
182  const metas = await Promise.all(
183    entries.filter((entry) => entry.kind === 'dir').map((entry) => readMeta($, jobsDir, entry.name)),
184  )
185  return metas
186    .filter((job): job is JobRecord => job !== null)
187    .sort((a, b) => b.startedAt - a.startedAt)
188    .slice(0, LIST_LIMIT)
189}
190
191async function mtimeMs($: CoreEngine, path: string): Promise<number | null> {
192  try {
193    return (await $.fs.stat(path)).mtimeMs
194  } catch {
195    return null
196  }
197}
198
199/** The exit file holds one whole number. Anything else, an empty file included, is -1, which counts as failed. */
200export function parseExit(text: string | null): number {
201  return text !== null && /^\s*\d+\s*$/.test(text) ? Number.parseInt(text, 10) : -1
202}
203
204export type Finished = {
205  status: 'done' | 'failed' | 'cancelled'
206  exitCode: number
207  endedAt: number | null
208  lastText: string
209  summary: StreamSummary | null
210  /** The question a job left in question.txt when it exited 0 and stopped there; null otherwise. */
211  question: string | null
212}
213
214/**
215 * The one rule for a job that has ended, shared by /orq list and /orq result. Exit 124 is cancelled.
216 * A clean exit 0 that left a question is done, and its answer is the question. Otherwise the job is
217 * done only when the exit is 0, last.txt holds a non-empty message, and the extractor reports no
218 * error; anything else is failed.
219 */
220export async function finishedOutcome($: CoreEngine, job: JobRecord, exitText: string | null): Promise<Finished> {
221  const exitCode = parseExit(exitText)
222  const endedAt = await mtimeMs($, `${job.dir}/exit`)
223  if (exitCode === 124) return { status: 'cancelled', exitCode, endedAt, lastText: '', summary: null, question: null }
224  const question = await readQuestion($, job)
225  if (exitCode === 0 && question !== null) {
226    return { status: 'done', exitCode, endedAt, lastText: '', summary: null, question }
227  }
228  const lastText = (await readText($, `${job.dir}/last.txt`)) ?? ''
229  if (exitCode !== 0 || lastText.trim() === '') {
230    return { status: 'failed', exitCode, endedAt, lastText, summary: null, question: null }
231  }
232  const summary = job.runner ? await summarizeEvents($, job, job.runner) : null
233  const status = summary?.error ? 'failed' : 'done'
234  return { status, exitCode, endedAt, lastText, summary, question: null }
235}
236
237export type JobView =
238  | { status: 'running' | 'cancelling' | 'interrupted' | 'waiting'; endedAt: null }
239  | Finished
240
241export async function jobView($: CoreEngine, job: JobRecord): Promise<JobView> {
242  const exit = await readText($, `${job.dir}/exit`)
243  if (exit !== null) return finishedOutcome($, job, exit)
244  const liveness = await jobLiveness($, job)
245  if (liveness === 'gone') return { status: 'interrupted', endedAt: null }
246  if (liveness === 'running') {
247    if ((await readQuestion($, job)) !== null) return { status: 'waiting', endedAt: null }
248    return { status: (await exists($, `${job.dir}/cancel`)) ? 'cancelling' : 'running', endedAt: null }
249  }
250  return finishedOutcome($, job, await readText($, `${job.dir}/exit`))
251}
252
253function mkdirScript(jobsDir: string, dir: string): string {
254  return `# orq:mkdir\nmkdir -p ${shellWord(jobsDir)} || exit 8\nmkdir ${shellWord(dir)} || exit 7`
255}
256
257/**
258 * Stands in for `setsid` where the binary is missing, as on macOS. Python starts with SIGPIPE and SIGXFSZ
259 * ignored, and an exec keeps an ignored signal ignored, so both go back to their defaults before the exec.
260 */
261const PYTHON_SETSID =
262  'import os, signal, sys; signal.signal(signal.SIGPIPE, signal.SIG_DFL); signal.signal(signal.SIGXFSZ, signal.SIG_DFL); os.setsid(); os.execvp(sys.argv[1], sys.argv[1:])'
263
264/**
265 * Start script for one job. The instance env is passed as --setenv for the
266 * systemd unit and as `env K=V` for the setsid launchers, so both reach the CLI.
267 */
268export function startScript(args: {
269  id: string
270  dir: string
271  supervise: string
272  runner: Runner
273  cwd: string
274  model: string
275  env: Record<string, string>
276}): string {
277  const supervisorArgs = [args.supervise, args.dir, args.runner, args.cwd, args.model]
278  const supervisorCmd = `bash ${supervisorArgs.map(shellWord).join(' ')} </dev/null >${shellWord(`${args.dir}/child.log`)} 2>&1`
279  const env = { ...args.env, ORQ_JOB_DIR: args.dir }
280  const pairs = Object.entries(env).map(([k, v]) => shellWord(`${k}=${v}`))
281  const setenv = pairs.map((pair) => `--setenv=${pair}`).join(' ')
282  const envAssign = pairs.join(' ')
283  const scope = shellWord(`${args.dir}/scope`)
284  // A job in the caller's cgroup dies with the terminal or session that launched it; a transient user
285  // unit gets its own, and stopping it also reaps children that escaped the process group. setsid is
286  // the fallback when no user manager is reachable.
287  return (
288    `# orq:start\n` +
289    `if systemd-run --user --quiet --collect --unit=${shellWord(`orq-${args.id}`)} --setenv=PATH="$PATH" --setenv=HOME="$HOME" ${setenv} ` +
290    `bash -c ${shellWord(`exec ${supervisorCmd}`)} 2>/dev/null; then echo systemd > ${scope}\n` +
291    `elif command -v setsid >/dev/null 2>&1; then echo setsid > ${scope}; setsid env ${envAssign} ${supervisorCmd} &\n` +
292    `else echo python-setsid > ${scope}; python3 -I -c ${shellWord(PYTHON_SETSID)} env ${envAssign} ${supervisorCmd} &\n` +
293    `fi\n` +
294    `for i in $(seq 1 10); do [ -s ${shellWord(`${args.dir}/pgid`)} ] && exit 0; sleep 0.5; done\n` +
295    `exit 11`
296  )
297}
298
299async function markLaunchFailed($: CoreEngine, dir: string): Promise<void> {
300  try {
301    await $.fs.write(`${dir}/exit`, '125')
302  } catch {
303    /* the refusal the caller returns is the signal */
304  }
305}
306
307async function debugLog($: CoreEngine, text: string): Promise<void> {
308  try {
309    await $.ui.log(text, { to: 'debug' })
310  } catch {
311    /* the refusal the caller returns is the signal */
312  }
313}
314
315/**
316 * Best effort after a launch failed once its worktree existed: remove the worktree, then its branch.
317 * A failed undo goes to the debug log, and the refusal still names the launch's own failure.
318 */
319async function unwindWorktree($: CoreEngine, plan: WorktreePlan): Promise<void> {
320  const undo: [string, string[]][] = [
321    ['remove the worktree', worktreeRemoveArgv(plan.repo, plan.worktree)],
322    ['delete the branch', branchDeleteArgv(plan.repo, plan.branch)],
323  ]
324  for (const [what, argv] of undo) {
325    try {
326      // A worktree can hold a copied node_modules, so its removal gets the copy's time.
327      const r = await $.process.run(argv, { timeoutMs: WORKTREE_INCLUDE_TIMEOUT_MS })
328      if (r.exitCode !== 0) await debugLog($, `orq: could not ${what} after a failed launch${gitDetail(r.stderr)}`)
329    } catch (err) {
330      await debugLog($, `orq: could not ${what} after a failed launch: ${String(err)}`)
331    }
332  }
333}
334
335/**
336 * The id of a live job that runs inside `checkout` or is a worktree of it, or null. A job is live while its
337 * supervisor runs. The lock covers the jobs that claimed one; this covers the rest, the jobs 0.1 launched and
338 * the worktree jobs, and it reads the job directory rather than memory, so it holds across sessions. A job
339 * stores the path its launch was given, so `spelling`, the path of this launch, finds a job stored under a
340 * symlink that leads to the checkout, as 0.1 did. `rearm` runs before each job is read, so a scan of a large
341 * jobs directory can keep the launch's lock young.
342 */
343export async function checkoutBusy(
344  $: CoreEngine,
345  jobsDir: string,
346  checkout: Checkout,
347  launch: { spelling?: string; rearm?: () => Promise<unknown> } = {},
348): Promise<string | null> {
349  let entries: { name: string; kind: string }[]
350  try {
351    entries = await $.fs.list(jobsDir)
352  } catch {
353    return null
354  }
355  const roots = launch.spelling === undefined ? [checkout.key] : [checkout.key, launch.spelling]
356  // Outside a repository the directory itself is the checkout, so only a job in that very directory counts.
357  const holds = (path: string, root: string): boolean => (checkout.repository ? isWithin(path, root) : path === root)
358  const inside = (path: string | undefined): boolean => path !== undefined && roots.some((root) => holds(path, root))
359  for (const entry of entries) {
360    if (entry.kind !== 'dir') continue
361    await launch.rearm?.()
362    const job = await readMeta($, jobsDir, entry.name)
363    if (job === null || !(inside(job.cwd) || inside(job.repo))) continue
364    if ((await jobLiveness($, job)) === 'running') return job.id
365  }
366  return null
367}
368
369/** Where the job that owns a lock stands: no exit file and no pgid yet is a launch that has not started its supervisor. */
370function ownerStateProbe($: CoreEngine, jobsDir: string): OwnerProbe {
371  return async (id) => {
372    const dir = `${jobsDir}/${id}`
373    if (await exists($, `${dir}/exit`)) return 'finished'
374    const pgid = ((await readText($, `${dir}/pgid`)) ?? '').trim()
375    return /^\d+$/.test(pgid) ? jobLiveness($, { dir }) : 'unstarted'
376  }
377}
378
379function busyRefusal(kind: string, cwd: string, owner: string | null): string {
380  return `orq: ${kind} refused: worktree_busy: ${cwd} is in use by ${owner === null ? 'a launch in progress' : `job ${owner}`}`
381}
382
383/**
384 * Best effort: the script stages the copy beside the worktree and renames it in only when it is complete, so a
385 * copy that fails leaves the worktree as git made it, which is what a checkout without .worktreeinclude gets.
386 */
387async function copyWorktreeIncludes($: CoreEngine, toplevel: string, worktree: string): Promise<void> {
388  const script = `${$.plugin.root}/scripts/worktree-include.py`
389  try {
390    const argv = ['python3', script, toplevel, worktree, String(WORKTREE_INCLUDE_DEADLINE_S)]
391    const r = await $.process.run(argv, { timeoutMs: WORKTREE_INCLUDE_TIMEOUT_MS })
392    if (r.exitCode === 1) {
393      await debugLog($, `orq: .worktreeinclude copy into ${worktree} failed and was undone${gitDetail(r.stderr)}`)
394      return
395    }
396    if (r.exitCode === 3) {
397      await debugLog($, `orq: .worktreeinclude copy into ${worktree} failed and part of it is still there${gitDetail(r.stderr)}`)
398      return
399    }
400    if (r.exitCode !== 0) throw new Error(`exit ${r.exitCode}`)
401    const counts = JSON.parse(r.stdout) as { copied?: unknown; skipped?: unknown }
402    if (typeof counts.copied === 'number' && typeof counts.skipped === 'number' && counts.copied + counts.skipped > 0) {
403      await debugLog($, `orq: .worktreeinclude copied ${counts.copied} and skipped ${counts.skipped} file(s) into ${worktree}`)
404    }
405  } catch (err) {
406    // A killed copy can leave its staging directory beside the worktree, never a partial tree inside it.
407    await debugLog($, `orq: .worktreeinclude copy into ${worktree} did not finish: ${String(err)}`)
408    try {
409      await $.process.run(['python3', script, '--discard', worktree], { timeoutMs: WORKTREE_INCLUDE_TIMEOUT_MS })
410    } catch {
411      /* the staging directory stays beside the worktree, outside it */
412    }
413  }
414}
415
416/**
417 * Plan and create the worktree of task TASK_ID: REPO_PARENT/REPO_NAME-wt/TASK_ID on a new branch
418 * orq/TASK_ID, cut from the checkout's HEAD. Refuses, and creates nothing, when cwd is not a git
419 * checkout. The path names the id the launch has just claimed, so no other job can hold it.
420 */
421export async function prepareWorktree($: CoreEngine, cwd: string, taskId: string, scope?: string[]): Promise<WorktreeResult> {
422  try {
423    const top = await $.process.run(revParseToplevelArgv(cwd), { timeoutMs: GIT_TIMEOUT_MS })
424    if (top.exitCode !== 0) {
425      return { failure: { code: 'not_a_repository', message: `${cwd} is not inside a git checkout${gitDetail(top.stderr)}` } }
426    }
427    const toplevel = top.stdout.trim()
428    // The worktree copies the whole checkout, so a checkout that reaches above the scope is refused.
429    if (scope !== undefined && !scope.some((root) => isWithin(toplevel, root))) {
430      return {
431        failure: {
432          code: 'cwd_out_of_scope',
433          message: `the git checkout ${toplevel} that holds ${cwd} is outside the session directory and the allowed roots`,
434        },
435      }
436    }
437    const head = await $.process.run(revParseHeadArgv(cwd), { timeoutMs: GIT_TIMEOUT_MS })
438    const baseSha = head.stdout.trim()
439    if (head.exitCode !== 0 || baseSha === '') {
440      return { failure: { code: 'git_failed', message: `${cwd} has no HEAD commit to start a worktree from${gitDetail(head.stderr)}` } }
441    }
442    const worktree = worktreePathFor(toplevel, taskId)
443    const branch = branchFor(taskId)
444    const subdir = cwd !== toplevel && isWithin(cwd, toplevel) ? cwd.slice(toplevel.length) : ''
445    const plan: WorktreePlan = { repo: cwd, toplevel, baseSha, branch, worktree, workdir: `${worktree}${subdir}` }
446    try {
447      const add = await $.process.run(worktreeAddArgv(cwd, worktree, branch, baseSha), { timeoutMs: GIT_TIMEOUT_MS })
448      if (add.exitCode !== 0) {
449        return { failure: { code: 'git_failed', message: `git worktree add ${worktree} failed${gitDetail(add.stderr)}` } }
450      }
451    } catch (err) {
452      // A timeout or a kill can leave a half-made worktree and its branch behind.
453      await unwindWorktree($, plan)
454      return { failure: { code: 'git_failed', message: `git worktree add ${worktree} did not finish: ${String(err)}` } }
455    }
456    // The worktree holds HEAD only, so an untracked or new cwd has no counterpart there.
457    if (!(await isDirectory($, plan.workdir))) {
458      await unwindWorktree($, plan)
459      return {
460        failure: {
461          code: 'invalid_request',
462          message: `${cwd} is not in HEAD of its checkout, so the worktree has no such directory; commit it, or use isolation: shared`,
463        },
464      }
465    }
466    await copyWorktreeIncludes($, toplevel, worktree)
467    return { plan }
468  } catch (err) {
469    return { failure: { code: 'git_failed', message: `git could not run: ${String(err)}` } }
470  }
471}
472
473export function newJobId(kind: string, now: number, agentId: string): string {
474  return `${kind}-${now}-${agentId.slice(-8)}-${rand36()}`
475}
476
477/** Create a job directory, write meta.json and prompt.txt, and start the supervisor detached. */
478export async function launchJob($: CoreEngine, config: OrqConfig, spec: LaunchSpec): Promise<LaunchResult> {
479  const resumed = spec.continuation
480  const now = await $.clock.now()
481  const supervise = `${$.plugin.root}/scripts/supervise.sh`
482  const ownerOf = ownerStateProbe($, config.jobsDir)
483  for (let attempt = 0; attempt < 2; attempt += 1) {
484    const id = spec.reservedId ?? newJobId(spec.kind, now, spec.agentId)
485    const dir = `${config.jobsDir}/${id}`
486    let mkdirCode: number
487    try {
488      const r = await $.process.run(['bash', '-c', mkdirScript(config.jobsDir, dir)], { timeoutMs: PROCESS_TIMEOUT_MS })
489      mkdirCode = r.exitCode
490    } catch (err) {
491      return { refusal: `orq: ${spec.kind} could not create its job directory: ${String(err)}` }
492    }
493    if (mkdirCode === 7 && attempt === 0 && spec.reservedId === undefined) continue
494    if (mkdirCode !== 0) {
495      return { refusal: `orq: ${spec.kind} could not create its job directory (exit ${mkdirCode})` }
496    }
497    // The id is claimed by the mkdir, so the worktree path can name it. Nothing is created before the claim.
498    let cwd = spec.cwd
499    let plan: WorktreePlan | null = null
500    let held: HeldLock | null = null
501    const abandon = async (refusal: string): Promise<LaunchResult> => {
502      await markLaunchFailed($, dir)
503      if (plan) await unwindWorktree($, plan)
504      if (held) await releaseLock($, held, ownerOf)
505      return { refusal }
506    }
507    if (resumed?.worktree !== undefined) {
508      // The worktree is the task's, and no other job runs in it, so there is nothing to create, lock or undo.
509    } else if (spec.isolation === 'worktree') {
510      const prepared = await prepareWorktree($, spec.cwd, id, spec.worktreeScope)
511      if ('failure' in prepared) {
512        return abandon(`orq: ${spec.kind} refused: ${prepared.failure.code}: ${prepared.failure.message}`)
513      }
514      plan = prepared.plan
515      cwd = plan.workdir
516    } else {
517      // The scan after the claim sees what holds no lock: jobs 0.1 launched, and worktree jobs.
518      const claim = await claimCheckout($, config.jobsDir, id, spec.cwd, ownerOf)
519      if (claim.kind === 'failed') return abandon(`orq: ${spec.kind} could not lock its checkout (${claim.detail})`)
520      if (claim.kind === 'busy') return abandon(busyRefusal(spec.kind, spec.cwd, claim.owner))
521      const lock = claim.held
522      held = lock
523      const busy = await checkoutBusy($, config.jobsDir, claim.checkout, { spelling: spec.cwd, rearm: () => rearmLock($, lock) })
524      if (busy !== null) return abandon(busyRefusal(spec.kind, spec.cwd, busy))
525      // A scan that outlasted the grace may have let another launcher break this lock and start its job.
526      if (!(await rearmLock($, lock, { force: true }))) return abandon(busyRefusal(spec.kind, spec.cwd, null))
527    }
528    const worktreeFields = plan
529      ? { repo: plan.repo, baseSha: plan.baseSha, branch: plan.branch, worktree: plan.worktree }
530      : (resumed?.worktree ?? {})
531    const startedAtIso = new Date(now).toISOString()
532    const meta = {
533      id,
534      kind: spec.kind,
535      agentId: spec.agentId,
536      cwd,
537      startedAt: now,
538      startedAtIso,
539      runner: spec.runner,
540      model: spec.model,
541      instanceId: spec.instanceId,
542      driver: spec.driver,
543      ...(spec.env.CODEX_HOME !== undefined ? { codexHome: spec.env.CODEX_HOME } : {}),
544      ...(spec.runtimeMode !== undefined ? { runtimeMode: spec.runtimeMode } : {}),
545      ...(spec.interactionMode !== undefined ? { interactionMode: spec.interactionMode } : {}),
546      ...(spec.effort !== undefined ? { effort: spec.effort } : {}),
547      ...(spec.questions !== undefined ? { questions: spec.questions } : {}),
548      ...(resumed !== undefined ? { parent: resumed.parent } : {}),
549      ...worktreeFields,
550    }
551    try {
552      await $.fs.write(`${dir}/meta.json`, JSON.stringify(meta))
553      await $.fs.write(`${dir}/prompt.txt`, spec.prompt)
554      if (spec.flags !== undefined && spec.flags.length > 0) {
555        await $.fs.write(`${dir}/flags`, spec.flags.map((flag) => `${flag}\n`).join(''))
556      }
557      if (resumed !== undefined) await $.fs.write(`${dir}/resume`, `${resumed.session}\n`)
558    } catch (err) {
559      return abandon(`orq: ${spec.kind} could not write its job directory: ${String(err)}`)
560    }
561    const script = startScript({
562      id,
563      dir,
564      supervise,
565      runner: spec.runner,
566      cwd,
567      model: spec.model,
568      env: spec.env,
569    })
570    // The job files above may have taken longer than the grace, so another launcher may have broken this lock since the last touch.
571    if (held && !(await rearmLock($, held, { force: true }))) return abandon(busyRefusal(spec.kind, spec.cwd, null))
572    let startCode: number
573    try {
574      const r = await $.process.run(['bash', '-c', script], { timeoutMs: PROCESS_TIMEOUT_MS })
575      startCode = r.exitCode
576    } catch (err) {
577      return abandon(`orq: ${spec.kind} could not start its job: ${String(err)}`)
578    }
579    if (startCode !== 0) {
580      return abandon(
581        startCode === 11
582          ? `orq: ${spec.kind} failed to start (supervisor did not report within 5 s)`
583          : `orq: ${spec.kind} failed to start (exit ${startCode})`,
584      )
585    }
586    const pgid = ((await readText($, `${dir}/pgid`)) ?? '').trim()
587    const job: JobRecord = {
588      id,
589      kind: spec.kind,
590      agentId: spec.agentId,
591      dir,
592      cwd,
593      startedAt: now,
594      startedAtIso,
595      runner: spec.runner,
596      model: spec.model,
597      instanceId: spec.instanceId,
598      driver: spec.driver,
599      ...(spec.env.CODEX_HOME !== undefined ? { codexHome: spec.env.CODEX_HOME } : {}),
600      ...(spec.effort !== undefined ? { effort: spec.effort } : {}),
601      ...(resumed !== undefined ? { parent: resumed.parent } : {}),
602      ...worktreeFields,
603    }
604    if (pgid) job.pgid = pgid
605    return { job }
606  }
607  return { refusal: `orq: ${spec.kind} could not pick a job id without collision` }
608}
609
610/**
611 * Launch one cc-orq:task job. The model and each runner flag reach the CLI as one argv item, so a value
612 * that could read as an option or span lines is refused before anything is written.
613 */
614export async function launchTaskJob($: CoreEngine, config: OrqConfig, spec: TaskLaunchSpec): Promise<LaunchResult> {
615  if (!plainArgument(spec.model)) {
616    return { refusal: `${TASK_REFUSAL}: the model must be one word that does not start with "-" and has no control characters.` }
617  }
618  if (spec.flags.some((flag) => flag === '' || hasControlCharacters(flag))) {
619    return { refusal: `${TASK_REFUSAL}: a runner flag is empty or contains a control character.` }
620  }
621  const kind = spec.kind ?? 'task'
622  const launched = await launchJob($, config, {
623    kind,
624    ...(spec.reservedId !== undefined ? { reservedId: spec.reservedId } : {}),
625    agentId: spec.agentId,
626    runner: DRIVER_RUNNER[spec.instance.driver],
627    model: spec.model,
628    cwd: spec.cwd,
629    prompt: spec.prompt,
630    env: spec.instance.env ?? {},
631    instanceId: spec.instance.id,
632    driver: spec.instance.driver,
633    flags: spec.flags,
634    runtimeMode: spec.runtimeMode,
635    interactionMode: spec.interactionMode,
636    ...(spec.effort !== undefined ? { effort: spec.effort } : {}),
637    ...(spec.questions !== undefined ? { questions: spec.questions } : {}),
638    ...(spec.isolation === 'worktree' ? { isolation: 'worktree' as const, worktreeScope: spec.scopeRoots } : {}),
639  })
640  if (!('refusal' in launched)) return launched
641  // launchJob words its refusals for any kind; a task's are answered as task refusals.
642  const own = `orq: ${kind} `
643  const refusal = launched.refusal.startsWith(own)
644    ? `${TASK_REFUSAL}: ${launched.refusal.slice(own.length).replace(/^refused: /, '')}`
645    : launched.refusal
646  const hint = spec.isolation === 'worktree' && refusal.includes('not_a_repository')
647    ? '. Pass `isolation: shared` to run the task in this directory as it is.'
648    : ''
649  return { refusal: refusal + hint }
650}
651
652export type TaskLaunchArgs = {
653  agentId: string
654  header: TaskHeader
655  ceiling: { runtimeMode: RuntimeMode; interactionMode: InteractionMode }
656  sessionDir: string
657  kind?: string
658  reservedId?: string
659  questions?: QuestionMode
660}
661
662/** A refused header comes back as `{ refusal }`, not as a throw. */
663export async function planTaskLaunch($: CoreEngine, config: OrqConfig, args: TaskLaunchArgs): Promise<{ spec: TaskLaunchSpec } | { refusal: string }> {
664  const { agentId, header, ceiling, sessionDir, kind, reservedId, questions } = args
665  const resolved = resolveTarget({
666    target: header.target,
667    instances: config.instances,
668    parentRuntimeMode: ceiling.runtimeMode,
669    parentInteractionMode: ceiling.interactionMode,
670    runtimeMode: header.runtimeMode ?? 'inherit',
671    interactionMode: header.interactionMode ?? 'inherit',
672  })
673  if (isFailure(resolved)) return { refusal: refusalText(resolved) }
674  const policy = runnerPolicy(resolved.instance.driver, resolved.runtimeMode, resolved.interactionMode)
675  if (isFailure(policy)) return { refusal: refusalText(policy) }
676  const scoped = await scopedCwd($, config, sessionDir, header.cwd)
677  if ('refusal' in scoped) return scoped
678  const effort = appliedEffort(resolved.instance.driver, resolved.options)
679  return {
680    spec: {
681      ...(kind !== undefined ? { kind } : {}),
682      ...(reservedId !== undefined ? { reservedId } : {}),
683      ...(questions !== undefined ? { questions } : {}),
684      agentId,
685      instance: resolved.instance,
686      model: resolved.model,
687      cwd: scoped.cwd,
688      prompt: taskPrompt(header.prompt, header.role),
689      flags: [...policy.flags, ...effortFlag(resolved.instance.driver, resolved.options)],
690      runtimeMode: resolved.runtimeMode,
691      interactionMode: resolved.interactionMode,
692      ...(effort !== undefined ? { effort } : {}),
693      isolation: header.isolation,
694      scopeRoots: scoped.roots,
695    },
696  }
697}
698
699export async function launchTaskFromHeader($: CoreEngine, config: OrqConfig, args: TaskLaunchArgs): Promise<LaunchResult> {
700  const planned = await planTaskLaunch($, config, args)
701  return 'refusal' in planned ? planned : launchTaskJob($, config, planned.spec)
702}
703
704/** finished | running | gone: the exit file, else the supervisor's process group. Rejects when the probe cannot run. */
705async function probeLiveness($: CoreEngine, job: Pick<JobRecord, 'dir'>): Promise<'finished' | 'running' | 'gone'> {
706  if (await exists($, `${job.dir}/exit`)) return 'finished'
707  const script =
708    `# orq:liveness\npgid=$(cat ${shellWord(`${job.dir}/pgid`)} 2>/dev/null | tr -d ' \\t\\n')\n` +
709    `case "$pgid" in ''|*[!0-9]*) exit 12 ;; esac\n` +
710    `kill -0 -"$pgid" 2>/dev/null && exit 11\n` +
711    `[ -e ${shellWord(`${job.dir}/exit`)} ] && exit 0 || exit 12`
712  const r = await $.process.run(['bash', '-c', script], { timeoutMs: 15000 })
713  if (r.exitCode === 0) return 'finished'
714  if (r.exitCode === 11) return 'running'
715  return 'gone'
716}
717
718/** finished | running | gone: the exit file, else the supervisor's process group. A probe that cannot run counts as running. */
719export async function jobLiveness($: CoreEngine, job: Pick<JobRecord, 'dir'>): Promise<'finished' | 'running' | 'gone'> {
720  try {
721    return await probeLiveness($, job)
722  } catch {
723    return 'running'
724  }
725}
726
727/**
728 * Whether a job may still be working, for a caller that must not act under a live job. jobLiveness reads a job
729 * with no pgid yet as gone, though its supervisor is about to write one, and a probe that cannot run as running
730 * for good. Here a job younger than LAUNCH_GRACE_MS counts as running in both states, and an older one as gone.
731 */
732export async function jobIsLive($: CoreEngine, job: JobRecord): Promise<boolean> {
733  if (await exists($, `${job.dir}/exit`)) return false
734  const starting = (await $.clock.now()) - job.startedAt < LAUNCH_GRACE_MS
735  const pgid = ((await readText($, `${job.dir}/pgid`)) ?? '').trim()
736  if (!/^\d+$/.test(pgid)) return starting
737  try {
738    return (await probeLiveness($, job)) === 'running'
739  } catch {
740    return starting
741  }
742}
743
744export type StreamSummary = { tid: string; in: number | null; out: number | null; error: boolean; model?: string }
745
746function parseSummary(raw: unknown): StreamSummary | null {
747  if (!isRecord(raw)) return null
748  return {
749    tid: typeof raw.tid === 'string' ? raw.tid : '',
750    in: typeof raw.in === 'number' ? raw.in : null,
751    out: typeof raw.out === 'number' ? raw.out : null,
752    error: raw.error === true,
753    ...(typeof raw.model === 'string' && raw.model !== '' ? { model: raw.model } : {}),
754  }
755}
756
757/** Run scripts/extract.py over the job's events; null when extraction fails. */
758export async function summarizeEvents($: CoreEngine, job: JobRecord, runner: Runner): Promise<StreamSummary | null> {
759  try {
760    const r = await $.process.run(
761      ['python3', `${$.plugin.root}/scripts/extract.py`, `${job.dir}/events.jsonl`, runner],
762      { timeoutMs: PROCESS_TIMEOUT_MS },
763    )
764    if (r.exitCode !== 0) return null
765    const line = r.stdout.trim().split('\n').pop() ?? ''
766    return parseSummary(JSON.parse(line))
767  } catch {
768    return null
769  }
770}
771
772/** The last 20 stderr lines of a job, redacted; empty when none can be read. */
773export async function stderrSnippet($: CoreEngine, job: JobRecord): Promise<string> {
774  try {
775    const script = `# orq:stderr\ntail -n 20 ${shellWord(`${job.dir}/stderr.txt`)} 2>/dev/null; exit 0`
776    const r = await $.process.run(['bash', '-c', script], { timeoutMs: 15000 })
777    const text = r.stdout.trim()
778    return text ? failureSnippet(text) : ''
779  } catch {
780    return ''
781  }
782}
783
784/** Model and effort of a Codex job, read from its rollout under the instance's CODEX_HOME. */
785async function codexModelEffort(
786  $: CoreEngine,
787  config: OrqConfig,
788  job: JobRecord,
789  threadId: string,
790): Promise<{ model: string; effort: string }> {
791  const fallback = { model: job.model ?? '', effort: '' }
792  if (!threadId) return fallback
793  try {
794    const home = (await $.env.get('HOME')) ?? ''
795    const instance = config.instances.find((i) => i.id === job.instanceId)
796    const codexHome = job.codexHome ?? instance?.env?.CODEX_HOME ?? (home ? `${home}/.codex` : '')
797    if (!codexHome) return fallback
798    const probe =
799      `# orq:rollout\nfind -L ${shellWord(`${codexHome}/sessions`)} -name ${shellWord(`rollout-*-${threadId}.jsonl`)} ` +
800      `2>/dev/null | sort | tail -n 1`
801    const r = await $.process.run(['bash', '-c', probe], { timeoutMs: PROCESS_TIMEOUT_MS })
802    const path = r.stdout.trim()
803    if (!path) return fallback
804    const parsed = parseRolloutModelEffort(await $.fs.read(path))
805    return { model: parsed.model || fallback.model, effort: parsed.effort }
806  } catch {
807    return fallback
808  }
809}
810
811/** The one-line answer for a job that has not ended, or was interrupted. */
812export function notEndedText(id: string, status: 'running' | 'cancelling' | 'interrupted' | 'waiting'): string {
813  if (status === 'running') return `orq: ${id} is still running`
814  if (status === 'cancelling') return `orq: ${id} is still cancelling`
815  if (status === 'waiting') return `orq: ${id} is waiting for an answer to its question`
816  return `orq: ${id} was interrupted: its supervisor is gone and it left no exit file`
817}
818
819/**
820 * The answer for a job that stopped on a question. The parent answers it with the answer tool, which resumes the same worker
821 * in a new job, so this job's text stays as it is. A job that recorded no session cannot be resumed, and the old way, a new
822 * task with the brief and the answer, is the one left.
823 */
824export function blockedText(id: string, question: string, state: { next: Next | null; form: boolean; resumable: boolean }): string {
825  const head = `BLOCKED on a question (taskId ${id}): ${question}`
826  if (state.next !== null) {
827    return `${head}\n\nThe question was answered: its worker continues in taskId ${state.next.jobId}. Read it with status; if it failed, delegate anew.`
828  }
829  if (state.form) {
830    return `${head}\n\nThe user is answering it in a form; the answer tool takes over once the form is declined, cancelled or 24 hours old.`
831  }
832  const way = state.resumable
833    ? 'Answer it with the answer tool; the same worker resumes its session.'
834    : 'This worker cannot be resumed (it recorded no session), so answer it in a new delegate with the original brief plus the answer.'
835  return `${head}\n\n${way}`
836}
837
838/** A form left open for this long ends the task with the BLOCKED text. */
839const FORM_WINDOW_MS = 24 * 60 * 60 * 1000
840
841/** The worker's question stays a form until it is declined, the task is cancelled, 24 hours pass, or it has no session to resume. */
842export async function formIsOpen($: CoreEngine, job: JobRecord, view: { endedAt: number | null }): Promise<boolean> {
843  if ((await readSession($, job)) === null || (await $.fs.exists(`${job.dir}/declined`))) return false
844  return (await $.clock.now()) - (view.endedAt ?? 0) < FORM_WINDOW_MS
845}
846
847/** The BLOCKED text as the answer tool will treat this question now: answered, open in a form, resumable, or not. */
848async function blockedAnswer($: CoreEngine, job: JobRecord, question: string, view: { endedAt: number | null }): Promise<string> {
849  return blockedText(job.id, question, {
850    next: await readNext($, job),
851    form: job.questions === 'human' && (await formIsOpen($, job, view)),
852    resumable: (await readSession($, job)) !== null,
853  })
854}
855
856export async function composeAnswer($: CoreEngine, config: OrqConfig, job: JobRecord, knownView?: JobView): Promise<string> {
857  const view = knownView ?? (await jobView($, job))
858  if (!('exitCode' in view)) return notEndedText(job.id, view.status)
859  if (view.status === 'cancelled') return `orq: ${job.id} was cancelled (exit 124)`
860  if (view.question !== null) return blockedAnswer($, job, view.question, view)
861  if (view.status === 'failed') {
862    const snippet = await stderrSnippet($, job)
863    return `orq: ${job.id} failed (exit ${view.exitCode})` + (snippet ? `\n${snippet}` : '')
864  }
865  const summary = view.summary
866  const now = await $.clock.now()
867  const durationMs = Math.max(0, (view.endedAt ?? now) - job.startedAt)
868  const runner = job.runner
869  let model = summary?.model || job.model || 'n/a'
870  let effort = job.effort || 'n/a'
871  if (runner === 'codex') {
872    const rollout = await codexModelEffort($, config, job, summary?.tid ?? '')
873    model = rollout.model || 'n/a'
874    effort = rollout.effort || effort
875  }
876  const footer = buildFooter({
877    kind: job.kind,
878    model,
879    effort,
880    durationMs,
881    inTokens: summary?.in ?? null,
882    outTokens: summary?.out ?? null,
883    taskId: job.id,
884    session: summary?.tid || job.id,
885  })
886  return `${view.lastText}\n\n${footer}`
887}
888
889/**
890 * Parse `[--worktree] INSTANCE MODEL -- PROMPT`; the prompt is everything after the first ` -- `.
891 * The flag sits before the instance and is reported as `worktree: true` only when given.
892 */
893export function parseLaunch(
894  rest: string,
895): { instance: string; model: string; prompt: string; worktree?: true } | { error: string } {
896  const trimmed = rest.trim()
897  const flag = /^--worktree(?:\s+|$)/.exec(trimmed)
898  const m = /^(\S+)\s+(\S+)\s+--(?:\s+([\s\S]*))?$/.exec(flag ? trimmed.slice(flag[0].length) : trimmed)
899  if (!m) return { error: 'usage: /orq launch [--worktree] INSTANCE MODEL -- PROMPT' }
900  const instance = m[1] ?? ''
901  const model = m[2] ?? ''
902  const prompt = (m[3] ?? '').trim()
903  if (model.startsWith('-')) return { error: 'the model must not start with "-"' }
904  if (prompt === '') return { error: 'the prompt after -- is empty' }
905  return flag ? { instance, model, prompt, worktree: true } : { instance, model, prompt }
906}
907
908/** The task status a job view answers with. A cancel marker on a live job still reads as running. */
909export function taskStatusOf(view: JobView): TaskStatus {
910  switch (view.status) {
911    case 'running':
912    case 'cancelling':
913      return 'running'
914    case 'done':
915      return 'completed'
916    case 'failed':
917      return 'failed'
918    case 'cancelled':
919      return 'cancelled'
920    case 'interrupted':
921      return 'interrupted'
922    case 'waiting':
923      return 'waiting'
924  }
925}
926
927/** A task that has ended has a final result or error for its parent to read: result_available. */
928function workStateOf(status: TaskStatus): WorkState {
929  if (status === 'running' || status === 'queued') return 'working'
930  // A waiting task waits on its parent's answer, not on children of its own.
931  if (status === 'waiting') return 'working'
932  return 'result_available'
933}
934
935const INTERRUPTED_ERROR = 'its supervisor is gone and it left no exit file'
936
937/** One job as the task tools answer it, read from its directory alone. Null for an unknown or unsafe id. */
938export async function readJobStatus($: CoreEngine, config: OrqConfig, jobId: string): Promise<TaskStatusResult | null> {
939  const job = await readMeta($, config.jobsDir, jobId)
940  if (!job) return null
941  const view = await jobView($, job)
942  const status = taskStatusOf(view)
943  const next = await readNext($, job)
944  const answer: TaskStatusResult = {
945    taskId: job.id,
946    status,
947    workState: workStateOf(status),
948    kind: job.kind,
949    startedAtIso: job.startedAtIso,
950    ...(job.instanceId ? { instanceId: job.instanceId } : {}),
951    ...(job.model ? { model: job.model } : {}),
952    ...(view.endedAt !== null ? { finishedAtIso: new Date(view.endedAt).toISOString() } : {}),
953    ...(job.parent !== undefined ? { parent: job.parent } : {}),
954    ...(next !== null ? { next: next.jobId } : {}),
955  }
956  if (view.status === 'done') {
957    return { ...answer, result: view.question !== null ? await blockedAnswer($, job, view.question, view) : view.lastText }
958  }
959  if (view.status === 'failed') {
960    const snippet = await stderrSnippet($, job)
961    return { ...answer, error: `failed (exit ${view.exitCode})` + (snippet ? `\n${snippet}` : '') }
962  }
963  if (view.status === 'interrupted') return { ...answer, error: INTERRUPTED_ERROR }
964  return answer
965}
966
967// ---------------------------------------------------------------------------------------------------------------------
968// Answering a question: the same worker resumes its session in a new job, on the same branch.
969
970export type NextVia = 'form' | 'answer'
971
972/** `next` in a job directory: the job that took over the question of this one, and how it was answered. */
973export type Next = { jobId: string; via: NextVia }
974
975const NEXT_FILE = 'next'
976const SESSION_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/
977
978/** The continuation that claimed this job's question; null when none did, or when the file cannot be read as one. */
979export async function readNext($: CoreEngine, job: Pick<JobRecord, 'dir'>): Promise<Next | null> {
980  const raw = await readText($, `${job.dir}/${NEXT_FILE}`)
981  if (raw === null) return null
982  try {
983    const parsed: unknown = JSON.parse(raw)
984    if (!isRecord(parsed) || typeof parsed.jobId !== 'string' || !isSafeJobId(parsed.jobId)) return null
985    const via = parsed.via === 'form' || parsed.via === 'answer' ? parsed.via : null
986    return via === null ? null : { jobId: parsed.jobId, via }
987  } catch {
988    return null
989  }
990}
991
992/** The session id the supervisor recorded for the worker of this job; null when it recorded none. */
993export async function readSession($: CoreEngine, job: Pick<JobRecord, 'dir'>): Promise<string | null> {
994  const text = ((await readText($, `${job.dir}/session`)) ?? '').trim()
995  return SESSION_ID.test(text) ? text : null
996}
997
998export type Claim = { claimed: true } | { claimed: false; existing: Next | null }
999
1000/**
1001 * The one answer that may continue a job. `fs.write` cannot create a file only when it is absent, so the claim is a hard
1002 * link: the content is written whole to a temporary file, and `ln` fails when `next` exists, so a reader never sees half of
1003 * it and two answers at once cannot both win.
1004 */
1005export async function claimNext($: CoreEngine, job: Pick<JobRecord, 'dir'>, next: Next): Promise<Claim> {
1006  const script =
1007    `# orq:claim\ntmp=$(mktemp "$1/.next.XXXXXX") || exit 12\n` +
1008    `printf '%s' "$2" > "$tmp" || { rm -f "$tmp"; exit 12; }\n` +
1009    `if ln "$tmp" "$1/${NEXT_FILE}" 2>/dev/null; then rm -f "$tmp"; exit 0; fi\n` +
1010    `rm -f "$tmp"\n[ -e "$1/${NEXT_FILE}" ] && exit 9\nexit 12`
1011  const r = await $.process.run(['bash', '-c', script, 'orq-claim', job.dir, JSON.stringify(next)], { timeoutMs: PROCESS_TIMEOUT_MS })
1012  if (r.exitCode === 0) return { claimed: true }
1013  if (r.exitCode === 9) return { claimed: false, existing: await readNext($, job) }
1014  throw new Error(`could not claim the question of ${job.dir} (exit ${r.exitCode})`)
1015}
1016
1017/** Gives the claim back after a launch that did not start, so the question can be answered again. */
1018async function releaseNext($: CoreEngine, job: Pick<JobRecord, 'dir'>): Promise<void> {
1019  try {
1020    await $.process.run(['rm', '-f', `${job.dir}/${NEXT_FILE}`], { timeoutMs: PROCESS_TIMEOUT_MS })
1021  } catch {
1022    /* a claim that stays reads as answered; the refusal that follows names the launch's own failure */
1023  }
1024}
1025
1026export function alreadyAnswered(jobId: string, existing: Next | null): Failure {
1027  return fail(
1028    'already_answered',
1029    existing === null
1030      ? `the question of ${jobId} was answered a moment ago by another call`
1031      : `the question of ${jobId} was already answered: its worker continues in taskId ${existing.jobId}. Read it with status; if it failed, delegate anew.`,
1032  )
1033}
1034
1035/** What a continuation needs of the job it resumes, checked against the config as it is now. */
1036export type ContinuationPlan = {
1037  parent: JobRecord
1038  session: string
1039  instance: InstanceConfig
1040  model: string
1041  modes: { runtimeMode: RuntimeMode; interactionMode: InteractionMode }
1042  flags: string[]
1043  /** `worktree` only for a worktree that git lists for the task; anything else runs, and is locked, as a shared task. */
1044  isolation: TaskIsolation
1045  /** The directory the task was asked to work in, resolved: the checkout of a worktree task, the directory itself of a shared one. */
1046  requestCwd: string
1047  /** Where the worker resumes: the resolved, checked directory, never the one meta.json names. */
1048  cwd: string
1049  worktree?: NonNullable<Continuation['worktree']>
1050}
1051
1052/** The real path of a path that exists, with every link followed; null when it does not exist. No realpath command: macOS has none with -e. */
1053async function realPath($: CoreEngine, path: string): Promise<string | null> {
1054  try {
1055    return (await $.fs.stat(path, { resolve: true })).realPath ?? null
1056  } catch {
1057    return null
1058  }
1059}
1060
1061/** Whether the `next` claims of the server lead from job `from` to job `to`: the chain of answers that a task is. */
1062async function chainReaches($: CoreEngine, jobsDir: string, from: string, to: string): Promise<boolean> {
1063  let id = from
1064  for (let hops = 0; hops < 64; hops += 1) {
1065    if (id === to) return true
1066    const next = await readNext($, { dir: `${jobsDir}/${id}` })
1067    if (next === null) return false
1068    id = next.jobId
1069  }
1070  return false
1071}
1072
1073/** Whether a live job other than `except` works on this branch, as far as its meta.json says. */
1074async function branchHeldByAnother($: CoreEngine, jobsDir: string, branch: string, except: string): Promise<boolean> {
1075  let entries: { name: string; kind: string }[]
1076  try {
1077    entries = await $.fs.list(jobsDir)
1078  } catch {
1079    return false
1080  }
1081  for (const entry of entries) {
1082    if (entry.kind !== 'dir' || entry.name === except) continue
1083    const other = await readMeta($, jobsDir, entry.name)
1084    if (other?.branch === branch && (await jobIsLive($, other))) return true
1085  }
1086  return false
1087}
1088
1089type OwnWorktree = { fields: NonNullable<Continuation['worktree']>; cwd: string }
1090
1091/**
1092 * The worktree of a job, when it is the one the server made for the task, and the directory to resume in. meta.json is the
1093 * job's own file, and a worker that can write outside its directory can change it, so nothing in it is taken as it stands: the
1094 * branch must be `orq/ID` of a job that does not run and whose `next` claims lead to this job (or of this job itself), the
1095 * checkout git reports must lie in scope as a launch requires, the worktree must be the path that a launch computes for that
1096 * ID, git must list it on that branch, and the directory must resolve, links followed, to where a launch from the recorded
1097 * directory would have put it, inside the worktree. Anything else is not the task's own worktree, and the job runs, is scoped
1098 * and is locked as a shared one.
1099 */
1100async function ownWorktree($: CoreEngine, config: OrqConfig, sessionDir: string, job: JobRecord): Promise<OwnWorktree | null> {
1101  const { repo, baseSha, branch } = job
1102  if (repo === undefined || baseSha === undefined || branch === undefined || job.worktree === undefined) return null
1103  const rootId = branch.startsWith('orq/') ? branch.slice('orq/'.length) : ''
1104  if (!isSafeJobId(rootId) || branchFor(rootId) !== branch) return null
1105  try {
1106    const top = await $.process.run(gitArgv(repo, ...NO_HOOKS, 'rev-parse', '--show-toplevel'), { timeoutMs: GIT_TIMEOUT_MS })
1107    if (top.exitCode !== 0) return null
1108    const toplevel = await realPath($, top.stdout.trim())
1109    const worktree = await realPath($, worktreePathFor(top.stdout.trim(), rootId))
1110    if (toplevel === null || worktree === null || (await realPath($, job.worktree)) !== worktree) return null
1111    if ('refusal' in (await scopedCwd($, config, sessionDir, toplevel))) return null
1112    const listed = await $.process.run(gitArgv(repo, ...NO_HOOKS, 'worktree', 'list', '--porcelain'), { timeoutMs: GIT_TIMEOUT_MS })
1113    if (listed.exitCode !== 0) return null
1114    let registered = false
1115    for (const block of listed.stdout.split('\n\n')) {
1116      const lines = block.split('\n')
1117      const path = lines.find((line) => line.startsWith('worktree '))?.slice('worktree '.length)
1118      if (path !== undefined && lines.includes(`branch refs/heads/${branch}`) && (await realPath($, path)) === worktree) registered = true
1119    }
1120    if (!registered) return null
1121    const asked = await realPath($, repo)
1122    if (asked === null) return null
1123    const subdir = asked !== toplevel && isWithin(asked, toplevel) ? asked.slice(toplevel.length) : ''
1124    const cwd = await realPath($, `${worktree}${subdir}`)
1125    if (cwd === null || !isWithin(cwd, worktree) || (await realPath($, job.cwd)) !== cwd) return null
1126    if (rootId !== job.id && !(await chainReaches($, config.jobsDir, rootId, job.id))) return null
1127    if (await branchHeldByAnother($, config.jobsDir, branch, job.id)) return null
1128    return { fields: { repo, baseSha, branch, worktree }, cwd }
1129  } catch {
1130    return null
1131  }
1132}
1133
1134/**
1135 * Whether the worker of a job that stopped on a question can be resumed, and with what. The job must have recorded its
1136 * session, its worker and its modes, its instance must still be the one it ran on (the session lives in that instance's
1137 * CODEX_HOME), and its directory must still be there. The flags are the ones a launch at the same modes would write.
1138 */
1139export async function continuationPlan(
1140  $: CoreEngine,
1141  config: OrqConfig,
1142  parent: JobRecord,
1143  sessionDir: string,
1144): Promise<{ plan: ContinuationPlan } | { failure: Failure }> {
1145  const unavailable = (why: string): { failure: Failure } => ({
1146    failure: fail('resume_unavailable', `${parent.id} cannot be resumed: ${why}. Delegate a new task that carries the brief and the answer instead.`),
1147  })
1148  const session = await readSession($, parent)
1149  if (session === null) return unavailable('its worker left no session id')
1150  const { driver, model, runtimeMode, interactionMode } = parent
1151  if (driver === undefined || model === undefined || runtimeMode === undefined || interactionMode === undefined) {
1152    return unavailable('the job did not record its worker or its permission modes')
1153  }
1154  const instance = config.instances.find((candidate) => candidate.id === parent.instanceId)
1155  if (instance === undefined || instance.driver !== driver) return unavailable(`the instance ${parent.instanceId ?? 'it ran on'} is no longer configured as it was`)
1156  if (instance.env?.CODEX_HOME !== parent.codexHome) return unavailable('the CODEX_HOME of its instance changed, so its session is not where the worker left it')
1157  if (!plainArgument(model)) return unavailable('its model is not one word')
1158  if (!(await isDirectory($, parent.cwd))) return unavailable(`its directory ${parent.cwd} is gone`)
1159  const own = await ownWorktree($, config, sessionDir, parent)
1160  // The scope is judged as a launch judges it: on the directory the task was asked to work in.
1161  const scoped = await scopedCwd($, config, sessionDir, own === null ? parent.cwd : own.fields.repo)
1162  if ('refusal' in scoped) {
1163    const body = scoped.refusal.slice(`${TASK_REFUSAL}: `.length)
1164    const outside = /^cwd_out_of_scope: ([\s\S]*)$/.exec(body)
1165    return outside?.[1] !== undefined ? { failure: fail('cwd_out_of_scope', outside[1]) } : unavailable(body)
1166  }
1167  const policy = runnerPolicy(driver, runtimeMode, interactionMode)
1168  if (isFailure(policy)) return unavailable(policy.error.message)
1169  const effort = parent.effort === undefined ? [] : effortFlag(driver, [{ id: EFFORT_OPTION, value: parent.effort }])
1170  return {
1171    plan: {
1172      parent,
1173      session,
1174      instance,
1175      model,
1176      modes: { runtimeMode, interactionMode },
1177      flags: [...policy.flags, ...effort],
1178      isolation: own === null ? 'shared' : 'worktree',
1179      requestCwd: scoped.cwd,
1180      cwd: own === null ? scoped.cwd : own.cwd,
1181      ...(own !== null ? { worktree: own.fields } : {}),
1182    },
1183  }
1184}
1185
1186export type ContinuationInput = { answer: string; via: NextVia; questions: QuestionMode; agentId: string; reservedId?: string }
1187
1188/**
1189 * Claims the question and starts the continuation under the id the claim names: the worker resumes its session on the same
1190 * branch, in the same directory, with the answer as its next message. A launch that does not start gives the claim back.
1191 */
1192export async function launchContinuation(
1193  $: CoreEngine,
1194  config: OrqConfig,
1195  plan: ContinuationPlan,
1196  input: ContinuationInput,
1197): Promise<{ job: JobRecord } | { refusal: string } | { failure: Failure }> {
1198  const { parent, session, instance, model, modes, flags, worktree } = plan
1199  const id = input.reservedId ?? newJobId('mcp', await $.clock.now(), input.agentId)
1200  const claim = await claimNext($, parent, { jobId: id, via: input.via })

1import { isRecord } from './types.ts'
2
3const TAIL_LINES = 30
4
5function shortText(s: string, n: number): string {
6  const one = s.replace(/\s+/g, ' ').trim()
7  return one.length > n ? `${one.slice(0, n - 1)}…` : one
8}
9
10function clockOf(o: Record<string, unknown>): string {
11  const ts = o.timestamp
12  if (typeof ts === 'number' && Number.isFinite(ts)) {
13    const ms = ts < 1e12 ? ts * 1000 : ts
14    return new Date(ms).toISOString().slice(11, 19)
15  }
16  return ''
17}
18
19function usageText(usage: unknown): string | null {
20  if (!isRecord(usage)) return null
21  const input = typeof usage.input_tokens === 'number' ? usage.input_tokens : null
22  const output = typeof usage.output_tokens === 'number' ? usage.output_tokens : null
23  if (input === null && output === null) return null
24  return `tokens ${input ?? '?'} in / ${output ?? '?'} out`
25}
26
27function lastContentItem(o: Record<string, unknown>): Record<string, unknown> | null {
28  const content: unknown[] = isRecord(o.message) && Array.isArray(o.message.content) ? o.message.content : []
29  const last: unknown = content[content.length - 1]
30  return isRecord(last) ? last : null
31}
32
33/** One compact line per event JSONL row: `HH:MM:SS kind text`. Null when the row carries nothing to show. */
34export function renderEventLine(raw: string): string | null {
35  const t = raw.trim()
36  if (!t || !t.startsWith('{')) return null
37  let o: Record<string, unknown>
38  try {
39    o = JSON.parse(t) as Record<string, unknown>
40  } catch {
41    return null
42  }
43  if (!o || typeof o !== 'object') return null
44  const clock = clockOf(o)
45  const pre = clock ? `${clock} ` : ''
46
47  // Codex `codex exec --json` rows.
48  if (typeof o.type === 'string') {
49    if (o.type === 'thread.started' && typeof o.thread_id === 'string') {
50      return `${pre}start thread ${o.thread_id.slice(0, 8)}`
51    }
52    if (o.type === 'item.completed') {
53      const item = o.item as Record<string, unknown> | undefined
54      const kind = item && typeof item.type === 'string' ? item.type : 'item'
55      if (item && item.type === 'agent_message' && typeof item.text === 'string') {
56        return `${pre}msg ${shortText(item.text, 160)}`
57      }
58      if (item && item.type === 'error' && typeof item.message === 'string') {
59        return `${pre}error ${shortText(item.message, 160)}`
60      }
61      return `${pre}${shortText(kind, 24)}`
62    }
63    if (o.type === 'turn.completed') {
64      const usage = o.usage as Record<string, unknown> | undefined
65      const input = usage && typeof usage.input_tokens === 'number' ? usage.input_tokens : null
66      const output = usage && typeof usage.output_tokens === 'number' ? usage.output_tokens : null
67      if (input !== null || output !== null) {
68        return `${pre}tokens ${input ?? '?'} in / ${output ?? '?'} out`
69      }
70      return `${pre}turn done`
71    }
72  }
73
74  // Claude `claude -p --output-format stream-json` rows.
75  if (o.type === 'system' && o.subtype === 'init') return `${pre}start`
76  if (o.type === 'assistant') {
77    const item = lastContentItem(o)
78    if (item?.type === 'text' && typeof item.text === 'string') return `${pre}msg ${shortText(item.text, 160)}`
79    if (item?.type === 'tool_use') return `${pre}tool ${shortText(typeof item.name === 'string' ? item.name : 'tool', 60)}`
80    return null
81  }
82  if (o.type === 'result') return `${pre}${usageText(o.usage) ?? 'result'}`
83  if (o.type === 'user') return null
84
85  // OpenCode `opencode run --format json` rows.
86  const part = o.part as Record<string, unknown> | undefined
87  if (part && typeof part === 'object') {
88    if (part.type === 'text' && typeof part.text === 'string' && part.text.trim()) {
89      return `${pre}msg ${shortText(part.text, 160)}`
90    }
91    if (part.type === 'tool') {
92      const name =
93        typeof part.name === 'string'
94          ? part.name
95          : typeof part.title === 'string'
96            ? part.title
97            : typeof part.tool === 'string'
98              ? part.tool
99              : 'tool'
100      return `${pre}tool ${shortText(name, 60)}`
101    }
102    if (part.type === 'step-finish') {
103      const tokens = part.tokens as Record<string, unknown> | undefined
104      const input = tokens && typeof tokens.input === 'number' ? tokens.input : null
105      const output = tokens && typeof tokens.output === 'number' ? tokens.output : null
106      if (input !== null || output !== null) {
107        return `${pre}tokens ${input ?? '?'} in / ${output ?? '?'} out`
108      }
109      return `${pre}step done`
110    }
111  }
112
113  // Antigravity (`agy`) stream rows.
114  if (o.event === 'init') return `${pre}start`
115  if (o.event === 'step_update') {
116    const update = isRecord(o.step_update) ? o.step_update : null
117    if (update && typeof update.step_type === 'string') return `${pre}step ${shortText(update.step_type, 24)}`
118    return null
119  }
120  if (o.event === 'result') return `${pre}${usageText(isRecord(o.result) ? o.result.usage : undefined) ?? 'result'}`
121  return null
122}
123
124export function renderTail(text: string, maxLines = TAIL_LINES): string[] {
125  const lines = text
126    .split('\n')
127    .map((raw) => renderEventLine(raw))
128    .filter((line): line is string => line !== null)
129  return lines.slice(-maxLines)
130}
131

1import type { DriverKind } from './config.ts'
2import { INTERACTION_MODES, RUNTIME_MODES, type Failure, type InteractionMode, type RuntimeMode } from './contract.ts'
3
4/** The CLI that runs a job; each provider driver maps to exactly one. */
5export type Runner = 'codex' | 'opencode' | 'agy' | 'claude'
6
7export const DRIVER_RUNNER: Record<DriverKind, Runner> = {
8  codex: 'codex',
9  opencode: 'opencode',
10  antigravity: 'agy',
11  claudeAgent: 'claude',
12}
13
14const RUNNERS: readonly Runner[] = ['codex', 'opencode', 'agy', 'claude']
15const DRIVERS: readonly DriverKind[] = ['codex', 'opencode', 'antigravity', 'claudeAgent']
16
17export type JobStatus = 'running' | 'cancelling' | 'waiting' | 'interrupted' | 'done' | 'failed' | 'cancelled'
18
19/** A job as its meta.json describes it. Optional fields are absent in older jobs. */
20export type JobRecord = {
21  id: string
22  kind: string
23  agentId: string
24  dir: string
25  cwd: string
26  startedAt: number
27  startedAtIso: string
28  pgid?: string
29  runner?: Runner
30  model?: string
31  instanceId?: string
32  driver?: DriverKind
33  /** The CODEX_HOME the job launched with, when its instance set one; the only env value meta.json keeps. */
34  codexHome?: string
35  /** The runtime and interaction modes a cc-orq:task job was resolved to; absent on older and manual jobs. */
36  runtimeMode?: RuntimeMode
37  interactionMode?: InteractionMode
38  /** The reasoning effort the job's CLI was given; absent when it got none, as for an OpenCode job. */
39  effort?: string
40  /** The checkout the task was launched from. Set only for a worktree task, where cwd is the worktree. */
41  repo?: string
42  /** The commit the task branch was cut from. */
43  baseSha?: string
44  /** The task branch, orq/TASK_ID. */
45  branch?: string
46  /** The task's worktree path: cwd, or the directory above cwd when the task ran in a subdirectory. */
47  worktree?: string
48  /** The job this one continues, when it was started to resume that job's worker. */
49  parent?: string
50  /** Who answers a question the worker asks: the parent session, or the user through a form. Absent on jobs older than `delegate`. */
51  questions?: 'parent' | 'human'
52}
53
54/** True when path is root or lies under it. Both are resolved paths. */
55export function isWithin(path: string, root: string): boolean {
56  return path === root || path.startsWith(root.endsWith('/') ? root : `${root}/`)
57}
58
59export function shortBranch(ref: string): string {
60  return ref.startsWith('refs/heads/') ? ref.slice('refs/heads/'.length) : ref
61}
62
63export function isRecord(value: unknown): value is Record<string, unknown> {
64  return typeof value === 'object' && value !== null && !Array.isArray(value)
65}
66
67/** Build a JobRecord from parsed meta.json, or null when the required fields are missing. */
68export function parseJobMeta(raw: unknown, jobsDir: string, id: string): JobRecord | null {
69  if (!isRecord(raw) || raw.id !== id) return null
70  if (typeof raw.kind !== 'string' || typeof raw.agentId !== 'string') return null
71  if (typeof raw.cwd !== 'string' || typeof raw.startedAt !== 'number') return null
72  const job: JobRecord = {
73    id,
74    kind: raw.kind,
75    agentId: raw.agentId,
76    dir: `${jobsDir}/${id}`,
77    cwd: raw.cwd,
78    startedAt: raw.startedAt,
79    startedAtIso: typeof raw.startedAtIso === 'string' ? raw.startedAtIso : new Date(raw.startedAt).toISOString(),
80  }
81  if (typeof raw.pgid === 'string' && raw.pgid !== '') job.pgid = raw.pgid
82  if (typeof raw.model === 'string') job.model = raw.model
83  if (typeof raw.instanceId === 'string') job.instanceId = raw.instanceId
84  const runner = RUNNERS.find((r) => r === raw.runner)
85  const driver = DRIVERS.find((d) => d === raw.driver)
86  if (driver) job.driver = driver
87  if (runner) job.runner = runner
88  else if (driver) job.runner = DRIVER_RUNNER[driver]
89  if (typeof raw.codexHome === 'string' && raw.codexHome !== '') job.codexHome = raw.codexHome
90  const runtimeMode = RUNTIME_MODES.find((mode) => mode === raw.runtimeMode)
91  if (runtimeMode) job.runtimeMode = runtimeMode
92  const interactionMode = INTERACTION_MODES.find((mode) => mode === raw.interactionMode)
93  if (interactionMode) job.interactionMode = interactionMode
94  if (typeof raw.effort === 'string' && raw.effort !== '') job.effort = raw.effort
95  // Worktree fields are optional and tolerated: a value of the wrong type is ignored, never a reason to drop the job.
96  if (typeof raw.repo === 'string' && raw.repo !== '') job.repo = raw.repo
97  if (typeof raw.baseSha === 'string' && raw.baseSha !== '') job.baseSha = raw.baseSha
98  if (typeof raw.branch === 'string' && raw.branch !== '') job.branch = raw.branch
99  if (typeof raw.worktree === 'string' && raw.worktree !== '') job.worktree = raw.worktree
100  if (typeof raw.parent === 'string' && isSafeJobId(raw.parent)) job.parent = raw.parent
101  if (raw.questions === 'parent' || raw.questions === 'human') job.questions = raw.questions
102  return job
103}
104
105/** Quote a word for bash only when it contains anything beyond a plain token. */
106export function shellWord(text: string): string {
107  if (/^[A-Za-z0-9_@%+=:,./-]+$/.test(text)) return text
108  return `'${text.replace(/'/g, `'\\''`)}'`
109}
110
111/** Format milliseconds as 45s, 6m, 12m05s. */
112export function formatDuration(ms: number): string {
113  const s = Math.max(0, Math.floor(ms / 1000))
114  const m = Math.floor(s / 60)
115  const rest = s % 60
116  if (m === 0) return `${s}s`
117  if (rest === 0) return `${m}m`
118  return `${m}m${String(rest).padStart(2, '0')}s`
119}
120
121/** Starts every refusal a task step answers. "taskId none" tells the parent there is no job to query. */
122export const TASK_REFUSAL = 'orq: task refused (taskId none)'
123
124export function refusalText(failure: Failure): string {
125  return `${TASK_REFUSAL}: ${failure.error.code}: ${failure.error.message}`
126}
127
128/**
129 * The one footer line of every answer. The taskId is the job id a parent passes to the job tools. Null token
130 * counts, when the extractor found none, render as `tokens n/a`.
131 */
132export function buildFooter(args: {
133  kind: string
134  model: string
135  effort: string
136  durationMs: number
137  inTokens: number | null
138  outTokens: number | null
139  taskId: string
140  session: string
141}): string {
142  const tokens = args.inTokens === null || args.outTokens === null ? 'tokens n/a' : `${args.inTokens}/${args.outTokens} tokens`
143  return `— orq · ${args.kind} · ${args.model} · effort ${args.effort} · ${formatDuration(args.durationMs)} · ${tokens} · taskId ${args.taskId} · session ${args.session}`
144}
145
146/** Read model and effort from the last turn_context of a Codex rollout file. */
147export function parseRolloutModelEffort(text: string): { model: string; effort: string } {
148  let model = ''
149  let effort = ''
150  for (const line of text.split('\n')) {
151    const t = line.trim()
152    if (!t || !t.includes('turn_context')) continue
153    let o: unknown
154    try {
155      o = JSON.parse(t)
156    } catch {
157      continue
158    }
159    if (!isRecord(o) || o.type !== 'turn_context' || !isRecord(o.payload)) continue
160    if (typeof o.payload.model === 'string') model = o.payload.model
161    if (typeof o.payload.effort === 'string') effort = o.payload.effort
162  }
163  return { model, effort }
164}
165
166const REDACTED = '[REDACTED]'
167
168/** Replace anything token-like with [REDACTED]. Order matters: specific shapes first. */
169export function redactSecrets(text: string): string {
170  let out = text
171  out = out.replace(/\bBearer\s+[A-Za-z0-9._~+/=-]+/gi, REDACTED)
172  out = out.replace(/\bBasic\s+[A-Za-z0-9._~+/=]+/gi, REDACTED)
173  out = out.replace(/\bsk-[A-Za-z0-9_-]{8,}/g, REDACTED)
174  out = out.replace(/\bghp_[A-Za-z0-9]{8,}/g, REDACTED)
175  out = out.replace(/\bgho_[A-Za-z0-9]{8,}/g, REDACTED)
176  out = out.replace(/\bgithub_pat_[A-Za-z0-9_]{8,}/g, REDACTED)
177  out = out.replace(/\bxox[abposr]-[A-Za-z0-9-]{8,}/g, REDACTED)
178  out = out.replace(/\bAKIA[0-9A-Z]{12,}/g, REDACTED)
179  out = out.replace(/\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}/g, REDACTED)
180  out = out.replace(
181    /((?:token|secret|password|passwd|api[_-]?key|authorization|cookie)\s*[:=]\s*)([^\s,;"']+)/gi,
182    `$1${REDACTED}`,
183  )
184  out = out.replace(/\b[A-Za-z0-9+/=_.-]{32,}/g, REDACTED)
185  return out
186}
187
188/** Failure log snippet: at most the last 20 lines, each cut to 300 chars, secrets redacted. */
189export function failureSnippet(stderrText: string): string {
190  const lines = stderrText.split('\n').slice(-20)
191  return lines
192    .map((l) => redactSecrets(l.length > 300 ? l.slice(0, 300) : l))
193    .join('\n')
194    .trim()
195}
196
197/** Random base36 part for job ids. */
198export function rand36(n = 6): string {
199  let s = ''
200  while (s.length < n) s += Math.random().toString(36).slice(2)
201  return s.slice(0, n)
202}
203
204/** Job ids reach the filesystem, so only this shape is accepted. */
205export function isSafeJobId(id: string): boolean {
206  return /^[A-Za-z0-9][A-Za-z0-9_.-]{0,128}$/.test(id) && !id.includes('..')
207}
208
209export type TableRow = { id: string; kind: string; status: string; elapsed: string; origin: string; cwd: string }
210
211/** Render the /orq list text table. */
212export function renderTable(rows: TableRow[]): string {
213  const head = ['id', 'kind', 'status', 'elapsed', 'origin', 'cwd']
214  const all = [head, ...rows.map((r) => [r.id, r.kind, r.status, r.elapsed, r.origin, r.cwd])]
215  const widths = head.map((_, i) => Math.max(...all.map((r) => (r[i] ?? '').length)))
216  const fmt = (r: string[]) => r.map((c, i) => (i === r.length - 1 ? c : c.padEnd(widths[i] ?? 0))).join('  ')
217  return all.map(fmt).join('\n')
218}
219

1/** The file a child writes its question to, inside its job directory. */
2export const QUESTION_FILE = 'question.txt'
3
4/** The paragraph every child prompt ends with. The child writes its question to the job directory and stops. */
5export const QUESTION_PROTOCOL =
6  'If you are blocked on a question that the brief does not answer, write the question to $ORQ_JOB_DIR/question.txt, print BLOCKED: followed by the question as your final message, and stop.'
7

1// The permission ceiling: the lower, on each axis, of what the config's `ceiling` allows and what the session's
2// permission mode allows. The cc-orq:task agent and the delegate tool both decide a request here.
3
4import {
5  INTERACTION_MODES,
6  RUNTIME_MODES,
7  fail,
8  type Failure,
9  type InteractionMode,
10  type RuntimeMode,
11  type TaskIsolation,
12} from './contract.ts'
13import { interactionModeRank, parentRuntimeModeFromPermissionMode, runtimeModeRank } from './target.ts'
14import { isRecord } from './types.ts'
15
16export type Ceiling = { runtimeMode: RuntimeMode; interactionMode: InteractionMode }
17
18export type CeilingField = 'runtime' | 'sharedRuntime' | 'interaction'
19
20/** The `ceiling` of the orq config once parsed. A field whose value was not valid is listed in `invalid` and holds the narrowest mode. */
21export type CeilingConfig = {
22  /** The most a task in its own worktree may run as. */
23  runtime: RuntimeMode
24  /** The most a task that runs in the checkout itself may run as. */
25  sharedRuntime: RuntimeMode
26  interaction: InteractionMode
27  invalid: CeilingField[]
28}
29
30type CeilingLimits = Omit<CeilingConfig, 'invalid'>
31
32const CEILING_FIELDS: readonly CeilingField[] = ['runtime', 'sharedRuntime', 'interaction']
33const DEFAULTS: CeilingLimits = { runtime: 'auto-accept-edits', sharedRuntime: 'approval-required', interaction: 'default' }
34const NARROWEST: CeilingLimits = { runtime: 'approval-required', sharedRuntime: 'approval-required', interaction: 'plan' }
35
36const CONFIG_FILE = '~/.config/orq/config.json'
37
38export function defaultCeilingConfig(): CeilingConfig {
39  return { ...DEFAULTS, invalid: [] }
40}
41
42function invalidWarning(field: CeilingField): string {
43  const choices = field === 'interaction' ? INTERACTION_MODES : RUNTIME_MODES
44  return `ceiling.${field} must be one of ${choices.join(', ')}; it counts as ${NARROWEST[field]}`
45}
46
47/** Reads the `ceiling` value of the config file. An absent value is the defaults; anything invalid fails closed and is named. */
48export function parseCeilingConfig(raw: unknown): { ceiling: CeilingConfig; warnings: string[] } {
49  if (raw === undefined) return { ceiling: defaultCeilingConfig(), warnings: [] }
50  if (!isRecord(raw)) {
51    const narrowest = CEILING_FIELDS.map((field) => `${field} ${NARROWEST[field]}`).join(', ')
52    return {
53      ceiling: { ...NARROWEST, invalid: [...CEILING_FIELDS] },
54      warnings: [`ceiling must be an object such as {"runtime": "auto"}; every field counts as its narrowest value (${narrowest})`],
55    }
56  }
57  const ceiling = defaultCeilingConfig()
58  const warnings: string[] = []
59  for (const key of Object.keys(raw)) {
60    if (!CEILING_FIELDS.some((field) => field === key)) {
61      warnings.push(`ceiling.${key} is not a ceiling field (${CEILING_FIELDS.join(', ')}), ignored`)
62    }
63  }
64  const reject = (field: CeilingField): void => {
65    ceiling.invalid.push(field)
66    warnings.push(invalidWarning(field))
67  }
68  for (const field of CEILING_FIELDS) {
69    const value = raw[field]
70    if (value === undefined) continue
71    if (field === 'interaction') {
72      const mode = INTERACTION_MODES.find((candidate) => candidate === value)
73      if (mode === undefined) {
74        ceiling.interaction = NARROWEST.interaction
75        reject(field)
76      } else {
77        ceiling.interaction = mode
78      }
79    } else {
80      const mode = RUNTIME_MODES.find((candidate) => candidate === value)
81      if (mode === undefined) {
82        ceiling[field] = NARROWEST[field]
83        reject(field)
84      } else {
85        ceiling[field] = mode
86      }
87    }
88  }
89  return { ceiling, warnings }
90}
91
92/** What the session's permission mode limits a task to; null when no mode reached orq, and then only the config limits. */
93export function sessionCeiling(sessionMode: string | undefined): Ceiling | null {
94  return sessionMode === undefined ? null : parentRuntimeModeFromPermissionMode(sessionMode)
95}
96
97export function configCeiling(config: CeilingConfig, isolation: TaskIsolation): Ceiling {
98  return { runtimeMode: isolation === 'shared' ? config.sharedRuntime : config.runtime, interactionMode: config.interaction }
99}
100
101export function lowerCeiling(a: Ceiling, b: Ceiling): Ceiling {
102  return {
103    runtimeMode: runtimeModeRank(a.runtimeMode) <= runtimeModeRank(b.runtimeMode) ? a.runtimeMode : b.runtimeMode,
104    interactionMode: interactionModeRank(a.interactionMode) <= interactionModeRank(b.interactionMode) ? a.interactionMode : b.interactionMode,
105  }
106}
107
108export function effectiveCeiling(config: CeilingConfig, session: Ceiling | null, isolation: TaskIsolation): Ceiling {
109  const limit = configCeiling(config, isolation)
110  return session === null ? limit : lowerCeiling(limit, session)
111}
112
113/** What a task asks for; a mode left out inherits the effective ceiling. */
114export type CeilingRequest = { runtimeMode?: RuntimeMode; interactionMode?: InteractionMode }
115
116export type CeilingDecision = {
117  effective: Ceiling
118  /** Why the request cannot run, as the typed failure the caller answers; null when it is within the ceiling. */
119  refusal: Failure | null
120}
121
122type Axis<Mode extends string> = {
123  code: 'runtime_mode_escalation_denied' | 'interaction_mode_escalation_denied'
124  label: string
125  field: string
126  rank: (mode: Mode) => number
127}
128
129/** The session's mode can be the lower limit, and then the config alone cannot lift it, so the refusal names each limit in the way. */
130function refuseAbove<Mode extends string>(axis: Axis<Mode>, asked: Mode, limits: { config: Mode; session: Mode | null; effective: Mode }): Failure {
131  const steps = [
132    ...(axis.rank(limits.config) < axis.rank(asked) ? [`raise ceiling.${axis.field} in ${CONFIG_FILE}`] : []),
133    ...(limits.session !== null && axis.rank(limits.session) < axis.rank(asked) ? ['use a session permission mode that allows it'] : []),
134  ]
135  const session = limits.session === null ? '' : ` and the session's permission mode allows ${limits.session}`
136  return fail(
137    axis.code,
138    `${axis.label} ${asked} is above the ceiling ${limits.effective} for this task: the config allows ${limits.config} (ceiling.${axis.field})${session}. ` +
139      `To allow ${asked}, ${steps.join(' and ')}; otherwise ask for ${limits.effective} at most.`,
140  )
141}
142
143const RUNTIME_AXIS = { code: 'runtime_mode_escalation_denied', label: 'Runtime mode', rank: runtimeModeRank } as const
144const INTERACTION_AXIS = { code: 'interaction_mode_escalation_denied', label: 'Interaction mode', field: 'interaction', rank: interactionModeRank } as const
145
146/** The effective ceiling of a task, and the refusal when its request is above it. The runtime mode is judged first, as the launch judges it. */
147export function resolveCeiling(input: {
148  config: CeilingConfig
149  /** What the session's permission mode limits to; null when none reached orq. */
150  session: Ceiling | null
151  isolation: TaskIsolation
152  request: CeilingRequest
153}): CeilingDecision {
154  const { config, session, isolation, request } = input
155  const fromConfig = configCeiling(config, isolation)
156  const effective = session === null ? fromConfig : lowerCeiling(fromConfig, session)
157  const { runtimeMode, interactionMode } = request
158  if (runtimeMode !== undefined && runtimeModeRank(runtimeMode) > runtimeModeRank(effective.runtimeMode)) {
159    const axis = { ...RUNTIME_AXIS, field: isolation === 'shared' ? 'sharedRuntime' : 'runtime' }
160    const limits = { config: fromConfig.runtimeMode, session: session?.runtimeMode ?? null, effective: effective.runtimeMode }
161    return { effective, refusal: refuseAbove(axis, runtimeMode, limits) }
162  }
163  if (interactionMode !== undefined && interactionModeRank(interactionMode) > interactionModeRank(effective.interactionMode)) {
164    const limits = { config: fromConfig.interactionMode, session: session?.interactionMode ?? null, effective: effective.interactionMode }
165    return { effective, refusal: refuseAbove(INTERACTION_AXIS, interactionMode, limits) }
166  }
167  return { effective, refusal: null }
168}
169
170/** What `capabilities` tells the model about the ceiling. `effective` has one entry per isolation, since the config limits them apart. */
171export type CeilingReport = {
172  config: CeilingLimits
173  session: (Ceiling & { mode: string }) | null
174  effective: Record<TaskIsolation, Ceiling>
175  warnings?: string[]
176}
177
178export function describeCeiling(config: CeilingConfig, sessionMode: string | undefined): CeilingReport {
179  const session = sessionMode === undefined ? null : { mode: sessionMode, ...parentRuntimeModeFromPermissionMode(sessionMode) }
180  const report: CeilingReport = {
181    config: { runtime: config.runtime, sharedRuntime: config.sharedRuntime, interaction: config.interaction },
182    session,
183    effective: { worktree: effectiveCeiling(config, session, 'worktree'), shared: effectiveCeiling(config, session, 'shared') },
184  }
185  if (config.invalid.length > 0) report.warnings = config.invalid.map(invalidWarning)
186  return report
187}
188