dsh-remote/lib/agent-ops.mjs

// Device-side operations the master's agent can invoke on any linked device:
// exec (shell), file read/write/list, and task (delegate a prompt to that device's own
// DSH agent and wait for the final answer). Also forge (git) credential install.
import { spawn } from 'node:child_process'
import crypto from 'node:crypto'
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'

const MAX_OUT = 256 * 1024
const MAX_FILE = 4 * 1024 * 1024

const expand = (p) => (p === '~' ? os.homedir() : String(p || '').replace(/^~(?=\/)/, os.homedir()))
const abs = (p, cwd) => path.resolve(expand(cwd || os.homedir()), expand(p))

/** Keep head + tail of long output so errors at the end survive. */
function clip(buf) {
  if (buf.length <= MAX_OUT) return { text: buf.toString('utf8'), truncated: false }
  const head = buf.subarray(0, MAX_OUT / 4).toString('utf8')
  const tail = buf.subarray(buf.length - (MAX_OUT * 3) / 4).toString('utf8')
  return { text: `${head}\n…[${buf.length - MAX_OUT} bytes omitted]…\n${tail}`, truncated: true }
}

/** Run one bash command; resolves with {exitCode, signal, stdout, stderr, durationMs, timedOut}. */
export async function execCommand({ command, cwd, timeoutMs = 120000, env = {}, stdin } = {}) {
  if (!command || typeof command !== 'string') throw Object.assign(new Error('command is required'), { status: 400 })
  const dir = abs('.', cwd)
  if (!fs.existsSync(dir)) throw Object.assign(new Error(`cwd does not exist: ${dir}`), { status: 400 })
  const t0 = Date.now()
  return new Promise((resolve) => {
    const child = spawn('bash', ['-lc', command], {
      cwd: dir,
      env: { ...process.env, ...env, GIT_TERMINAL_PROMPT: '0' },
      stdio: ['pipe', 'pipe', 'pipe'],
      detached: true,
    })
    const out = []
    const err = []
    let outLen = 0
    let errLen = 0
    child.stdout.on('data', (d) => { if (outLen < MAX_OUT * 4) { out.push(d); outLen += d.length } })
    child.stderr.on('data', (d) => { if (errLen < MAX_OUT * 4) { err.push(d); errLen += d.length } })
    let timedOut = false
    const limit = Math.min(Math.max(Number(timeoutMs) || 120000, 1000), 30 * 60 * 1000)
    const timer = setTimeout(() => {
      timedOut = true
      try { process.kill(-child.pid, 'SIGTERM') } catch { /* gone */ }
      setTimeout(() => { try { process.kill(-child.pid, 'SIGKILL') } catch { /* gone */ } }, 3000).unref()
    }, limit)
    if (stdin) child.stdin.end(String(stdin)); else child.stdin.end()
    child.on('error', (e) => { clearTimeout(timer); resolve({ exitCode: -1, signal: null, stdout: '', stderr: String(e.message), durationMs: Date.now() - t0, timedOut: false, cwd: dir }) })
    child.on('close', (code, signal) => {
      clearTimeout(timer)
      const o = clip(Buffer.concat(out))
      const e = clip(Buffer.concat(err))
      resolve({ exitCode: code, signal, stdout: o.text, stderr: e.text, truncated: o.truncated || e.truncated, durationMs: Date.now() - t0, timedOut, cwd: dir })
    })
  })
}

export function readFileOp({ path: p, cwd, offset = 0, limit = MAX_FILE } = {}) {
  const file = abs(p, cwd)
  const st = fs.statSync(file)
  if (st.isDirectory()) throw Object.assign(new Error(`${file} is a directory (use list)`), { status: 400 })
  const len = Math.min(Math.max(0, Number(limit) || MAX_FILE), MAX_FILE)
  const fd = fs.openSync(file, 'r')
  try {
    const buf = Buffer.alloc(Math.min(len, Math.max(0, st.size - offset)))
    fs.readSync(fd, buf, 0, buf.length, Number(offset) || 0)
    const binary = buf.includes(0)
    return { path: file, size: st.size, offset: Number(offset) || 0, encoding: binary ? 'base64' : 'utf8', content: buf.toString(binary ? 'base64' : 'utf8'), truncated: (Number(offset) || 0) + buf.length < st.size }
  } finally { fs.closeSync(fd) }
}

export function writeFileOp({ path: p, cwd, content = '', encoding = 'utf8', append = false, mode } = {}) {
  const file = abs(p, cwd)
  const data = Buffer.from(String(content), encoding === 'base64' ? 'base64' : 'utf8')
  if (data.length > MAX_FILE) throw Object.assign(new Error('content too large (4 MiB max)'), { status: 413 })
  fs.mkdirSync(path.dirname(file), { recursive: true })
  if (append) fs.appendFileSync(file, data)
  else {
    const tmp = `${file}.dshr-${process.pid}.tmp`
    fs.writeFileSync(tmp, data, mode ? { mode: parseInt(String(mode), 8) } : undefined)
    fs.renameSync(tmp, file)
  }
  return { path: file, bytes: data.length, appended: !!append }
}

export function listDirOp({ path: p = '.', cwd, hidden = false } = {}) {
  const dir = abs(p, cwd)
  const entries = fs.readdirSync(dir, { withFileTypes: true })
    .filter((e) => hidden || !e.name.startsWith('.'))
    .slice(0, 2000)
    .map((e) => {
      let size
      try { size = e.isFile() ? fs.statSync(path.join(dir, e.name)).size : undefined } catch { size = undefined }
      return { name: e.name, type: e.isDirectory() ? 'dir' : e.isSymbolicLink() ? 'link' : e.isFile() ? 'file' : 'other', ...(size !== undefined ? { size } : {}) }
    })
    .sort((a, b) => (a.type === 'dir') - (b.type === 'dir') || a.name.localeCompare(b.name)).reverse()
  return { path: dir, entries }
}

// ---------------------------------------------------------------------------
// task: delegate to this device's own DSH agent
// ---------------------------------------------------------------------------

function lastAssistantText(events, sinceSeq) {
  for (let i = events.length - 1; i >= 0; i--) {
    const e = events[i]
    if (e.seq <= sinceSeq) break
    if (e.type !== 'assistant/message') continue
    const parts = (e.data?.message?.content || []).filter((c) => c.type === 'text').map((c) => c.text)
    if (parts.length) return parts.join('\n')
  }
  return null
}

function summarizeTools(events, sinceSeq) {
  const calls = []
  for (const e of events) {
    if (e.seq <= sinceSeq || e.type !== 'tool/call') continue
    calls.push(e.data?.name)
  }
  const counts = {}
  for (const n of calls) counts[n] = (counts[n] || 0) + 1
  return counts
}

/**
 * Prompt a session (new or existing) on this device and wait for the turn to end.
 * @param sc ctx.sessionController
 * @param wr ctx.workspaceRegistry (optional; used to attach new sessions to a workspace)
 */
export async function runTask(sc, wr, { prompt, sessionId, cwd, workspaceId, timeoutMs = 20 * 60 * 1000, wait = true, title, permission } = {}, signal, presets) {
  if (!sc) throw Object.assign(new Error('session controller unavailable on this device'), { status: 503 })
  if (!prompt || !String(prompt).trim()) throw Object.assign(new Error('prompt is required'), { status: 400 })
  let sid = sessionId
  let created = false
  if (!sid) {
    const dir = abs('.', cwd)
    fs.mkdirSync(dir, { recursive: true })
    let wsId = workspaceId
    if (!wsId && wr) {
      try { wsId = (wr.resolveByPath?.(dir) || await wr.create(dir, path.basename(dir)))?.id } catch { wsId = undefined }
    }
    const r = await sc.create(wsId ? { workspaceId: wsId } : { cwd: dir })
    sid = r.sessionId
    created = true
    if (title) { try { await sc.rename({ sessionId: sid, title }) } catch { /* optional */ } }
  }
  let appliedPermission
  if (permission) {
    // Switch the session's permission preset (e.g. danger-full-access for test suites that
    // spawn processes or write outside the workspace). Needs the live agent's session.
    if (!presets) throw Object.assign(new Error('permission presets are not available on this device'), { status: 501 })
    const res = await sc.resolveAgent(sid)
    if (!res.agent) throw Object.assign(new Error(`cannot resolve session agent: ${res.error?.message || 'unknown'}`), { status: 409 })
    presets.set(res.agent.session, String(permission))
    appliedPermission = presets.current(res.agent.session)
  }
  const before = await sc.inspect(sid, signal)
  const sinceSeq = before.events.length ? before.events[before.events.length - 1].seq : -1
  const requestId = crypto.randomUUID()
  await sc.prompt({ requestId, sessionId: sid, mode: 'queue', content: [{ type: 'text', text: String(prompt) }] }, signal ?? new AbortController().signal)
  if (!wait) return { sessionId: sid, created, status: 'started', permission: appliedPermission }

  const deadline = Date.now() + Math.min(Math.max(Number(timeoutMs) || 0, 10000), 2 * 60 * 60 * 1000)
  let delay = 1000
  while (Date.now() < deadline) {
    if (signal?.aborted) break
    await new Promise((r) => setTimeout(r, delay))
    delay = Math.min(delay * 1.5, 5000)
    const ins = await sc.inspect(sid, signal)
    const ev = ins.events
    const end = ev.findLast?.((e) => e.seq > sinceSeq && e.type === 'turn/end') ?? [...ev].reverse().find((e) => e.seq > sinceSeq && e.type === 'turn/end')
    if (end) {
      return {
        sessionId: sid,
        created,
        status: end.data?.reason?.kind || 'completed',
        answer: lastAssistantText(ev, sinceSeq),
        tools: summarizeTools(ev, sinceSeq),
        permission: appliedPermission,
        reason: end.data?.reason?.kind !== 'completed' ? end.data?.reason : undefined,
      }
    }
  }
  return { sessionId: sid, created, status: signal?.aborted ? 'cancelled-wait' : 'timeout', answer: null, note: 'The task keeps running on the device; call remote_task again with action=status and this sessionId to collect the result.' }
}

/** Latest finished-turn answer of a session (for status polling). */
export async function taskStatus(sc, { sessionId } = {}, signal) {
  const ins = await sc.inspect(sessionId, signal)
  const ev = ins.events
  const lastStart = [...ev].reverse().find((e) => e.type === 'turn/start')
  const lastEnd = [...ev].reverse().find((e) => e.type === 'turn/end')
  const running = !!lastStart && (!lastEnd || lastEnd.seq < lastStart.seq)
  const sinceSeq = lastStart ? lastStart.seq - 1 : -1
  return {
    sessionId,
    status: running ? 'running' : (lastEnd?.data?.reason?.kind || 'idle'),
    answer: running ? null : lastAssistantText(ev, sinceSeq),
    tools: summarizeTools(ev, sinceSeq),
  }
}

// ---------------------------------------------------------------------------
// forge (git) credentials
// ---------------------------------------------------------------------------

/**
 * Read forge logins on the master (tea config: Gitea/Forgejo). Returned only to devices
 * that opted in to the secret named `FORGE_CREDENTIALS`.
 */
export function readForgeLogins() {
  const base = process.env.XDG_CONFIG_HOME || path.join(os.homedir(), '.config')
  let text = ''
  try { text = fs.readFileSync(path.join(base, 'tea', 'config.yml'), 'utf8') } catch { return [] }
  const logins = []
  let inLogins = false
  let cur = null
  for (const raw of text.split(/\r?\n/)) {
    if (!raw.trim() || raw.trim().startsWith('#')) continue
    if (/^[^\s-]/.test(raw)) { inLogins = /^logins\s*:/.test(raw); cur = null; continue }
    if (!inLogins) continue
    let line = raw
    const item = /^(\s*)-\s+(.*)$/.exec(line)
    if (item) { cur = {}; logins.push(cur); line = item[2] } else line = line.trim()
    const kv = /^([A-Za-z0-9_]+)\s*:\s*(.*)$/.exec(line)
    if (cur && kv) cur[kv[1]] = kv[2].replace(/^["']|["']$/g, '')
  }
  return logins.filter((l) => l.url && l.token).map((l) => ({ name: l.name, url: l.url.replace(/\/+$/, ''), user: l.user || l.name, token: l.token }))
}

/**
 * Install forge logins on this device: tea config (so the git-forge plugin sees the
 * account), git credential store + helper (so plain git clone/push over https works).
 */
export async function installForgeLogins(logins) {
  const home = os.homedir()
  const base = process.env.XDG_CONFIG_HOME || path.join(home, '.config')
  // tea config.yml
  const teaDir = path.join(base, 'tea')
  fs.mkdirSync(teaDir, { recursive: true, mode: 0o700 })
  const teaYml = ['logins:', ...logins.flatMap((l, i) => [
    `- name: ${l.name}`,
    `  url: ${l.url}`,
    `  token: ${l.token}`,
    `  default: ${i === 0}`,
    '  ssh_host: ""',
    '  ssh_key: ""',
    '  insecure: false',
    `  user: ${l.user}`,
  ]), 'preferences:', '  editor: false', '  flag_defaults:', '    remote: ""', ''].join('\n')
  fs.writeFileSync(path.join(teaDir, 'config.yml'), teaYml, { mode: 0o600 })
  // git credential store (~/.git-credentials, 0600) — one line per host.
  const credFile = path.join(home, '.git-credentials')
  let existing = []
  try { existing = fs.readFileSync(credFile, 'utf8').split('\n').filter(Boolean) } catch { existing = [] }
  const hosts = new Set(logins.map((l) => new URL(l.url).host))
  const kept = existing.filter((line) => { try { return !hosts.has(new URL(line).host) } catch { return true } })
  const lines = logins.map((l) => { const u = new URL(l.url); return `${u.protocol}//${encodeURIComponent(l.user)}:${encodeURIComponent(l.token)}@${u.host}` })
  fs.writeFileSync(credFile, [...kept, ...lines].join('\n') + '\n', { mode: 0o600 })
  // Point git at the store (global config; does not override an existing helper).
  const r = await execCommand({ command: 'git config --global --get credential.helper || git config --global credential.helper store', timeoutMs: 10000 })
  const name = await execCommand({ command: 'git config --global --get user.name >/dev/null || git config --global user.name "dsh-remote agent"; git config --global --get user.email >/dev/null || git config --global user.email "agent@$(hostname)"', timeoutMs: 10000 })
  return { installed: logins.map((l) => `${l.name}@${new URL(l.url).host}`), credentialHelper: r.stdout.trim() || 'store', gitIdentity: name.exitCode === 0 }
}