// 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 } }