From 25ff9efcce8abcaa65147a5d78991ccad5078726 Mon Sep 17 00:00:00 2001 From: Bo Wu Date: Fri, 21 Aug 2026 15:57:17 -0700 Subject: [PATCH 1/3] Add container-native multiagent control server --- .dockerignore | 10 + .github/workflows/container.yml | 39 ++++ .gitignore | 1 + Dockerfile | 31 +++ README.md | 42 ++++ bin/container-entrypoint.sh | 15 ++ bin/git-askpass.sh | 6 + bin/hash-password.mjs | 17 ++ bin/sync-repositories.mjs | 25 +++ control-server/package-lock.json | 36 +++ control-server/package.json | 12 + control-server/public/app.js | 95 ++++++++ control-server/public/index.html | 57 +++++ control-server/public/styles.css | 43 ++++ control-server/src/server.mjs | 370 +++++++++++++++++++++++++++++++ 15 files changed, 799 insertions(+) create mode 100644 .dockerignore create mode 100644 .github/workflows/container.yml create mode 100644 Dockerfile create mode 100644 bin/container-entrypoint.sh create mode 100644 bin/git-askpass.sh create mode 100644 bin/hash-password.mjs create mode 100644 bin/sync-repositories.mjs create mode 100644 control-server/package-lock.json create mode 100644 control-server/package.json create mode 100644 control-server/public/app.js create mode 100644 control-server/public/index.html create mode 100644 control-server/public/styles.css create mode 100644 control-server/src/server.mjs diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..6734515 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +.git +.multiagent +.worktrees +InternalServices +benchmarks +evaluation +prod-mcp +patch.txt +**/node_modules +**/target diff --git a/.github/workflows/container.yml b/.github/workflows/container.yml new file mode 100644 index 0000000..8b85392 --- /dev/null +++ b/.github/workflows/container.yml @@ -0,0 +1,39 @@ +name: container + +on: + push: + branches: + - main + workflow_dispatch: + +permissions: + contents: read + packages: write + +jobs: + publish: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + - uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3 + - uses: docker/login-action@184bdaa0721073962dff0199f1fb9940f07167d1 # v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5 + id: metadata + with: + images: ghcr.io/${{ github.repository_owner }}/multiagent + tags: | + type=raw,value=main,enable={{is_default_branch}} + type=sha,prefix=sha- + - uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6 + with: + context: . + push: true + tags: ${{ steps.metadata.outputs.tags }} + labels: ${{ steps.metadata.outputs.labels }} + platforms: linux/amd64 + provenance: mode=max + sbom: true diff --git a/.gitignore b/.gitignore index b52eb4c..7039c2a 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ evaluation/runs/ evaluation/reports/* !evaluation/reports/.gitkeep !evaluation/reports/swe-bench-pro-prod-multiagent-first50-summary.md +control-server/node_modules/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..acd015f --- /dev/null +++ b/Dockerfile @@ -0,0 +1,31 @@ +FROM node:22-bookworm-slim + +ARG CODEX_VERSION=0.145.0 +ARG CLAUDE_CODE_VERSION=2.1.239 + +RUN apt-get update \ + && apt-get install -y --no-install-recommends awscli ca-certificates git python3 tmux \ + && rm -rf /var/lib/apt/lists/* \ + && npm install --global "@openai/codex@${CODEX_VERSION}" "@anthropic-ai/claude-code@${CLAUDE_CODE_VERSION}" + +WORKDIR /opt/multiagent +COPY control-server/package*.json control-server/ +RUN cd control-server && npm install --omit=dev +COPY launch.sh orchestrator_prompt.md README.md ./ +COPY bin/ bin/ +COPY control-server/ control-server/ +RUN chmod +x launch.sh bin/*.sh bin/*.mjs \ + && useradd --create-home --home-dir /var/lib/multiagent --uid 10001 multiagent \ + && mkdir -p /var/lib/multiagent/state /var/lib/multiagent/repositories \ + && chown -R multiagent:multiagent /var/lib/multiagent + +USER 10001:10001 +ENV HOME=/var/lib/multiagent/home \ + CODEX_HOME=/var/lib/multiagent/codex \ + CLAUDE_CONFIG_DIR=/var/lib/multiagent/claude \ + MULTIAGENT_LAUNCHER_ROOT=/opt/multiagent \ + MULTIAGENT_STATE_DIR=/var/lib/multiagent/state \ + MULTIAGENT_REPOSITORY_ROOT=/var/lib/multiagent/repositories \ + PORT=8080 +EXPOSE 8080 +ENTRYPOINT ["/opt/multiagent/bin/container-entrypoint.sh"] diff --git a/README.md b/README.md index 90d147c..0886358 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,48 @@ does not implement another coding agent or model loop. It runs Codex, Claude Code, and Qwen Code in explicit roles, records durable workflow state, and accepts work only when reviewer evidence matches the exact final Git diff. +## Stateful control server + +The container image runs a same-origin web UI and authenticated WebSocket gateway as PID 1. The server bootstraps configured repositories, starts or resumes tmux orchestrator sessions, streams orchestrator pane output, accepts user messages, checkpoints session output, and mirrors durable state to S3. + +Users are configured in a mounted JSON file. Passwords must be scrypt hashes, never plaintext: + +```bash +node bin/hash-password.mjs operator +``` + +The mounted file has this shape: + +```json +{ + "sessionSecret": "at-least-32-random-characters", + "users": [ + {"username": "operator", "passwordHash": "scrypt$16384$8$1$..."} + ] +} +``` + +Repositories are allowlisted in `MULTIAGENT_REPOSITORIES_FILE`: + +```json +{ + "repositories": [ + {"name": "example", "url": "https://github.com/example/repository.git", "ref": "main"} + ] +} +``` + +Important container variables: + +- `MULTIAGENT_USERS_FILE`: mounted login configuration, default `/run/secrets/multiagent/users.json`. +- `MULTIAGENT_REPOSITORIES_FILE`: mounted repository allowlist. +- `MULTIAGENT_BOOTSTRAP_REPOSITORY`: repository used for the initial orchestrator session. +- `MULTIAGENT_BOOTSTRAP_SESSION`: initial session name, default `orchestrator`. +- `MULTIAGENT_STATE_S3_URI`: S3 prefix used for recovery snapshots. +- `MULTIAGENT_PUBLIC_URL`: canonical HTTPS origin accepted for browser and WebSocket requests. + +The PVC mounted at `/var/lib/multiagent` is the primary store for repositories, CLI conversation history, checkpoints, and session metadata. S3 is the durable recovery and inspection copy. + ## Requirements From a source checkout you need Rust 1.75+, Cargo, Bash, Git, and tmux. Install diff --git a/bin/container-entrypoint.sh b/bin/container-entrypoint.sh new file mode 100644 index 0000000..99f721a --- /dev/null +++ b/bin/container-entrypoint.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +set -euo pipefail +export HOME="${HOME:-/var/lib/multiagent/home}" +export CODEX_HOME="${CODEX_HOME:-/var/lib/multiagent/codex}" +export CLAUDE_CONFIG_DIR="${CLAUDE_CONFIG_DIR:-/var/lib/multiagent/claude}" +export MULTIAGENT_STATE_DIR="${MULTIAGENT_STATE_DIR:-/var/lib/multiagent/state}" +export MULTIAGENT_REPOSITORY_ROOT="${MULTIAGENT_REPOSITORY_ROOT:-/var/lib/multiagent/repositories}" +export GIT_ASKPASS="${GIT_ASKPASS:-/opt/multiagent/bin/git-askpass.sh}" +export GIT_TERMINAL_PROMPT=0 +mkdir -p "$HOME" "$CODEX_HOME" "$CLAUDE_CONFIG_DIR" "$MULTIAGENT_STATE_DIR" "$MULTIAGENT_REPOSITORY_ROOT" +if [[ -n "${MULTIAGENT_STATE_S3_URI:-}" && ! -f "$MULTIAGENT_STATE_DIR/control-server/sessions.json" ]]; then + aws s3 sync "$MULTIAGENT_STATE_S3_URI" "$MULTIAGENT_STATE_DIR" --only-show-errors || true +fi +node /opt/multiagent/bin/sync-repositories.mjs +exec node /opt/multiagent/control-server/src/server.mjs diff --git a/bin/git-askpass.sh b/bin/git-askpass.sh new file mode 100644 index 0000000..e873367 --- /dev/null +++ b/bin/git-askpass.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +case "$1" in + *sername*) printf '%s\n' "x-access-token" ;; + *assword*) printf '%s\n' "${GITHUB_TOKEN:?GITHUB_TOKEN is required}" ;; + *) exit 1 ;; +esac diff --git a/bin/hash-password.mjs b/bin/hash-password.mjs new file mode 100644 index 0000000..1ddc488 --- /dev/null +++ b/bin/hash-password.mjs @@ -0,0 +1,17 @@ +#!/usr/bin/env node +import crypto from "node:crypto"; +import readline from "node:readline"; + +const username = process.argv[2]; +if (!username || !/^[a-zA-Z0-9._-]{1,64}$/.test(username)) { + console.error("usage: bin/hash-password.mjs USERNAME"); + process.exit(2); +} +const terminal = readline.createInterface({ input: process.stdin, output: process.stderr, terminal: true }); +terminal.question("Password: ", (password) => { + terminal.close(); + if (password.length < 12) { console.error("password must be at least 12 characters"); process.exit(2); } + const salt = crypto.randomBytes(16); + const hash = crypto.scryptSync(password, salt, 64, { N: 16384, r: 8, p: 1, maxmem: 64 * 1024 * 1024 }); + console.log(JSON.stringify({ username, passwordHash: `scrypt$16384$8$1$${salt.toString("base64url")}$${hash.toString("base64url")}` })); +}); diff --git a/bin/sync-repositories.mjs b/bin/sync-repositories.mjs new file mode 100644 index 0000000..1cce366 --- /dev/null +++ b/bin/sync-repositories.mjs @@ -0,0 +1,25 @@ +#!/usr/bin/env node +import fs from "node:fs"; +import path from "node:path"; +import { execFileSync } from "node:child_process"; + +const configFile = process.env.MULTIAGENT_REPOSITORIES_FILE || "/etc/multiagent/repositories.json"; +const root = path.resolve(process.env.MULTIAGENT_REPOSITORY_ROOT || "/var/lib/multiagent/repositories"); +if (!fs.existsSync(configFile)) process.exit(0); +const config = JSON.parse(fs.readFileSync(configFile, "utf8")); +fs.mkdirSync(root, { recursive: true }); +for (const repository of config.repositories || []) { + if (!/^[a-z0-9][a-z0-9-]{0,62}$/.test(repository.name)) throw new Error(`invalid repository name: ${repository.name}`); + if (!/^https:\/\/github\.com\/[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+(?:\.git)?$/.test(repository.url)) throw new Error(`unsupported repository URL: ${repository.url}`); + const destination = path.join(root, repository.name); + if (!fs.existsSync(path.join(destination, ".git"))) { + execFileSync("git", ["clone", "--origin", "origin", repository.url, destination], { stdio: "inherit" }); + } else { + execFileSync("git", ["-C", destination, "fetch", "origin", "--prune"], { stdio: "inherit" }); + } + const status = execFileSync("git", ["-C", destination, "status", "--porcelain"], { encoding: "utf8" }); + if (!status.trim() && repository.ref) { + execFileSync("git", ["-C", destination, "checkout", repository.ref], { stdio: "inherit" }); + execFileSync("git", ["-C", destination, "pull", "--ff-only", "origin", repository.ref], { stdio: "inherit" }); + } +} diff --git a/control-server/package-lock.json b/control-server/package-lock.json new file mode 100644 index 0000000..c81aa04 --- /dev/null +++ b/control-server/package-lock.json @@ -0,0 +1,36 @@ +{ + "name": "multiagent-control-server", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "multiagent-control-server", + "version": "0.1.0", + "dependencies": { + "ws": "8.21.3" + } + }, + "node_modules/ws": { + "version": "8.21.3", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.3.tgz", + "integrity": "sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + } + } +} diff --git a/control-server/package.json b/control-server/package.json new file mode 100644 index 0000000..6931234 --- /dev/null +++ b/control-server/package.json @@ -0,0 +1,12 @@ +{ + "name": "multiagent-control-server", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "start": "node src/server.mjs" + }, + "dependencies": { + "ws": "8.21.3" + } +} diff --git a/control-server/public/app.js b/control-server/public/app.js new file mode 100644 index 0000000..0d60576 --- /dev/null +++ b/control-server/public/app.js @@ -0,0 +1,95 @@ +const $ = (selector) => document.querySelector(selector); +let active = null; +let socket = null; +let sessions = []; + +async function api(url, options = {}) { + const response = await fetch(url, { headers: { "content-type": "application/json", ...(options.headers || {}) }, ...options }); + const body = await response.json().catch(() => ({})); + if (!response.ok) throw new Error(body.error || `Request failed (${response.status})`); + return body; +} + +function showLogin() { $("#login").hidden = false; $("#console").hidden = true; } +function showConsole(username) { $("#username").textContent = username; $("#login").hidden = true; $("#console").hidden = false; } + +async function refreshSessions() { + sessions = (await api("/api/sessions")).sessions; + const target = $("#sessions"); + target.replaceChildren(...sessions.map((session) => { + const button = document.createElement("button"); + button.className = `session${active === session.id ? " active" : ""}`; + button.innerHTML = `${session.id}${session.live ? "live" : session.status} · ${session.repository}`; + button.onclick = () => selectSession(session.id); + return button; + })); + if (active && !sessions.some((session) => session.id === active)) selectSession(null); +} + +function selectSession(id) { + if (socket) socket.close(); + active = id; + const session = sessions.find((candidate) => candidate.id === id); + $("#active-name").textContent = session?.id || "Select a session"; + $("#active-repo").textContent = session?.repository || ""; + $("#live-dot").classList.toggle("live", Boolean(session?.live)); + for (const id of ["restart", "checkpoint", "stop", "message", "send"]) $("#" + id).disabled = !session; + if (!session) { $("#terminal").textContent = "No orchestrator selected."; return refreshSessions(); } + $("#terminal").textContent = "Connecting to orchestrator…"; + const protocol = location.protocol === "https:" ? "wss" : "ws"; + socket = new WebSocket(`${protocol}://${location.host}/api/sessions/${id}/terminal`); + socket.onmessage = ({ data }) => { + const message = JSON.parse(data); + if (message.type === "output") { + const terminal = $("#terminal"); + const atBottom = terminal.scrollHeight - terminal.scrollTop - terminal.clientHeight < 80; + terminal.textContent = message.output; + $("#live-dot").classList.toggle("live", message.live); + if (atBottom) terminal.scrollTop = terminal.scrollHeight; + } + if (message.type === "error") $("#terminal").textContent += `\n[control error] ${message.error}`; + }; + socket.onclose = () => $("#live-dot").classList.remove("live"); + refreshSessions(); +} + +$("#login-form").onsubmit = async (event) => { + event.preventDefault(); + const data = Object.fromEntries(new FormData(event.currentTarget)); + try { const me = await api("/api/login", { method: "POST", body: JSON.stringify(data) }); showConsole(me.username); await refreshSessions(); } + catch (error) { $("#login-error").textContent = error.message; } +}; +$("#logout").onclick = async () => { await api("/api/logout", { method: "POST" }); if (socket) socket.close(); showLogin(); }; +$("#message-form").onsubmit = (event) => { + event.preventDefault(); + const text = $("#message").value; + if (!text.trim() || socket?.readyState !== WebSocket.OPEN) return; + socket.send(JSON.stringify({ type: "input", text })); + $("#message").value = ""; +}; +for (const action of ["restart", "checkpoint", "stop"]) $("#" + action).onclick = async () => { + if (!active) return; + await api(`/api/sessions/${active}/${action}`, { method: "POST" }); + await refreshSessions(); + if (action === "restart") selectSession(active); +}; +$("#new-session").onclick = async () => { + const repositories = (await api("/api/repositories")).repositories; + const select = $("#create-form select"); + select.replaceChildren(...repositories.map((name) => new Option(name, name))); + $("#create-dialog").showModal(); +}; +$("#cancel-create").onclick = () => $("#create-dialog").close(); +$("#create-form").onsubmit = async (event) => { + event.preventDefault(); + try { + const body = Object.fromEntries(new FormData(event.currentTarget)); + await api("/api/sessions", { method: "POST", body: JSON.stringify(body) }); + $("#create-dialog").close(); + await refreshSessions(); + selectSession(body.id); + } catch (error) { $("#create-error").textContent = error.message; } +}; + +try { const me = await api("/api/me"); showConsole(me.username); await refreshSessions(); } catch { showLogin(); } +setInterval(() => $("#console").hidden || refreshSessions().catch(() => {}), 5000); diff --git a/control-server/public/index.html b/control-server/public/index.html new file mode 100644 index 0000000..13d42b3 --- /dev/null +++ b/control-server/public/index.html @@ -0,0 +1,57 @@ + + + + + + Multiagent Control + + + +
+
+ +
+ +
+
+

Multiagent

Session Control

+
+
+
+ +
+
+
Select a session
+
+
+ 
No orchestrator selected.
+
+ + +
+
+
+
+ + +
+

Bootstrap

New session

+ + +
+

+
+ + + + diff --git a/control-server/public/styles.css b/control-server/public/styles.css new file mode 100644 index 0000000..178840d --- /dev/null +++ b/control-server/public/styles.css @@ -0,0 +1,43 @@ +:root { --ink:#171811; --paper:#e9e3d2; --acid:#d8ff3e; --rust:#bb3e24; --muted:#817d70; --line:#3a3b31; } +* { box-sizing:border-box; } +body { margin:0; min-height:100vh; background:var(--ink); color:var(--paper); font-family:"IBM Plex Mono","Courier New",monospace; } +.grain { position:fixed; inset:0; opacity:.12; pointer-events:none; background-image:repeating-linear-gradient(0deg,transparent,transparent 3px,#fff 4px); mix-blend-mode:overlay; } +.eyebrow { margin:0 0 .5rem; color:var(--acid); font-size:.72rem; letter-spacing:.2em; text-transform:uppercase; } +h1,h2 { margin:0; font-family:Georgia,"Times New Roman",serif; font-weight:400; } +.login-shell { min-height:100vh; display:grid; place-items:center; padding:2rem; background:radial-gradient(circle at 20% 10%,#45492b 0,transparent 32%),linear-gradient(135deg,#14150f,#25261c); } +.login-card { width:min(430px,100%); padding:3.4rem; border:1px solid #565844; box-shadow:14px 14px 0 #0b0c08; background:#202119; } +.login-card h1 { font-size:4rem; line-height:.82; margin-bottom:3rem; } +label { display:grid; gap:.5rem; margin:1.15rem 0; color:#b9b4a6; font-size:.75rem; text-transform:uppercase; letter-spacing:.12em; } +input,select,textarea { width:100%; border:1px solid #555748; border-radius:0; background:#11120d; color:var(--paper); padding:.85rem; font:inherit; } +input:focus,select:focus,textarea:focus { outline:2px solid var(--acid); outline-offset:1px; } +button { border:1px solid var(--acid); background:var(--acid); color:#11120d; padding:.72rem 1rem; font:700 .72rem/1 inherit; letter-spacing:.08em; text-transform:uppercase; cursor:pointer; } +button:disabled { opacity:.35; cursor:not-allowed; } +button.quiet,.actions button { background:transparent; color:var(--paper); border-color:#5f6150; } +.login-card > button { width:100%; margin-top:1.4rem; } +.error { min-height:1.2em; color:#ff876e; font-size:.78rem; } +main { min-height:100vh; } +header { height:92px; display:flex; align-items:center; justify-content:space-between; padding:0 2rem; border-bottom:1px solid var(--line); background:#1c1d16; } +header h1 { font-size:2rem; } +.operator { display:flex; align-items:center; gap:1rem; font-size:.8rem; } +.layout { display:grid; grid-template-columns:280px 1fr; height:calc(100vh - 92px); } +aside { border-right:1px solid var(--line); background:#202119; overflow:auto; } +.aside-head { display:flex; align-items:center; justify-content:space-between; padding:1.25rem; border-bottom:1px solid var(--line); } +.aside-head h2 { font-size:1.4rem; } +.aside-head button { width:34px; height:34px; padding:0; font-size:1.2rem; } +.session { width:100%; display:block; text-align:left; padding:1rem 1.2rem; background:transparent; color:var(--paper); border:0; border-bottom:1px solid #303126; text-transform:none; letter-spacing:0; } +.session.active { background:#303226; box-shadow:inset 4px 0 var(--acid); } +.session small { display:block; color:var(--muted); margin-top:.35rem; } +.terminal-panel { min-width:0; display:grid; grid-template-rows:auto 1fr auto; background:#10110d; } +.terminal-head { display:flex; align-items:center; justify-content:space-between; padding:1rem 1.25rem; border-bottom:1px solid var(--line); } +.terminal-head small { display:block; color:var(--muted); margin:.35rem 0 0 1.2rem; } +.dot { display:inline-block; width:8px; height:8px; margin-right:.7rem; border-radius:50%; background:#555; } +.dot.live { background:var(--acid); box-shadow:0 0 12px var(--acid); } +.actions { display:flex; gap:.5rem; } +#terminal { margin:0; padding:1.5rem; overflow:auto; white-space:pre-wrap; word-break:break-word; color:#d6d2c4; font:13px/1.52 "IBM Plex Mono","Courier New",monospace; } +#message-form { display:grid; grid-template-columns:1fr auto; gap:.75rem; padding:1rem; border-top:1px solid var(--line); background:#1b1c15; } +#message { resize:vertical; min-height:66px; } +dialog { width:min(480px,calc(100% - 2rem)); border:1px solid #696b57; background:#202119; color:var(--paper); padding:2rem; box-shadow:12px 12px 0 #090a07; } +dialog::backdrop { background:#080906cc; } +dialog h2 { font-size:2.2rem; margin-bottom:1.5rem; } +.dialog-actions { display:flex; justify-content:flex-end; gap:.75rem; margin-top:1.5rem; } +@media (max-width:760px) { header { padding:0 1rem; } .layout { grid-template-columns:1fr; grid-template-rows:190px 1fr; } aside { border-right:0; border-bottom:1px solid var(--line); } #sessions { display:flex; overflow:auto; } .session { min-width:180px; border-right:1px solid #303126; } .terminal-head { align-items:flex-start; gap:1rem; } .actions { flex-wrap:wrap; justify-content:flex-end; } #message-form { grid-template-columns:1fr; } } diff --git a/control-server/src/server.mjs b/control-server/src/server.mjs new file mode 100644 index 0000000..369563d --- /dev/null +++ b/control-server/src/server.mjs @@ -0,0 +1,370 @@ +import crypto from "node:crypto"; +import fs from "node:fs"; +import http from "node:http"; +import path from "node:path"; +import { execFile, execFileSync, spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { WebSocketServer } from "ws"; + +const here = path.dirname(fileURLToPath(import.meta.url)); +const publicDir = path.resolve(here, "../public"); +const launcherRoot = path.resolve(process.env.MULTIAGENT_LAUNCHER_ROOT || path.join(here, "../..")); +const stateRoot = path.resolve(process.env.MULTIAGENT_STATE_DIR || "/var/lib/multiagent/state"); +const repositoryRoot = path.resolve(process.env.MULTIAGENT_REPOSITORY_ROOT || "/var/lib/multiagent/repositories"); +const usersFile = path.resolve(process.env.MULTIAGENT_USERS_FILE || "/run/secrets/multiagent/users.json"); +const port = Number(process.env.PORT || "8080"); +const host = process.env.HOST || "0.0.0.0"; +const cookieSecure = process.env.MULTIAGENT_COOKIE_SECURE !== "false"; +const sessionTtlSeconds = Number(process.env.MULTIAGENT_LOGIN_TTL_SECONDS || "43200"); +const captureLines = Math.min(Number(process.env.MULTIAGENT_CAPTURE_LINES || "1200"), 5000); +const s3StateUri = (process.env.MULTIAGENT_STATE_S3_URI || "").replace(/\/$/, ""); +const snapshotIntervalMs = Math.max(Number(process.env.MULTIAGENT_SNAPSHOT_INTERVAL_SECONDS || "60"), 15) * 1000; +const defaultSession = process.env.MULTIAGENT_BOOTSTRAP_SESSION || "orchestrator"; +const defaultRepository = process.env.MULTIAGENT_BOOTSTRAP_REPOSITORY || ""; +const idPattern = /^[a-z0-9][a-z0-9-]{0,62}$/; +const registryFile = path.join(stateRoot, "control-server", "sessions.json"); + +fs.mkdirSync(path.dirname(registryFile), { recursive: true }); +fs.mkdirSync(repositoryRoot, { recursive: true }); + +function loadUsers() { + const parsed = JSON.parse(fs.readFileSync(usersFile, "utf8")); + if (!Array.isArray(parsed.users) || typeof parsed.sessionSecret !== "string" || parsed.sessionSecret.length < 32) { + throw new Error("users file requires users[] and a sessionSecret of at least 32 characters"); + } + return parsed; +} + +let authConfig = loadUsers(); +fs.watchFile(usersFile, { interval: 5000 }, () => { + try { authConfig = loadUsers(); } catch (error) { console.error("users reload failed", error); } +}); + +function json(response, status, value, headers = {}) { + const body = JSON.stringify(value); + response.writeHead(status, { "content-type": "application/json", "content-length": Buffer.byteLength(body), ...headers }); + response.end(body); +} + +function parseCookies(request) { + return Object.fromEntries((request.headers.cookie || "").split(";").map((item) => item.trim()).filter(Boolean).map((item) => { + const split = item.indexOf("="); + return split < 0 ? [item, ""] : [item.slice(0, split), decodeURIComponent(item.slice(split + 1))]; + })); +} + +function base64url(value) { + return Buffer.from(value).toString("base64url"); +} + +function issueSession(username) { + const payload = base64url(JSON.stringify({ username, expiresAt: Date.now() + sessionTtlSeconds * 1000, nonce: crypto.randomBytes(12).toString("hex") })); + const signature = crypto.createHmac("sha256", authConfig.sessionSecret).update(payload).digest("base64url"); + return `${payload}.${signature}`; +} + +function verifySession(token) { + if (!token || !token.includes(".")) return null; + const [payload, signature] = token.split(".", 2); + const expected = crypto.createHmac("sha256", authConfig.sessionSecret).update(payload).digest(); + let supplied; + try { supplied = Buffer.from(signature, "base64url"); } catch { return null; } + if (supplied.length !== expected.length || !crypto.timingSafeEqual(supplied, expected)) return null; + try { + const session = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")); + return session.expiresAt > Date.now() ? session : null; + } catch { return null; } +} + +function currentUser(request) { + return verifySession(parseCookies(request).multiagent_session)?.username || null; +} + +function verifyPassword(password, encoded) { + const [scheme, n, r, p, salt, expected] = String(encoded).split("$"); + if (scheme !== "scrypt") return false; + try { + const actual = crypto.scryptSync(password, Buffer.from(salt, "base64url"), Buffer.from(expected, "base64url").length, { + N: Number(n), r: Number(r), p: Number(p), maxmem: 64 * 1024 * 1024, + }); + return crypto.timingSafeEqual(actual, Buffer.from(expected, "base64url")); + } catch { return false; } +} + +function validOrigin(request) { + const origin = request.headers.origin; + if (!origin) return true; + const forwardedProto = String(request.headers["x-forwarded-proto"] || (request.socket.encrypted ? "https" : "http")).split(",")[0].trim(); + const expected = `${forwardedProto}://${request.headers.host}`; + return origin === expected || origin === process.env.MULTIAGENT_PUBLIC_URL; +} + +async function readBody(request) { + const chunks = []; + let size = 0; + for await (const chunk of request) { + size += chunk.length; + if (size > 1024 * 1024) throw new Error("request body too large"); + chunks.push(chunk); + } + if (!chunks.length) return {}; + return JSON.parse(Buffer.concat(chunks).toString("utf8")); +} + +function run(command, args, options = {}) { + return execFileSync(command, args, { encoding: "utf8", timeout: 30000, ...options }).trim(); +} + +function tmuxAlive(id) { + try { run("tmux", ["has-session", "-t", id]); return true; } catch { return false; } +} + +function loadRegistry() { + try { return JSON.parse(fs.readFileSync(registryFile, "utf8")); } catch { return { sessions: {} }; } +} + +let registry = loadRegistry(); +let registryWrite = Promise.resolve(); + +function saveRegistry() { + const serialized = JSON.stringify(registry, null, 2) + "\n"; + registryWrite = registryWrite.then(async () => { + const temporary = `${registryFile}.${process.pid}.tmp`; + await fs.promises.writeFile(temporary, serialized, { mode: 0o600 }); + await fs.promises.rename(temporary, registryFile); + }); + return registryWrite; +} + +function repositoryPath(name) { + if (!idPattern.test(name)) throw new Error("invalid repository name"); + const candidate = path.resolve(repositoryRoot, name); + if (!candidate.startsWith(`${repositoryRoot}${path.sep}`) || !fs.existsSync(path.join(candidate, ".git"))) { + throw new Error(`repository is not bootstrapped: ${name}`); + } + return candidate; +} + +function sessionStateDir(id) { + return path.join(stateRoot, "sessions", id); +} + +function launchSession(id, repository, resume, actor) { + if (!idPattern.test(id)) throw new Error("invalid session id"); + if (tmuxAlive(id)) throw new Error("session already running"); + const root = repositoryPath(repository); + const persistent = sessionStateDir(id); + fs.mkdirSync(persistent, { recursive: true }); + const args = [path.join(launcherRoot, "launch.sh"), "--session", id, "--root", root, "--no-attach"]; + if (resume) args.push("--resume"); + const env = { + ...process.env, + MULTIAGENT_SESSION: id, + MULTIAGENT_ROOT: root, + MULTIAGENT_STATE_DIR: persistent, + MULTIAGENT_WRITE_POLICY: path.join(persistent, "write-policy.paths"), + MULTIAGENT_PROMPT: path.join(launcherRoot, "orchestrator_prompt.md"), + }; + run("bash", args, { cwd: launcherRoot, env }); + registry.sessions[id] = { + id, repository, status: "running", autoResume: true, createdBy: actor, + createdAt: registry.sessions[id]?.createdAt || new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + saveRegistry(); + return sessionView(id); +} + +function sessionView(id) { + const record = registry.sessions[id]; + if (!record) return null; + return { ...record, live: tmuxAlive(id) }; +} + +function capture(id) { + if (!tmuxAlive(id)) return ""; + return run("tmux", ["capture-pane", "-p", "-J", "-S", `-${captureLines}`, "-t", `${id}:orchestrator`], { maxBuffer: 8 * 1024 * 1024 }); +} + +function sendInput(id, text) { + if (!tmuxAlive(id)) throw new Error("session is not running"); + if (typeof text !== "string" || !text.trim() || text.length > 32768) throw new Error("message must contain 1 to 32768 characters"); + execFileSync("tmux", ["load-buffer", "-"], { input: text, timeout: 5000 }); + run("tmux", ["paste-buffer", "-d", "-t", `${id}:orchestrator`]); + run("tmux", ["send-keys", "-t", `${id}:orchestrator`, "Enter"]); + registry.sessions[id].updatedAt = new Date().toISOString(); + saveRegistry(); +} + +function checkpoint(id) { + const destination = path.join(sessionStateDir(id), "control-server"); + fs.mkdirSync(destination, { recursive: true }); + if (tmuxAlive(id)) { + fs.writeFileSync(path.join(destination, "orchestrator.log"), capture(id), { mode: 0o600 }); + } + fs.writeFileSync(path.join(destination, "checkpoint.json"), JSON.stringify({ capturedAt: new Date().toISOString(), live: tmuxAlive(id) }, null, 2) + "\n", { mode: 0o600 }); +} + +function syncS3() { + if (!s3StateUri) return; + execFile("aws", ["s3", "sync", stateRoot, s3StateUri, "--only-show-errors", "--exclude", "worktrees/*/.git/objects/*"], (error) => { + if (error) console.error("S3 state sync failed", error.message); + }); +} + +function checkpointAll() { + for (const id of Object.keys(registry.sessions)) { + try { checkpoint(id); } catch (error) { console.error(`checkpoint failed for ${id}`, error); } + } + syncS3(); +} + +const loginAttempts = new Map(); +function loginAllowed(address) { + const entry = loginAttempts.get(address); + return !entry || entry.blockedUntil < Date.now(); +} + +function recordLoginFailure(address) { + const previous = loginAttempts.get(address) || { failures: 0, blockedUntil: 0 }; + previous.failures += 1; + if (previous.failures >= 5) previous.blockedUntil = Date.now() + 15 * 60 * 1000; + loginAttempts.set(address, previous); +} + +function staticFile(response, file, type) { + const body = fs.readFileSync(path.join(publicDir, file)); + response.writeHead(200, { "content-type": type, "content-length": body.length, "cache-control": "no-store" }); + response.end(body); +} + +const server = http.createServer(async (request, response) => { + const url = new URL(request.url, `http://${request.headers.host || "localhost"}`); + try { + if (request.method === "GET" && url.pathname === "/healthz") return json(response, 200, { ok: true }); + if (request.method === "GET" && url.pathname === "/readyz") return json(response, 200, { ready: fs.existsSync(usersFile) }); + if (request.method === "GET" && url.pathname === "/") return staticFile(response, "index.html", "text/html; charset=utf-8"); + if (request.method === "GET" && url.pathname === "/app.js") return staticFile(response, "app.js", "text/javascript; charset=utf-8"); + if (request.method === "GET" && url.pathname === "/styles.css") return staticFile(response, "styles.css", "text/css; charset=utf-8"); + + if (!validOrigin(request)) return json(response, 403, { error: "origin rejected" }); + if (request.method === "POST" && url.pathname === "/api/login") { + const address = request.socket.remoteAddress || "unknown"; + if (!loginAllowed(address)) return json(response, 429, { error: "too many login attempts" }); + const body = await readBody(request); + const user = authConfig.users.find((candidate) => candidate.username === body.username && candidate.disabled !== true); + if (!user || !verifyPassword(String(body.password || ""), user.passwordHash)) { + recordLoginFailure(address); + return json(response, 401, { error: "invalid username or password" }); + } + loginAttempts.delete(address); + const cookie = `multiagent_session=${encodeURIComponent(issueSession(user.username))}; HttpOnly; SameSite=Strict; Path=/; Max-Age=${sessionTtlSeconds}${cookieSecure ? "; Secure" : ""}`; + return json(response, 200, { username: user.username }, { "set-cookie": cookie }); + } + + const username = currentUser(request); + if (!username) return json(response, 401, { error: "authentication required" }); + if (request.method === "POST" && url.pathname === "/api/logout") { + return json(response, 200, { ok: true }, { "set-cookie": "multiagent_session=; HttpOnly; SameSite=Strict; Path=/; Max-Age=0" }); + } + if (request.method === "GET" && url.pathname === "/api/me") return json(response, 200, { username }); + if (request.method === "GET" && url.pathname === "/api/repositories") { + const repositories = fs.readdirSync(repositoryRoot, { withFileTypes: true }).filter((entry) => entry.isDirectory() && fs.existsSync(path.join(repositoryRoot, entry.name, ".git"))).map((entry) => entry.name).sort(); + return json(response, 200, { repositories }); + } + if (request.method === "GET" && url.pathname === "/api/sessions") { + return json(response, 200, { sessions: Object.keys(registry.sessions).sort().map(sessionView) }); + } + if (request.method === "POST" && url.pathname === "/api/sessions") { + const body = await readBody(request); + return json(response, 201, launchSession(String(body.id || ""), String(body.repository || ""), Boolean(body.resume), username)); + } + const match = url.pathname.match(/^\/api\/sessions\/([a-z0-9-]+)\/(restart|stop|checkpoint)$/); + if (request.method === "POST" && match) { + const [, id, action] = match; + if (!registry.sessions[id]) return json(response, 404, { error: "unknown session" }); + if (action === "checkpoint") checkpoint(id); + if (action === "stop") { + checkpoint(id); + if (tmuxAlive(id)) run("tmux", ["kill-session", "-t", id]); + registry.sessions[id].status = "stopped"; + registry.sessions[id].autoResume = false; + await saveRegistry(); + } + if (action === "restart") { + checkpoint(id); + if (tmuxAlive(id)) run("tmux", ["kill-session", "-t", id]); + return json(response, 200, launchSession(id, registry.sessions[id].repository, true, username)); + } + return json(response, 200, sessionView(id)); + } + return json(response, 404, { error: "not found" }); + } catch (error) { + console.error(error); + return json(response, 400, { error: error.message || "request failed" }); + } +}); + +const sockets = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024 }); +server.on("upgrade", (request, socket, head) => { + const url = new URL(request.url, `http://${request.headers.host || "localhost"}`); + const match = url.pathname.match(/^\/api\/sessions\/([a-z0-9-]+)\/terminal$/); + if (!match || !validOrigin(request) || !currentUser(request) || !registry.sessions[match[1]]) { + socket.write("HTTP/1.1 401 Unauthorized\r\nConnection: close\r\n\r\n"); + return socket.destroy(); + } + request.sessionId = match[1]; + sockets.handleUpgrade(request, socket, head, (websocket) => sockets.emit("connection", websocket, request)); +}); + +sockets.on("connection", (socket, request) => { + const id = request.sessionId; + let previous = ""; + const publish = () => { + try { + const output = capture(id); + if (output !== previous && socket.readyState === socket.OPEN) { + previous = output; + socket.send(JSON.stringify({ type: "output", output, live: tmuxAlive(id), capturedAt: new Date().toISOString() })); + } + } catch (error) { + if (socket.readyState === socket.OPEN) socket.send(JSON.stringify({ type: "error", error: error.message })); + } + }; + publish(); + const interval = setInterval(publish, 750); + socket.on("message", (message) => { + try { + const payload = JSON.parse(message.toString()); + if (payload.type !== "input") throw new Error("unsupported WebSocket message"); + sendInput(id, payload.text); + publish(); + } catch (error) { socket.send(JSON.stringify({ type: "error", error: error.message })); } + }); + socket.on("close", () => clearInterval(interval)); +}); + +for (const record of Object.values(registry.sessions)) { + if (record.autoResume && !tmuxAlive(record.id)) { + try { launchSession(record.id, record.repository, true, "system"); } catch (error) { console.error(`restore failed for ${record.id}`, error); } + } +} +if (defaultRepository && !registry.sessions[defaultSession]) { + try { launchSession(defaultSession, defaultRepository, false, "system"); } catch (error) { console.error("bootstrap session failed", error); } +} + +const snapshotTimer = setInterval(checkpointAll, snapshotIntervalMs); +server.listen(port, host, () => console.log(`multiagent control server listening on ${host}:${port}`)); + +let shuttingDown = false; +function shutdown(signal) { + if (shuttingDown) return; + shuttingDown = true; + console.log(`received ${signal}; checkpointing sessions`); + clearInterval(snapshotTimer); + checkpointAll(); + setTimeout(() => server.close(() => process.exit(0)), 1000).unref(); + setTimeout(() => process.exit(1), 25000).unref(); +} +process.on("SIGTERM", () => shutdown("SIGTERM")); +process.on("SIGINT", () => shutdown("SIGINT")); From 1e28fcf0c6660cc0f3bf6251234456bbba607962 Mon Sep 17 00:00:00 2001 From: Bo Wu Date: Fri, 21 Aug 2026 16:29:33 -0700 Subject: [PATCH 2/3] Model orchestrators as resumable task sessions --- Dockerfile | 13 ++++-- README.md | 17 ++------ bin/container-entrypoint.sh | 3 -- bin/git-askpass.sh | 6 --- bin/sync-repositories.mjs | 25 ----------- control-server/public/app.js | 14 ++++-- control-server/public/index.html | 10 ++--- control-server/src/server.mjs | 75 +++++++++++++++++++++++--------- 8 files changed, 83 insertions(+), 80 deletions(-) delete mode 100644 bin/git-askpass.sh delete mode 100644 bin/sync-repositories.mjs diff --git a/Dockerfile b/Dockerfile index acd015f..f34e528 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,3 +1,9 @@ +FROM rust:1-bookworm AS multiagent-builder + +WORKDIR /src +COPY . . +RUN cargo build --release --locked + FROM node:22-bookworm-slim ARG CODEX_VERSION=0.145.0 @@ -10,10 +16,9 @@ RUN apt-get update \ WORKDIR /opt/multiagent COPY control-server/package*.json control-server/ -RUN cd control-server && npm install --omit=dev -COPY launch.sh orchestrator_prompt.md README.md ./ -COPY bin/ bin/ -COPY control-server/ control-server/ +RUN cd control-server && npm ci --omit=dev +COPY . . +COPY --from=multiagent-builder /src/target/release/multiagent /opt/multiagent/bin/multiagent RUN chmod +x launch.sh bin/*.sh bin/*.mjs \ && useradd --create-home --home-dir /var/lib/multiagent --uid 10001 multiagent \ && mkdir -p /var/lib/multiagent/state /var/lib/multiagent/repositories \ diff --git a/README.md b/README.md index 0886358..0455f0c 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ accepts work only when reviewer evidence matches the exact final Git diff. ## Stateful control server -The container image runs a same-origin web UI and authenticated WebSocket gateway as PID 1. The server bootstraps configured repositories, starts or resumes tmux orchestrator sessions, streams orchestrator pane output, accepts user messages, checkpoints session output, and mirrors durable state to S3. +The container image runs a same-origin web UI and authenticated WebSocket gateway as PID 1. Each task owns an isolated tmux orchestrator session. Paused, completed, and archived tasks retain their workflow state and terminal transcript without retaining an active tmux process; resuming reconstructs the session from that state. Users are configured in a mounted JSON file. Passwords must be scrypt hashes, never plaintext: @@ -26,22 +26,13 @@ The mounted file has this shape: } ``` -Repositories are allowlisted in `MULTIAGENT_REPOSITORIES_FILE`: - -```json -{ - "repositories": [ - {"name": "example", "url": "https://github.com/example/repository.git", "ref": "main"} - ] -} -``` +Task repositories must be provisioned as Git worktrees below `MULTIAGENT_REPOSITORY_ROOT`. Repository provisioning is owned by the deployment rather than the control server, so restarting the UI never fetches or mutates source checkouts. Important container variables: - `MULTIAGENT_USERS_FILE`: mounted login configuration, default `/run/secrets/multiagent/users.json`. -- `MULTIAGENT_REPOSITORIES_FILE`: mounted repository allowlist. -- `MULTIAGENT_BOOTSTRAP_REPOSITORY`: repository used for the initial orchestrator session. -- `MULTIAGENT_BOOTSTRAP_SESSION`: initial session name, default `orchestrator`. +- `MULTIAGENT_REPOSITORY_ROOT`: deployment-provisioned Git worktrees available for new tasks. +- `MULTIAGENT_IDLE_TIMEOUT_SECONDS`: inactivity period after which a running task is checkpointed and paused. - `MULTIAGENT_STATE_S3_URI`: S3 prefix used for recovery snapshots. - `MULTIAGENT_PUBLIC_URL`: canonical HTTPS origin accepted for browser and WebSocket requests. diff --git a/bin/container-entrypoint.sh b/bin/container-entrypoint.sh index 99f721a..0de2b30 100644 --- a/bin/container-entrypoint.sh +++ b/bin/container-entrypoint.sh @@ -5,11 +5,8 @@ export CODEX_HOME="${CODEX_HOME:-/var/lib/multiagent/codex}" export CLAUDE_CONFIG_DIR="${CLAUDE_CONFIG_DIR:-/var/lib/multiagent/claude}" export MULTIAGENT_STATE_DIR="${MULTIAGENT_STATE_DIR:-/var/lib/multiagent/state}" export MULTIAGENT_REPOSITORY_ROOT="${MULTIAGENT_REPOSITORY_ROOT:-/var/lib/multiagent/repositories}" -export GIT_ASKPASS="${GIT_ASKPASS:-/opt/multiagent/bin/git-askpass.sh}" -export GIT_TERMINAL_PROMPT=0 mkdir -p "$HOME" "$CODEX_HOME" "$CLAUDE_CONFIG_DIR" "$MULTIAGENT_STATE_DIR" "$MULTIAGENT_REPOSITORY_ROOT" if [[ -n "${MULTIAGENT_STATE_S3_URI:-}" && ! -f "$MULTIAGENT_STATE_DIR/control-server/sessions.json" ]]; then aws s3 sync "$MULTIAGENT_STATE_S3_URI" "$MULTIAGENT_STATE_DIR" --only-show-errors || true fi -node /opt/multiagent/bin/sync-repositories.mjs exec node /opt/multiagent/control-server/src/server.mjs diff --git a/bin/git-askpass.sh b/bin/git-askpass.sh deleted file mode 100644 index e873367..0000000 --- a/bin/git-askpass.sh +++ /dev/null @@ -1,6 +0,0 @@ -#!/usr/bin/env bash -case "$1" in - *sername*) printf '%s\n' "x-access-token" ;; - *assword*) printf '%s\n' "${GITHUB_TOKEN:?GITHUB_TOKEN is required}" ;; - *) exit 1 ;; -esac diff --git a/bin/sync-repositories.mjs b/bin/sync-repositories.mjs deleted file mode 100644 index 1cce366..0000000 --- a/bin/sync-repositories.mjs +++ /dev/null @@ -1,25 +0,0 @@ -#!/usr/bin/env node -import fs from "node:fs"; -import path from "node:path"; -import { execFileSync } from "node:child_process"; - -const configFile = process.env.MULTIAGENT_REPOSITORIES_FILE || "/etc/multiagent/repositories.json"; -const root = path.resolve(process.env.MULTIAGENT_REPOSITORY_ROOT || "/var/lib/multiagent/repositories"); -if (!fs.existsSync(configFile)) process.exit(0); -const config = JSON.parse(fs.readFileSync(configFile, "utf8")); -fs.mkdirSync(root, { recursive: true }); -for (const repository of config.repositories || []) { - if (!/^[a-z0-9][a-z0-9-]{0,62}$/.test(repository.name)) throw new Error(`invalid repository name: ${repository.name}`); - if (!/^https:\/\/github\.com\/[A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+(?:\.git)?$/.test(repository.url)) throw new Error(`unsupported repository URL: ${repository.url}`); - const destination = path.join(root, repository.name); - if (!fs.existsSync(path.join(destination, ".git"))) { - execFileSync("git", ["clone", "--origin", "origin", repository.url, destination], { stdio: "inherit" }); - } else { - execFileSync("git", ["-C", destination, "fetch", "origin", "--prune"], { stdio: "inherit" }); - } - const status = execFileSync("git", ["-C", destination, "status", "--porcelain"], { encoding: "utf8" }); - if (!status.trim() && repository.ref) { - execFileSync("git", ["-C", destination, "checkout", repository.ref], { stdio: "inherit" }); - execFileSync("git", ["-C", destination, "pull", "--ff-only", "origin", repository.ref], { stdio: "inherit" }); - } -} diff --git a/control-server/public/app.js b/control-server/public/app.js index 0d60576..6fc5cd1 100644 --- a/control-server/public/app.js +++ b/control-server/public/app.js @@ -33,7 +33,15 @@ function selectSession(id) { $("#active-name").textContent = session?.id || "Select a session"; $("#active-repo").textContent = session?.repository || ""; $("#live-dot").classList.toggle("live", Boolean(session?.live)); - for (const id of ["restart", "checkpoint", "stop", "message", "send"]) $("#" + id).disabled = !session; + const running = Boolean(session?.live); + $("#resume").disabled = !session || running; + $("#restart").disabled = !running; + $("#checkpoint").disabled = !running; + $("#pause").disabled = !running; + $("#complete").disabled = !running; + $("#archive").disabled = !session || running || session.status === "archived"; + $("#message").disabled = !running; + $("#send").disabled = !running; if (!session) { $("#terminal").textContent = "No orchestrator selected."; return refreshSessions(); } $("#terminal").textContent = "Connecting to orchestrator…"; const protocol = location.protocol === "https:" ? "wss" : "ws"; @@ -67,11 +75,11 @@ $("#message-form").onsubmit = (event) => { socket.send(JSON.stringify({ type: "input", text })); $("#message").value = ""; }; -for (const action of ["restart", "checkpoint", "stop"]) $("#" + action).onclick = async () => { +for (const action of ["restart", "resume", "checkpoint", "pause", "complete", "archive"]) $("#" + action).onclick = async () => { if (!active) return; await api(`/api/sessions/${active}/${action}`, { method: "POST" }); await refreshSessions(); - if (action === "restart") selectSession(active); + if (action === "restart" || action === "resume") selectSession(active); }; $("#new-session").onclick = async () => { const repositories = (await api("/api/repositories")).repositories; diff --git a/control-server/public/index.html b/control-server/public/index.html index 13d42b3..d048e89 100644 --- a/control-server/public/index.html +++ b/control-server/public/index.html @@ -21,18 +21,18 @@

Operator
Console

-

Multiagent

Session Control

+

Multiagent

Task Control

Select a session
-
+ 
No orchestrator selected.
@@ -45,8 +45,8 @@

Operator
Console

-

Bootstrap

New session

- +

Bootstrap

New task

+

diff --git a/control-server/src/server.mjs b/control-server/src/server.mjs index 369563d..2ee531b 100644 --- a/control-server/src/server.mjs +++ b/control-server/src/server.mjs @@ -2,9 +2,9 @@ import crypto from "node:crypto"; import fs from "node:fs"; import http from "node:http"; import path from "node:path"; -import { execFile, execFileSync, spawn } from "node:child_process"; +import { execFile, execFileSync } from "node:child_process"; import { fileURLToPath } from "node:url"; -import { WebSocketServer } from "ws"; +import { WebSocket, WebSocketServer } from "ws"; const here = path.dirname(fileURLToPath(import.meta.url)); const publicDir = path.resolve(here, "../public"); @@ -19,8 +19,7 @@ const sessionTtlSeconds = Number(process.env.MULTIAGENT_LOGIN_TTL_SECONDS || "43 const captureLines = Math.min(Number(process.env.MULTIAGENT_CAPTURE_LINES || "1200"), 5000); const s3StateUri = (process.env.MULTIAGENT_STATE_S3_URI || "").replace(/\/$/, ""); const snapshotIntervalMs = Math.max(Number(process.env.MULTIAGENT_SNAPSHOT_INTERVAL_SECONDS || "60"), 15) * 1000; -const defaultSession = process.env.MULTIAGENT_BOOTSTRAP_SESSION || "orchestrator"; -const defaultRepository = process.env.MULTIAGENT_BOOTSTRAP_REPOSITORY || ""; +const idleTimeoutMs = Math.max(Number(process.env.MULTIAGENT_IDLE_TIMEOUT_SECONDS || "86400"), 300) * 1000; const idPattern = /^[a-z0-9][a-z0-9-]{0,62}$/; const registryFile = path.join(stateRoot, "control-server", "sessions.json"); @@ -152,6 +151,9 @@ function sessionStateDir(id) { function launchSession(id, repository, resume, actor) { if (!idPattern.test(id)) throw new Error("invalid session id"); if (tmuxAlive(id)) throw new Error("session already running"); + const existing = registry.sessions[id]; + if (resume && !existing) throw new Error("cannot resume an unknown task"); + if (!resume && existing) throw new Error("task id already exists"); const root = repositoryPath(repository); const persistent = sessionStateDir(id); fs.mkdirSync(persistent, { recursive: true }); @@ -166,10 +168,12 @@ function launchSession(id, repository, resume, actor) { MULTIAGENT_PROMPT: path.join(launcherRoot, "orchestrator_prompt.md"), }; run("bash", args, { cwd: launcherRoot, env }); + const now = new Date().toISOString(); registry.sessions[id] = { - id, repository, status: "running", autoResume: true, createdBy: actor, - createdAt: registry.sessions[id]?.createdAt || new Date().toISOString(), - updatedAt: new Date().toISOString(), + ...existing, id, repository, status: "running", autoResume: true, + createdBy: existing?.createdBy || actor, createdAt: existing?.createdAt || now, + resumedBy: resume ? actor : undefined, resumedAt: resume ? now : undefined, + updatedAt: now, lastActivityAt: now, }; saveRegistry(); return sessionView(id); @@ -182,7 +186,9 @@ function sessionView(id) { } function capture(id) { - if (!tmuxAlive(id)) return ""; + if (!tmuxAlive(id)) { + try { return fs.readFileSync(path.join(sessionStateDir(id), "control-server", "orchestrator.log"), "utf8"); } catch { return ""; } + } return run("tmux", ["capture-pane", "-p", "-J", "-S", `-${captureLines}`, "-t", `${id}:orchestrator`], { maxBuffer: 8 * 1024 * 1024 }); } @@ -193,6 +199,7 @@ function sendInput(id, text) { run("tmux", ["paste-buffer", "-d", "-t", `${id}:orchestrator`]); run("tmux", ["send-keys", "-t", `${id}:orchestrator`, "Enter"]); registry.sessions[id].updatedAt = new Date().toISOString(); + registry.sessions[id].lastActivityAt = registry.sessions[id].updatedAt; saveRegistry(); } @@ -219,6 +226,22 @@ function checkpointAll() { syncS3(); } +async function retireSession(id, status, actor) { + const record = registry.sessions[id]; + if (!record) throw new Error("unknown task"); + checkpoint(id); + if (tmuxAlive(id)) run("tmux", ["kill-session", "-t", id]); + const now = new Date().toISOString(); + record.status = status; + record.autoResume = false; + record.updatedAt = now; + record[`${status}At`] = now; + record[`${status}By`] = actor; + await saveRegistry(); + syncS3(); + return sessionView(id); +} + const loginAttempts = new Map(); function loginAllowed(address) { const entry = loginAttempts.get(address); @@ -279,17 +302,20 @@ const server = http.createServer(async (request, response) => { const body = await readBody(request); return json(response, 201, launchSession(String(body.id || ""), String(body.repository || ""), Boolean(body.resume), username)); } - const match = url.pathname.match(/^\/api\/sessions\/([a-z0-9-]+)\/(restart|stop|checkpoint)$/); + const match = url.pathname.match(/^\/api\/sessions\/([a-z0-9-]+)\/(restart|resume|pause|complete|archive|checkpoint)$/); if (request.method === "POST" && match) { const [, id, action] = match; if (!registry.sessions[id]) return json(response, 404, { error: "unknown session" }); if (action === "checkpoint") checkpoint(id); - if (action === "stop") { - checkpoint(id); - if (tmuxAlive(id)) run("tmux", ["kill-session", "-t", id]); - registry.sessions[id].status = "stopped"; - registry.sessions[id].autoResume = false; - await saveRegistry(); + if (action === "pause") return json(response, 200, await retireSession(id, "paused", username)); + if (action === "complete") return json(response, 200, await retireSession(id, "completed", username)); + if (action === "archive") { + if (tmuxAlive(id)) throw new Error("pause or complete the task before archiving"); + return json(response, 200, await retireSession(id, "archived", username)); + } + if (action === "resume") { + if (tmuxAlive(id)) throw new Error("task is already running"); + return json(response, 200, launchSession(id, registry.sessions[id].repository, true, username)); } if (action === "restart") { checkpoint(id); @@ -323,12 +349,12 @@ sockets.on("connection", (socket, request) => { const publish = () => { try { const output = capture(id); - if (output !== previous && socket.readyState === socket.OPEN) { + if (output !== previous && socket.readyState === WebSocket.OPEN) { previous = output; socket.send(JSON.stringify({ type: "output", output, live: tmuxAlive(id), capturedAt: new Date().toISOString() })); } } catch (error) { - if (socket.readyState === socket.OPEN) socket.send(JSON.stringify({ type: "error", error: error.message })); + if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ type: "error", error: error.message })); } }; publish(); @@ -345,15 +371,21 @@ sockets.on("connection", (socket, request) => { }); for (const record of Object.values(registry.sessions)) { - if (record.autoResume && !tmuxAlive(record.id)) { + if (record.status === "running" && record.autoResume && !tmuxAlive(record.id)) { try { launchSession(record.id, record.repository, true, "system"); } catch (error) { console.error(`restore failed for ${record.id}`, error); } } } -if (defaultRepository && !registry.sessions[defaultSession]) { - try { launchSession(defaultSession, defaultRepository, false, "system"); } catch (error) { console.error("bootstrap session failed", error); } -} const snapshotTimer = setInterval(checkpointAll, snapshotIntervalMs); +const retirementTimer = setInterval(() => { + const now = Date.now(); + for (const record of Object.values(registry.sessions)) { + const lastActivity = Date.parse(record.lastActivityAt || record.updatedAt || record.createdAt); + if (record.status === "running" && tmuxAlive(record.id) && Number.isFinite(lastActivity) && now - lastActivity >= idleTimeoutMs) { + retireSession(record.id, "paused", "idle-timeout").catch((error) => console.error(`idle retirement failed for ${record.id}`, error)); + } + } +}, 60000); server.listen(port, host, () => console.log(`multiagent control server listening on ${host}:${port}`)); let shuttingDown = false; @@ -362,6 +394,7 @@ function shutdown(signal) { shuttingDown = true; console.log(`received ${signal}; checkpointing sessions`); clearInterval(snapshotTimer); + clearInterval(retirementTimer); checkpointAll(); setTimeout(() => server.close(() => process.exit(0)), 1000).unref(); setTimeout(() => process.exit(1), 25000).unref(); From 88db3c1f9835634b4a5d8688b4d9a51ad78f037c Mon Sep 17 00:00:00 2001 From: Bo Wu Date: Fri, 21 Aug 2026 16:34:51 -0700 Subject: [PATCH 3/3] Persist concise task reports with agent traces --- README.md | 2 +- control-server/public/app.js | 4 ++ control-server/public/index.html | 1 + control-server/src/server.mjs | 120 +++++++++++++++++++++++++++++-- 4 files changed, 121 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 0455f0c..4e60da3 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ Important container variables: - `MULTIAGENT_STATE_S3_URI`: S3 prefix used for recovery snapshots. - `MULTIAGENT_PUBLIC_URL`: canonical HTTPS origin accepted for browser and WebSocket requests. -The PVC mounted at `/var/lib/multiagent` is the primary store for repositories, CLI conversation history, checkpoints, and session metadata. S3 is the durable recovery and inspection copy. +The PVC mounted at `/var/lib/multiagent` is the primary store for repositories, CLI conversation history, checkpoints, and session metadata. Final reports, a bounded terminal tail, and the transcript index live under each task's existing `logs` trace root. They reference immutable agent event traces instead of duplicating full transcripts. S3 is the durable recovery and inspection copy. ## Requirements diff --git a/control-server/public/app.js b/control-server/public/app.js index 6fc5cd1..80ef489 100644 --- a/control-server/public/app.js +++ b/control-server/public/app.js @@ -42,7 +42,11 @@ function selectSession(id) { $("#archive").disabled = !session || running || session.status === "archived"; $("#message").disabled = !running; $("#send").disabled = !running; + $("#report").textContent = ""; if (!session) { $("#terminal").textContent = "No orchestrator selected."; return refreshSessions(); } + api(`/api/sessions/${id}/report`).then(({ report, transcript }) => { + $("#report").textContent = report || (transcript ? `Trace references\n${transcript.traceReferences.join("\n")}` : ""); + }).catch(() => {}); $("#terminal").textContent = "Connecting to orchestrator…"; const protocol = location.protocol === "https:" ? "wss" : "ws"; socket = new WebSocket(`${protocol}://${location.host}/api/sessions/${id}/terminal`); diff --git a/control-server/public/index.html b/control-server/public/index.html index d048e89..fdf2d70 100644 --- a/control-server/public/index.html +++ b/control-server/public/index.html @@ -34,6 +34,7 @@

Operator
Console

Select a session
+ 

         
No orchestrator selected.

         
           
diff --git a/control-server/src/server.mjs b/control-server/src/server.mjs
index 2ee531b..d95f033 100644
--- a/control-server/src/server.mjs
+++ b/control-server/src/server.mjs
@@ -148,6 +148,76 @@ function sessionStateDir(id) {
   return path.join(stateRoot, "sessions", id);
 }
 
+function traceRoot(id) {
+  return path.join(sessionStateDir(id), "logs");
+}
+
+function conciseTail(value, lines = 80, characters = 12000) {
+  return String(value || "").split("\n").slice(-lines).join("\n").slice(-characters);
+}
+
+function activeWorkflow(id) {
+  try { return fs.readFileSync(path.join(sessionStateDir(id), "runtime_state", "active-workflow-id"), "utf8").trim(); } catch { return ""; }
+}
+
+function workflowPhase(id) {
+  const workflow = activeWorkflow(id);
+  if (!workflow) return "";
+  try {
+    const lifecycle = fs.readFileSync(path.join(sessionStateDir(id), "workflows", workflow, "lifecycle", "lifecycle.env"), "utf8");
+    return lifecycle.split("\n").find((line) => line.startsWith("phase="))?.slice(6).trim() || "";
+  } catch { return ""; }
+}
+
+function traceReferences(id) {
+  const root = traceRoot(id);
+  const references = [];
+  const workflow = activeWorkflow(id);
+  if (workflow) references.push(`../workflows/${workflow}/lifecycle/events.log`);
+  const agents = path.join(root, "agents");
+  try {
+    for (const entry of fs.readdirSync(agents, { withFileTypes: true }).filter((item) => item.isDirectory())) {
+      const base = path.join(agents, entry.name);
+      let attempt = "";
+      try { attempt = fs.readFileSync(path.join(base, "latest"), "utf8").trim(); } catch {}
+      const events = path.join("agents", entry.name, attempt, "events.jsonl");
+      if (attempt && fs.existsSync(path.join(root, events))) references.push(events);
+    }
+  } catch {}
+  return references;
+}
+
+function writeTraceSummary(id, status) {
+  const root = traceRoot(id);
+  fs.mkdirSync(root, { recursive: true });
+  let finalMessage = "";
+  try { finalMessage = conciseTail(fs.readFileSync(path.join(sessionStateDir(id), "orchestrator-last-message.txt"), "utf8"), 40, 6000); } catch {}
+  const references = traceReferences(id);
+  const report = {
+    taskId: id,
+    workflowId: activeWorkflow(id) || null,
+    status,
+    completedAt: registry.sessions[id]?.completedAt || null,
+    finalMessage,
+    traceReferences: references,
+  };
+  const markdown = [
+    `# ${id}`,
+    "",
+    `Status: ${status}`,
+    report.workflowId ? `Workflow: ${report.workflowId}` : null,
+    "",
+    finalMessage ? "## Final agent message" : null,
+    finalMessage || null,
+    "",
+    "## Trace references",
+    ...references.map((reference) => `- ${reference}`),
+    "",
+  ].filter((line) => line !== null).join("\n");
+  fs.writeFileSync(path.join(root, "final-report.json"), JSON.stringify(report, null, 2) + "\n", { mode: 0o600 });
+  fs.writeFileSync(path.join(root, "final-report.md"), markdown, { mode: 0o600 });
+}
+
 function launchSession(id, repository, resume, actor) {
   if (!idPattern.test(id)) throw new Error("invalid session id");
   if (tmuxAlive(id)) throw new Error("session already running");
@@ -187,7 +257,7 @@ function sessionView(id) {
 
 function capture(id) {
   if (!tmuxAlive(id)) {
-    try { return fs.readFileSync(path.join(sessionStateDir(id), "control-server", "orchestrator.log"), "utf8"); } catch { return ""; }
+    try { return fs.readFileSync(path.join(traceRoot(id), "terminal-tail.log"), "utf8"); } catch { return ""; }
   }
   return run("tmux", ["capture-pane", "-p", "-J", "-S", `-${captureLines}`, "-t", `${id}:orchestrator`], { maxBuffer: 8 * 1024 * 1024 });
 }
@@ -205,11 +275,23 @@ function sendInput(id, text) {
 
 function checkpoint(id) {
   const destination = path.join(sessionStateDir(id), "control-server");
+  const traces = traceRoot(id);
   fs.mkdirSync(destination, { recursive: true });
+  fs.mkdirSync(traces, { recursive: true });
+  let terminalTail = "";
   if (tmuxAlive(id)) {
-    fs.writeFileSync(path.join(destination, "orchestrator.log"), capture(id), { mode: 0o600 });
+    terminalTail = conciseTail(capture(id));
+    fs.writeFileSync(path.join(traces, "terminal-tail.log"), terminalTail, { mode: 0o600 });
+    const digest = crypto.createHash("sha256").update(terminalTail).digest("hex");
+    if (registry.sessions[id]?.lastOutputSha256 !== digest) {
+      registry.sessions[id].lastOutputSha256 = digest;
+      registry.sessions[id].lastActivityAt = new Date().toISOString();
+      saveRegistry();
+    }
   }
-  fs.writeFileSync(path.join(destination, "checkpoint.json"), JSON.stringify({ capturedAt: new Date().toISOString(), live: tmuxAlive(id) }, null, 2) + "\n", { mode: 0o600 });
+  const references = traceReferences(id);
+  fs.writeFileSync(path.join(traces, "transcript-index.json"), JSON.stringify({ taskId: id, capturedAt: new Date().toISOString(), terminalTail: "terminal-tail.log", traceReferences: references }, null, 2) + "\n", { mode: 0o600 });
+  fs.writeFileSync(path.join(destination, "checkpoint.json"), JSON.stringify({ capturedAt: new Date().toISOString(), live: tmuxAlive(id), transcriptIndex: "../logs/transcript-index.json" }, null, 2) + "\n", { mode: 0o600 });
 }
 
 function syncS3() {
@@ -238,6 +320,7 @@ async function retireSession(id, status, actor) {
   record[`${status}At`] = now;
   record[`${status}By`] = actor;
   await saveRegistry();
+  writeTraceSummary(id, status);
   syncS3();
   return sessionView(id);
 }
@@ -298,6 +381,16 @@ const server = http.createServer(async (request, response) => {
     if (request.method === "GET" && url.pathname === "/api/sessions") {
       return json(response, 200, { sessions: Object.keys(registry.sessions).sort().map(sessionView) });
     }
+    const reportMatch = url.pathname.match(/^\/api\/sessions\/([a-z0-9-]+)\/report$/);
+    if (request.method === "GET" && reportMatch) {
+      const id = reportMatch[1];
+      if (!registry.sessions[id]) return json(response, 404, { error: "unknown session" });
+      try {
+        const report = fs.readFileSync(path.join(traceRoot(id), "final-report.md"), "utf8");
+        const transcript = JSON.parse(fs.readFileSync(path.join(traceRoot(id), "transcript-index.json"), "utf8"));
+        return json(response, 200, { report, transcript });
+      } catch { return json(response, 200, { report: "", transcript: null }); }
+    }
     if (request.method === "POST" && url.pathname === "/api/sessions") {
       const body = await readBody(request);
       return json(response, 201, launchSession(String(body.id || ""), String(body.repository || ""), Boolean(body.resume), username));
@@ -371,7 +464,16 @@ sockets.on("connection", (socket, request) => {
 });
 
 for (const record of Object.values(registry.sessions)) {
-  if (record.status === "running" && record.autoResume && !tmuxAlive(record.id)) {
+  if (record.status === "running" && workflowPhase(record.id) === "complete") {
+    const now = new Date().toISOString();
+    record.status = "completed";
+    record.autoResume = false;
+    record.completedAt = now;
+    record.completedBy = "workflow-supervisor";
+    record.updatedAt = now;
+    writeTraceSummary(record.id, "completed");
+    saveRegistry();
+  } else if (record.status === "running" && record.autoResume && !tmuxAlive(record.id)) {
     try { launchSession(record.id, record.repository, true, "system"); } catch (error) { console.error(`restore failed for ${record.id}`, error); }
   }
 }
@@ -380,12 +482,20 @@ const snapshotTimer = setInterval(checkpointAll, snapshotIntervalMs);
 const retirementTimer = setInterval(() => {
   const now = Date.now();
   for (const record of Object.values(registry.sessions)) {
+    if (record.status === "running" && workflowPhase(record.id) === "complete") {
+      retireSession(record.id, "completed", "workflow-supervisor").catch((error) => console.error(`completion retirement failed for ${record.id}`, error));
+      continue;
+    }
+    if (record.status === "running" && !tmuxAlive(record.id)) {
+      retireSession(record.id, "failed", "process-exit").catch((error) => console.error(`failed retirement failed for ${record.id}`, error));
+      continue;
+    }
     const lastActivity = Date.parse(record.lastActivityAt || record.updatedAt || record.createdAt);
     if (record.status === "running" && tmuxAlive(record.id) && Number.isFinite(lastActivity) && now - lastActivity >= idleTimeoutMs) {
       retireSession(record.id, "paused", "idle-timeout").catch((error) => console.error(`idle retirement failed for ${record.id}`, error));
     }
   }
-}, 60000);
+}, 5000);
 server.listen(port, host, () => console.log(`multiagent control server listening on ${host}:${port}`));
 
 let shuttingDown = false;