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

@@ -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:** "<device>: '<title>' 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
+      secrets: []           # slave: credential refs it accepts from the master (+ FORGE_CREDENTIALS)
-      offerSecrets: [OMNI_API_KEY]   # master: refs it may hand out (per-device opt-in on the Devices page)
+      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
 ```
 

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

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)}`)
+}

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)
+    if (r.meta.to) return r.mux.call('forward', { device, method, params, timeoutMs }, timeoutMs + 5000)
-    return r.mux.call(method, params || {}, 20000)
+    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`)
 }

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

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",

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 })
+  }
+})