From 9deae55275da076749b8dd572fe0495556f8305f Mon Sep 17 00:00:00 2001 From: Vikas Singhal Date: Mon, 13 Jul 2026 15:13:20 +0530 Subject: [PATCH] feat(scheduler): concurrency cap on by default + fail-open fix (v0.143.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of docs/concurrency-cap-plan.md. Every live session holds a claude process (hundreds of MB); an unbounded burst of scheduled work can OOM the box. - The scheduler cap now defaults to a RAM-derived value (max(3, floor(GB/1.5))) instead of 0/unlimited. Resolved live as env → operator Settings → derived by the new single-source-of-truth Automations.concurrencyCap() (the static `maxConcurrent` field is gone), so a Settings change lands next tick, no restart. - Fix the fail-open: aliveSessionCount() fell back to 0 when tmux liveness was unpollable (always on the Linux LauncherSessionBackend), silently disabling the cap under the exact load it's for. It now uses a DB count of running rows (new runningSessionCount()); the crash sweep keeps that set honest. - Settings.maxConcurrentSessions get/set (null=unset→derived, 0=unlimited, N=cap) + GET/PUT /api/settings/concurrency (owner/admin, audited) + a Settings → Runtime defaults → "Concurrency cap" panel (live count + effective cap + source; env-pinned installs are read-only). - scripts/concurrency-cap-test.cjs: 20 assertions (derived math, Settings semantics, resolver order, DB fallback). typecheck + web build + 68/68 governance all pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 20 ++++ docs/concurrency-cap-plan.md | 162 +++++++++++++++++++++++++++++++ package-lock.json | 4 +- package.json | 2 +- scripts/concurrency-cap-test.cjs | 78 +++++++++++++++ src/edge/automations.ts | 41 ++++++-- src/governance/settings.ts | 28 ++++++ src/server.ts | 28 +++++- src/terminal.ts | 20 +++- web/src/App.tsx | 76 ++++++++++++++- web/src/lib/api.ts | 15 +++ 11 files changed, 454 insertions(+), 20 deletions(-) create mode 100644 docs/concurrency-cap-plan.md create mode 100644 scripts/concurrency-cap-test.cjs diff --git a/CHANGELOG.md b/CHANGELOG.md index 4323021..67eed25 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,26 @@ new version heading in the same commit. ## [Unreleased] +## [0.143.0] — 2026-07-13 +### Added +- **Whole-box concurrency cap is now ON by default (Phase 1 of `docs/concurrency-cap-plan.md`).** Every live + session holds a `claude` process (hundreds of MB), so an unbounded burst of scheduled work can OOM the box. + The scheduler cap — previously opt-in via `AOS_MAX_CONCURRENT_SESSIONS` and **0 (unlimited) by default** — + now defaults to a **RAM-derived** value: `max(3, floor(totalGB / 1.5))` (a 2 GB droplet → 3; a 32 GB Mac + Mini → ~21). Resolved live as **env override → operator Settings value → derived default** by the new + `Automations.concurrencyCap()` (single source of truth; the old static `maxConcurrent` field is gone), so a + change takes effect on the next scheduler tick with no restart. +- **Settings → Runtime defaults → "Concurrency cap".** A new owner/admin panel shows the live running-session + count and the effective cap + where it came from (env / operator / box default), and lets you set it: blank + = the RAM-derived default, `0` = unlimited, `N` = cap at N. Env-pinned installs show the value read-only. + Backed by `GET/PUT /api/settings/concurrency` (audited `settings.concurrency.updated`). +### Fixed +- **The concurrency cap no longer silently disables itself when tmux liveness can't be polled.** + `aliveSessionCount()` used to fail-open to `0` when `aliveNames()` returned null (always on the Linux + `LauncherSessionBackend`; transient local hiccups) — turning the cap off under exactly the load it exists + for. It now falls back to a pure DB count of `running` rows (new `runningSessionCount()`); the crash sweep + keeps that set honest, so it's a safe proxy. + ## [0.142.0] — 2026-07-13 ### Added - **Settings → System → Host resources now breaks down RAM by agent session.** Alongside the host diff --git a/docs/concurrency-cap-plan.md b/docs/concurrency-cap-plan.md new file mode 100644 index 0000000..7a97a6b --- /dev/null +++ b/docs/concurrency-cap-plan.md @@ -0,0 +1,162 @@ +# Fleet concurrency cap — plan (too many live sessions) + +**Problem.** Every live session holds a tmux pane + a `claude` process (several hundred MB each). Enough +concurrent ones and the box swaps / OOMs. Automations, tasks, chat mentions, thread follow-ups, and +console clicks can all spawn — but only the *scheduler* honours a cap. The event-driven chat paths pour +straight in ungated. + +**Decision (chosen):** *Chat admission gate + turn the cap on; humans still pass; queue-and-ack for the +rest.* Minimal scope — no per-agent/tenant fairness or adaptive reaping in this pass (noted as futures). + +--- + +## What already exists (reuse it) + +- **Cap + defer-and-retry** — `AOS_MAX_CONCURRENT_SESSIONS` (`automations.ts:267`), enforced only in + `tick()` (`automations.ts:720-763`): over cap → a cron/`once`/task isn't stamped `lastFiredAt`, so it + re-fires next tick. Clean backpressure, no queue. +- **Global live count** — `aliveSessionCount()` (`terminal.ts:701`): `status='running'` rows confirmed + alive in tmux. +- **Per-trigger pile-up guards** — `isAlive(lastSessionId)` on automations/tasks (`automations.ts:424,469`). +- **Reclamation** — headless Stop-hook teardown, resident idle reaper (~30 min), crash sweep. + +## The three holes + +1. **Cap is off by default** (`0` = unlimited) → nothing protects the box today. +2. **Chat/event spawns bypass the cap.** `fireSlack`/`fireDiscord`/`fireComposio`/`fireWebhook` and the + `/agent` router `spawnChatAgent` call `fire(..., {guard:false})` — the cap is never consulted (it lives + in `tick()`, not `fire()`). This is the actual flood path: @mentions + fresh chats fan out unbounded, + each leaving a resident session for ~30 min. +3. **`aliveSessionCount()` fail-opens.** When the backend can't poll tmux (`aliveNames()===null` — always + true on the Linux `LauncherSessionBackend`; transient `spawnSync` errors on local) it returns `0`, so + the cap silently disables under exactly the load it's for. + +Note: `continueSlackThread` (`automations.ts:629`) delivers into an existing session (send-keys) or revives +the same row — **no new session**, so it correctly stays ungated. Only *brand-new* spawns queue. + +--- + +## Phase 1 — turn the cap on, correctly ✅ SHIPPED (v0.143.0) + +1. **Default is no longer 0.** Resolve `maxConcurrent` as `env override → Settings runtime default → + RAM-derived default`. Derived: `Math.max(3, Math.floor(totalMemGB / 1.5))` (2 GB droplet → 3; 32 GB + Mac Mini → ~21). Adapts per box, env still wins for tuning. Expose it as a **Settings → runtime** + value alongside the existing runtime defaults so it's tunable from the console without a restart. +2. **Fix the fail-open.** When `aliveNames()` returns `null`, `aliveSessionCount()` should fall back to a + pure DB count of `status='running'` rows instead of `0`, so the cap engages on launcher backends and + through transient poll failures. The crash sweep already reaps stale `running` rows, so the DB count is + a safe proxy. (Keep the count cheap — it runs every tick and per admission check.) + +**As shipped:** +- `derivedConcurrencyCap(totalBytes?)` (`automations.ts`, exported) = `max(3, floor(GB/1.5))`. +- `Automations.concurrencyCap()` — the single source of truth resolver (env → Settings → derived); + `tick()` and `sweepStuckGoals()` now read it (the old `private readonly maxConcurrent` field is gone). +- `Settings.maxConcurrentSessions()` / `setMaxConcurrentSessions()` — operator override (`null` = unset → + derived; `0` = unlimited; `N>0` = cap), key `max_concurrent_sessions`. +- `TerminalManager.aliveSessionCount()` now falls back to the new `runningSessionCount()` (DB count) when + `aliveNames()` is null, instead of returning 0. +- `GET/PUT /api/settings/concurrency` (owner/admin; audited `settings.concurrency.updated`) → console + **Settings → Runtime defaults → Concurrency cap** panel: shows `alive` running / effective cap + source, + editable unless env-pinned. +- Verified by `scripts/` logic test (derived math, Settings semantics, resolver order, DB fallback). + +**Multi-tenant caveat (document, don't fix here):** each tenant runtime has its own `Automations` + +`TerminalManager` + tmux socket, so the cap is **per-tenant**, not per-box. On the shared mac-mini +process, N hot tenants can still oversubscribe. A true box-wide cap needs a shared cross-tenant counter in +`tenant-registry.ts` — listed under Futures. + +## Phase 2 — admission gate + chat queue (the core) + +**One admission decision.** A small classifier at the spawn boundary: + +- **interactive human** (console `POST /api/sessions`) → **always admit**, never queue (someone's waiting; + few of them). Phase 3 surfaces capacity but doesn't block. +- **chat / webhook / composio** (event-driven, no natural retry) → subject to cap. Under cap → fire now + (unchanged). Over cap → **enqueue + ack**. +- **cron / `once` / task** → already deferred by `tick()`; leave as-is. + +**Durable queue.** New `pending_spawns` table (+ migration in `state/db.ts`) so a restart resumes draining: + +| column | note | +|---|---| +| `id`, `tenant`, `enqueued_at` | | +| `agent`, `title`, `task` | the full built prompt (`extra`) | +| `spawned_by` | provenance (`automation:` / `chat:`) | +| `run_as` | accountable member | +| `headless`, `resident` | lane flags | +| `slack_channel`/`slack_thread`, `discord_channel`/`discord_message` | thread binding for the reply | +| `resume_claude_id` | for self-scheduled follow-ups | +| `source` | `chat` / `webhook` / `composio` (for audit + drain ordering) | + +**API surface (in `Automations`):** +- `private admit(): boolean` — `cap<=0 || aliveSessionCount() < cap` (single source of truth; `tick()` + reuses it too). +- `private enqueueSpawn(row)` — insert; returns queue depth for the ack. Bounded by `MAX_PENDING_SPAWNS` + (e.g. 50) — over that, reject with "at capacity, try again shortly" rather than grow unbounded. +- Drain in `tick()` (see Phase-2 ordering) — FIFO, up to remaining budget, dropping entries older than + `PENDING_SPAWN_TTL` (e.g. 15 min) so a queued reply never fires hours later out of context (drop → + best-effort "took too long, ask again" back to the thread). + +**Wire the event paths.** In `fireSlack`/`fireDiscord`/`fireComposio`/`fireWebhook` and `spawnChatAgent`: +replace the direct `fire()`/`createSession()` with `admit() ? fire()/createSession() : enqueueSpawn(...)`. +When enqueued, return an ack string: +- Slack/Discord: fold "🕒 You're in line (~N ahead) — I'll start shortly" into the in-thread ack the socket + already posts on mention. +- Webhook/composio HTTP: `202 { queued: true, position: N }` instead of `200 { sessionId }`. + +**Drain (`tick()` ordering)** — after the existing cron/`once` loop, before `dispatchTasks`, so humans +waiting in chat aren't starved behind the task board: +1. cron / `once` (time-critical) — unchanged +2. **`drainPendingSpawns(remaining)`** — FIFO, TTL-pruned +3. `dispatchTasks(remaining)` — unchanged + +All three share the one `cap - running` budget already computed in `tick()`. + +**Observability.** Audit `spawn.queued` (on enqueue) and `spawn.drained` (on drain), and extend the +existing `scheduler.deferred` line. Optional: a "N queued" counter on the console header next to the +version. + +## Phase 3 — console stays passing but honest + +`POST /api/sessions` (`server.ts:1698`) never blocks a human, but returns a `capacity: { alive, cap }` +hint so the UI can show "fleet busy — 11/12 running" and the operator understands why automated work is +lagging. No queue, no new gate. + +--- + +## Interactions / edge cases + +- **Resident chat dwell.** A resident Slack session counts against the cap for ~30 min. Cap + idle reaper + together bound the pool; if a low cap + many warm chats makes everything queue, the RAM-derived default + is meant to be generous, and both the cap and `chatIdleTimeoutMinutes` are tunable in Settings. +- **Thread follow-ups never queue** — `continueSlackThread` reuses the bound session (no new spawn), so an + in-flight conversation always flows even at capacity. Only first-message spawns are gated. +- **Restart** — persisted queue resumes draining; boot prunes entries past TTL. + +## Futures (out of scope this pass) + +- Per-agent / per-tenant sub-caps (fairness — one hot agent can't eat every slot). +- Pressure-aware reaping (shorten resident idle timeout as we approach the cap). +- True **box-wide** cap across tenants (shared counter in `tenant-registry.ts`). + +## Test / validation + +Per CLAUDE.md (no test runner): `npm run typecheck`; `cd web && npm run build`; an isolated in-process +Node script (`export AGENT_OS_HOME=`) that sets a low `AOS_MAX_CONCURRENT_SESSIONS`, drives +`fireSlack` past the cap, and asserts: (1) over-cap calls enqueue + return an ack, (2) `tick()` drains +FIFO as `aliveSessionCount()` drops, (3) TTL/`MAX_PENDING_SPAWNS` bounds hold, (4) interactive +`POST /api/sessions` never queues. Add a governance-suite case if one fits. + +## Touch list + +- `src/edge/automations.ts` — `maxConcurrent` resolution, `admit()`, `enqueueSpawn`, `drainPendingSpawns`, + wire `fireSlack`/`fireDiscord`/`fireComposio`/`fireWebhook`/`spawnChatAgent`, `tick()` drain step. +- `src/terminal.ts` — `aliveSessionCount()` DB fallback; a `runningSessionCount()` helper. +- `src/state/db.ts` — `pending_spawns` table + migration. +- `src/governance/settings.ts` — optional `maxConcurrent` runtime default. +- `src/server.ts` — `capacity` hint on `POST /api/sessions`; `202` on queued webhook/composio. +- `src/edge/slack-socket.ts` / `discord-socket.ts` — surface the queued ack in the mention reply. +- `web/src` — capacity indicator (optional). +- `CHANGELOG.md` + version bump on the shipping PR. + + diff --git a/package-lock.json b/package-lock.json index 3c6bfe7..f9a897b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "agent-os", - "version": "0.142.0", + "version": "0.143.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "agent-os", - "version": "0.142.0", + "version": "0.143.0", "license": "MIT", "bin": { "agent-os": "bin/agent-os" diff --git a/package.json b/package.json index 9adabc4..e5c1f39 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "agent-os", - "version": "0.142.0", + "version": "0.143.0", "description": "A generic, governed operating system for running autonomous agents safely across brands. Ships with a local web console.", "license": "MIT", "type": "commonjs", diff --git a/scripts/concurrency-cap-test.cjs b/scripts/concurrency-cap-test.cjs new file mode 100644 index 0000000..bfb8cfd --- /dev/null +++ b/scripts/concurrency-cap-test.cjs @@ -0,0 +1,78 @@ +#!/usr/bin/env node +/* Phase-1 concurrency-cap logic test — isolated home, no server/tmux/claude. + * Covers: derivedConcurrencyCap math, Settings get/set/clear semantics, the + * Automations.concurrencyCap() resolution order (env → setting → derived), and + * TerminalManager.runningSessionCount / aliveSessionCount DB fallback. */ +const fs = require('fs'); +const os = require('os'); +const path = require('path'); + +const ROOT = path.resolve(__dirname, '..'); +const HOME = fs.mkdtempSync(path.join(os.tmpdir(), 'aos-conc-test-')); +process.env.AGENT_OS_HOME = HOME; +process.env.AGENT_OS_TENANT = 'testco'; +delete process.env.AGENT_OS_SECRET_KEY; +delete process.env.AOS_MAX_CONCURRENT_SESSIONS; + +let pass = 0, fail = 0; +const assert = (cond, name, detail) => cond ? (pass++, console.log(` \x1b[32m✓\x1b[0m ${name}`)) : (fail++, console.log(` \x1b[31m✗ ${name}\x1b[0m${detail ? ' — ' + detail : ''}`)); + +const { loadAgentOS } = require(path.join(ROOT, 'dist/kernel.js')); +const { TerminalManager } = require(path.join(ROOT, 'dist/terminal.js')); +const { Automations, derivedConcurrencyCap } = require(path.join(ROOT, 'dist/edge/automations.js')); + +const aos = loadAgentOS(); +const tm = new TerminalManager(aos, 'http://127.0.0.1:0', path.join(HOME, 'tmux.sock')); +const autos = new Automations(aos, tm); +const GB = 1024 ** 3; + +console.log('\n\x1b[1m1) derivedConcurrencyCap (RAM → cap)\x1b[0m'); +assert(derivedConcurrencyCap(2 * GB) === 3, '2 GB → 3 (floor)'); +assert(derivedConcurrencyCap(1 * GB) === 3, '1 GB → 3 (floor, never below 3)'); +assert(derivedConcurrencyCap(32 * GB) === 21, '32 GB → 21'); +assert(derivedConcurrencyCap(48 * GB) === 32, '48 GB → 32'); + +console.log('\n\x1b[1m2) Settings.maxConcurrentSessions get/set/clear\x1b[0m'); +assert(aos.settings.maxConcurrentSessions() === null, 'unset → null'); +aos.settings.setMaxConcurrentSessions(7, 'tester'); +assert(aos.settings.maxConcurrentSessions() === 7, 'set 7 → 7'); +aos.settings.setMaxConcurrentSessions(0, 'tester'); +assert(aos.settings.maxConcurrentSessions() === 0, 'set 0 → 0 (explicit unlimited)'); +aos.settings.setMaxConcurrentSessions(null, 'tester'); +assert(aos.settings.maxConcurrentSessions() === null, 'clear (null) → null again'); +aos.settings.setMaxConcurrentSessions(-4, 'tester'); +assert(aos.settings.maxConcurrentSessions() === null, 'negative → cleared (null)'); +aos.settings.setMaxConcurrentSessions(12.9, 'tester'); +assert(aos.settings.maxConcurrentSessions() === 12, 'floors 12.9 → 12'); +aos.settings.setMaxConcurrentSessions(null, 'tester'); // reset for the resolver tests + +console.log('\n\x1b[1m3) Automations.concurrencyCap() resolution order\x1b[0m'); +delete process.env.AOS_MAX_CONCURRENT_SESSIONS; +assert(autos.concurrencyCap() === derivedConcurrencyCap(), 'no env, no setting → derived default'); +aos.settings.setMaxConcurrentSessions(5, 'tester'); +assert(autos.concurrencyCap() === 5, 'setting 5 (no env) → 5'); +aos.settings.setMaxConcurrentSessions(0, 'tester'); +assert(autos.concurrencyCap() === 0, 'setting 0 (no env) → 0 (unlimited)'); +process.env.AOS_MAX_CONCURRENT_SESSIONS = '9'; +assert(autos.concurrencyCap() === 9, 'env 9 overrides setting 0 → 9'); +process.env.AOS_MAX_CONCURRENT_SESSIONS = '0'; +assert(autos.concurrencyCap() === 0, 'env 0 → 0 (explicit unlimited, wins)'); +process.env.AOS_MAX_CONCURRENT_SESSIONS = ' '; +aos.settings.setMaxConcurrentSessions(4, 'tester'); +assert(autos.concurrencyCap() === 4, 'blank/whitespace env is ignored → falls to setting 4'); +process.env.AOS_MAX_CONCURRENT_SESSIONS = 'abc'; +assert(autos.concurrencyCap() === 4, 'non-numeric env ignored → setting 4'); +delete process.env.AOS_MAX_CONCURRENT_SESSIONS; +aos.settings.setMaxConcurrentSessions(null, 'tester'); + +console.log('\n\x1b[1m4) runningSessionCount / aliveSessionCount fallback\x1b[0m'); +assert(typeof tm.runningSessionCount() === 'number', 'runningSessionCount returns a number'); +assert(tm.runningSessionCount() === 0, 'no sessions → 0'); +// aliveSessionCount on the local backend with no live tmux: aliveNames() returns a Set (empty) → 0, +// and with no running rows the DB fallback is also 0. Either way it must be a finite number ≥ 0. +const ac = tm.aliveSessionCount(); +assert(Number.isFinite(ac) && ac >= 0, `aliveSessionCount finite ≥ 0 (got ${ac})`); + +console.log(`\n${fail === 0 ? '\x1b[32m' : '\x1b[31m'}PHASE-1 CONCURRENCY: ${pass}/${pass + fail} passed\x1b[0m`); +try { fs.rmSync(HOME, { recursive: true, force: true }); } catch {} +process.exit(fail === 0 ? 0 : 1); diff --git a/src/edge/automations.ts b/src/edge/automations.ts index 59a63ba..6066dc9 100644 --- a/src/edge/automations.ts +++ b/src/edge/automations.ts @@ -258,13 +258,17 @@ export function buildTaskPrompt(t: { id: string; title: string; body: string; cr import { claudeSupportsGoal } from './claude-cli'; export { claudeSupportsGoal }; +/** RAM-derived default concurrency cap: ~1 session per 1.5 GB of host memory, floored at 3. Adapts to the + * box (2 GB droplet → 3; 32 GB Mac Mini → ~21) so the cap protects small boxes without throttling big ones. + * Parameterized for testing. See docs/concurrency-cap-plan.md Phase 1. */ +export function derivedConcurrencyCap(totalBytes = os.totalmem()): number { + const gb = totalBytes / (1024 ** 3); + return Math.max(3, Math.floor(gb / 1.5)); +} + export class Automations { private readonly db: Db; private timer?: NodeJS.Timeout; - // Whole-box concurrency cap: the scheduler stops firing NEW cron/task spawns once this many sessions - // are alive (defense-in-depth against an OOM-inducing burst — see #137). Interactive/chat spawns are - // never gated (a human is waiting; a chat spawn has no natural retry). 0 = unlimited (opt-in per box). - private readonly maxConcurrent = Number(process.env.AOS_MAX_CONCURRENT_SESSIONS) || 0; constructor( private readonly os: AgentOS, @@ -712,16 +716,35 @@ export class Automations { this.timer = undefined; } + /** + * The effective whole-box concurrency cap, resolved LIVE each tick so an operator change takes effect + * without a restart. Order: `AOS_MAX_CONCURRENT_SESSIONS` env override → operator Settings value → + * RAM-derived default. Returns `0` for UNLIMITED (env or Settings explicitly set to 0). Single source + * of truth — the scheduler cap AND the Settings observability route both read this. Phase 2 will reuse + * it for the chat/webhook admission gate (`admit()`). See docs/concurrency-cap-plan.md. + */ + concurrencyCap(): number { + const env = process.env.AOS_MAX_CONCURRENT_SESSIONS; + if (env !== undefined && env.trim() !== '') { + const n = Number(env); + if (Number.isFinite(n) && n >= 0) return Math.floor(n); // env wins; 0 = unlimited + } + const setting = this.os.settings.maxConcurrentSessions(); + if (setting != null) return setting; // operator value; 0 = unlimited + return derivedConcurrencyCap(); // RAM-based default + } + /** One scheduler pass — public so tests (and a future "catch-up" boot pass) can drive it. */ tick(now: Date): void { // Advance any in-flight async video renders (poll → ingest on completion). Fire-and-forget: it's // async, tick is sync, and a poll error must never break the scheduler loop. void this.tm.pollVideoJobs().catch(() => {}); const minute = Math.floor(now.getTime() / 60_000); - // Whole-box concurrency cap (#137). When set, count sessions already alive and stop firing NEW - // scheduler spawns once we hit the ceiling — a deferred cron/one-shot isn't stamped `lastFiredAt` - // (and a `once` isn't disabled), so it simply RE-FIRES next tick. No queue needed. 0 = unlimited. - const cap = this.maxConcurrent; + // Whole-box concurrency cap (#137). Count sessions already alive and stop firing NEW scheduler spawns + // once we hit the ceiling — a deferred cron/one-shot isn't stamped `lastFiredAt` (and a `once` isn't + // disabled), so it simply RE-FIRES next tick. No queue needed. 0 = unlimited. The cap is now ON by + // default (RAM-derived) rather than opt-in — resolved live so a Settings change needs no restart. + const cap = this.concurrencyCap(); let running = cap > 0 ? this.tm.aliveSessionCount() : 0; let deferred = 0; const overCap = (): boolean => cap > 0 && running >= cap; @@ -772,7 +795,7 @@ export class Automations { private sweepStuckGoals(now: Date): void { if (!this.os.settings.autoPlanGoals()) return; try { - const cap = this.maxConcurrent; + const cap = this.concurrencyCap(); let spawned = 0; for (const g of this.os.goals.stuck(this.os.tenant, GOAL_AUTOPLAN_GRACE_MS, now.getTime())) { if (spawned >= GOAL_AUTOPLAN_MAX_PER_TICK) break; diff --git a/src/governance/settings.ts b/src/governance/settings.ts index d95bb82..d78d113 100644 --- a/src/governance/settings.ts +++ b/src/governance/settings.ts @@ -56,6 +56,7 @@ export const DEFAULT_GOVERNANCE_THRESHOLDS: GovernanceThresholds = { moneyCapUsd const EMAIL_ORG_DOMAINS_KEY = 'email_org_domains'; // internal email domains (JSON string[]); email.send to these is green const CHAT_ROUTER_KEY = 'chat_router_enabled'; // generic Slack/Discord `/agent` router fallback ('off' disables) const CHAT_IDLE_MIN_KEY = 'chat_idle_timeout_min'; // resident (warm) chat session idle-kill, minutes +const MAX_CONCURRENT_KEY = 'max_concurrent_sessions'; // whole-box concurrency cap override; unset → RAM-derived default, 0 → unlimited const KILL_SWITCH_KEY = 'kill_switch'; // workspace-wide emergency stop (JSON KillSwitchState) const SUPPRESSED_BUILTINS_KEY = 'suppressed_builtins'; // built-in agent ids an admin deleted (JSON string[]); boot won't re-seed them @@ -641,6 +642,33 @@ export class SettingsStore { return this.chatIdleTimeoutMinutes(); } + // ── whole-box concurrency cap (docs/concurrency-cap-plan.md) ───────────────────── + // The max number of live sessions the scheduler will let run at once. Every live session holds a + // tmux pane + a `claude` process (hundreds of MB), so enough of them swap/OOM the box. This is the + // OPERATOR override; the effective cap is resolved live by `Automations.concurrencyCap()` as + // env → this setting → RAM-derived default, so a change here takes effect on the next tick with no + // restart. Distinct from the model/effort runtime defaults (those are per-agent tuning) but surfaced + // in the same Settings → Runtime panel. + + /** The operator-set concurrency cap, or `null` when unset (→ the resolver uses the RAM-derived + * default). An explicit `0` means UNLIMITED (opt out of the cap); `N>0` caps live sessions at N. */ + maxConcurrentSessions(): number | null { + const raw = this.getRow(MAX_CONCURRENT_KEY)?.value; + if (raw == null || raw.trim() === '') return null; // unset → derived default + const n = Number(raw); + if (!Number.isFinite(n) || n < 0) return null; // garbage → treat as unset + return Math.floor(n); // 0 = unlimited; N>0 = cap + } + + /** Set (or clear, with `null`) the operator concurrency cap. `0` = unlimited; `N>0` = cap; `null`/ + * negative/garbage clears the override so the RAM-derived default applies again. */ + setMaxConcurrentSessions(n: number | null, by?: string): number | null { + const v = Number(n); + if (n == null || !Number.isFinite(v) || v < 0) this.set(MAX_CONCURRENT_KEY, '', by); // clear → derived default + else this.set(MAX_CONCURRENT_KEY, String(Math.floor(v)), by); + return this.maxConcurrentSessions(); + } + // ── kill switch (workspace emergency stop) ─────────────────────────────────────── // A single boolean that, when engaged, makes the gate deny EVERY action across the whole workspace // (governance-model.md — the operational control we lacked). Reversible: clear it and agents resume. diff --git a/src/server.ts b/src/server.ts index 58a8192..d5e9e08 100644 --- a/src/server.ts +++ b/src/server.ts @@ -16,7 +16,7 @@ import { exampleCapabilities } from './capabilities/examples'; import { evaluate } from './observability/evaluation'; import { TerminalManager, AGENT_OS_OPERATING_NOTES } from './terminal'; import { classifyActivity, clipText, ActivityCategory, ActivityEffect } from './state/session-activity'; -import { Automation, Automations, nextCronRun } from './edge/automations'; +import { Automation, Automations, nextCronRun, derivedConcurrencyCap } from './edge/automations'; import { SlackSocket } from './edge/slack-socket'; import { DiscordSocket } from './edge/discord-socket'; import { DreamingEngine } from './edge/dreaming'; @@ -2672,6 +2672,32 @@ async function handle(os: AgentOS, tm: TerminalManager, autos: Automations, req: return sendJson(res, 200, { ok: true, ...saved }); } + // ── whole-box concurrency cap (docs/concurrency-cap-plan.md Phase 1) ── + // The effective cap resolves env → operator Settings → RAM-derived default (single source of truth = + // Automations.concurrencyCap). Reports the resolved value + its source + the live running count so the + // console can show "N / cap running" and whether an env var is pinning it. + if (method === 'GET' && p === '/api/settings/concurrency') { + if (!isAdmin(me)) return sendJson(res, 403, { error: 'owner or admin required' }); + const env = process.env.AOS_MAX_CONCURRENT_SESSIONS; + const envLocked = env !== undefined && env.trim() !== '' && Number.isFinite(Number(env)) && Number(env) >= 0; + const value = os.settings.maxConcurrentSessions(); // operator override (null = unset) + const resolved = autos.concurrencyCap(); // effective cap the scheduler enforces (0 = unlimited) + const source = envLocked ? 'env' : value != null ? 'setting' : 'derived'; + return sendJson(res, 200, { value, resolved, derived: derivedConcurrencyCap(), source, envLocked, alive: tm.aliveSessionCount() }); + } + if (method === 'PUT' && p === '/api/settings/concurrency') { + if (!isAdmin(me)) return sendJson(res, 403, { error: 'owner or admin required' }); + const b = await readBody(req) as { value?: unknown }; + // `null`/'' clears the override (→ derived default); 0 = unlimited; N>0 = cap. Reject non-numeric junk. + const raw = b.value; + const clear = raw === null || raw === undefined || raw === ''; + const n = clear ? null : Number(raw); + if (!clear && (!Number.isFinite(n as number) || (n as number) < 0)) return sendJson(res, 400, { error: 'value must be a non-negative integer, 0 (unlimited), or null (use default)' }); + const saved = os.settings.setMaxConcurrentSessions(clear ? null : (n as number), me.email); + os.audit.append({ ts: Date.now(), runId: '-', tenant: os.tenant, principal: me.email, type: 'settings.concurrency.updated', data: { value: saved } }); + return sendJson(res, 200, { ok: true, value: saved, resolved: autos.concurrencyCap(), derived: derivedConcurrencyCap() }); + } + // ── UI branding (per-tenant accent colour + favicon badge) — owner/admin edits ── // The public read is GET /api/branding (above the member gate); this is the admin editor surface. if (method === 'GET' && p === '/api/settings/branding') { diff --git a/src/terminal.ts b/src/terminal.ts index ef6b0f5..385a2c5 100644 --- a/src/terminal.ts +++ b/src/terminal.ts @@ -694,20 +694,30 @@ export class TerminalManager { /** * How many sessions have a `running` row AND a live tmux pane right now — the whole-box concurrency - * measure for the scheduler cap (`AOS_MAX_CONCURRENT_SESSIONS`). Counts every provenance (interactive, - * chat, automation, task) since they all consume memory, so the scheduler backs off when a human is - * already loading the box. FAIL-OPEN: if liveness can't be polled (launcher backend / transient tmux - * failure) it returns 0 so a hiccup never freezes the scheduler into deferring everything. + * measure for the scheduler cap. Counts every provenance (interactive, chat, automation, task) since + * they all consume memory, so the scheduler backs off when a human is already loading the box. + * + * When liveness CAN'T be polled (`aliveNames()===null` — always on the Linux LauncherSessionBackend, + * or a transient tmux hiccup) it falls back to a pure DB count of `running` rows rather than 0. The old + * fail-open-to-0 silently DISABLED the cap under exactly the load it's for (the launcher backend never + * polls) — a DB proxy keeps the cap engaged. The crash sweep reaps stale `running` rows, so the count + * is a safe upper-bound. (docs/concurrency-cap-plan.md Phase 1.) */ aliveSessionCount(): number { const alive = this.backend.aliveNames(); - if (!alive) return 0; + if (!alive) return this.runningSessionCount(); const rows = this.db.prepare("SELECT tmux FROM term_sessions WHERE status = 'running'").all<{ tmux: string }>(); let n = 0; for (const r of rows) if (alive.has(r.tmux)) n++; return n; } + /** Pure DB count of `running` sessions — the cap's fallback when tmux liveness can't be polled. Cheap + * (runs per tick + per admission check); the crash sweep keeps the `running` set honest. */ + runningSessionCount(): number { + return this.db.prepare("SELECT COUNT(*) AS c FROM term_sessions WHERE status = 'running'").get<{ c: number }>()!.c; + } + /** * Per-session resident memory for the live running set — what each agent session's process tree * (shell → claude/node → MCP subprocesses) currently occupies. Joins the running rows against the diff --git a/web/src/App.tsx b/web/src/App.tsx index 1071bc5..f50a85d 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -12,7 +12,7 @@ import { Input } from '@/components/ui/input' import { Separator } from '@/components/ui/separator' import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from '@/components/ui/select' import { DropdownMenu, DropdownMenuTrigger, DropdownMenuContent, DropdownMenuItem, DropdownMenuSeparator } from '@/components/ui/dropdown-menu' -import { api, EFFORTS, PERMISSION_MODES, type PermissionMode, type StateResp, type AgentInfo, type Session, type Msg, type Member, type Role, type TeamResp, type MemberIdentity, type IdentityProvider, IDENTITY_PROVIDERS, type Automation, type Task, type TaskEvent, type TaskAttachment, type TaskStatus, type AddTaskReq, type Goal, type GoalEvent, type GoalStatus, type GoalCounts, type GoalProgress, type AddGoalReq, type MemoryRecord, type MemoryHealth, type MemoryBackend, type MemorySettings, type MemorySettingsReq, type OllamaStatus, type KbPage, type KbRevision, type AgentRevision, type AgentStats, type Recommendation, type PolicyDocument, type PolicyRule, type PolicyOutcome, type PolicyOp, type DirListing, type FileEntry, type FileContent, type Artifact, type SkillSummary, type SkillsResp, type CatalogSkill, type CatalogAgent, type SkillSource, type RemoteSkill, type SkillshHit, type SkillRequest, type IntegrationsResp, type SlackStatus, type DiscordStatus, type AuditEvent, type Effort, type RuntimeTuning, type SecretMeta, type UpdateStatus, type UpdateApplyResult, type ActivityEvent, type ActivitySummaryRow, type SystemMetrics } from '@/lib/api' +import { api, EFFORTS, PERMISSION_MODES, type PermissionMode, type StateResp, type AgentInfo, type Session, type Msg, type Member, type Role, type TeamResp, type MemberIdentity, type IdentityProvider, IDENTITY_PROVIDERS, type Automation, type Task, type TaskEvent, type TaskAttachment, type TaskStatus, type AddTaskReq, type Goal, type GoalEvent, type GoalStatus, type GoalCounts, type GoalProgress, type AddGoalReq, type MemoryRecord, type MemoryHealth, type MemoryBackend, type MemorySettings, type MemorySettingsReq, type OllamaStatus, type KbPage, type KbRevision, type AgentRevision, type AgentStats, type Recommendation, type PolicyDocument, type PolicyRule, type PolicyOutcome, type PolicyOp, type DirListing, type FileEntry, type FileContent, type Artifact, type SkillSummary, type SkillsResp, type CatalogSkill, type CatalogAgent, type SkillSource, type RemoteSkill, type SkillshHit, type SkillRequest, type IntegrationsResp, type SlackStatus, type DiscordStatus, type AuditEvent, type Effort, type RuntimeTuning, type Concurrency, type SecretMeta, type UpdateStatus, type UpdateApplyResult, type ActivityEvent, type ActivitySummaryRow, type SystemMetrics } from '@/lib/api' import { type Branding, type PublicBranding, type NotificationPrefs, DEFAULT_NOTIFICATION_PREFS } from '@/lib/api' import { applyAccent, applyFavicon, faviconDataUri, readableOn } from '@/lib/branding' import { ConnectorsPage } from '@/connectors' @@ -7966,7 +7966,7 @@ function SettingsPage({ me, state, tab: tabParam, onTab }: { me: Member; state:
{tab === 'company' ? - : tab === 'runtime' ? + : tab === 'runtime' ?
: tab === 'theme' ? : tab === 'secrets' ? : tab === 'memory' ? @@ -8615,6 +8615,78 @@ function RuntimeDefaultsSettings({ me }: { me: Member }) { ) } +/** Settings → Runtime → Concurrency cap. The max number of agent sessions the scheduler will run at once + * (each holds a claude process ≈ hundreds of MB). The cap is ON by default (RAM-derived); an operator can + * raise/lower it or lift it entirely. Effective value resolves env → this setting → derived default. */ +function ConcurrencySettings({ me }: { me: Member }) { + const [data, setData] = useState(null) + const [input, setInput] = useState('') // '' = use default; '0' = unlimited; N = cap + const [busy, setBusy] = useState(false) + const [hint, setHint] = useState('') + const canEdit = me.role === 'owner' || me.role === 'admin' + + const load = () => api.concurrency().then((r) => { + if (r.error) return + setData(r) + setInput(r.value == null ? '' : String(r.value)) // box reflects only an explicit operator override + }).catch(() => {}) + useEffect(() => { load() }, []) + + const dirty = data != null && input.trim() !== (data.value == null ? '' : String(data.value)) + const save = async () => { + setBusy(true); setHint('') + const r = await api.saveConcurrency(input.trim() === '' ? null : Number(input)) + setBusy(false) + if (r.error) return setHint('⚠ ' + r.error) + setHint('saved — applies on the next scheduler tick'); setTimeout(() => setHint(''), 2500) + load() + } + const capLabel = (n: number) => (n <= 0 ? 'unlimited' : String(n)) + + return ( + + +
+

Concurrency cap

+

+ The most agent sessions allowed to run at once. Each live session holds a{' '} + claude process (hundreds of MB), so this guards the box against an + OOM-inducing burst of scheduled work. Over the cap, the scheduler defers cron/task spawns and retries next tick — + interactive and chat sessions are never blocked. +

+
+ {data && ( +
+ {data.alive} running now · effective cap{' '} + {capLabel(data.resolved)}{' '} + + ({data.source === 'env' ? 'from AOS_MAX_CONCURRENT_SESSIONS' : data.source === 'setting' ? 'operator-set' : `default for this box (~${data.derived})`}) + +
+ )} +
+ +
+ setInput(e.target.value)} + placeholder={data ? `default (${data.derived})` : 'default'} + disabled={!canEdit || !!data?.envLocked} + className="h-8 w-40 font-mono text-xs" + /> + + {hint && {hint}} +
+

+ Blank = the RAM-derived default{data ? ` (${data.derived})` : ''}. 0 = unlimited (no cap). + {data?.envLocked && <> Pinned by the AOS_MAX_CONCURRENT_SESSIONS env var — clear it to edit here.} +

+
+ + + ) +} + /** * Settings → Memory — pick the persistent-memory backend for every agent and apply it live. * sqlite — zero-infra default (FTS5 keyword recall), ships in the box. diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 45864ed..865d027 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -39,6 +39,19 @@ export interface RuntimeTuning { permissionMode?: PermissionMode } +/** Whole-box concurrency cap state (Settings → Runtime). `value` = operator override (null = unset); + * `resolved` = effective cap the scheduler enforces (0 = unlimited); `derived` = the RAM-based default; + * `source` = which of the three won; `envLocked` = pinned by the AOS_MAX_CONCURRENT_SESSIONS env var; + * `alive` = live running-session count right now. */ +export interface Concurrency { + value: number | null + resolved: number + derived: number + source: 'env' | 'setting' | 'derived' + envLocked: boolean + alive: number +} + export interface AgentInfo { id: string description: string @@ -1046,6 +1059,8 @@ export const api = { agentRevert: (id: string, rev: number) => call<{ ok: boolean; id?: string; toRev?: number; rev?: number; error?: string }>('POST', `/api/agents/${encodeURIComponent(id)}/revert`, { rev }), runtimeDefaults: () => call('GET', '/api/settings/runtime-defaults'), saveRuntimeDefaults: (tuning: RuntimeTuning) => call<{ ok: boolean; error?: string } & RuntimeTuning>('PUT', '/api/settings/runtime-defaults', tuning), + concurrency: () => call('GET', '/api/settings/concurrency'), + saveConcurrency: (value: number | null) => call<{ ok: boolean; error?: string; value?: number | null; resolved?: number; derived?: number }>('PUT', '/api/settings/concurrency', { value }), governance: () => call('GET', '/api/settings/governance'), saveGovernance: (t: GovernanceThresholds & { hostGovernanceEnabled?: boolean }) => call<{ ok: boolean; error?: string; hostGovernanceEnabled?: boolean } & GovernanceThresholds>('PUT', '/api/settings/governance', t),