From d1e134196d5aaf51ef53b0db63b4ea059f5e6dad Mon Sep 17 00:00:00 2001 From: lyuksovannyy Date: Sun, 11 Oct 2026 10:08:04 +0000 Subject: [PATCH] dsh-remote 0.2.0: agent tools to drive slaves from the master - remote_devices / remote_exec / remote_file / remote_task tools on the master (exec: bash -lc with timeout + output clipping; files: read/write/list; task: delegate a prompt to the device's own AI, wait for turn end, return its answer; optional permission preset, status/cancel) - only the master may control devices (allowPeerControl to relay between slaves), agentTools.enabled=false opts a device out - FORGE_CREDENTIALS opt-in pseudo-secret: ships the master's tea logins to a slave (tea config + git credential store + identity) - /dsh-remote/debug/inspect + deploy/session-log.mjs for transcripts - tests for exec/files/task/forge install --- README.md | 42 ++++++- cordis.patch.yml | 9 ++ deploy/session-log.mjs | 31 +++++ index.js | 196 +++++++++++++++++++++++++++++- lib/agent-ops.mjs | 268 +++++++++++++++++++++++++++++++++++++++++ package.json | 2 +- test/ops.test.mjs | 85 +++++++++++++ 7 files changed, 625 insertions(+), 8 deletions(-) create mode 100644 deploy/session-log.mjs create mode 100644 lib/agent-ops.mjs create mode 100644 test/ops.test.mjs diff --git a/README.md b/README.md index b3e6974..9f3359d 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,37 @@ The slave gets its own device key; the master stores only its hash. **Revoke** c - **Open DSH ↗** opens a device's whole UI. - **Toasts:** ": '' finished", with an *Open* action. +## Agent tools: the master's AI drives the slaves + +On the master, sessions get four tools. Every paired, online device can be targeted by id or name; `self` means the master itself. + +| Tool | What it does | +| --- | --- | +| `remote_devices` | Lists devices with online state, user, home and kernel. With `sessions: true` it also returns each device's workspaces and recent sessions. | +| `remote_exec` | Runs `bash -lc <command>` on the device as its DSH user and returns exit code, stdout and stderr. It is non-interactive (`GIT_TERMINAL_PROMPT=0`). Options: `cwd`, `timeoutMs` (up to 30 min), `stdin`. Use it for quick deterministic steps: clone/pull, build, run tests. | +| `remote_file` | `read`, `write` (utf8 or base64, with append) and `list` for files on the device. Up to 4 MiB per call. | +| `remote_task` | Delegates a job to the device's **own AI**. It creates a session there (attached to the `cwd` workspace), or continues `sessionId`, sends the prompt and waits for the turn to end, up to 2 h. It returns the final answer, a count of tools used, and a link to the session (`/r/<device>/#/session/…`). `permission: "danger-full-access"` runs the session without the workspace sandbox. `action: status` / `cancel` poll or stop a running task. | + +Example, typed into any master session: + +> On ai-pc, clone https://git.rizzy.dev/Python/ai-pc into ~/work/ai-pc (or git pull), run its unit tests and report the result. + +The master's AI calls `remote_exec` (clone, venv, `pytest`) and reports "20 passed, 5 deselected, commit 32d4434". With `remote_task` the AI on ai-pc does the work itself, and its session stays browsable under Devices. + +**Choosing a tool:** `remote_exec` is cheap and exact; you get the command output. `remote_task` uses the slave's model tokens and judgement, which suits open-ended work ("fix the failing test"). The full transcript stays on the slave. + +**Who can control whom:** only the master controls devices. Slaves reject agent ops from anyone except their master, and the master rejects them from slaves. `agentTools.allowPeerControl: true` on the master lets slaves relay ops to each other. `agentTools.enabled: false` on a device refuses all agent control of that device. + +### Git credentials on slaves (`FORGE_CREDENTIALS`) + +Private repos need a login on the slave. `FORGE_CREDENTIALS` is a pseudo-secret that works like the other opt-in secrets: list it in the master's `offerSecrets`, enable it for the device on the Devices page, and add it to the slave's `sync.secrets`. On the next sync the slave gets the master's Gitea/Forgejo logins from `tea-cli`, written as: + +- `~/.config/tea/config.yml` (0600), so the synced git-forge plugin's `forge_*` tools work there; +- `~/.git-credentials` (0600) plus `git config --global credential.helper store`, so plain `git clone`/`push` over https works; +- a default git identity, unless one is already set. + +GitHub `gh` logins are not shipped. + ## Configuration (`cordis.patch.yml`) ```yaml @@ -57,10 +88,14 @@ The slave gets its own device key; the master stores only its hash. **Revoke** c auto: true # slave: sync on link-up and when the master pushes allow: [] # slave: only these bundles (exact or prefix*) deny: [] # slave: never these bundles (wins over allow) - secrets: [] # slave: credential refs it accepts from the master - offerSecrets: [OMNI_API_KEY] # master: refs it may hand out (per-device opt-in on the Devices page) + secrets: [] # slave: credential refs it accepts from the master (+ FORGE_CREDENTIALS) + offerSecrets: [OMNI_API_KEY, FORGE_CREDENTIALS] # master: what it may hand out (per-device opt-in on the Devices page) excludeBundles: [] # master: never sync these bundles excludePatchIds: [] # master: never sync these cordis.patch.yml entry ids + agentTools: + tools: true # register remote_* tools (default: on for the master, off for slaves) + enabled: true # false = refuse agent control of THIS device + allowPeerControl: false # master: let slaves relay exec/file/task ops to other devices ``` Never synced: the master's `dsh-remote` and `dsh-lanmode` entries, and `connection`/`webserver` settings, since these are machine-local. @@ -83,6 +118,7 @@ Local plugins are compared by content hash and re-shipped only when they change. - The proxy strips the browser's DSH cookies. The **owning** device mints a short-lived cookie for its own loopback authority, signed with its own browser-session secret, and rewrites `Host`/`Origin`. A remote DSH therefore only trusts its own minted cookie, and credentials never leave the machine they belong to, except secrets you explicitly opted into. - `/dsh-remote/enroll` is the only unauthenticated route. It accepts a single-use pairing code. - Use TLS (`https://` master URL) whenever the link crosses an untrusted network. +- **Agent control is real remote code execution.** `remote_exec`, `remote_file` and `remote_task` run as the device's DSH user. Anyone who can prompt a master session can run code on every paired device. Revoke a device, or set `agentTools.enabled: false` on it, to opt it out. ## Public access via rizzy.dev @@ -109,6 +145,7 @@ For example, add a second Traefik router with a higher priority, `Host(\`dsh.riz | `POST /dsh-remote/sync` | slave: pull now · master: push to all | | `GET /dsh-remote/sync/status` | manifest (master) / last result (slave) | | `POST /dsh-remote/new-session` | `{device, workspaceId? , cwd?}` | +| `GET /dsh-remote/debug/inspect` | `?sessionId=…&device=…`: last events of a session on any device | | `* /r/<device>/…` | that device's DSH | ## Development @@ -119,6 +156,7 @@ node deploy/dshr-cli.mjs GET state # call the local API as the DSH owner node deploy/e2e-ws.mjs ai-pc # Remote WebSocket through /r/ai-pc/ node deploy/e2e-events.mjs 30 # watch the SSE event stream node deploy/e2e-ui.mjs <chrome> /tmp/out # headless UI check + screenshots +node deploy/session-log.mjs <sessionId> # compact transcript (DEVICE=ai-pc for a remote one) deploy/push-slave.sh agent@192.168.0.110 # copy to a slave + restart its dsh ``` diff --git a/cordis.patch.yml b/cordis.patch.yml index b198248..8bd2e4b 100644 --- a/cordis.patch.yml +++ b/cordis.patch.yml @@ -16,6 +16,7 @@ allow: [] deny: [] # slave: names of credential refs to accept from the master (opt-in, per secret). + # FORGE_CREDENTIALS = the master's Gitea/Forgejo (tea) logins + git https credentials. secrets: [] # master: credential refs offered to devices (each device still opts in on the master # Devices page AND locally via `secrets` above). @@ -23,3 +24,11 @@ # master: never sync these bundles / cordis.patch.yml entry ids. excludeBundles: [] excludePatchIds: [] + agentTools: + # Register remote_devices / remote_exec / remote_file / remote_task. + # Unset = on for the master only; true = also on a slave. + # tools: true + # false = refuse agent control of THIS device (exec/files/tasks from the master). + enabled: true + # master: let slaves relay agent ops to other devices. + allowPeerControl: false diff --git a/deploy/session-log.mjs b/deploy/session-log.mjs new file mode 100644 index 0000000..fc40a44 --- /dev/null +++ b/deploy/session-log.mjs @@ -0,0 +1,31 @@ +#!/usr/bin/env node +// Print a compact transcript of a local DSH session (tool calls, results, answers). +// node deploy/session-log.mjs <sessionId> [sinceSeq] +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { mintCookie } from '../lib/auth-cookie.mjs' + +const [sid, since = '-1'] = process.argv.slice(2) +const port = process.env.DSH_PORT || '3081' +const home = process.env.DSH_HOME || path.join(os.homedir(), '.dsh') +const secret = fs.readFileSync(path.join(home, '.credentials.yaml'), 'utf8').match(/browser-session:[\s\S]*?secret:\s*([A-Za-z0-9_-]{43})/)[1] +const authority = `127.0.0.1:${port}` +const device = process.env.DEVICE ? `&device=${encodeURIComponent(process.env.DEVICE)}` : '' +const res = await fetch(`http://${authority}/dsh-remote/debug/inspect?sessionId=${encodeURIComponent(sid)}${device}`, { headers: { cookie: mintCookie(authority, secret, 1) } }) +const j = await res.json() +if (!res.ok) { console.error(res.status, j); process.exit(1) } +const short = (s, n = 300) => { const t = String(s ?? '').replace(/\s+/g, ' '); return t.length > n ? t.slice(0, n) + '…' : t } +for (const e of j.events) { + if (e.seq <= Number(since)) continue + const d = e.data || {} + if (e.type === 'user/message') console.log(`#${e.seq} USER: ${short(d.content?.map((c) => c.text).join(' '), 200)}`) + else if (e.type === 'tool/call') console.log(`#${e.seq} CALL ${d.name}: ${short(JSON.stringify(d.arguments), 260)}`) + else if (e.type === 'tool/result') { + const c = d.message?.content || [] + console.log(`#${e.seq} → ${short(c.map((x) => x.text ?? JSON.stringify(x)).join(' '), 360)}`) + } else if (e.type === 'assistant/message') { + const t = (d.message?.content || []).filter((c) => c.type === 'text').map((c) => c.text).join(' ') + if (t) console.log(`#${e.seq} AI: ${short(t, 1200)}`) + } else if (e.type === 'turn/end') console.log(`#${e.seq} TURN END ${JSON.stringify(d.reason)}`) +} diff --git a/index.js b/index.js index efa7af1..ddc596b 100644 --- a/index.js +++ b/index.js @@ -32,12 +32,14 @@ import { Store, randomId, setProfileDir, slug } from './lib/store.mjs' import { CookieJar } from './lib/auth-cookie.mjs' import { ownHttp, ownUpgrade, viewHttp, viewUpgrade } from './lib/proxy.mjs' import * as sync from './lib/sync.mjs' +import * as ops from './lib/agent-ops.mjs' export const name = 'dsh-remote' export const inject = { webServer: { required: true }, connection: { required: true }, credentials: { required: false }, + tools: { required: false }, } /** Optional services, read at call time (they may load after this plugin). */ @@ -49,6 +51,10 @@ const MAX_BODY = 256 * 1024 const PREFIX = '/r/' const LINK_PATH = '/dsh-remote/link' const MUX_PATH = '/api/remote.mux' +/** Link methods that act on a device on behalf of an agent. */ +const OPS = new Set(['exec', 'readFile', 'writeFile', 'listDir', 'task', 'taskStatus', 'taskCancel', 'inspectSession']) +/** Pseudo-secret name: opting into it ships the master's forge (git) logins. */ +const FORGE_SECRET = 'FORGE_CREDENTIALS' function sendJson(res, status, payload) { res.statusCode = status @@ -153,12 +159,44 @@ export function apply(ctx, config = {}) { return { name, value: r.value } }, // Forwarded calls: master answers for another device by asking it. - forward: async (p) => { + forward: async (p, mux) => { if (role !== 'master') throw httpErr(400, 'not a master') const peer = links.get(p.device) if (!peer) throw httpErr(404, `device ${p.device} offline`) - return peer.mux.call(p.method, p.params || {}, 20000) + // Agent control (exec/files/task) is master → device only, unless explicitly allowed. + if (OPS.has(p.method) && config.agentTools?.allowPeerControl !== true) throw httpErr(403, `${p.method} on other devices is only allowed from the master`) + return peer.mux.call(p.method, p.params || {}, p.timeoutMs || 20000) }, + // ---- agent operations (served by the target device) ---- + exec: async (p, mux) => { guardOps(mux); return ops.execCommand(p) }, + readFile: async (p, mux) => { guardOps(mux); return ops.readFileOp(p) }, + writeFile: async (p, mux) => { guardOps(mux); return ops.writeFileOp(p) }, + listDir: async (p, mux) => { guardOps(mux); return ops.listDirOp(p) }, + task: async (p, mux) => { guardOps(mux); return ops.runTask(svc(ctx, 'sessionController'), svc(ctx, 'workspaceRegistry'), p, undefined, svc(ctx, 'permissionPresets')) }, + taskStatus: async (p, mux) => { guardOps(mux); return ops.taskStatus(svc(ctx, 'sessionController'), p) }, + taskCancel: async (p, mux) => { guardOps(mux); return svc(ctx, 'sessionController').cancel({ sessionId: p.sessionId }) }, + inspectSession: async (p, mux) => { + guardOps(mux) + const ins = await svc(ctx, 'sessionController').inspect(p.sessionId) + return { sessionId: p.sessionId, events: ins.events.slice(-Math.min(Number(p.last) || 400, 2000)) } + }, + // Master → slave: hand over forge logins (only names the slave accepted). + forgeLogins: async (p, mux) => { + if (role !== 'master') throw httpErr(400, 'not a master') + const dev = store.devices()[mux.deviceId] + if (!offerSecretsFor(dev).includes(FORGE_SECRET)) throw httpErr(403, `${FORGE_SECRET} not offered to this device`) + return { logins: ops.readForgeLogins() } + }, + } + + /** + * Agent ops arrive over a link. On a slave, only the master may call them; on the master, + * slaves may not (the master is the controller) unless allowPeerControl is set. + */ + function guardOps(mux) { + if (config.agentTools?.enabled === false) throw httpErr(403, `agent control is disabled on ${self.name}`) + if (role === 'slave' && mux?.deviceId !== 'master') throw httpErr(403, 'agent control only from the master') + if (role === 'master' && mux && config.agentTools?.allowPeerControl !== true) throw httpErr(403, 'slaves may not control the master') } function offerSecretsFor(dev) { @@ -266,12 +304,23 @@ export function apply(ctx, config = {}) { return { mux: masterLink.mux, meta: { to: device } } // relay through master } - async function callDevice(device, method, params) { + async function callDevice(device, method, params, timeoutMs = 20000) { const r = routeTo(device) if (!r) throw httpErr(404, `device ${device} unreachable`) if (r.local) return handlers[method](params || {}) - if (r.meta.to) return r.mux.call('forward', { device, method, params }, 25000) - return r.mux.call(method, params || {}, 20000) + if (r.meta.to) return r.mux.call('forward', { device, method, params, timeoutMs }, timeoutMs + 5000) + return r.mux.call(method, params || {}, timeoutMs) + } + + /** Resolve a device by id or (case-insensitive) name; "self"/"local" = this device. */ + function resolveDevice(ref) { + const want = String(ref || '').trim().toLowerCase() + if (!want || want === 'self' || want === 'local' || want === 'this') return self.id + const all = devicesView() + const hit = all.find((d) => d.id === want) || all.find((d) => String(d.name).toLowerCase() === want) || all.find((d) => slug(d.name) === slug(want)) + if (!hit) throw httpErr(404, `unknown device "${ref}". Known: ${all.map((d) => `${d.id}${d.online ? '' : ' (offline)'}`).join(', ')}`) + if (!hit.online) throw httpErr(503, `device ${hit.id} is offline`) + return hit.id } async function allSessions() { @@ -483,6 +532,14 @@ export function apply(ctx, config = {}) { for (const name of manifest.secrets || []) { if (!(local.secrets || []).includes(name)) continue try { + if (name === FORGE_SECRET) { + // Not a credential ref: the master's forge logins (tea + git https credentials). + const { logins } = await mux.call('forgeLogins', {}, 15000) + if (!logins?.length) throw new Error('master has no forge logins') + const r = await ops.installForgeLogins(logins) + result.secrets.push(`${name} (${r.installed.join(', ')})`) + continue + } const { value } = await mux.call('secret', { name }, 15000) const creds = svc(ctx, 'credentials') if (!creds) throw new Error('credentials service unavailable') @@ -627,6 +684,15 @@ export function apply(ctx, config = {}) { return { role, lastSync, synced: store.data.synced || {} } }) + // Session events of a session on any device (same auth as the UI; used by tooling). + route('/dsh-remote/debug/inspect', ['GET'], async (_req, url) => { + const sessionId = url.searchParams.get('sessionId') + const device = url.searchParams.get('device') || self.id + if (!sessionId) throw httpErr(400, 'sessionId required') + if (device !== self.id) return callDevice(device, 'inspectSession', { sessionId }, 30000) + return handlers.inspectSession({ sessionId }) + }) + route('/dsh-remote/new-session', ['POST'], async (req) => { const body = await readJson(req) const device = String(body.device || self.id) @@ -723,5 +789,125 @@ export function apply(ctx, config = {}) { upgradeRoutes.clear() }, 'dsh-remote: upgrade routes') + // ------------------------------------------------------------------ agent tools + // Registered only where ctx.tools exists (agent compositions). By default they are + // offered on the master; a slave can enable them too (agentTools.enabled: true), in which + // case it can only reach itself and, if the master allows peer control, other devices. + const tools = svc(ctx, 'tools') + const toolsOn = config.agentTools?.tools !== false && (role === 'master' || config.agentTools?.tools === true) + if (tools && typeof tools.register === 'function' && toolsOn) { + const T = (def, fn) => ({ + name: def.name, + description: def.description, + parameters: { type: 'object', properties: def.properties, required: def.required ?? [], additionalProperties: false }, + output: { schema: {}, render: (_a, v) => [{ type: 'text', text: typeof v === 'string' ? v : JSON.stringify(v, null, 1) }] }, + timeoutMs: def.timeoutMs, + async execute(args, exec) { + try { return JSON.parse(JSON.stringify((await fn(args ?? {}, exec)) ?? null)) } catch (e) { return { error: String(e?.message ?? e) } } + }, + }) + const S = (type, description, extra = {}) => ({ type, description, ...extra }) + const DEVICE = S('string', 'Device id or name from remote_devices (e.g. "ai-pc"); "self" = this machine.') + const CWD = S('string', 'Working directory on the device (absolute or ~/…; default: the device user\'s home).') + const toolDefs = [ + T({ + name: 'remote_devices', + description: 'List the DSH devices linked through dsh-remote (this master + paired slaves): id, name, online state, user, home, platform, and optionally their workspaces and recent sessions. Use the ids with remote_exec / remote_file / remote_task.', + properties: { sessions: S('boolean', 'Include each online device\'s workspaces and recent sessions (default false).') }, + }, async (args) => { + const devices = devicesView().filter((d) => !d.revoked) + const out = [] + for (const d of devices) { + const row = { id: d.id, name: d.name, role: d.role, online: !!d.online, self: !!d.self, platform: d.platform || (d.self ? `${process.platform}-${process.arch}` : undefined) } + if (d.online) { + try { + const info = await callDevice(d.id, 'exec', { command: 'printf "%s\\n%s\\n%s" "$(id -un)" "$HOME" "$(uname -sr)"', timeoutMs: 8000 }, 12000) + const [user, home, kernel] = String(info.stdout || '').split('\n') + Object.assign(row, { user, home, kernel }) + } catch (e) { row.execError = e.message } + if (args.sessions) { + try { + const s = await callDevice(d.id, 'sessions', {}, 15000) + row.workspaces = s.workspaces + row.sessions = (s.items || []).slice(0, 15).map(({ sessionId, title, running, cwd, updatedAt }) => ({ sessionId, title, running, cwd, updatedAt: new Date(updatedAt).toISOString() })) + } catch (e) { row.sessionsError = e.message } + } + } + out.push(row) + } + return { devices: out } + }), + T({ + name: 'remote_exec', + description: 'Run a bash command on another DSH device (e.g. a slave) and return exit code, stdout and stderr. Runs as that device\'s DSH user, in `bash -lc`, non-interactive (no TTY, GIT_TERMINAL_PROMPT=0). Use for quick deterministic steps: git clone/pull, builds, test runs, inspecting files. For long multi-step work prefer remote_task.', + properties: { + device: DEVICE, + command: S('string', 'The bash command line.'), + cwd: CWD, + timeoutMs: S('integer', 'Kill the command after this many ms (default 120000, max 1800000).'), + stdin: S('string', 'Optional data written to stdin.'), + }, + required: ['device', 'command'], + timeoutMs: 31 * 60 * 1000, + }, async (args) => { + const device = resolveDevice(args.device) + const t = Math.min(Math.max(Number(args.timeoutMs) || 120000, 1000), 1800000) + const r = await callDevice(device, 'exec', { command: args.command, cwd: args.cwd, timeoutMs: t, stdin: args.stdin }, t + 15000) + return { device, ...r } + }), + T({ + name: 'remote_file', + description: 'Files on another DSH device. action=read (path, offset, limit; binary comes back base64), write (path, content, encoding utf8|base64, append), list (directory entries).', + properties: { + device: DEVICE, + action: S('string', 'read | write | list', { enum: ['read', 'write', 'list'] }), + path: S('string', 'File or directory path (absolute or relative to cwd / home).'), + cwd: CWD, + content: S('string', 'write: file content.'), + encoding: S('string', 'write: utf8 (default) or base64.', { enum: ['utf8', 'base64'] }), + append: S('boolean', 'write: append instead of replace.'), + offset: S('integer', 'read: byte offset.'), + limit: S('integer', 'read: max bytes (default/max 4 MiB).'), + hidden: S('boolean', 'list: include dotfiles.'), + }, + required: ['device', 'action', 'path'], + }, async (args) => { + const device = resolveDevice(args.device) + const method = { read: 'readFile', write: 'writeFile', list: 'listDir' }[args.action] + if (!method) throw httpErr(400, 'action must be read, write or list') + return { device, ...(await callDevice(device, method, args, 30000)) } + }), + T({ + name: 'remote_task', + description: 'Delegate a task to the AI agent running on another DSH device, so the work executes there with that device\'s own tools and model, and get its final answer. action=run (default): create a session on the device (in cwd, attached to that workspace) or continue sessionId, send prompt, and wait up to timeoutMs for the turn to finish. action=status: collect the latest answer of sessionId. action=cancel: stop it. The session stays visible on that device (Devices panel / /r/<device>/). Write a complete, self-contained prompt — the remote agent has no access to this conversation.', + properties: { + device: DEVICE, + action: S('string', 'run (default) | status | cancel', { enum: ['run', 'status', 'cancel'] }), + prompt: S('string', 'run: the full task for the remote agent.'), + sessionId: S('string', 'Continue / poll / cancel this session on the device.'), + cwd: CWD, + title: S('string', 'run, new session: session title.'), + permission: S('string', 'run: permission preset for the remote session, e.g. "danger-full-access" when the job must run tools outside the workspace sandbox (test suites that spawn processes, git credential writes). Default: the device\'s default preset.'), + wait: S('boolean', 'run: wait for completion (default true). false = start and return the sessionId.'), + timeoutMs: S('integer', 'run: how long to wait (default 1200000 = 20 min, max 2 h). On timeout the task keeps running; poll with action=status.'), + }, + required: ['device'], + timeoutMs: 2 * 60 * 60 * 1000 + 60000, + }, async (args, exec) => { + const device = resolveDevice(args.device) + const action = args.action || 'run' + if (action === 'status') return { device, ...(await callDevice(device, 'taskStatus', { sessionId: args.sessionId }, 30000)) } + if (action === 'cancel') return { device, ...(await callDevice(device, 'taskCancel', { sessionId: args.sessionId }, 15000)) } + const t = Math.min(Math.max(Number(args.timeoutMs) || 1200000, 10000), 2 * 60 * 60 * 1000) + const r = await callDevice(device, 'task', { + prompt: args.prompt, sessionId: args.sessionId, cwd: args.cwd, title: args.title, permission: args.permission, + wait: args.wait !== false, timeoutMs: t, + }, t + 30000) + return { device, ...r, url: device === self.id ? `#/session/${r.sessionId}` : `/r/${device}/#/session/${r.sessionId}` } + }), + ] + ctx.effect(function* () { for (const t of toolDefs) yield tools.register(t) }, 'dsh-remote: agent tools') + } + log.info?.(`dsh-remote: ${role} "${self.name}" (${self.id}) ready`) } diff --git a/lib/agent-ops.mjs b/lib/agent-ops.mjs new file mode 100644 index 0000000..27a3e1c --- /dev/null +++ b/lib/agent-ops.mjs @@ -0,0 +1,268 @@ +// 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 } +} diff --git a/package.json b/package.json index 4af2c03..2fccae5 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@local/dsh-remote", - "version": "0.1.0", + "version": "0.2.0", "description": "Claude-Desktop-style remote control for DSH: master/slave link, every device's sessions on every device (/r/<device>/), plugin + config sync with opt-in secrets.", "private": true, "type": "module", diff --git a/test/ops.test.mjs b/test/ops.test.mjs new file mode 100644 index 0000000..3d798f6 --- /dev/null +++ b/test/ops.test.mjs @@ -0,0 +1,85 @@ +// Device-side agent ops: exec, file read/write/list, task delegation (with a fake +// session controller), forge login install into a temp HOME. +import test from 'node:test' +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import * as ops from '../lib/agent-ops.mjs' + +test('exec: exit code, stdout/stderr, cwd, timeout', async () => { + const r = await ops.execCommand({ command: 'echo out; echo err >&2; pwd; exit 3', cwd: '/tmp' }) + assert.equal(r.exitCode, 3) + assert.match(r.stdout, /^out\n\/tmp\n$/) + assert.equal(r.stderr, 'err\n') + const t = await ops.execCommand({ command: 'sleep 5', timeoutMs: 1000 }) + assert.equal(t.timedOut, true) + assert.notEqual(t.exitCode, 0) + await assert.rejects(ops.execCommand({ command: 'true', cwd: '/nonexistent/dir' }), /cwd does not exist/) +}) + +test('files: write, append, read (text + binary), list', () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'dshr-ops-')) + ops.writeFileOp({ path: 'a/b.txt', cwd: dir, content: 'hello' }) + ops.writeFileOp({ path: 'a/b.txt', cwd: dir, content: ' world', append: true }) + assert.equal(ops.readFileOp({ path: path.join(dir, 'a/b.txt') }).content, 'hello world') + ops.writeFileOp({ path: 'bin', cwd: dir, content: Buffer.from([0, 1, 2]).toString('base64'), encoding: 'base64' }) + const b = ops.readFileOp({ path: 'bin', cwd: dir }) + assert.equal(b.encoding, 'base64') + assert.equal(b.content, 'AAEC') + const l = ops.listDirOp({ path: dir }) + assert.deepEqual(l.entries.map((e) => `${e.type}:${e.name}`).sort(), ['dir:a', 'file:bin']) + fs.rmSync(dir, { recursive: true, force: true }) +}) + +test('task: creates a session, prompts, waits for turn/end, returns last answer', async () => { + const events = [{ seq: 0, type: 'session/start' }] + let prompted = null + const sc = { + async create(req) { assert.ok(req.cwd || req.workspaceId); return { sessionId: 'session-x' } }, + async inspect() { return { events: [...events] } }, + async prompt(req) { + prompted = req + setTimeout(() => { + events.push({ seq: 1, type: 'turn/start', data: { turn: 1 } }) + events.push({ seq: 2, type: 'tool/call', data: { name: 'bash' } }) + events.push({ seq: 3, type: 'assistant/message', data: { message: { content: [{ type: 'text', text: 'all 12 tests passed' }] } } }) + events.push({ seq: 4, type: 'turn/end', data: { reason: { kind: 'completed' } } }) + }, 300) + return { accepted: true } + }, + } + const r = await ops.runTask(sc, null, { prompt: 'run the tests', cwd: os.tmpdir(), timeoutMs: 15000 }) + assert.equal(prompted.content[0].text, 'run the tests') + assert.equal(r.sessionId, 'session-x') + assert.equal(r.status, 'completed') + assert.equal(r.answer, 'all 12 tests passed') + assert.deepEqual(r.tools, { bash: 1 }) + const s = await ops.taskStatus(sc, { sessionId: 'session-x' }) + assert.equal(s.status, 'completed') + assert.equal(s.answer, 'all 12 tests passed') +}) + +test('forge logins: tea config + git credential store in a temp HOME', async () => { + const home = fs.mkdtempSync(path.join(os.tmpdir(), 'dshr-home-')) + const prev = { HOME: process.env.HOME, XDG: process.env.XDG_CONFIG_HOME } + process.env.HOME = home + delete process.env.XDG_CONFIG_HOME + const origHomedir = os.homedir + os.homedir = () => home + try { + const r = await ops.installForgeLogins([{ name: 'Agent', url: 'https://git.example.com', user: 'Agent', token: 'tok123' }]) + assert.deepEqual(r.installed, ['Agent@git.example.com']) + const tea = fs.readFileSync(path.join(home, '.config/tea/config.yml'), 'utf8') + assert.match(tea, /token: tok123/) + assert.equal(fs.statSync(path.join(home, '.git-credentials')).mode & 0o777, 0o600) + assert.match(fs.readFileSync(path.join(home, '.git-credentials'), 'utf8'), /^https:\/\/Agent:tok123@git\.example\.com$/m) + const helper = await ops.execCommand({ command: 'HOME=' + home + ' git config --global credential.helper' }) + assert.equal(helper.stdout.trim(), 'store') + } finally { + os.homedir = origHomedir + process.env.HOME = prev.HOME + if (prev.XDG) process.env.XDG_CONFIG_HOME = prev.XDG + fs.rmSync(home, { recursive: true, force: true }) + } +})