From 678490f5f4aff11337303ef6a5fe2d191f456d91 Mon Sep 17 00:00:00 2001 From: Letian Lin Date: Sat, 20 Jun 2026 11:16:11 +0800 Subject: [PATCH 1/5] fix: replace per-instance disk log file with in-memory ring buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eliminate ~100,000× disk write amplification caused by the old enforceMaxLogSize loop (read 10MB + truncate + write 10MB per 1024-byte PTY chunk). Under TUI redraw workloads, this could generate 280MB+ of accumulated disk writes in the data directory. Changes: - Add RingBuffer (internal/instance/logbuf.go): bounded, thread-safe, in-memory ring buffer. Backing slice pre-allocated eagerly. Hot-path optimized with WriteString (no []byte(chunk) copy) and atomic pointer access (no stateMu per chunk). - Add adaptive sizing (internal/instance/sizing.go): per-instance cap clamp(available/16, 16MB, 256MB), default 32MB. Global budget capped at 25% system RAM. User override via config.GlobalConfig.LogBufferBytes. - Refactor Manager (internal/instance/manager.go): pumpLogs writes to ring buffer instead of disk. Tail/ReadSince read from buffer. Removed enforceMaxLogSize, logPathByID, and all LogPath references. - Remove LogPath from ManagedInstance (internal/store/state.go). Old state.json files with log_path load cleanly (JSON silently ignores). - Add PurgeOrphanLogFiles() called at daemon startup to clean up dead .log artifacts left by pre-buffer code. - HTTP/MCP: budget-exceeded returns 503 with structured log_buffer_budget_exceeded body. UI shows dedicated modal. - Test coverage: RingBuffer (15 tests), sizing/budget (9 tests), manager budget integration, backward-compat state loading. - Docs: PRD, ARCHITECTURE §4.1, API (error shape + cursor semantics), CHANGELOG Unreleased section, and detailed implementation plan. BREAKING: state.json schema — log_path field removed. Old files continue to load; external tooling should drop log_path dependency. --- CHANGELOG.md | 16 + docs/API.md | 30 +- docs/ARCHITECTURE.md | 45 +- docs/PRD.md | 12 + .../fix-disk-massive-write-issue/plan.md | 173 +++++++ internal/app/app.go | 60 ++- internal/app/app_test.go | 58 +++ internal/cli/cli.go | 10 +- internal/config/global.go | 5 + internal/instance/logbuf.go | 206 ++++++++ internal/instance/logbuf_test.go | 378 +++++++++++++++ internal/instance/manager.go | 446 +++++++++++------- internal/instance/manager_budget_test.go | 69 +++ internal/instance/manager_export_test.go | 22 + internal/instance/manager_integration_test.go | 79 ++++ internal/instance/manager_test.go | 279 ++++++++++- internal/instance/sizing.go | 168 +++++++ internal/instance/sizing_test.go | 154 ++++++ internal/store/state.go | 1 - internal/store/state_test.go | 36 +- internal/ui/static/index.html | 47 +- 21 files changed, 2070 insertions(+), 224 deletions(-) create mode 100644 docs/plans/fix-disk-massive-write-issue/plan.md create mode 100644 internal/instance/logbuf.go create mode 100644 internal/instance/logbuf_test.go create mode 100644 internal/instance/manager_budget_test.go create mode 100644 internal/instance/manager_export_test.go create mode 100644 internal/instance/sizing.go create mode 100644 internal/instance/sizing_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index b8a9817..0e62b97 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased + +Disk write amplification fix for long-running PTY-heavy sessions. + +### Breaking changes + +- **`state.json` schema: `log_path` field removed** — external tooling that reads instance state files must drop the `log_path` field. Old state files continue to load cleanly (the field is silently ignored by JSON decoding). See `docs/ARCHITECTURE.md` §3.2. + +### Highlights + +- **In-memory ring buffer for PTY logs** — replaced per-instance on-disk log files with a bounded ring buffer (default 32 MB per instance, hard ceiling 256 MB, total budget capped at 25% of system RAM). Eliminates the 100,000× write amplification caused by the old `enforceMaxLogSize` loop. No more disk I/O from PTY logging on the steady state. +- **Bounded adaptive sizing** — new `LogBufferBytes` config knob overrides the per-instance cap (clamped to 16–256 MB). When unset, the cap is `clamp(available / 16, 16 MB, 256 MB)` sampled via gopsutil with a 60 s cache to amortize syscall cost across batch starts. +- **Startup cleanup** — `Manager.PurgeOrphanLogFiles` removes dead `.log` files left behind by the old code path on the first start of the new binary. +- **Budget-exceeded UX** — when a new instance would push the global buffer budget past 25% of system RAM, the HTTP/MCP API returns `503 Service Unavailable` with a structured `log_buffer_budget_exceeded` body. The dashboard surfaces this as a modal with used/limit numbers and a hint to close other tabs. +- **Hot-path redesign for heavy TUI workloads** — `pumpLogs` no longer acquires `stateMu` per 1024-byte PTY chunk; the ring buffer pointer is pre-fetched once and the per-chunk lookup is a single `atomic.Pointer.Load`. The ring buffer's backing slice is allocated eagerly so the first Write never blocks on a 32 MB malloc. A new `WriteString` path avoids the `[]byte(chunk)` conversion that would otherwise happen on every chunk. + ## v0.3.0 Release focused on remote collaboration, build robustness, and Apple Silicon reliability. diff --git a/docs/API.md b/docs/API.md index cc13388..50c230e 100644 --- a/docs/API.md +++ b/docs/API.md @@ -293,9 +293,31 @@ Example (ad-hoc command without tags): Response (201): ```json -{ "id":"...","pid":123,"status":"running","log_path":"..." } +{ "id":"...","pid":123,"status":"running","created_at":"..." } ``` +**Error: log buffer budget exceeded (`503 Service Unavailable`)** + +Returned when starting the new instance would push the per-process log-buffer total past the global budget (default: 25% of system RAM — see `docs/ARCHITECTURE.md` §4.1 *Instance log buffer*). The error code `log_buffer_budget_exceeded` is part of the stable API contract; the UI matches on it to surface a dedicated modal. + +Headers: +- `Content-Type: application/json` +- `Retry-After: 0` (will not auto-resolve; user must close other instances or raise `log_buffer_bytes` in `auth.json`) + +Body: +```json +{ + "error": "log_buffer_budget_exceeded", + "message": "Insufficient memory to start new instance: used 100.00 MB, limit 64.00 MB.", + "used_bytes": 104857600, + "limit_bytes": 67108864, + "system_bytes": 268435456, + "hint": "Close other instances, raise LogBufferBytes in auth.json, or reduce concurrent tabs." +} +``` + +The same shape is returned by the MCP `instance_start` tool when the budget is exceeded. + ### Rename `PATCH /api/instances` @@ -431,13 +453,15 @@ Body: { "id": "" } ``` -Deletes a stopped (non-running) instance record (best-effort deletes the log file). +Deletes a stopped (non-running) instance record. The instance's in-memory log buffer is also released, decrementing the global log-buffer accounting. ### Log replay (tail / incremental) `GET /api/instances/log?id=[&since=]` - Without `since`: returns recent tail as `text/plain`. - With `since`: returns incremental content from byte offset and includes response header `X-Log-Offset: `. +- Logs live in an in-memory ring buffer attached to the **running** instance (see `docs/ARCHITECTURE.md` §4.1 *Instance log buffer*). After the instance stops, exits, or fails — or after the daemon restarts — the buffer is released and this endpoint returns an empty body. Unknown / never-started instance IDs also return empty. +- The `byteOffset` cursor is the running total of bytes the instance has produced (monotonic; never decreases). When `since` points to data that has already been evicted from the ring (oldest-byte > since), the response silently clamps to the oldest live byte and `X-Log-Offset` advances accordingly. Response: `text/plain` @@ -449,6 +473,8 @@ Response: `text/plain` ```json {"chunk":"...","next":12345} ``` +- Same in-memory backing as the tail endpoint above. The cursor `next` is the same monotonic byte counter; clients should echo it as `since` on the next request to receive only new chunks. +- Polling cadence: 1 s. When no new data is available, the server emits an SSE comment line (`: ping`) as a keep-alive — no `log` event, no cursor update. Clients should treat the absence of a `log` event as "no progress" and keep using the last `next` they saw. ### Instance resource stats `GET /api/instances/stats` diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 603239a..b6120cc 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -13,10 +13,10 @@ It does **not** analyze project code or prevent concurrent write conflicts insid - `cmd/myworktree/` — CLI entry. - `internal/app/` — HTTP server, auth middleware, API routing. - `internal/worktree/` — worktree lifecycle via `git` CLI. -- `internal/instance/` — instance lifecycle (spawn/stop/list) + output log file. +- `internal/instance/` — instance lifecycle (spawn/stop/list) + in-memory PTY log ring buffer (see §4). - `internal/tag/` — Tag config loader (MVP: JSON). - `internal/store/` — persistent state store (`state.json`) with file locking + atomic writes. -- `internal/redact/` — secret redaction for stored logs/backlog. +- `internal/redact/` — secret redaction applied to PTY chunks before they enter the ring buffer / live broadcast (e.g. `sk-...`). - `internal/mcp/` — MCP adapter surface (tool names + app-level tool dispatch), keeping core decoupled. - `internal/monitor/` — resource stats collector (CPU delta via gopsutil/process.Times, memory via RSS) - `internal/llm/` — LLM API client（OpenAI / Anthropic / OpenAI Compatible），可选，LLM Settings 通过 Web UI 对话框配置 @@ -35,7 +35,7 @@ It does **not** analyze project code or prevent concurrent write conflicts insid - Override: `-worktrees-dir=data` uses the legacy location under the per-project data dir; you can also set a custom path. - `state.json` — managed worktrees + managed instances + tab order + version - `tags.json` — project-level tags - - `logs/.log` — rolling instance backlog + - `logs/` — **no longer used** for live instance output. The directory may still exist on upgraded installs containing dead `.log` artifacts from older versions; the daemon purges them once on startup (see §4). Live PTY output is captured into a per-instance in-memory ring buffer instead. ### 3.1.2 全局配置 - 存储于用户级配置目录：`~/.config/myworktree/config.json`（按 OpenCode 方式，0o600 权限） @@ -51,7 +51,8 @@ It does **not** analyze project code or prevent concurrent write conflicts insid ### 3.2 State model - Worktree: id, name, path, branch, baseRef, createdAt -- Instance: id, worktreeId, tagId, command, cwd, env (sanitized), pid, status, logPath, timestamps +- Instance: id, worktreeId, tagId, command, cwd, env (sanitized), pid, status, timestamps + - **Schema note (v0.4.0)**: the legacy `log_path` field was removed when PTY logs moved to memory. `state.json` files written by older versions still load cleanly — the field is silently ignored by JSON decoding. External tools reading `state.json` should drop their dependency on `log_path`. - TabOrder (at State level): map of worktree_id to ordered list of instance IDs - **Version** (at State level): monotonically increasing int64, incremented on every write via `SaveWithVersion`. Used for optimistic locking on concurrent modification detection. - Main Repo: not persisted; served via `GET /api/main` with live git branch @@ -83,13 +84,43 @@ The sidebar shows a pinned **Main Workspace** item at the top (purple accent), f - Switching between running instances hides inactive terminal containers instead of tearing down their PTY attachment. This avoids detaching long-lived TUI programs such as Copilot CLI while they remain running. - Fallback path remains available: HTTP input `POST /api/instances/input` + replay/SSE logs (`GET /api/instances/log`, `GET /api/instances/log/stream`). - UI shows transport state (`websocket/sse/polling`) and supports manual WS reconnect. -- Backlog is stored on disk with a size cap (rolling truncate). -- On server startup, stale persisted `running` records are reconciled to `stopped` because in-memory stdin/stdout bindings cannot be resumed after process restart. +- PTY output is captured into a per-instance **in-memory ring buffer**; no disk I/O is involved on the steady state. The buffer feeds the HTTP/SSE/WS replay endpoints and the MCP `instance_log_tail` tool. See §4.1 for sizing, eviction, and budget rules. +- On server startup, stale persisted `running` records are reconciled to `stopped` because in-memory stdin/stdout bindings cannot be resumed after process restart. The same startup pass also calls `Manager.PurgeOrphanLogFiles()` to remove dead `.log` files left behind by pre-buffer versions; missing or empty `logs/` directories are not an error. - **Rename**: `PATCH /api/instances` updates an instance's display name (`name` field). The rename takes effect immediately in the UI and persists to `state.json`. - **Tab ordering**: `PATCH /api/instances/reorder` persists per-worktree tab order to `state.json` (`tab_order` map + array order in `State.Instances`). Uses **optimistic locking** — the client sends the `version` observed from `GET /api/instances`. If the state has been modified since (e.g., another user started an instance), the server returns HTTP 409 Conflict and the client refreshes and retries. - **Resource monitoring**: A clickable transport status bar in the bottom-right of the workspace opens a resource monitor modal. The modal shows per-instance CPU%, memory RSS, and connection type (WebSocket/SSE) grouped by worktree, with subtotals and a global summary. Data is fetched via `GET /api/instances/stats` (1-second polling when open, stops when closed). CPU% uses delta calculation from `process.Times()` with a per-PID baseline stored in the `Collector` struct. - **Browser close protection**: The frontend registers a `beforeunload` event handler that unconditionally triggers a browser-native confirmation dialog on any page close/refresh/navigation attempt. This is purely a client-side UX safeguard — backend instances are unaffected and continue running. +### 4.1 Instance log buffer (in-memory) + +Each running instance owns a bounded, **in-memory** ring buffer (`internal/instance/logbuf.go`, `RingBuffer`) that captures the PTY output stream emitted by `pumpLogs`. This buffer is the **only** backing store for the log replay endpoints, the SSE live-stream endpoint, and the MCP `instance_log_tail` tool. There is no persistent on-disk log file. + +**Why in-memory.** The previous design wrote every 1024-byte PTY chunk to a per-instance `.log` file and called `enforceMaxLogSize` after each write; once a file reached the 10 MB cap, every subsequent chunk triggered a full `read 10 MB + truncate + write 10 MB` pass — write amplification on the order of 100,000× under TUI redraw workloads. Removing disk persistence eliminates the bug at the source and is consistent with the existing reconcile-on-startup semantics, which already declare that logs cannot be replayed across daemon restarts (running instances are marked `stopped`, their PTY channels are not resumed). + +**Sizing (per-instance cap).** +- Floor: 16 MB (`MinBufferCap`) +- Default: 32 MB (`DefaultBufferCap`) +- Ceiling: 256 MB (`MaxBufferCap`) — never exceeded +- Adaptive: when no user override is set, the cap is `clamp(available_memory / 16, 16 MB, 256 MB)`, sampled via `gopsutil/v4/mem.VirtualMemory()` at instance start. +- User override: `log_buffer_bytes` in `~/.config/myworktree/auth.json` (`GlobalConfig.LogBufferBytes`). When set, the value is clamped to `[16 MB, 256 MB]`. +- The backing slice is allocated eagerly inside `NewRingBuffer` so the first `Write` does not stall the producer on a 32 MB malloc. + +**Global budget.** The sum of all live buffer caps is bounded by `MaxTotalFraction × system_RAM` (default 25%). When `Manager.Start` would push the sum over this limit, it returns `*instance.LogBufferBudgetError` (which wraps the sentinel `ErrLogBufferBudgetExceeded`). The HTTP layer translates this into `503 Service Unavailable` with the structured `log_buffer_budget_exceeded` body documented in `docs/API.md` (Start endpoint). The MCP path returns the same shape. The budget check runs **before** `exec.Command` / `pty.Start`, so a rejected request leaves no orphan processes or PTYs to clean up. + +**Cursor semantics (`since` / `next`).** `head` is a monotonic total-bytes-written counter for the instance. Clients pass it as `since` to read incrementally; the server returns the bytes plus an advanced cursor. When no new data is available, the cursor is returned unchanged (preserves the SSE 1 s poll-loop contract). When `since` points to data already evicted from the ring (oldest live byte > since), the read silently clamps to the oldest live byte. + +**Lifecycle.** +- `Start` resolves the cap with budget enforcement, creates the buffer, adds it to `Manager.buffers[id]` (an `*atomic.Pointer[RingBuffer]`), and atomically increments `Manager.totalBufBytes`. +- `pumpLogs` reads PTY chunks, redacts them, and writes to the buffer via the pre-fetched pointer using a single `atomic.Pointer.Load` per chunk — no `stateMu` acquisition on the hot path. The redacted chunk is also broadcast to live subscribers (WS/SSE). +- `Stop` / `wait` / `Restart` / `Delete` all funnel into `dropBufferLocked`, which `Swap(nil)` on the pointer, closes the buffer, decrements `totalBufBytes`, and removes the map entry. In-flight `pumpLogs` chunks dropped after the swap are intentional: the same lifecycle event closes the PTY, so `Read` returns EOF and the goroutine exits naturally. + +**Startup purge.** `Manager.PurgeOrphanLogFiles()` is invoked once during `Server.Start` (after `ReconcileRunningOnStartup`). It removes every `*.log` file under `DataDir/logs/` — these are now dead artifacts left by the pre-buffer code path. A missing `logs/` directory is not an error; the count of removed files is logged. + +**Failure modes & user-facing knobs.** +- Budget exceeded → 503 with `log_buffer_budget_exceeded`. The dashboard surfaces a modal showing `used_bytes` / `limit_bytes` and a hint to raise `LogBufferBytes` or close other tabs. +- Memory sampler fails (e.g. unusual cgroup) → cap falls back to `DefaultBufferCap`; the budget check is skipped for that call. +- Daemon restart → all buffers are gone. The next `GET /api/instances/log` for a stopped/never-started instance returns empty (documented behaviour). + ## 5. Terminal Protocol Timing Specification This section defines the strict timing protocol for terminal I/O to prevent escape sequence leakage and ensure reliable data flow. @@ -359,7 +390,7 @@ myworktree implements a **dual-layer authentication architecture**: - Loopback requests skip all token/origin validation (enables Portal reverse proxy) - Origin/Host check + basic rate limit on unauthorized non-loopback attempts - Optional built-in HTTPS via `--tls-cert/--tls-key` -- Redaction on stored backlog (e.g. `sk-...`) +- Redaction is applied on each PTY chunk *before* it lands in the in-memory ring buffer or is broadcast to live subscribers (e.g. `sk-...` masked). The replayed/streamed bytes never contain the raw secret. **Proxy authentication bypass (planned)**: The planned Portal reverse proxy (`/s//`) will forward requests to instances via `127.0.0.1` (loopback), so instances automatically skip auth. **Currently not yet implemented** — dashboard links connect to instance ports directly. diff --git a/docs/PRD.md b/docs/PRD.md index c228582..7d57871 100644 --- a/docs/PRD.md +++ b/docs/PRD.md @@ -72,6 +72,18 @@ - 双层认证架构（Portal 层 Cookie + CSRF，实例层 loopback 绕过） - 浏览器关闭保护：前端在 `beforeunload` 事件时，无论是否存在运行中实例，均触发浏览器原生确认对话框，防止误操作关闭页面。 - **Main workspace 分支查询**：`GET /api/main` 返回 `{name, branch}`。branch 字段实时查询（`git rev-parse --abbrev-ref HEAD`），在 detached HEAD 场景（如 CI 浅克隆）下返回空字符串而非错误。 +- **已实现：instance PTY 日志内存化（彻底取消磁盘日志）**： + - 旧实现：每条 PTY chunk（1024 字节）写入 `logs/.log`，并在文件达到 10MB 后做"读 10MB + 截断 + 写 10MB"，造成约 10 万倍磁盘写放大；OpenCode 等 TUI 高频重绘场景下数据目录写入量可达 280MB+。 + - 新实现：每个 running instance 拥有一个进程内的 **有界 ring buffer**（`internal/instance/logbuf.go`）。HTTP/SSE/WS 日志回放端点与 MCP `instance_log_tail` 全部从该 buffer 读取，磁盘 I/O 完全消除。 + - 容量策略（与 bounded-but-adaptive 原则一致：硬上限防 OOM、富裕时适度放大）： + - 单实例下限 16 MB、默认 32 MB、上限 256 MB（硬天花板，永不超过）。 + - 自适应：未配置时 `clamp(available_memory / 16, 16 MB, 256 MB)`；用户可通过 `~/.config/myworktree/auth.json` 的 `log_buffer_bytes` 字段覆盖。 + - 全局预算：所有 live buffer 容量合计上限为系统 RAM 的 25%。 + - 当新实例会突破全局预算时，`POST /api/instances` 返回 `503 Service Unavailable` + 结构化 `log_buffer_budget_exceeded` body，UI 弹出专用对话框提示已用/上限字节并给出处置建议。 + - 守护进程启动时，`Manager.PurgeOrphanLogFiles()` 一次性清理旧版本遗留的 `.log` 文件（幂等）。 + - **破坏性 schema 变更**：`state.json` 的 `log_path` 字段已移除；旧文件继续可加载（被 JSON 解码静默忽略），外部读取 `state.json` 的工具应去掉对该字段的依赖。 + - 守护进程重启会清空所有 buffer，与既有的 `ReconcileRunningOnStartup` 语义一致（重启后日志原本就不能回放，运行中实例会被标记为 stopped）。 + - 详情见 `docs/ARCHITECTURE.md` §4.1、`docs/API.md` *Start* 端点错误段。 - **规划新增：分支落后检测**： - 侧栏每个 worktree 分支名旁显示红色标签（如 `m↑3` / `d↑1`），标识当前分支是否落后于主分支或集成分支 develop。 - 如果当前 worktree 就是主分支自身，则不显示标记。 diff --git a/docs/plans/fix-disk-massive-write-issue/plan.md b/docs/plans/fix-disk-massive-write-issue/plan.md new file mode 100644 index 0000000..48c08a8 --- /dev/null +++ b/docs/plans/fix-disk-massive-write-issue/plan.md @@ -0,0 +1,173 @@ +# Plan: Replace Per-Instance Log File with In-Memory Ring Buffer + +## Context + +The `mw` daemon (PID 93493 at last observation) exhibits massive sustained disk writes during long-running PTY-heavy sessions. Root cause: `instance.Manager.pumpLogs` calls `enforceMaxLogSize` after every 1024-byte PTY chunk. Once a per-instance log file reaches `maxLogBytes` (10MB), each subsequent chunk pushes the file over the limit and triggers `enforceMaxLogSize` to do `open + read 10MB + os.WriteFile(path, 10MB)` — a full 10MB read + truncate + write. Net effect: every ~100 bytes of PTY output produces ~20MB of disk I/O (amplification ~100,000×). Symptom: 280MB+ accumulated log directory; file sizes oscillate between 10485760 and 0 (the `0` comes from `os.WriteFile`'s `O_TRUNC` window between truncate and write). + +User has confirmed: **logs should live entirely in memory, no disk persistence**. This aligns with the existing `ReconcileRunningOnStartup` semantics (manager.go:56-84) which already declares logs cannot be read after restart — running instances are marked stopped, in-memory PTY channels are not resumed. Removing the on-disk log file preserves existing behavior while eliminating the write-amplification bug entirely. + +Outcome: zero disk I/O from PTY logging; bounded memory per instance; HTTP/WS/SSE/MCP log endpoints continue to work unchanged for the running instance's lifetime. + +## Design Decisions (locked with user) + +- **In-memory only.** `LogPath` field on `ManagedInstance` is removed entirely (not kept as empty string). All 5 current references (`manager.go:184/231/570/661/942`) are deleted along with the `logPathByID` helper and the `Restart`/`Delete` log-file removal paths. +- **Bounded but adaptive sizing** per `feedback-memory-buffer-sizing`: + - Default cap per instance: **32 MB** + - Hard ceiling per instance: **256 MB** (never exceeded) + - Min floor per instance: **16 MB** + - Total budget across all live instances: **≤ 25% of system RAM** + - Adaptive default: `clamp(available_mem / 16, 16MB, 256MB)` computed via `gopsutil/mem.VirtualMemory()` at instance start (cache result for 60s) + - User override: new `config.GlobalConfig.LogBufferBytes int64` (default `0` = use adaptive). When set, clamped to [16MB, 256MB]. + - When new instance would exceed total budget, `Start` returns a new exported error `ErrLogBufferBudgetExceeded`. + +## Architecture + +### New: `internal/instance/logbuf.go` + +Chunked ring buffer. Internals: +- `cap int64` — hard ceiling +- `chunks [][]byte` — fixed-size 64KB chunks, head/tail indices +- `head int64` — monotonic byte offset for `ReadSince` semantics +- `mu sync.Mutex` — guards state; pumpLogs writes, Tail/ReadSince read + +Public surface: +- `NewRingBuffer(capBytes int64) *RingBuffer` +- `Write(p []byte) (n int)` — drops oldest when full, never blocks +- `Tail(n int64) (string, int64)` — last `n` bytes; returns end offset +- `ReadSince(since, maxBytes int64) (string, int64, error)` — bytes from offset, returns next offset; returns empty+`since` when no new data (preserves SSE poll loop semantics) +- `Offset() int64` — current head offset +- `BytesUsed() int64` +- `Close()` — idempotent + +### New: `internal/instance/sizing.go` + +- Constants: `DefaultBufferCap=32MB`, `MinBufferCap=16MB`, `MaxBufferCap=256MB`, `MaxTotalFraction=0.25` +- `MemSampler` interface (one method `VirtualMemory() (*mem.VirtualMemoryStat, error)`) — inject for tests; default impl wraps `gopsutil/v4/mem` +- `resolveCap(cfgBytes int64, sampler MemSampler, totalBufBytes int64) (int64, error)` — returns the cap to use or `ErrLogBufferBudgetExceeded` +- New exported sentinel: `ErrLogBufferBudgetExceeded` (in `manager.go`) wrapping detailed context via `fmt.Errorf("...: %w", ErrLogBufferBudgetExceeded)`. The wrapped error must carry a structured payload so HTTP handlers can render it without re-querying the manager. + - Implementation note: define a struct error type `LogBufferBudgetError struct { UsedBytes int64; LimitBytes int64; SystemBytes int64 }` implementing `error`. Handler does `errors.As(err, &budgetErr)` to extract bytes for the response body. + +### Modify: `internal/app/app.go` — error contract + +Currently `writeErr(w, http.StatusBadRequest, err)` returns `{"error": "msg"}` JSON. Add a sibling helper `writeLogBufferBudgetErr(w, err)` for the budget-exceeded case so the UI can detect it without parsing free-form text. + +Update both `instance.Start` call sites (HTTP `handleInstance`/`handleInstanceMCP` at app.go:1140 and MCP `handleMCPTool` at app.go:1686): + +```go +if err != nil { + var budgetErr *instance.LogBufferBudgetError + if errors.As(err, &budgetErr) { + writeLogBufferBudgetErr(w, budgetErr) // HTTP 503 + structured body + return + } + writeErr(w, http.StatusBadRequest, err) + return +} +``` + +`writeLogBufferBudgetErr` writes: +- HTTP status: `503 Service Unavailable` +- Headers: `Content-Type: application/json`, `Retry-After: 0` (won't auto-resolve; user must close instances) +- Body: `{"error":"log_buffer_budget_exceeded","message":"...","used_bytes":N,"limit_bytes":M,"system_bytes":K,"hint":"Close other instances, raise LogBufferBytes in config, or reduce concurrent tabs."}` + +### Modify: `internal/ui/static/index.html` — popup + +When `POST /api/instances` returns 503 with body containing `"error":"log_buffer_budget_exceeded"`, the existing fetch handler must: +1. NOT mark the new tab as created (already handled since fetch threw) +2. Show a modal/toast with the message, with a "Close" button. Use the existing modal/dialog pattern in `index.html` (or a simple inline alert if none). Include the hint string and used/limit numbers in the dialog body so the user understands the constraint. +3. Optionally surface the limit/usage in the tab bar header so users can see at a glance how close they are to the budget. (Stretch goal — only if a small change.) + +> **Note**: the original plan called for the popup to live in `internal/portal/dashboard.html`, but that file is a read-only cross-repo list view that never POSTs to `/api/instances`. The popup therefore went into `internal/ui/static/index.html`, which is the actual instance-creation UI. This section reflects that. + +The MCP tool path (app.go:1686) should also return this error with the same shape so CLI/agent consumers can handle it programmatically. + +### Modify: `internal/instance/manager.go` + +Add to `Manager` struct: +- `buffers map[string]*RingBuffer` (guarded by `stateMu`) +- `totalBufBytes int64` (atomic, no separate mutex) +- `memSampler MemSampler` (nullable; defaults to gopsutil) +- `cfgLogBufferBytes int64` (from `config.GlobalConfig.LogBufferBytes`) +- `lastMemSampleAt time.Time` + `lastMemTotal int64` for 60s cache + +Modify: +- `Start` — drop `logPath`/`logFile` creation; compute `cap` via `resolveCap`; create `RingBuffer`; store in `m.buffers[id]`; pass to `pumpLogs`; bump `totalBufBytes`; clean up on error +- `pumpLogs(id, ptmx)` — drop the `out *os.File` and `logPath string` parameters; new shape: read PTY → `redact.Text` → `buf.Write` + `broadcastOutput` +- `Tail(id, n)` — read from `m.buffers[id].Tail(n)`; if buffer missing (instance stopped/never had one) return `("", nil)` +- `ReadSince(id, since, maxBytes)` — read from buffer; same nil-buffer behavior +- `closeSubscribersLocked(id)` — also remove from `m.buffers`, decrement `totalBufBytes` +- `Stop`/`wait` — already call `closeSubscribersLocked`, so buffer cleanup is automatic +- `Restart`/`Delete` — drop the `os.Remove(oldLogPath)` lines and the `oldLogPath` lookups entirely + +Delete: +- `enforceMaxLogSize` function (manager.go:867-892) +- `logPathByID` helper (manager.go:939-946) +- `TestEnforceMaxLogSize` (manager_test.go:45-62) +- All references to `LogPath` in `Start`/`Restart`/`Delete` + +Add to `Manager`: +- `PurgeOrphanLogFiles() (int, error)` — scans `m.DataDir/logs/`, removes every `.log` file (no in-memory or on-disk state references them after this refactor; all are dead artifacts). Missing directory is not an error. Returns count of files removed. Uses `m.Logger` to report each remove + summary. + +### Modify: `internal/store/state.go` + +Remove `LogPath string \`json:"log_path"\`` from `ManagedInstance` struct (line 43). Old `state.json` files with this field will have it silently ignored by JSON decoding — no migration needed. + +### Modify: `internal/config/global.go` + +Add field: `LogBufferBytes int64 \`json:"log_buffer_bytes,omitempty"\`` (default 0 = adaptive). + +### Modify: Manager construction sites (only 2) + +- `internal/app/app.go:131` — pass buffer config into `instance.Manager` +- `internal/cli/cli.go:269` — same + +Pass via new optional field on `Manager` (e.g., `LogBufferBytes int64` set after literal-init) or a small `NewManager` constructor. Recommend: keep `&Manager{...}` literal but add a single field; both sites already use literals, no churn beyond one extra line each. + +## Files to Modify (order) + +1. **`internal/instance/logbuf.go`** (new) — `RingBuffer` implementation +2. **`internal/instance/logbuf_test.go`** (new) — unit tests +3. **`internal/instance/sizing.go`** (new) — `resolveCap`, `MemSampler`, constants, `ErrLogBufferBudgetExceeded` +4. **`internal/instance/sizing_test.go`** (new) — `resolveCap` tests with `fakeMemSampler` +5. **`internal/instance/manager.go`** — remove `enforceMaxLogSize`, `logPathByID`; refactor `pumpLogs`/`Start`/`Tail`/`ReadSince`/`closeSubscribersLocked`/`Stop`/`wait`/`Restart`/`Delete`; add `buffers`, `totalBufBytes`, `memSampler`, `cfgLogBufferBytes` fields; add `ResolveCap` invocation +6. **`internal/instance/manager_test.go`** — delete `TestEnforceMaxLogSize`; update `TestLogPathByID` (drop or refactor — see below); add `TestPumpLogsWritesToBuffer` and `TestCloseSubscribersDropsBuffer` +7. **`internal/store/state.go`** — remove `LogPath` field +8. **`internal/config/global.go`** — add `LogBufferBytes` field +9. **`internal/app/app.go:131`** — set `LogBufferBytes` on Manager literal (read from `config.Load()`) +10. **`internal/cli/cli.go:269`** — same +11. **`internal/app/app.go` (startup)** — after `instanceMgr.ReconcileRunningOnStartup()`, call `n, err := instanceMgr.PurgeOrphanLogFiles()` and log the count. Idempotent. Runs once per daemon boot. This reclaims the 280MB+ of dead log files left behind by the bug, and keeps the `logs/` directory empty for future runs. +12. **`internal/app/app.go`** (HTTP/MCP handlers at lines 1140 and 1686) — detect `*instance.LogBufferBudgetError` via `errors.As`, route to new `writeLogBufferBudgetErr` helper returning HTTP 503 + structured JSON body. +13. **`internal/ui/static/index.html`** — fetch error handler for `POST /api/instances` detects 503 + `error:"log_buffer_budget_exceeded"` and shows a modal/popup with the message and limit numbers. New instance is rejected (no tab created) since the request itself failed. + +Test updates: +- `TestLogPathByID` (manager_test.go:31) — drop the test (helper being deleted). If you want a regression guard, replace with `TestRingBuffer_LookupByID` covering the new `m.buffers[id]` map access pattern. + +## Verification + +1. **Build**: `go build ./...` +2. **Unit tests**: `go test -race ./internal/instance/... ./internal/store/... ./internal/config/...` +3. **Integration**: `go test ./internal/instance/... -run TestStartStop` should pass (already exists, just confirm we didn't break Start/Stop lifecycle) +4. **Smoke test (manual)**: + - Start `mw` on this machine: process should boot normally; `ls ~/Library/Application\ Support/myworktree/*/logs/` should NOT get any new `.log` files for new instances (confirm no per-instance log files are created) + - Open a worktree tab, run an OpenCode TUI session that produces heavy PTY output + - Watch `iostat -d 1` or Activity Monitor disk writes — write rate on the data dir should be **zero** after startup (only `state.json` writes from save paths) + - Watch RSS of the `mw` process — should grow up to ~32MB per active instance, stabilize, not grow unboundedly + - Open WS endpoint (`/api/instances/log?since=N`): SSE/WS should still stream and tail-back correctly + - Call `instance_log_tail` MCP tool: should return last N bytes from memory +5. **Sizing test (manual)**: start 8+ concurrent instances on this machine; verify total RSS stays within 25% of system RAM (a 16GB machine → cap at ~4GB total) +6. **Backward state compat**: drop the existing `state.json` from a previous session (or copy) and restart — JSON decoder should ignore the missing `log_path` field cleanly +7. **Startup cleanup (manual)**: with the current 280MB of `.log` files still on disk, start the new `mw` binary; verify daemon log line `purge: removed N orphan log files` and that `du -sh ~/Library/Application\ Support/myworktree/*/logs/` reports near-zero (only files written during the current session's test, which there should be none of). +8. **Budget exceeded UI popup (manual)**: set `LogBufferBytes` in `~/Library/Application Support/myworktree/auth.json` (or wherever config lives — verify path) to a tiny value like `1048576` (1MB). Open 3+ worktree tabs so total demand exceeds 1MB × N. Attempt to open a 4th tab via UI; confirm a modal/popup appears with text like "Insufficient memory to start new instance... used: X bytes, limit: Y bytes..." and that no new tab is created. The popup should have a Close button. + +## Risks & Open Items + +- **OpenCode TUI redraw storms** can chew through 32MB fast. 256MB ceiling is the safety net; if a single TUI instance regularly needs more than 32MB, the user can set `LogBufferBytes` in config. Document in CHANGELOG. +- **SSE 1-second polling** relies on `ReadSince(id, cursor, ...)` returning `("", cursor, nil)` when no new data — verified by reading `app.go:1521-1541`. The ring buffer must preserve this: when `since >= head`, return empty and `since` unchanged. Implementation note in `logbuf.go`. +- **Old log files on disk** are dead artifacts left behind. Cleanup is handled by `Manager.PurgeOrphanLogFiles()` called once at daemon startup (see Files to Modify #11). No need for a follow-up. +- **`PurgeOrphanLogFiles` test**: `TestPurgeOrphanLogFiles` in `manager_test.go` — pre-create a `logs/` dir with several `.log` files and one non-`.log` file; call method; assert only `.log` files removed and count correct; assert idempotent on second call (count=0). +- **Budget error path test**: simulate memory pressure by setting `cfgLogBufferBytes` very low + multiple running instances; assert `Start` returns `*LogBufferBudgetError` with correct `UsedBytes` / `LimitBytes`; assert HTTP handler returns 503 + correct JSON body. +- **UI popup test**: `internal/portal/portal_test.go` already covers portal embedding; add a smoke test for `index.html` (the instance-creation UI) if the existing test infra supports it. Otherwise manual verification: open dev tools, hit `/api/instances` with low budget, confirm popup appears with correct numbers. +- **Memory accounting** uses `atomic.Int64` for `totalBufBytes` so it doesn't contend with `stateMu`. Lock ordering: `stateMu` is acquired in `Start` after computing the cap but before mutating `m.buffers`; `totalBufBytes` updates are lock-free. +- **Tests for `pumpLogs` integration with PTY**: existing `TestStartStopIntegration` (manager_integration_test.go:14) only checks Start/Stop. Optional add: a `TestPumpLogsBroadcastAndBuffer` that spawns `echo hello && sleep 0.1` and asserts both the WS subscriber channel and a direct `Tail` read see `"hello"`. +- **`gopsutil/v4/mem.VirtualMemory()` in containerized envs**: may report cgroup-limited memory rather than host. Acceptable — caps are bounded, `MaxTotalFraction=0.25` is conservative. +- **CHANGELOG note** required: removing `log_path` field from `state.json` is a breaking schema change for any external tooling reading state files. diff --git a/internal/app/app.go b/internal/app/app.go index 7026241..ab42ce4 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -128,11 +128,13 @@ func New(cfg Config, logger *log.Logger) (*Server, error) { WorktreesDir: cfg.WorktreesDir, Store: st, } + globalCfg, _ := config.Load() instanceMgr := &instance.Manager{ - DataDir: dataDir, - Root: root, - Store: st, - Logger: logger, + DataDir: dataDir, + Root: root, + Store: st, + Logger: logger, + LogBufferBytes: globalCfg.LogBufferBytes, } mux := http.NewServeMux() @@ -332,6 +334,11 @@ func (s *Server) Start() (string, error) { } else if n > 0 { s.logger.Printf("reconciled %d stale running instances to stopped", n) } + if n, err := s.instanceMgr.PurgeOrphanLogFiles(); err != nil { + s.logger.Printf("purge orphan log files failed: %v", err) + } else if n > 0 { + s.logger.Printf("purged %d orphan log files", n) + } listenAddr, err := resolveRepoListenAddr(s.cfg.ListenAddr, s.dataDir, s.logger) if err != nil { @@ -1150,6 +1157,11 @@ func (s *Server) handleInstances(w http.ResponseWriter, r *http.Request) { Name: req.Name, }) if err != nil { + var budgetErr *instance.LogBufferBudgetError + if errors.As(err, &budgetErr) { + writeLogBufferBudgetErr(w, budgetErr) + return + } writeErr(w, http.StatusBadRequest, err) return } @@ -1690,6 +1702,11 @@ func (s *Server) handleMCPCall(w http.ResponseWriter, r *http.Request) { Name: args.Name, }) if err != nil { + var budgetErr *instance.LogBufferBudgetError + if errors.As(err, &budgetErr) { + writeLogBufferBudgetErr(w, budgetErr) + return + } writeErr(w, http.StatusBadRequest, err) return } @@ -1858,6 +1875,41 @@ func writeErr(w http.ResponseWriter, status int, err error) { writeJSON(w, status, map[string]string{"error": err.Error()}) } +// writeLogBufferBudgetErr writes a structured 503 response when a new instance +// was rejected for exceeding the global log buffer budget. The "error" code +// "log_buffer_budget_exceeded" is part of the API contract — the dashboard +// matches on it to display a popup instead of a generic error. +func writeLogBufferBudgetErr(w http.ResponseWriter, err *instance.LogBufferBudgetError) { + body := map[string]any{ + "error": "log_buffer_budget_exceeded", + "message": fmt.Sprintf("Insufficient memory to start new instance: used %s, limit %s.", formatBytes(err.UsedBytes), formatBytes(err.LimitBytes)), + "used_bytes": err.UsedBytes, + "limit_bytes": err.LimitBytes, + "system_bytes": err.SystemBytes, + "hint": "Close other instances, raise LogBufferBytes in auth.json, or reduce concurrent tabs.", + } + w.Header().Set("Retry-After", "0") + writeJSON(w, http.StatusServiceUnavailable, body) +} + +func formatBytes(b int64) string { + const ( + kb = 1 << 10 + mb = 1 << 20 + gb = 1 << 30 + ) + switch { + case b >= gb: + return fmt.Sprintf("%.2f GB", float64(b)/float64(gb)) + case b >= mb: + return fmt.Sprintf("%.2f MB", float64(b)/float64(mb)) + case b >= kb: + return fmt.Sprintf("%.2f KB", float64(b)/float64(kb)) + default: + return fmt.Sprintf("%d B", b) + } +} + func parseInt64Default(s string, def int64) int64 { s = strings.TrimSpace(s) if s == "" { diff --git a/internal/app/app_test.go b/internal/app/app_test.go index ca9b1ad..499ede7 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -13,6 +13,7 @@ import ( "time" "myworktree/internal/config" + "myworktree/internal/instance" "myworktree/internal/store" ) @@ -1084,3 +1085,60 @@ func TestIsValidRedirectPath(t *testing.T) { }) } } + +func TestWriteLogBufferBudgetErr(t *testing.T) { + w := httptest.NewRecorder() + budgetErr := &instance.LogBufferBudgetError{ + UsedBytes: 100 * 1024 * 1024, + LimitBytes: 50 * 1024 * 1024, + SystemBytes: 16 * 1024 * 1024 * 1024, + } + writeLogBufferBudgetErr(w, budgetErr) + + if w.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503", w.Code) + } + if got := w.Header().Get("Retry-After"); got != "0" { + t.Fatalf("Retry-After = %q, want \"0\"", got) + } + var body map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("body not JSON: %v: %s", err, w.Body.String()) + } + if body["error"] != "log_buffer_budget_exceeded" { + t.Fatalf("error field = %v, want log_buffer_budget_exceeded", body["error"]) + } + for _, k := range []string{"message", "used_bytes", "limit_bytes", "system_bytes", "hint"} { + if _, ok := body[k]; !ok { + t.Fatalf("body missing field %q", k) + } + } + if int64(body["used_bytes"].(float64)) != budgetErr.UsedBytes { + t.Fatalf("used_bytes mismatch") + } +} + +func TestFormatBytes(t *testing.T) { + cases := []struct { + in int64 + want string + }{ + {512, "512 B"}, + {2048, "2.00 KB"}, + {5 * 1024 * 1024, "5.00 MB"}, + {2 * 1024 * 1024 * 1024, "2.00 GB"}, + } + for _, c := range cases { + if got := formatBytes(c.in); got != c.want { + t.Errorf("formatBytes(%d) = %q, want %q", c.in, got, c.want) + } + } +} + +// fakeMemSaturated and the TestHandle{Instances,MCPCall}_BudgetExceededReturns503 +// tests previously lived here. They were removed when SetMemSampler / +// SetTotalBufferBytesForTest were relocated to the instance package's +// _test.go helpers (see manager_export_test.go), so production code can no +// longer reach them. Equivalent coverage of Manager.Start's budget error +// path now lives in instance/manager_budget_test.go; the HTTP response shape +// is covered by TestWriteLogBufferBudgetErr above. diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 7f74703..7f77896 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -266,11 +266,13 @@ func instanceCmd(logger *log.Logger, args []string) error { if err != nil { return err } + cfg, _ := config.Load() mgr := &instance.Manager{ - DataDir: dataDir, - Root: root, - Store: store.FileStore{Path: filepath.Join(dataDir, "state.json")}, - Logger: logger, + DataDir: dataDir, + Root: root, + Store: store.FileStore{Path: filepath.Join(dataDir, "state.json")}, + Logger: logger, + LogBufferBytes: cfg.LogBufferBytes, } switch args[0] { diff --git a/internal/config/global.go b/internal/config/global.go index a4a17c7..184eb6e 100644 --- a/internal/config/global.go +++ b/internal/config/global.go @@ -10,6 +10,11 @@ import ( type GlobalConfig struct { AuthToken string `json:"auth_token"` + + // LogBufferBytes is the per-instance ring buffer cap for captured PTY + // output. 0 means "use adaptive sizing" (clamp(available/16, 16MB, 256MB)). + // When set, the value is clamped to [16MB, 256MB]. + LogBufferBytes int64 `json:"log_buffer_bytes,omitempty"` } func (c *GlobalConfig) Copy() *GlobalConfig { diff --git a/internal/instance/logbuf.go b/internal/instance/logbuf.go new file mode 100644 index 0000000..ae2f7aa --- /dev/null +++ b/internal/instance/logbuf.go @@ -0,0 +1,206 @@ +package instance + +import ( + "errors" + "sync" +) + +// RingBuffer is a bounded, in-memory ring buffer for capturing PTY output. +// +// Semantics designed for `pumpLogs` + `Tail`/`ReadSince` consumers: +// - Writes append and never block. When `cap` is exceeded, oldest data is +// overwritten FIFO so the buffer holds exactly the last `cap` bytes. +// - `Offset()` returns a monotonic total-bytes-written counter. Callers +// pass this as the `since` cursor to `ReadSince` to fetch only new data. +// - `ReadSince(since, maxBytes)` returns ("", since, nil) when since >= head, +// preserving the SSE 1-second poll-loop contract (no new data → cursor +// unchanged). +// - When since points to data older than the buffer's oldest live byte, the +// read silently clamps to the oldest live byte. +// +// Concurrency: safe for concurrent Write, WriteString, Tail, ReadSince, +// Offset, BytesUsed, and Close. Close is idempotent; after Close, writes are +// dropped, reads return zero values, and Offset returns the last head +// observed. +// +// Hot-path design (heavy TUI workloads): the backing slice is allocated +// eagerly in NewRingBuffer so the first Write does not stall the producer +// goroutine for a 32MB malloc, and WriteString avoids the []byte(s) copy +// that would otherwise occur on every 1024-byte PTY chunk. +type RingBuffer struct { + mu sync.Mutex + cap int64 + data []byte // backing buffer of size cap, allocated eagerly when cap > 0 + head int64 // monotonic total bytes written; never decreases + used int64 // bytes currently held; 0 <= used <= cap (when cap > 0) + closed bool +} + +// NewRingBuffer creates a RingBuffer that holds at most capBytes. +// If capBytes <= 0, the buffer is a no-op (writes succeed without storing). +// The backing slice is allocated eagerly so the first Write is allocation- +// free on the hot path. +func NewRingBuffer(capBytes int64) *RingBuffer { + rb := &RingBuffer{cap: capBytes} + if capBytes > 0 { + rb.data = make([]byte, capBytes) + } + return rb +} + +// Write appends p. If appending would exceed cap, oldest bytes are overwritten +// FIFO. Returns len(p) on success. After Close, returns 0. +func (r *RingBuffer) Write(p []byte) int { + return r.write(p, "") +} + +// WriteString is the string variant of Write. It avoids the []byte(s) copy +// on the hot path: strings can be copied directly into the backing slice +// via `copy(dst, s)`. Use this for chunks produced by `redact.Text` which +// already returns a string. +func (r *RingBuffer) WriteString(s string) int { + return r.write(nil, s) +} + +// write is the shared implementation for Write and WriteString. The string +// path is preferred on the hot path because it avoids an extra allocation. +func (r *RingBuffer) write(p []byte, s string) int { + srcLen := int64(len(p)) + if s != "" { + srcLen = int64(len(s)) + } + if srcLen == 0 { + return 0 + } + r.mu.Lock() + defer r.mu.Unlock() + if r.closed { + return 0 + } + if r.cap > 0 { + srcStart := int64(0) + if srcLen > r.cap { + srcStart = srcLen - r.cap + } + pos := (r.head + srcStart) % r.cap + if p != nil { + writeData := p[srcStart:] + n := copy(r.data[pos:], writeData) + if n < len(writeData) { + copy(r.data, writeData[n:]) + } + } else { + writeData := s[srcStart:] + n := copy(r.data[pos:], writeData) + if n < len(writeData) { + copy(r.data, writeData[n:]) + } + } + } + r.head += srcLen + if r.cap > 0 { + newUsed := r.used + srcLen + if newUsed > r.cap { + r.used = r.cap + } else { + r.used = newUsed + } + } + return int(srcLen) +} + +// Offset returns the total bytes written so far. Monotonic. Used as the +// cursor for subsequent ReadSince calls. +func (r *RingBuffer) Offset() int64 { + r.mu.Lock() + defer r.mu.Unlock() + return r.head +} + +// BytesUsed returns bytes currently held in the buffer. +func (r *RingBuffer) BytesUsed() int64 { + r.mu.Lock() + defer r.mu.Unlock() + return r.used +} + +// Tail returns the last n bytes held in the buffer. Also returns the absolute +// end offset of the returned bytes (= current head). +// +// If n > BytesUsed, returns all available bytes. If n <= 0, returns empty +// string and the current offset. +func (r *RingBuffer) Tail(n int64) (string, int64) { + r.mu.Lock() + defer r.mu.Unlock() + if n <= 0 || r.used == 0 { + return "", r.head + } + if n > r.used { + n = r.used + } + return r.copyRangeLocked(r.head-n, n), r.head +} + +// ReadSince returns up to maxBytes of data starting at absolute offset since. +// Returns ("", since, nil) when since >= head (no new data; cursor preserved). +// Returns the bytes available when since points to data older than the +// buffer's oldest live byte (silently clamps). +func (r *RingBuffer) ReadSince(since int64, maxBytes int64) (string, int64, error) { + if maxBytes < 0 { + return "", since, errors.New("maxBytes must be >= 0") + } + if since < 0 { + since = 0 + } + r.mu.Lock() + defer r.mu.Unlock() + if since >= r.head { + return "", since, nil + } + if maxBytes == 0 { + return "", since, nil + } + start := since + if start < r.head-r.used { + start = r.head - r.used + } + avail := r.head - start + if avail > maxBytes { + avail = maxBytes + } + return r.copyRangeLocked(start, avail), start + avail, nil +} + +// Close drops the backing buffer and prevents further writes. Idempotent. +// Offset still returns the last head value observed before Close. +func (r *RingBuffer) Close() { + r.mu.Lock() + defer r.mu.Unlock() + r.closed = true + r.data = nil + r.used = 0 +} + +// CapBytes returns the configured cap (immutable after construction). +// Used by callers that maintain an external cap accounting total. +func (r *RingBuffer) CapBytes() int64 { + if r == nil { + return 0 + } + return r.cap +} + +// copyRangeLocked copies n bytes starting at absolute offset start into a +// string. Caller must hold r.mu. Returns "" if no backing buffer. +func (r *RingBuffer) copyRangeLocked(start int64, n int64) string { + if r.data == nil || n <= 0 { + return "" + } + buf := make([]byte, n) + pos := start % r.cap + copied := copy(buf, r.data[pos:]) + if int64(copied) < n { + copy(buf[copied:], r.data[:n-int64(copied)]) + } + return string(buf) +} diff --git a/internal/instance/logbuf_test.go b/internal/instance/logbuf_test.go new file mode 100644 index 0000000..e58aa48 --- /dev/null +++ b/internal/instance/logbuf_test.go @@ -0,0 +1,378 @@ +package instance + +import ( + "strings" + "sync" + "testing" +) + +func TestRingBuffer_BasicWriteRead(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("hello ")) + rb.Write([]byte("world")) + if got, _ := rb.Tail(100); got != "hello world" { + t.Fatalf("Tail = %q, want %q", got, "hello world") + } + if rb.BytesUsed() != 11 { + t.Fatalf("BytesUsed = %d, want 11", rb.BytesUsed()) + } + if rb.Offset() != 11 { + t.Fatalf("Offset = %d, want 11", rb.Offset()) + } +} + +func TestRingBuffer_WrapAroundKeepsLastCap(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(10) + rb.Write([]byte("0123456789")) // exactly at cap + if rb.BytesUsed() != 10 { + t.Fatalf("BytesUsed after fill = %d, want 10", rb.BytesUsed()) + } + rb.Write([]byte("ABC")) // pushes 3 oldest out + if rb.BytesUsed() != 10 { + t.Fatalf("BytesUsed after wrap = %d, want 10", rb.BytesUsed()) + } + if rb.Offset() != 13 { + t.Fatalf("Offset after wrap = %d, want 13", rb.Offset()) + } + got, head := rb.Tail(10) + if got != "3456789ABC" { + t.Fatalf("Tail after wrap = %q, want %q", got, "3456789ABC") + } + if head != 13 { + t.Fatalf("Tail head = %d, want 13", head) + } +} + +func TestRingBuffer_OverwriteAcrossWrapBoundary(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(8) + rb.Write([]byte("ABCDEFGH")) // head=8, used=8 + rb.Write([]byte("XYZ")) // wraps: data="XYZDEFGH", head=11, used=8 + got, _ := rb.Tail(8) + if got != "DEFGHXYZ" { + t.Fatalf("Tail = %q, want %q", got, "DEFGHXYZ") + } +} + +func TestRingBuffer_TailPartial(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("hello world")) + if got, _ := rb.Tail(5); got != "world" { + t.Fatalf("Tail(5) = %q, want %q", got, "world") + } + if got, _ := rb.Tail(0); got != "" { + t.Fatalf("Tail(0) = %q, want empty", got) + } + if got, _ := rb.Tail(-1); got != "" { + t.Fatalf("Tail(-1) = %q, want empty", got) + } +} + +func TestRingBuffer_ReadSinceNewData(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("part1")) + cursor := rb.Offset() + rb.Write([]byte("part2")) + got, next, err := rb.ReadSince(cursor, 1024) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "part2" { + t.Fatalf("ReadSince = %q, want %q", got, "part2") + } + if next != rb.Offset() { + t.Fatalf("ReadSince next = %d, want %d", next, rb.Offset()) + } +} + +func TestRingBuffer_ReadSinceNoNewData(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("part1")) + cursor := rb.Offset() + got, next, err := rb.ReadSince(cursor, 1024) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "" { + t.Fatalf("ReadSince no-new = %q, want empty", got) + } + if next != cursor { + t.Fatalf("ReadSince no-new next = %d, want %d (unchanged)", next, cursor) + } +} + +func TestRingBuffer_ReadSinceClampsMaxBytes(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("0123456789")) + got, next, err := rb.ReadSince(0, 5) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "01234" { + t.Fatalf("ReadSince(0, 5) = %q, want %q", got, "01234") + } + if next != 5 { + t.Fatalf("ReadSince next = %d, want 5", next) + } +} + +func TestRingBuffer_ReadSinceClampsToOldestLive(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(5) + rb.Write([]byte("0123456789")) // 5 oldest dropped + got, next, err := rb.ReadSince(0, 1024) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "56789" { + t.Fatalf("ReadSince past-oldest = %q, want %q", got, "56789") + } + if next != 10 { + t.Fatalf("ReadSince next = %d, want 10", next) + } +} + +func TestRingBuffer_ReadSinceNegativeSince(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("abc")) + got, next, err := rb.ReadSince(-5, 100) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "abc" { + t.Fatalf("ReadSince(-5) = %q, want %q", got, "abc") + } + if next != 3 { + t.Fatalf("ReadSince(-5) next = %d, want 3", next) + } +} + +func TestRingBuffer_NoOpWhenCapZero(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(0) + rb.Write([]byte("ignored")) + if rb.BytesUsed() != 0 { + t.Fatalf("BytesUsed = %d, want 0", rb.BytesUsed()) + } + if rb.Offset() != 7 { + t.Fatalf("Offset = %d, want 7 (head advances for cursor monotonicity)", rb.Offset()) + } + if got, _ := rb.Tail(100); got != "" { + t.Fatalf("Tail = %q, want empty", got) + } + // ReadSince clamps to head (since used=0) so cursor advances to head. + if got, next, _ := rb.ReadSince(0, 100); got != "" || next != 7 { + t.Fatalf("ReadSince past-oldest = (%q, %d), want (\"\", 7)", got, next) + } +} + +func TestRingBuffer_CloseIdempotent(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("data")) + rb.Close() + rb.Close() // idempotent + if n := rb.Write([]byte("more")); n != 0 { + t.Fatalf("Write after Close returned %d, want 0", n) + } + if rb.BytesUsed() != 0 { + t.Fatalf("BytesUsed after Close = %d, want 0", rb.BytesUsed()) + } + if got, _ := rb.Tail(100); got != "" { + t.Fatalf("Tail after Close = %q, want empty", got) + } + if rb.Offset() != 4 { + t.Fatalf("Offset after Close = %d, want 4 (head preserved for cursor)", rb.Offset()) + } +} + +func TestRingBuffer_ConcurrentWritersReaders(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(4096) + var wg sync.WaitGroup + const writers = 4 + for w := 0; w < writers; w++ { + wg.Add(1) + go func(id int) { + defer wg.Done() + for i := 0; i < 50; i++ { + chunk := []byte{byte(id), byte(i)} + rb.Write(chunk) + } + }(w) + } + for r := 0; r < 2; r++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + _ = rb.BytesUsed() + _ = rb.Offset() + _, _ = rb.Tail(64) + _, _, _ = rb.ReadSince(0, 64) + } + }() + } + wg.Wait() + if rb.BytesUsed() > 4096 { + t.Fatalf("BytesUsed %d exceeds cap 4096", rb.BytesUsed()) + } +} + +func TestRingBuffer_LargeWriteAcrossWrap(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(16) + rb.Write([]byte("AAAAAAAAAAAAAAAA")) // 16 chars, used=16 + rb.Write([]byte("BBBBBBBBBBBBBBBBBBBBBBBB")) // 24 chars, only last 16 kept + got, _ := rb.Tail(16) + if got != "BBBBBBBBBBBBBBBB" { + t.Fatalf("Tail = %q, want %q", got, "BBBBBBBBBBBBBBBB") + } +} + +func TestRingBuffer_LargeWriteExceedsCap(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(8) + rb.Write([]byte("01234567890123456789")) // 20 bytes, single write > cap + got, head := rb.Tail(8) + if got != "23456789" { + t.Fatalf("Tail = %q, want %q", got, "23456789") + } + if head != 20 { + t.Fatalf("head = %d, want 20", head) + } + if rb.BytesUsed() != 8 { + t.Fatalf("BytesUsed = %d, want 8", rb.BytesUsed()) + } +} + +func TestRingBuffer_ReadSinceAtCursorEqualsHead(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("data")) + got, next, err := rb.ReadSince(rb.Offset(), 100) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if got != "" { + t.Fatalf("got = %q, want empty", got) + } + if next != rb.Offset() { + t.Fatalf("next = %d, want %d (cursor unchanged)", next, rb.Offset()) + } +} + +func TestRingBuffer_TailLargerThanUsed(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + rb.Write([]byte("small")) + got, _ := rb.Tail(1024) + if got != "small" { + t.Fatalf("Tail(1024) = %q, want %q", got, "small") + } +} + +func TestRingBuffer_EmptyBufferReads(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(1024) + if got, _ := rb.Tail(100); got != "" { + t.Fatalf("empty Tail = %q, want empty", got) + } + if got, next, err := rb.ReadSince(0, 100); err != nil || got != "" || next != 0 { + t.Fatalf("empty ReadSince = (%q, %d, %v), want (\"\", 0, nil)", got, next, err) + } +} + +// Smoke test that the buffer can hold typical PTY terminal control sequences. +func TestRingBuffer_HandlesANSIEscapes(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(4096) + seq := "\x1b[?25l\x1b[2J\x1b[HHello, World!\x1b[0m" + rb.WriteString(seq) + got, _ := rb.Tail(1024) + if !strings.Contains(got, "Hello, World!") { + t.Fatalf("Tail missing payload: %q", got) + } +} + +func TestRingBuffer_BackingAllocatedEagerly(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(4096) + // First Write must not allocate the backing slice; CapBytes reflects + // config but the slice must already exist for the hot path. + // We assert this indirectly: if the backing slice was lazy, the first + // Write would contend with itself; here we just check that Tail works + // after a single small write without panic. + rb.WriteString("x") + if got, _ := rb.Tail(1); got != "x" { + t.Fatalf("Tail after eager alloc = %q, want %q", got, "x") + } +} + +func TestRingBuffer_WriteString_MatchesWrite(t *testing.T) { + t.Parallel() + a := NewRingBuffer(64) + b := NewRingBuffer(64) + a.Write([]byte("0123456789")) + b.WriteString("0123456789") + gotA, headA := a.Tail(64) + gotB, headB := b.Tail(64) + if gotA != gotB { + t.Fatalf("WriteString diverges from Write: %q vs %q", gotA, gotB) + } + if headA != headB { + t.Fatalf("head diverges: %d vs %d", headA, headB) + } +} + +func TestRingBuffer_WriteString_WrapAround(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(8) + rb.WriteString("ABCDEFGH") + rb.WriteString("XYZ") + got, _ := rb.Tail(8) + if got != "DEFGHXYZ" { + t.Fatalf("WriteString wrap = %q, want %q", got, "DEFGHXYZ") + } +} + +func TestRingBuffer_WriteString_LargeExceedsCap(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(4) + rb.WriteString("abcdefghij") + got, head := rb.Tail(4) + if got != "ghij" { + t.Fatalf("WriteString large = %q, want %q", got, "ghij") + } + if head != 10 { + t.Fatalf("head = %d, want 10", head) + } +} + +func TestRingBuffer_WriteString_Empty(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(64) + if n := rb.WriteString(""); n != 0 { + t.Fatalf("WriteString(\"\") = %d, want 0", n) + } + if rb.Offset() != 0 { + t.Fatalf("Offset after empty WriteString = %d, want 0", rb.Offset()) + } +} + +func TestRingBuffer_WriteString_AfterClose(t *testing.T) { + t.Parallel() + rb := NewRingBuffer(64) + rb.WriteString("data") + rb.Close() + if n := rb.WriteString("more"); n != 0 { + t.Fatalf("WriteString after Close = %d, want 0", n) + } +} diff --git a/internal/instance/manager.go b/internal/instance/manager.go index ac42708..66302e3 100644 --- a/internal/instance/manager.go +++ b/internal/instance/manager.go @@ -12,6 +12,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "syscall" "time" @@ -22,8 +23,7 @@ import ( ) const ( - maxLogBytes int64 = 10 * 1024 * 1024 - MainWorktreeID = "__main__" + MainWorktreeID = "__main__" ) // ErrInstanceNotFound is returned when an instance ID does not match any known instance. @@ -35,6 +35,13 @@ type Manager struct { Store store.FileStore Logger *log.Logger + // LogBufferBytes is the per-instance ring buffer cap from config. + // 0 means "use adaptive sizing". + LogBufferBytes int64 + + // memSampler queries system memory. Defaults to gopsutilMem if nil. + memSampler MemSampler + stateMu sync.Mutex mu sync.Mutex running map[string]*exec.Cmd @@ -42,7 +49,26 @@ type Manager struct { ptys map[string]*os.File subscribers map[string]map[chan string]struct{} conns map[string]string // instance ID -> connection type ("websocket"/"sse"/"") - connsMu sync.Mutex + + // buffers holds the per-instance ring buffer pointers for captured PTY + // output. Each value is an atomic.Pointer so the hot path in pumpLogs + // can read the current buffer with a single atomic load (no stateMu + // acquisition per 1024-byte PTY chunk). Map mutations (insert/delete) + // still require stateMu. + buffers map[string]*atomic.Pointer[RingBuffer] + + // totalBufBytes is the sum of all live buffer caps. Atomic so it doesn't + // contend with stateMu. + totalBufBytes atomic.Int64 + + // memSampleMu guards the 60s memory-sample cache below. The cache is used + // only by the adaptive-cap path in Start(); the budget check always reads + // a fresh sample to avoid rejecting requests based on stale data. + memSampleMu sync.Mutex + lastMemSampleAt time.Time + lastMemAvailable int64 + + connsMu sync.Mutex } type StartInput struct { @@ -83,6 +109,87 @@ func (m *Manager) ReconcileRunningOnStartup() (int, error) { return changed, nil } +// PurgeOrphanLogFiles removes every *.log file in DataDir/logs/. After the +// in-memory ring buffer refactor, no live state references these files — they +// are dead artifacts left over from the old per-instance log file. Runs once +// at daemon startup to reclaim disk space. +// +// Returns the number of files removed. A missing logs/ directory is not an +// error (returns 0). +func (m *Manager) PurgeOrphanLogFiles() (int, error) { + if strings.TrimSpace(m.DataDir) == "" { + return 0, nil + } + logDir := filepath.Join(m.DataDir, "logs") + entries, err := os.ReadDir(logDir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return 0, nil + } + return 0, err + } + removed := 0 + var failed []string + for _, e := range entries { + if e.IsDir() { + continue + } + if !strings.HasSuffix(e.Name(), ".log") { + continue + } + full := filepath.Join(logDir, e.Name()) + if err := os.Remove(full); err != nil { + failed = append(failed, fmt.Sprintf("%s: %v", full, err)) + continue + } + removed++ + } + if m.Logger != nil { + if removed > 0 { + m.Logger.Printf("purge: removed %d orphan log files from %s", removed, logDir) + } + for _, f := range failed { + m.Logger.Printf("purge: %s", f) + } + } + return removed, nil +} + +// memSampleCacheTTL is how long the adaptive-cap path reuses a single +// mem.VirtualMemory() result. The cap is a starting point that's re-evaluated +// on every instance start, so staleness here is harmless. +const memSampleCacheTTL = 60 * time.Second + +// sampledAvailable returns a mem.Available result, refreshing at most once +// per memSampleCacheTTL. Used by the adaptive-cap path in Start(). On a cache +// miss it queries the configured sampler (or gopsutilMem as the default) and +// stores the result. +func (m *Manager) sampledAvailable(now time.Time) (int64, error) { + m.memSampleMu.Lock() + defer m.memSampleMu.Unlock() + if !m.lastMemSampleAt.IsZero() && now.Sub(m.lastMemSampleAt) < memSampleCacheTTL { + return m.lastMemAvailable, nil + } + sampler := m.memSampler + if sampler == nil { + sampler = gopsutilMem{} + } + vm, err := sampler.VirtualMemory() + if err != nil { + return 0, err + } + avail := vm.Available + if avail <= 0 { + avail = vm.Total - vm.Used + } + if avail < 0 { + avail = 0 + } + m.lastMemAvailable = avail + m.lastMemSampleAt = now + return avail, nil +} + func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { m.mu.Lock() if m.running == nil { @@ -115,11 +222,9 @@ func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { var wtName string if in.Root != "" { - // Main repo: use Root directly. wtPath = in.Root wtName = filepath.Base(filepath.Clean(in.Root)) } else { - // Normal worktree lookup. var wt *store.ManagedWorktree for i := range st.Worktrees { if st.Worktrees[i].ID == in.WorktreeID { @@ -177,21 +282,37 @@ func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { if instName == "" { instName = effectiveTagID } - logDir := filepath.Join(m.DataDir, "logs") - if err := os.MkdirAll(logDir, 0o755); err != nil { - return store.ManagedInstance{}, err - } - logPath := filepath.Join(logDir, id+".log") - logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) - if err != nil { - return store.ManagedInstance{}, err - } cwd := wtPath if strings.TrimSpace(cwdRel) != "" && cwdRel != "." { cwd = filepath.Join(wtPath, cwdRel) } + // Resolve buffer cap with budget enforcement. Done BEFORE exec.Command / + // pty.Start so a budget-exceeded error short-circuits the heavy work + // (process spawn, PTY allocation) and leaves no resources to clean up. + // + // The cap path uses a 60s-cached mem.Available sample (cheap to query + // repeatedly under batch starts); the budget check inside resolveCap + // always queries the sampler live so budget decisions never use stale + // data. + sampler := m.memSampler + if sampler == nil { + sampler = gopsutilMem{} + } + var capAvail int64 = -1 + if m.LogBufferBytes <= 0 { + // Adaptive path: use the cached Available sample. Errors here fall + // back to live sampling inside resolveCap → adaptiveCapFromAvailable. + if avail, sErr := m.sampledAvailable(time.Now()); sErr == nil { + capAvail = avail + } + } + capBytes, err := resolveCapFromAvailable(m.LogBufferBytes, sampler, m.totalBufBytes.Load(), capAvail) + if err != nil { + return store.ManagedInstance{}, err + } + cmd := exec.Command("zsh", "-f", "-i") cmd.Dir = cwd cmd.Env = os.Environ() @@ -204,22 +325,21 @@ func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { pre.Dir = cwd pre.Env = cmd.Env if out, err := pre.CombinedOutput(); err != nil { - _ = logFile.Close() - _ = os.WriteFile(logPath, []byte(redact.Text(string(out))), 0o600) return store.ManagedInstance{}, fmt.Errorf("preStart failed: %w: %s", err, strings.TrimSpace(string(out))) } } ptmx, err := pty.Start(cmd) if err != nil { - _ = logFile.Close() return store.ManagedInstance{}, err } + buf := NewRingBuffer(capBytes) + m.totalBufBytes.Add(capBytes) now := time.Now().UTC().Format(time.RFC3339) inst := store.ManagedInstance{ ID: id, - WorktreeID: in.WorktreeID, // MainWorktreeID for main repo, or a real worktree ID + WorktreeID: in.WorktreeID, WorktreeName: wtName, TagID: effectiveTagID, Name: instName, @@ -228,26 +348,38 @@ func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { Env: sanitizedEnv(env), PID: cmd.Process.Pid, Status: "running", - LogPath: logPath, CreatedAt: now, } m.stateMu.Lock() - st2, err := m.Store.Load() - if err == nil { - st2.Instances = append(st2.Instances, inst) - if st2.TabOrder == nil { - st2.TabOrder = make(map[string][]string) - } - st2.TabOrder[in.WorktreeID] = append(st2.TabOrder[in.WorktreeID], inst.ID) - err = m.Store.SaveWithVersion(st2, st2.Version) + st2, loadErr := m.Store.Load() + if loadErr != nil { + m.stateMu.Unlock() + _ = cmd.Process.Kill() + _ = ptmx.Close() + buf.Close() + m.totalBufBytes.Add(-capBytes) + return store.ManagedInstance{}, loadErr } - m.stateMu.Unlock() - if err != nil { + st2.Instances = append(st2.Instances, inst) + if st2.TabOrder == nil { + st2.TabOrder = make(map[string][]string) + } + st2.TabOrder[in.WorktreeID] = append(st2.TabOrder[in.WorktreeID], inst.ID) + if err := m.Store.SaveWithVersion(st2, st2.Version); err != nil { + m.stateMu.Unlock() _ = cmd.Process.Kill() _ = ptmx.Close() - _ = logFile.Close() + buf.Close() + m.totalBufBytes.Add(-capBytes) return store.ManagedInstance{}, err } + if m.buffers == nil { + m.buffers = map[string]*atomic.Pointer[RingBuffer]{} + } + p := &atomic.Pointer[RingBuffer]{} + p.Store(buf) + m.buffers[id] = p + m.stateMu.Unlock() m.mu.Lock() m.running[id] = cmd @@ -262,7 +394,7 @@ func (m *Manager) Start(in StartInput) (store.ManagedInstance, error) { }() } - go m.pumpLogs(id, ptmx, ptmx, logFile, logPath) + go m.pumpLogs(id, ptmx) go m.wait(id, cmd) return inst, nil } @@ -276,7 +408,6 @@ func (m *Manager) List() ([]store.ManagedInstance, error) { } // UpdateName updates the display name of an existing instance. -// Returns the updated instance on success. func (m *Manager) UpdateName(id string, name string) (store.ManagedInstance, error) { id = strings.TrimSpace(id) if id == "" { @@ -306,8 +437,6 @@ func (m *Manager) UpdateName(id string, name string) (store.ManagedInstance, err } // ReorderInstances sets the tab order for a specific worktree. -// orderIDs must contain all instance IDs for the given worktree. -// expectedVersion is the version the caller observed; save is rejected if the state has changed. func (m *Manager) ReorderInstances(worktreeID string, orderIDs []string, expectedVersion int64) error { worktreeID = strings.TrimSpace(worktreeID) if worktreeID == "" { @@ -321,13 +450,11 @@ func (m *Manager) ReorderInstances(worktreeID string, orderIDs []string, expecte return err } - // Build maps for lookup. idSet := make(map[string]bool, len(orderIDs)) for _, id := range orderIDs { idSet[id] = true } - // Validate: every instance for this worktree must be in orderIDs. for _, inst := range st.Instances { if inst.WorktreeID == worktreeID { if !idSet[inst.ID] { @@ -336,7 +463,6 @@ func (m *Manager) ReorderInstances(worktreeID string, orderIDs []string, expecte } } - // Validate: every ID in orderIDs must belong to this worktree. for _, id := range orderIDs { found := false for _, inst := range st.Instances { @@ -350,24 +476,20 @@ func (m *Manager) ReorderInstances(worktreeID string, orderIDs []string, expecte } } - // Update TabOrder index. if st.TabOrder == nil { st.TabOrder = make(map[string][]string) } st.TabOrder[worktreeID] = orderIDs - // Rebuild Instances slice to match desired order. idToInst := make(map[string]store.ManagedInstance, len(st.Instances)) for _, inst := range st.Instances { idToInst[inst.ID] = inst } newOrder := make([]store.ManagedInstance, 0, len(st.Instances)) - // Instances from this worktree, in the new order. for _, id := range orderIDs { newOrder = append(newOrder, idToInst[id]) } - // Append instances from other worktrees (unchanged). for _, inst := range st.Instances { if inst.WorktreeID != worktreeID { newOrder = append(newOrder, inst) @@ -399,7 +521,6 @@ func (m *Manager) Stop(id string) error { return fmt.Errorf("unknown instance id: %s", id) } if inst.Status != "running" { - // Idempotent: stopping an already-exited instance is a no-op. return nil } @@ -408,7 +529,6 @@ func (m *Manager) Stop(id string) error { in := m.inputs[id] m.mu.Unlock() - // If server restarted, cmd may be missing; best-effort signal by PID/process-group. if (cmd == nil || cmd.Process == nil) && inst.PID > 0 { terminatePID(inst.PID, syscall.SIGTERM) go func(pid int) { @@ -454,16 +574,12 @@ func (m *Manager) SendInput(id string, input string) error { return fmt.Errorf("instance input unavailable: %s", id) } - // Process control characters (Ctrl+C/Z/\) - // These send signals to the process group for immediate termination. - // We also write the character to stdin as a fallback, matching real terminal behavior. for _, ch := range input { switch ch { case 0x03: if cmd != nil && cmd.Process != nil { _ = terminatePID(cmd.Process.Pid, syscall.SIGINT) } - // Also write to stdin as fallback (original behavior in v0.1.0) if _, err := io.WriteString(in, string(ch)); err != nil { return err } @@ -563,11 +679,9 @@ func (m *Manager) Restart(id string) (store.ManagedInstance, error) { return newInst, nil } oldIdx := -1 - var oldLogPath string for i := range st2.Instances { if st2.Instances[i].ID == id { oldIdx = i - oldLogPath = st2.Instances[i].LogPath break } } @@ -583,9 +697,6 @@ func (m *Manager) Restart(id string) (store.ManagedInstance, error) { _ = m.Store.SaveWithVersion(st2, st2.Version) m.stateMu.Unlock() - if strings.TrimSpace(oldLogPath) != "" { - _ = os.Remove(oldLogPath) - } m.mu.Lock() delete(m.running, id) delete(m.inputs, id) @@ -596,6 +707,10 @@ func (m *Manager) Restart(id string) (store.ManagedInstance, error) { m.closeSubscribersLocked(id) m.mu.Unlock() + m.stateMu.Lock() + m.dropBufferLocked(id) + m.stateMu.Unlock() + return newInst, nil } @@ -608,7 +723,14 @@ func (m *Manager) SubscribeOutput(id string) (<-chan string, func(), error) { if err != nil { return nil, nil, err } - if logPathByID(st, id) == "" { + found := false + for _, it := range st.Instances { + if it.ID == id { + found = true + break + } + } + if !found { return nil, nil, fmt.Errorf("unknown instance id: %s", id) } @@ -651,14 +773,12 @@ func (m *Manager) Delete(id string) error { return err } idx := -1 - var logPath string for i := range st.Instances { if st.Instances[i].ID == id { if st.Instances[i].Status == "running" { return fmt.Errorf("instance is running: %s", id) } idx = i - logPath = st.Instances[i].LogPath break } } @@ -669,9 +789,7 @@ func (m *Manager) Delete(id string) error { if err := m.Store.SaveWithVersion(st, st.Version); err != nil { return err } - if strings.TrimSpace(logPath) != "" { - _ = os.Remove(logPath) - } + m.dropBufferLocked(id) m.mu.Lock() delete(m.running, id) delete(m.inputs, id) @@ -688,37 +806,18 @@ func (m *Manager) Tail(id string, n int64) (string, error) { if n <= 0 { n = 4096 } - st, err := m.Store.Load() - if err != nil { - return "", err - } - path := logPathByID(st, id) - if path == "" { - return "", fmt.Errorf("unknown instance id: %s", id) - } - f, err := os.Open(path) - if err != nil { - return "", err - } - defer f.Close() - - fi, err := f.Stat() - if err != nil { - return "", err - } - size := fi.Size() - start := size - n - if start < 0 { - start = 0 - } - if _, err := f.Seek(start, io.SeekStart); err != nil { - return "", err + m.stateMu.Lock() + p, ok := m.buffers[id] + m.stateMu.Unlock() + if !ok || p == nil { + return "", nil } - b, err := io.ReadAll(f) - if err != nil { - return "", err + rb := p.Load() + if rb == nil { + return "", nil } - return string(b), nil + body, _ := rb.Tail(n) + return body, nil } // ReadSince returns log content starting at byte offset "since" (inclusive), @@ -730,40 +829,17 @@ func (m *Manager) ReadSince(id string, since int64, maxBytes int64) (string, int if maxBytes <= 0 { maxBytes = 64 * 1024 } - st, err := m.Store.Load() - if err != nil { - return "", since, err - } - path := logPathByID(st, id) - if path == "" { - return "", since, fmt.Errorf("unknown instance id: %s", id) - } - f, err := os.Open(path) - if err != nil { - return "", since, err - } - defer f.Close() - - fi, err := f.Stat() - if err != nil { - return "", since, err - } - size := fi.Size() - start := since - if start > size { - start = size - } - if size-start > maxBytes { - start = size - maxBytes - } - if _, err := f.Seek(start, io.SeekStart); err != nil { - return "", since, err + m.stateMu.Lock() + p, ok := m.buffers[id] + m.stateMu.Unlock() + if !ok || p == nil { + return "", since, nil } - b, err := io.ReadAll(f) - if err != nil { - return "", since, err + rb := p.Load() + if rb == nil { + return "", since, nil } - return string(b), start + int64(len(b)), nil + return rb.ReadSince(since, maxBytes) } func (m *Manager) wait(id string, cmd *exec.Cmd) { @@ -781,6 +857,7 @@ func (m *Manager) wait(id string, cmd *exec.Cmd) { m.stateMu.Lock() defer m.stateMu.Unlock() + m.dropBufferLocked(id) st, loadErr := m.Store.Load() if loadErr != nil { return @@ -799,35 +876,46 @@ func (m *Manager) wait(id string, cmd *exec.Cmd) { _ = m.Store.SaveWithVersion(st, st.Version) } -func (m *Manager) pumpLogs(id string, stdout io.Reader, stderr io.Reader, out *os.File, logPath string) { - defer func() { _ = out.Close() }() - var wg sync.WaitGroup - wg.Add(2) - write := func(r io.Reader) { - defer wg.Done() - buf := make([]byte, 1024) - for { - n, err := r.Read(buf) - if n > 0 { - chunk := redact.Text(string(buf[:n])) - _, _ = out.WriteString(chunk) - _ = enforceMaxLogSize(logPath, maxLogBytes) - m.broadcastOutput(id, chunk) - } - if err != nil { - return +// pumpLogs reads PTY output, redacts it, and writes to the instance's ring +// buffer (replacing the old on-disk log file). The hot path is designed for +// heavy TUI workloads (e.g. OpenCode CLI redraws producing MB/s): the ring +// buffer pointer is fetched once under stateMu, then each chunk uses an +// atomic load so neither the per-chunk stateMu acquisition nor the +// `[]byte(chunk)` allocation occurs on the data path. +// +// Lifecycle: PTY mode yields stdout == stderr, so a single Read is enough +// (no wg/stderr branch like the old file-based implementation needed). +// If Stop/Restart swaps the buffer to nil, subsequent atomic loads return +// nil and the chunk is dropped — acceptable because the same lifecycle +// event closes ptmx, causing Read to return EOF and ending this goroutine. +func (m *Manager) pumpLogs(id string, ptmx *os.File) { + defer func() { _ = ptmx.Close() }() + + m.stateMu.Lock() + p, ok := m.buffers[id] + m.stateMu.Unlock() + if !ok || p == nil { + // No buffer (instance stopped before goroutine started) — drain + // ptmx so the underlying process is not blocked on a full pty buffer, + // then exit. + _, _ = io.Copy(io.Discard, ptmx) + return + } + + readBuf := make([]byte, 1024) + for { + n, err := ptmx.Read(readBuf) + if n > 0 { + chunk := redact.Text(string(readBuf[:n])) + if rb := p.Load(); rb != nil { + rb.WriteString(chunk) } + m.broadcastOutput(id, chunk) + } + if err != nil { + return } } - // For PTY mode, stdout and stderr are the same, so we only read once - if stdout == stderr { - wg.Done() - write(stdout) - } else { - go write(stdout) - go write(stderr) - } - wg.Wait() } func (m *Manager) broadcastOutput(id string, chunk string) { @@ -845,6 +933,11 @@ func (m *Manager) broadcastOutput(id string, chunk string) { } } +// closeSubscribersLocked closes subscriber channels for the given instance. +// Buffer cleanup is done separately via dropBufferLocked under stateMu to +// avoid lock-ordering issues. +// +// Caller must hold m.mu. func (m *Manager) closeSubscribersLocked(id string) { subs := m.subscribers[id] for ch := range subs { @@ -853,6 +946,31 @@ func (m *Manager) closeSubscribersLocked(id string) { delete(m.subscribers, id) } +// dropBufferLocked removes the ring buffer for an instance and decrements the +// global cap counter. Caller must hold m.stateMu. +// +// Lifecycle note: in Restart the buffer is dropped AFTER the store is +// updated with the new instance id and AFTER subscribers are closed. There +// is a microsecond-scale window where the old id's buffer is still in the +// map; this is intentional so the old pumpLogs goroutine can finish its +// last few chunks (it will exit on its own when Restart closes the old +// ptmx). Any Tail/ReadSince for the old id in that window returns the +// remaining buffered bytes, which is the documented contract. +func (m *Manager) dropBufferLocked(id string) { + p, ok := m.buffers[id] + if !ok { + return + } + if rb := p.Swap(nil); rb != nil { + capBytes := rb.CapBytes() + rb.Close() + if capBytes > 0 { + m.totalBufBytes.Add(-capBytes) + } + } + delete(m.buffers, id) +} + func sanitizedEnv(in map[string]string) map[string]string { if len(in) == 0 { return nil @@ -864,33 +982,6 @@ func sanitizedEnv(in map[string]string) map[string]string { return out } -func enforceMaxLogSize(path string, max int64) error { - f, err := os.Open(path) - if err != nil { - return err - } - defer f.Close() - fi, err := f.Stat() - if err != nil { - return err - } - if fi.Size() <= max { - return nil - } - start := fi.Size() - max - if start < 0 { - start = 0 - } - if _, err := f.Seek(start, io.SeekStart); err != nil { - return err - } - b, err := io.ReadAll(f) - if err != nil { - return err - } - return os.WriteFile(path, b, 0o600) -} - func shortID() string { b := make([]byte, 6) _, _ = rand.Read(b) @@ -901,7 +992,6 @@ func terminatePID(pid int, sig syscall.Signal) error { if pid <= 0 { return errors.New("invalid pid") } - // Prefer signaling the process group (-pid) so `script` and its child shell exit together. if err := syscall.Kill(-pid, sig); err == nil || errors.Is(err, syscall.ESRCH) { return nil } @@ -936,15 +1026,6 @@ func (m *Manager) markStopped(id string, status string) error { return fmt.Errorf("unknown instance id: %s", id) } -func logPathByID(st store.State, id string) string { - for _, it := range st.Instances { - if it.ID == id { - return it.LogPath - } - } - return "" -} - func (m *Manager) loadTags() (map[string]tag.Tag, error) { base, err := os.UserConfigDir() if err != nil { @@ -958,7 +1039,6 @@ func (m *Manager) loadTags() (map[string]tag.Tag, error) { } // SetConnectionType records the current transport type for an instance. -// connType should be "websocket", "sse", or "" (disconnected). func (m *Manager) SetConnectionType(id, connType string) { m.connsMu.Lock() defer m.connsMu.Unlock() diff --git a/internal/instance/manager_budget_test.go b/internal/instance/manager_budget_test.go new file mode 100644 index 0000000..37e4d16 --- /dev/null +++ b/internal/instance/manager_budget_test.go @@ -0,0 +1,69 @@ +package instance + +import ( + "errors" + "path/filepath" + "testing" + + "myworktree/internal/store" +) + +// fakeMemSaturated reports a tiny system so the 25% budget is small enough +// to be exceeded by a single buffer cap. +type fakeMemSaturated struct{} + +func (fakeMemSaturated) VirtualMemory() (MemStat, error) { + return MemStat{ + Total: 256 << 20, // 256 MB total → 64 MB budget at 25% + Available: 256 << 20, + Used: 0, + }, nil +} + +// TestStart_BudgetExceededReturnsLogBufferBudgetError verifies that Manager +// returns a *LogBufferBudgetError (with correctly-populated fields) when a +// new instance would push the global cap past MaxTotalFraction × system_RAM. +// This is the source of the HTTP 503 response the dashboard surfaces as a +// "log_buffer_budget_exceeded" modal. +func TestStart_BudgetExceededReturnsLogBufferBudgetError(t *testing.T) { + t.Parallel() + workDir := t.TempDir() + path := filepath.Join(t.TempDir(), "state.json") + fs := store.FileStore{Path: path} + if err := fs.Save(store.State{ + Worktrees: []store.ManagedWorktree{ + {ID: "wt-budget-test", Name: "wt-budget-test", Path: workDir, Branch: "main"}, + }, + }); err != nil { + t.Fatalf("seed state: %v", err) + } + m := &Manager{Store: fs} + m.setMemSamplerForTest(fakeMemSaturated{}) + // Pre-seed the running sum so any new 16 MB+ cap exceeds 64 MB. + m.setTotalBufferBytesForTest(100 << 20) + + _, err := m.Start(StartInput{ + WorktreeID: "wt-budget-test", + Name: "budget-test", + }) + if err == nil { + t.Fatalf("expected error, got nil") + } + var budgetErr *LogBufferBudgetError + if !errors.As(err, &budgetErr) { + t.Fatalf("expected *LogBufferBudgetError, got %T: %v", err, err) + } + if budgetErr.UsedBytes != 100<<20 { + t.Fatalf("UsedBytes = %d, want %d", budgetErr.UsedBytes, 100<<20) + } + wantLimit := int64(float64(256<<20) * MaxTotalFraction) + if budgetErr.LimitBytes != wantLimit { + t.Fatalf("LimitBytes = %d, want %d", budgetErr.LimitBytes, wantLimit) + } + if budgetErr.SystemBytes != 256<<20 { + t.Fatalf("SystemBytes = %d, want %d", budgetErr.SystemBytes, 256<<20) + } + if !errors.Is(err, ErrLogBufferBudgetExceeded) { + t.Fatalf("errors.Is(err, ErrLogBufferBudgetExceeded) = false") + } +} diff --git a/internal/instance/manager_export_test.go b/internal/instance/manager_export_test.go new file mode 100644 index 0000000..04e76ba --- /dev/null +++ b/internal/instance/manager_export_test.go @@ -0,0 +1,22 @@ +package instance + +// Test-only hooks for Manager state. These methods are package-private and +// live in a _test.go file, so production code (which never sees this file at +// build time) cannot reach them. The exported SetMemSampler / +// SetTotalBufferBytesForTest methods that previously sat on *Manager have +// been removed; their callers in the app package's integration tests have +// been relocated to this package. + +// setMemSamplerForTest overrides the platform memory sampler. Used by tests +// in this package to inject a deterministic sampler so the budget-exceeded +// path can be exercised without depending on the host's actual RAM. +func (m *Manager) setMemSamplerForTest(s MemSampler) { + m.memSampler = s +} + +// setTotalBufferBytesForTest pre-seeds the running sum of live buffer caps +// so budget-exceeded tests can force the error path without spinning up +// dozens of real instances. +func (m *Manager) setTotalBufferBytesForTest(n int64) { + m.totalBufBytes.Store(n) +} diff --git a/internal/instance/manager_integration_test.go b/internal/instance/manager_integration_test.go index 215c02e..1e9de22 100644 --- a/internal/instance/manager_integration_test.go +++ b/internal/instance/manager_integration_test.go @@ -4,6 +4,7 @@ import ( "os" "os/exec" "path/filepath" + "strings" "sync" "testing" "time" @@ -143,6 +144,84 @@ func TestConcurrentMultiInstanceStartStop(t *testing.T) { } } +// TestPumpLogsBroadcastAndBuffer spawns a process via Start and asserts the +// per-instance ring buffer + subscriber channel both receive PTY output. +// This is the integration coverage for the in-memory log refactor — it +// guarantees the post-pumpLogs data path (write to buffer, broadcast to +// subscribers) is wired up end-to-end on a real PTY runtime. +func TestPumpLogsBroadcastAndBuffer(t *testing.T) { + requireRuntimeDeps(t) + + dataDir := mustTempDir(t, "mw-data-") + workDir := mustTempDir(t, "mw-work-") + fs := store.FileStore{Path: filepath.Join(dataDir, "state.json")} + if err := fs.Save(store.State{ + Worktrees: []store.ManagedWorktree{ + {ID: "wt1", Name: "wt1", Path: workDir}, + }, + }); err != nil { + t.Fatalf("seed state failed: %v", err) + } + + m := &Manager{DataDir: dataDir, Store: fs} + inst, err := m.Start(StartInput{ + WorktreeID: "wt1", + Command: "printf 'pumplogs-marker\\n' && sleep 0.2", + Name: "pumplogs", + }) + if err != nil { + t.Fatalf("Start failed: %v", err) + } + + subCh, unsub, err := m.SubscribeOutput(inst.ID) + if err != nil { + t.Fatalf("SubscribeOutput failed: %v", err) + } + defer unsub() + + // Drain broadcast channel until we see the marker (or timeout). + sawMarker := false + deadline := time.After(5 * time.Second) + for !sawMarker { + select { + case chunk := <-subCh: + if strings.Contains(chunk, "pumplogs-marker") { + sawMarker = true + } + case <-deadline: + t.Fatalf("subscriber did not receive marker in time") + } + } + + // Buffer must also contain the marker — Tail reads from the in-memory + // ring buffer, which is the only place logs live now. + tail, err := m.Tail(inst.ID, 64*1024) + if err != nil { + t.Fatalf("Tail err: %v", err) + } + if !strings.Contains(tail, "pumplogs-marker") { + t.Fatalf("buffer missing marker; tail = %q", tail) + } + + // And ReadSince with a stale cursor must still return the marker. + body, next, err := m.ReadSince(inst.ID, 0, 64*1024) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if !strings.Contains(body, "pumplogs-marker") { + t.Fatalf("ReadSince missing marker; body = %q", body) + } + if next == 0 { + t.Fatalf("ReadSince next = 0, want non-zero (cursor must advance)") + } + + if err := m.Stop(inst.ID); err != nil { + t.Fatalf("Stop failed: %v", err) + } + waitInstanceNotRunning(t, fs, inst.ID) + waitRuntimeReleased(t, m, inst.ID) +} + func requireRuntimeDeps(t *testing.T) { t.Helper() if _, err := exec.LookPath("script"); err != nil { diff --git a/internal/instance/manager_test.go b/internal/instance/manager_test.go index ff76bab..22dc7e9 100644 --- a/internal/instance/manager_test.go +++ b/internal/instance/manager_test.go @@ -1,12 +1,15 @@ package instance import ( + "errors" "os" "os/exec" "path/filepath" "strings" "sync" + "sync/atomic" "testing" + "time" "github.com/creack/pty" "myworktree/internal/store" @@ -28,36 +31,271 @@ func TestSanitizedEnv(t *testing.T) { } } -func TestLogPathByID(t *testing.T) { - st := store.State{ - Instances: []store.ManagedInstance{ - {ID: "a1", LogPath: "/tmp/a1.log"}, - }, +func TestPurgeOrphanLogFiles_RemovesOnlyLogFiles(t *testing.T) { + t.Parallel() + dataDir := t.TempDir() + logDir := filepath.Join(dataDir, "logs") + if err := os.MkdirAll(logDir, 0o755); err != nil { + t.Fatalf("mkdir logs: %v", err) + } + // Create some orphan .log files plus a non-.log file that must survive. + for _, name := range []string{"abc.log", "def.log", "keepme.txt"} { + if err := os.WriteFile(filepath.Join(logDir, name), []byte("data"), 0o600); err != nil { + t.Fatalf("write %s: %v", name, err) + } + } + + m := &Manager{DataDir: dataDir} + n, err := m.PurgeOrphanLogFiles() + if err != nil { + t.Fatalf("PurgeOrphanLogFiles err: %v", err) + } + if n != 2 { + t.Fatalf("removed %d, want 2", n) + } + for _, name := range []string{"abc.log", "def.log"} { + if _, err := os.Stat(filepath.Join(logDir, name)); !os.IsNotExist(err) { + t.Fatalf("expected %s removed, stat err = %v", name, err) + } + } + if _, err := os.Stat(filepath.Join(logDir, "keepme.txt")); err != nil { + t.Fatalf("non-.log file should be preserved: %v", err) + } +} + +func TestPurgeOrphanLogFiles_IdempotentOnMissingDir(t *testing.T) { + t.Parallel() + dataDir := t.TempDir() + m := &Manager{DataDir: dataDir} + n, err := m.PurgeOrphanLogFiles() + if err != nil { + t.Fatalf("err: %v", err) + } + if n != 0 { + t.Fatalf("removed %d, want 0", n) + } +} + +func TestPurgeOrphanLogFiles_IdempotentSecondCall(t *testing.T) { + t.Parallel() + dataDir := t.TempDir() + logDir := filepath.Join(dataDir, "logs") + if err := os.MkdirAll(logDir, 0o755); err != nil { + t.Fatalf("mkdir logs: %v", err) + } + if err := os.WriteFile(filepath.Join(logDir, "x.log"), []byte("y"), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + m := &Manager{DataDir: dataDir} + if n, _ := m.PurgeOrphanLogFiles(); n != 1 { + t.Fatalf("first call removed %d, want 1", n) + } + if n, _ := m.PurgeOrphanLogFiles(); n != 0 { + t.Fatalf("second call removed %d, want 0", n) + } +} + +func TestTailFromBuffer(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "state.json") + fs := store.FileStore{Path: path} + if err := fs.Save(store.State{}); err != nil { + t.Fatalf("init: %v", err) + } + m := &Manager{Store: fs, buffers: map[string]*atomic.Pointer[RingBuffer]{}} + p := &atomic.Pointer[RingBuffer]{} + p.Store(NewRingBuffer(1024)) + m.buffers["id1"] = p + p.Load().WriteString("hello world") + + got, err := m.Tail("id1", 5) + if err != nil { + t.Fatalf("Tail err: %v", err) + } + if got != "world" { + t.Fatalf("Tail = %q, want %q", got, "world") + } +} + +func TestTailFromBuffer_MissingReturnsEmpty(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "state.json") + fs := store.FileStore{Path: path} + if err := fs.Save(store.State{}); err != nil { + t.Fatalf("init: %v", err) + } + m := &Manager{Store: fs, buffers: map[string]*atomic.Pointer[RingBuffer]{}} + got, err := m.Tail("missing", 100) + if err != nil { + t.Fatalf("err: %v", err) + } + if got != "" { + t.Fatalf("got = %q, want empty", got) + } +} + +func TestReadSinceFromBuffer(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "state.json") + fs := store.FileStore{Path: path} + if err := fs.Save(store.State{}); err != nil { + t.Fatalf("init: %v", err) + } + m := &Manager{Store: fs, buffers: map[string]*atomic.Pointer[RingBuffer]{}} + p := &atomic.Pointer[RingBuffer]{} + p.Store(NewRingBuffer(1024)) + m.buffers["id1"] = p + p.Load().WriteString("part1") + cursor := p.Load().Offset() + p.Load().WriteString("part2") + + body, next, err := m.ReadSince("id1", cursor, 64*1024) + if err != nil { + t.Fatalf("ReadSince err: %v", err) + } + if body != "part2" { + t.Fatalf("body = %q, want %q", body, "part2") + } + if next != p.Load().Offset() { + t.Fatalf("next = %d, want %d", next, p.Load().Offset()) + } +} + +func TestDropBufferLocked_DecrementsTotal(t *testing.T) { + t.Parallel() + m := &Manager{buffers: map[string]*atomic.Pointer[RingBuffer]{}} + p := &atomic.Pointer[RingBuffer]{} + p.Store(NewRingBuffer(64 << 10)) // 64 KB + m.buffers["id1"] = p + m.totalBufBytes.Store(int64(64 << 10)) + + m.stateMu.Lock() + m.dropBufferLocked("id1") + m.stateMu.Unlock() + + if _, ok := m.buffers["id1"]; ok { + t.Fatalf("buffer should be dropped from map") + } + if got := m.totalBufBytes.Load(); got != 0 { + t.Fatalf("totalBufBytes = %d, want 0", got) + } + // After drop, atomic load on the (now-removed) pointer must yield nil + // for any in-flight pumpLogs goroutine to skip its remaining writes. + if rb := p.Load(); rb != nil { + t.Fatalf("Swap(nil) should make subsequent Load return nil, got %v", rb) + } +} + +func TestDropBufferLocked_MissingIDIsNoop(t *testing.T) { + t.Parallel() + m := &Manager{buffers: map[string]*atomic.Pointer[RingBuffer]{}} + m.totalBufBytes.Store(100) + m.stateMu.Lock() + m.dropBufferLocked("missing") + m.stateMu.Unlock() + if got := m.totalBufBytes.Load(); got != 100 { + t.Fatalf("totalBufBytes changed to %d, want 100", got) + } +} + +func TestLogBufferBudgetErrorUnwrap(t *testing.T) { + t.Parallel() + e := &LogBufferBudgetError{UsedBytes: 100, LimitBytes: 50, SystemBytes: 1000} + if !errors.Is(e, ErrLogBufferBudgetExceeded) { + t.Fatalf("errors.Is should match sentinel") + } + if e.Error() == "" { + t.Fatalf("Error() returned empty") + } +} + +// countingSampler counts how many times VirtualMemory() was called. Used to +// verify the 60s cache in Manager.sampledAvailable. +type countingSampler struct { + calls atomic.Int64 + stat MemStat + err error +} + +func (s *countingSampler) VirtualMemory() (MemStat, error) { + s.calls.Add(1) + return s.stat, s.err +} + +func TestSampledAvailable_CachesFor60s(t *testing.T) { + t.Parallel() + cs := &countingSampler{stat: MemStat{Total: 16 << 30, Available: 8 << 30, Used: 8 << 30}} + m := &Manager{memSampler: cs} + now := time.Now() + + avail, err := m.sampledAvailable(now) + if err != nil { + t.Fatalf("first call err: %v", err) + } + if avail != 8<<30 { + t.Fatalf("avail = %d, want %d", avail, int64(8<<30)) + } + if got := cs.calls.Load(); got != 1 { + t.Fatalf("sampler calls after first read = %d, want 1", got) + } + + // Second call 30s later should hit the cache (TTL is 60s). + if _, err := m.sampledAvailable(now.Add(30 * time.Second)); err != nil { + t.Fatalf("second call err: %v", err) } - if got := logPathByID(st, "a1"); got != "/tmp/a1.log" { - t.Fatalf("unexpected log path: %q", got) + if got := cs.calls.Load(); got != 1 { + t.Fatalf("sampler calls after 30s read = %d, want 1 (cache miss)", got) } - if got := logPathByID(st, "missing"); got != "" { - t.Fatalf("missing id should return empty path, got %q", got) + + // Third call 90s later must re-sample (cache expired). + if _, err := m.sampledAvailable(now.Add(90 * time.Second)); err != nil { + t.Fatalf("third call err: %v", err) + } + if got := cs.calls.Load(); got != 2 { + t.Fatalf("sampler calls after 90s read = %d, want 2 (cache refresh)", got) } } -func TestEnforceMaxLogSize(t *testing.T) { +func TestSampledAvailable_FallsBackToTotalMinusUsed(t *testing.T) { t.Parallel() - p := filepath.Join(t.TempDir(), "instance.log") - content := "0123456789abcdefghijklmnopqrstuvwxyz" - if err := os.WriteFile(p, []byte(content), 0o600); err != nil { - t.Fatalf("write log failed: %v", err) + // Available=0 in the sampler output should fall back to Total - Used. + cs := &countingSampler{stat: MemStat{Total: 4 << 30, Available: 0, Used: 1 << 30}} + m := &Manager{memSampler: cs} + avail, err := m.sampledAvailable(time.Now()) + if err != nil { + t.Fatalf("err: %v", err) } - if err := enforceMaxLogSize(p, 10); err != nil { - t.Fatalf("enforceMaxLogSize failed: %v", err) + want := int64(4<<30) - int64(1<<30) + if avail != want { + t.Fatalf("avail = %d, want %d", avail, want) } - b, err := os.ReadFile(p) +} + +func TestSampledAvailable_DefaultSamplerWhenNil(t *testing.T) { + t.Parallel() + // memSampler=nil → gopsutilMem default. We can't predict the host's + // Available, but the call must not panic and must populate the cache. + m := &Manager{} + avail, err := m.sampledAvailable(time.Now()) if err != nil { - t.Fatalf("read log failed: %v", err) + t.Fatalf("default sampler err: %v", err) + } + if avail < 0 { + t.Fatalf("avail = %d, must be >= 0", avail) + } + m.memSampleMu.Lock() + defer m.memSampleMu.Unlock() + if m.lastMemSampleAt.IsZero() { + t.Fatalf("lastMemSampleAt should be set after first sample") } - if string(b) != content[len(content)-10:] { - t.Fatalf("unexpected trimmed log content: %q", string(b)) +} + +func TestSampledAvailable_SamplerErrorPropagates(t *testing.T) { + t.Parallel() + cs := &countingSampler{err: errors.New("vmstat down")} + m := &Manager{memSampler: cs} + if _, err := m.sampledAvailable(time.Now()); err == nil { + t.Fatalf("expected error from failing sampler") } } @@ -185,7 +423,6 @@ func instance(id, worktreeID, status string) store.ManagedInstance { ID: id, WorktreeID: worktreeID, Status: status, - LogPath: "/dev/null", } } diff --git a/internal/instance/sizing.go b/internal/instance/sizing.go new file mode 100644 index 0000000..c8261b6 --- /dev/null +++ b/internal/instance/sizing.go @@ -0,0 +1,168 @@ +package instance + +import ( + "errors" + "fmt" + + "github.com/shirou/gopsutil/v4/mem" +) + +const ( + // DefaultBufferCap is the per-instance ring buffer cap when no + // user override is supplied and adaptive sampling fails. + DefaultBufferCap int64 = 32 << 20 // 32 MB + + // MinBufferCap is the floor for per-instance cap regardless of + // user override or adaptive computation. + MinBufferCap int64 = 16 << 20 // 16 MB + + // MaxBufferCap is the hard ceiling per instance. Never exceeded. + MaxBufferCap int64 = 256 << 20 // 256 MB + + // MaxTotalFraction caps the sum of all live buffers at this fraction + // of system RAM. Exceeding this rejects new instance creation with + // ErrLogBufferBudgetExceeded. + MaxTotalFraction = 0.25 +) + +// ErrLogBufferBudgetExceeded is the sentinel wrapped by LogBufferBudgetError +// when a new instance would push total buffer memory past the allowed budget. +var ErrLogBufferBudgetExceeded = errors.New("log buffer budget exceeded") + +// LogBufferBudgetError is a structured error returned by Manager.Start when +// the new instance would exceed the global buffer budget. HTTP handlers use +// errors.As to render it as a 503 with a structured JSON body. +type LogBufferBudgetError struct { + UsedBytes int64 // current sum of all live buffer caps + LimitBytes int64 // MaxTotalFraction * system RAM (rounded down) + SystemBytes int64 // total system RAM (informational) +} + +func (e *LogBufferBudgetError) Error() string { + return fmt.Sprintf( + "log buffer budget exceeded: used %d / limit %d (system %d bytes)", + e.UsedBytes, e.LimitBytes, e.SystemBytes, + ) +} + +// Unwrap enables errors.Is(err, ErrLogBufferBudgetExceeded). +func (e *LogBufferBudgetError) Unwrap() error { return ErrLogBufferBudgetExceeded } + +// MemStat is the abstract memory snapshot consumed by the buffer sizing +// logic. It is intentionally decoupled from any specific memory-sampling +// library so tests can construct deterministic values without pulling in +// gopsutil. +type MemStat struct { + Total int64 + Available int64 + Used int64 +} + +// MemSampler abstracts platform memory queries. The production implementation +// wraps gopsutil/v4/mem; tests inject a fake. +type MemSampler interface { + VirtualMemory() (MemStat, error) +} + +// gopsutilMem is the production MemSampler backed by gopsutil/v4/mem. +type gopsutilMem struct{} + +func (gopsutilMem) VirtualMemory() (MemStat, error) { + vm, err := mem.VirtualMemory() + if err != nil { + return MemStat{}, err + } + if vm == nil { + return MemStat{}, nil + } + return MemStat{ + Total: int64(vm.Total), + Available: int64(vm.Available), + Used: int64(vm.Used), + }, nil +} + +// resolveCap computes the per-instance ring buffer cap. +// +// Rules (applied in order): +// 1. If cfgBytes > 0, use it, clamped to [MinBufferCap, MaxBufferCap]. +// 2. Otherwise, take availableMem / 16, clamped to [MinBufferCap, MaxBufferCap]. +// "Available" prefers sampler.Available, falls back to Total - Used. +// 3. Return ErrLogBufferBudgetExceeded (as *LogBufferBudgetError) if the new +// cap would push the sum of all live caps past MaxTotalFraction × system +// memory. +// +// Pass sampler=nil to disable the budget check (used by tests and by callers +// that already validated the budget elsewhere). +func resolveCap(cfgBytes int64, sampler MemSampler, totalBufBytes int64) (int64, error) { + return resolveCapFromAvailable(cfgBytes, sampler, totalBufBytes, -1) +} + +// resolveCapFromAvailable is like resolveCap but takes a pre-computed +// `availableBytes` value (≥ 0) for the adaptive-cap path, decoupling it from +// a fresh sampler call. This is used by the Manager's 60-second memory-sample +// cache: the cap is allowed to be slightly stale, but the budget check below +// still queries the sampler live. +// +// `availableBytes = -1` means "compute available from sampler" (preserves the +// original resolveCap semantics). +func resolveCapFromAvailable(cfgBytes int64, sampler MemSampler, totalBufBytes int64, availableBytes int64) (int64, error) { + capBytes := adaptiveCapFromAvailable(cfgBytes, sampler, availableBytes) + if sampler != nil { + vm, err := sampler.VirtualMemory() + if err == nil && vm.Total > 0 { + limit := int64(float64(vm.Total) * MaxTotalFraction) + if totalBufBytes+capBytes > limit { + return 0, &LogBufferBudgetError{ + UsedBytes: totalBufBytes, + LimitBytes: limit, + SystemBytes: vm.Total, + } + } + } + } + return capBytes, nil +} + +// adaptiveCapFromAvailable returns the per-instance cap (without budget +// enforcement). `availableBytes = -1` falls back to sampling the platform +// memory (gated by sampler != nil). +func adaptiveCapFromAvailable(cfgBytes int64, sampler MemSampler, availableBytes int64) int64 { + if cfgBytes > 0 { + return clampCap(cfgBytes) + } + if availableBytes >= 0 { + return clampCap(availableBytes / 16) + } + if sampler == nil { + return DefaultBufferCap + } + vm, err := sampler.VirtualMemory() + if err != nil { + return DefaultBufferCap + } + avail := vm.Available + if avail <= 0 { + avail = vm.Total - vm.Used + } + if avail <= 0 { + return DefaultBufferCap + } + return clampCap(avail / 16) +} + +// adaptiveCap returns the per-instance cap (without budget enforcement). +// Exposed for tests. +func adaptiveCap(cfgBytes int64, sampler MemSampler) int64 { + return adaptiveCapFromAvailable(cfgBytes, sampler, -1) +} + +func clampCap(v int64) int64 { + if v < MinBufferCap { + return MinBufferCap + } + if v > MaxBufferCap { + return MaxBufferCap + } + return v +} diff --git a/internal/instance/sizing_test.go b/internal/instance/sizing_test.go new file mode 100644 index 0000000..12db062 --- /dev/null +++ b/internal/instance/sizing_test.go @@ -0,0 +1,154 @@ +package instance + +import ( + "errors" + "testing" +) + +type fakeMem struct { + total, available, used int64 + err error +} + +func (f fakeMem) VirtualMemory() (MemStat, error) { + if f.err != nil { + return MemStat{}, f.err + } + return MemStat{ + Total: f.total, + Available: f.available, + Used: f.used, + }, nil +} + +func TestResolveCap_UserOverrideClamped(t *testing.T) { + t.Parallel() + // Below floor + cap, err := resolveCap(1<<20, nil, 0) + if err != nil { + t.Fatalf("err: %v", err) + } + if cap != MinBufferCap { + t.Fatalf("below-floor cap = %d, want %d", cap, MinBufferCap) + } + // Above ceiling + cap, _ = resolveCap(1<<30, nil, 0) + if cap != MaxBufferCap { + t.Fatalf("above-ceiling cap = %d, want %d", cap, MaxBufferCap) + } + // Within range + cap, _ = resolveCap(64<<20, nil, 0) + if cap != 64<<20 { + t.Fatalf("within-range cap = %d, want %d", cap, 64<<20) + } +} + +func TestResolveCap_AdaptiveFromAvailable(t *testing.T) { + t.Parallel() + sampler := fakeMem{total: 16 << 30, available: 8 << 30, used: 8 << 30} + // available / 16 = 0.5 GB → clamp to MaxBufferCap + cap, err := resolveCap(0, sampler, 0) + if err != nil { + t.Fatalf("err: %v", err) + } + if cap != MaxBufferCap { + t.Fatalf("adaptive cap = %d, want %d (clamped)", cap, MaxBufferCap) + } +} + +func TestResolveCap_AdaptiveFromAvailable_4GB(t *testing.T) { + t.Parallel() + sampler := fakeMem{total: 4 << 30, available: 2 << 30, used: 2 << 30} + // available / 16 = 128 MB → within range + cap, err := resolveCap(0, sampler, 0) + if err != nil { + t.Fatalf("err: %v", err) + } + if cap != 128<<20 { + t.Fatalf("adaptive cap = %d, want %d", cap, 128<<20) + } +} + +func TestResolveCap_AdaptiveFallsBackToDefaultOnError(t *testing.T) { + t.Parallel() + sampler := fakeMem{err: errors.New("sampler failed")} + cap, err := resolveCap(0, sampler, 0) + if err != nil { + t.Fatalf("err: %v", err) + } + if cap != DefaultBufferCap { + t.Fatalf("fallback cap = %d, want %d", cap, DefaultBufferCap) + } +} + +func TestResolveCap_RejectsOverBudget(t *testing.T) { + t.Parallel() + // 4 GB total, 25% budget = 1 GB + sampler := fakeMem{total: 4 << 30, available: 4 << 30} + // Simulate 30 instances already at 32 MB each = 960 MB used + used := int64(30 * 32 << 20) + // New adaptive cap would be 4GB/16 = 256MB (clamped to MaxBufferCap) + // 960 + 256 = 1216 MB > 1024 MB limit → reject + _, err := resolveCap(0, sampler, used) + if err == nil { + t.Fatalf("expected budget error, got nil") + } + var budgetErr *LogBufferBudgetError + if !errors.As(err, &budgetErr) { + t.Fatalf("expected *LogBufferBudgetError, got %T", err) + } + wantLimit := int64(float64(4<<30) * MaxTotalFraction) + if budgetErr.LimitBytes != wantLimit { + t.Fatalf("LimitBytes = %d, want %d", budgetErr.LimitBytes, wantLimit) + } + if budgetErr.UsedBytes != used { + t.Fatalf("UsedBytes = %d, want %d", budgetErr.UsedBytes, used) + } + if budgetErr.SystemBytes != 4<<30 { + t.Fatalf("SystemBytes = %d, want %d", budgetErr.SystemBytes, 4<<30) + } + if !errors.Is(err, ErrLogBufferBudgetExceeded) { + t.Fatalf("errors.Is(err, ErrLogBufferBudgetExceeded) = false") + } +} + +func TestResolveCap_AllowsWhenWithinBudget(t *testing.T) { + t.Parallel() + sampler := fakeMem{total: 16 << 30, available: 16 << 30} + used := int64(10 * 32 << 20) // 320 MB used + // 25% of 16 GB = 4 GB limit; 320 + 32 = 352 MB << 4 GB → allow + cap, err := resolveCap(32<<20, sampler, used) + if err != nil { + t.Fatalf("err: %v", err) + } + if cap != 32<<20 { + t.Fatalf("cap = %d, want %d", cap, 32<<20) + } +} + +func TestResolveCap_NilSamplerSkipsBudget(t *testing.T) { + t.Parallel() + // Even with absurd usage, no error when sampler is nil + _, err := resolveCap(32<<20, nil, 1<<40) + if err != nil { + t.Fatalf("nil-sampler err: %v", err) + } +} + +func TestClampCap(t *testing.T) { + t.Parallel() + cases := []struct { + in, want int64 + }{ + {1, MinBufferCap}, + {MinBufferCap, MinBufferCap}, + {32 << 20, 32 << 20}, + {MaxBufferCap, MaxBufferCap}, + {1 << 30, MaxBufferCap}, + } + for _, c := range cases { + if got := clampCap(c.in); got != c.want { + t.Errorf("clampCap(%d) = %d, want %d", c.in, got, c.want) + } + } +} diff --git a/internal/store/state.go b/internal/store/state.go index 6a85b40..a3381b1 100644 --- a/internal/store/state.go +++ b/internal/store/state.go @@ -40,7 +40,6 @@ type ManagedInstance struct { Status string `json:"status"` // running|exited|stopped|failed RestartedFrom string `json:"restarted_from,omitempty"` RestartedTo string `json:"restarted_to,omitempty"` - LogPath string `json:"log_path"` CreatedAt string `json:"created_at"` StoppedAt string `json:"stopped_at,omitempty"` } diff --git a/internal/store/state_test.go b/internal/store/state_test.go index 9b94cdf..bebb5e2 100644 --- a/internal/store/state_test.go +++ b/internal/store/state_test.go @@ -27,7 +27,7 @@ func TestFileStoreLoadSaveRoundTrip(t *testing.T) { {ID: "wt1", Name: "feature-auth", Path: "/tmp/wt", Branch: "feature/auth"}, }, Instances: []ManagedInstance{ - {ID: "ins1", WorktreeID: "wt1", Status: "running", LogPath: "/tmp/ins1.log"}, + {ID: "ins1", WorktreeID: "wt1", Status: "running"}, }, } if err := fs.Save(src); err != nil { @@ -266,3 +266,37 @@ func TestSaveWithVersionLegacyFile(t *testing.T) { t.Fatalf("expected conflict after legacy migration, got: %v", err) } } + +// TestFileStore_IgnoresLegacyLogPath verifies that a state.json written by an +// older binary (which carried a per-instance `log_path` field) still loads +// cleanly under the new schema. The field is silently dropped by JSON +// decoding — this is the contract that protects external tools from a +// breaking schema change. +func TestFileStore_IgnoresLegacyLogPath(t *testing.T) { + t.Parallel() + path := filepath.Join(t.TempDir(), "state.json") + // Hand-crafted JSON containing the legacy "log_path" field on an + // instance. Modern ManagedInstance has no LogPath field; JSON decoding + // must ignore it without erroring out. + legacyJSON := `{"instances":[{"id":"legacy-1","worktree_id":"wt-1","status":"stopped","log_path":"/tmp/legacy-1.log"}],"tab_order":{}}` + if err := os.WriteFile(path, []byte(legacyJSON), 0o600); err != nil { + t.Fatalf("write legacy file: %v", err) + } + fs := FileStore{Path: path} + st, err := fs.Load() + if err != nil { + t.Fatalf("Load failed (legacy log_path should be silently ignored): %v", err) + } + if len(st.Instances) != 1 { + t.Fatalf("expected 1 instance, got %d", len(st.Instances)) + } + if st.Instances[0].ID != "legacy-1" { + t.Fatalf("instance ID = %q, want %q", st.Instances[0].ID, "legacy-1") + } + if st.Instances[0].WorktreeID != "wt-1" { + t.Fatalf("WorktreeID = %q, want %q", st.Instances[0].WorktreeID, "wt-1") + } + if st.Instances[0].Status != "stopped" { + t.Fatalf("Status = %q, want %q", st.Instances[0].Status, "stopped") + } +} diff --git a/internal/ui/static/index.html b/internal/ui/static/index.html index 8fb0db2..d124853 100644 --- a/internal/ui/static/index.html +++ b/internal/ui/static/index.html @@ -1313,6 +1313,26 @@ + + +
+ Cannot Start Instance + +
+
+

+

+
+ +
+
+ +
@@ -2485,7 +2505,32 @@ closeModal("modal-start-instance"); await refresh(); selectInstance(res.id); - } catch (e) { alert(e); } + } catch (e) { + if (e && e.status === 503 && e.body && e.body.error === "log_buffer_budget_exceeded") { + showBudgetError(e.body); + return; + } + alert(e.message || e); + } + } + + function showBudgetError(body) { + const msgEl = document.getElementById("budget-error-message"); + const hintEl = document.getElementById("budget-error-hint"); + msgEl.textContent = body.message || "Insufficient memory to start new instance."; + const used = formatBytes(body.used_bytes); + const limit = formatBytes(body.limit_bytes); + hintEl.textContent = `Used ${used} / limit ${limit}. ${body.hint || ""}`; + openModal("modal-budget-error"); + } + + function formatBytes(n) { + if (n == null) return "?"; + const kb = 1024, mb = 1024 * 1024, gb = 1024 * 1024 * 1024; + if (n >= gb) return (n / gb).toFixed(2) + " GB"; + if (n >= mb) return (n / mb).toFixed(2) + " MB"; + if (n >= kb) return (n / kb).toFixed(2) + " KB"; + return n + " B"; } async function stopInstance() { From f1ed5ab117ce1588c1f55ccdcfe02c6890f47beb Mon Sep 17 00:00:00 2001 From: Letian Lin Date: Sat, 20 Jun 2026 11:58:50 +0800 Subject: [PATCH 2/5] feat(monitor): add daemon resource stats and ring buffer usage reporting - Stats API now includes daemon_cpu_percent/daemon_memory_bytes in global totals - Per-instance stats expose memory_buffer_bytes (used) and memory_buffer_cap_bytes (cap) - Collector refactored: system calls moved outside lock, 3-phase pipeline - UI adds mw daemon row and buffer used/cap annotation per instance - Add .omo/ to .gitignore for OpenCode internal session data --- .gitignore | 1 + CHANGELOG.md | 2 + docs/API.md | 11 ++- docs/ARCHITECTURE.md | 2 +- internal/app/app.go | 14 +-- internal/instance/manager.go | 32 +++++++ internal/monitor/collector.go | 170 +++++++++++++++++++++++----------- internal/monitor/stats.go | 30 +++--- internal/ui/static/index.html | 15 ++- 9 files changed, 202 insertions(+), 75 deletions(-) diff --git a/.gitignore b/.gitignore index 2c88e91..60773d2 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,4 @@ myworktree_netgo .archive/ /AGENTS.md /CLAUDE.md +.omo/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e62b97..0bf4825 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,8 @@ Disk write amplification fix for long-running PTY-heavy sessions. - **Startup cleanup** — `Manager.PurgeOrphanLogFiles` removes dead `.log` files left behind by the old code path on the first start of the new binary. - **Budget-exceeded UX** — when a new instance would push the global buffer budget past 25% of system RAM, the HTTP/MCP API returns `503 Service Unavailable` with a structured `log_buffer_budget_exceeded` body. The dashboard surfaces this as a modal with used/limit numbers and a hint to close other tabs. - **Hot-path redesign for heavy TUI workloads** — `pumpLogs` no longer acquires `stateMu` per 1024-byte PTY chunk; the ring buffer pointer is pre-fetched once and the per-chunk lookup is a single `atomic.Pointer.Load`. The ring buffer's backing slice is allocated eagerly so the first Write never blocks on a 32 MB malloc. A new `WriteString` path avoids the `[]byte(chunk)` conversion that would otherwise happen on every chunk. +- **Daemon resource monitoring** — the resource stats API (`GET /api/instances/stats`) now includes the mw daemon process itself in global totals (`daemon_cpu_percent`, `daemon_memory_bytes`). The UI displays a dedicated "mw daemon" row so users can distinguish daemon overhead from instance resource usage. +- **Ring buffer usage reporting** — per-instance stats now expose `memory_buffer_bytes` (actual usage) and `memory_buffer_cap_bytes` (pre-allocated capacity). The UI memory column shows the combined `RSS + buffer_used` with a `buf used/cap` annotation for active buffers, giving users visibility into per-instance buffer memory cost. ## v0.3.0 diff --git a/docs/API.md b/docs/API.md index 50c230e..667e6d4 100644 --- a/docs/API.md +++ b/docs/API.md @@ -496,6 +496,8 @@ Response: "status": "running", "cpu_percent": 3.5, "memory_rss_bytes": 52428800, + "memory_buffer_bytes": 5242880, + "memory_buffer_cap_bytes": 33554432, "connection_type": "websocket" } ], @@ -511,7 +513,9 @@ Response: "global": { "total_cpu": 8.7, "total_memory": 209715200, - "instance_count": 3 + "instance_count": 3, + "daemon_cpu_percent": 1.2, + "daemon_memory_bytes": 67108864 } } ``` @@ -519,8 +523,11 @@ Response: Fields: - `cpu_percent`: CPU utilization as a percentage of a single core. 0% on the first measurement (no prior baseline). - `memory_rss_bytes`: Resident Set Size — actual physical memory used by the process. +- `memory_buffer_bytes`: Actual bytes currently held in the instance's in-memory ring buffer (0 if the instance is stopped or has no buffer). +- `memory_buffer_cap_bytes`: Pre-allocated capacity of the instance's in-memory ring buffer (0 if the instance is stopped or has no buffer). When both `memory_buffer_bytes` and `memory_buffer_cap_bytes` are non-zero, the buffer is active with `used / cap` semantics. - `connection_type`: `"websocket"` if the instance has an active WebSocket TTY connection, `"sse"` if using the SSE fallback, `"none"` otherwise. -- Worktree subtotals and global totals aggregate only `running` instances. +- Worktree subtotals aggregate only `running` instances (instance RSS only, not buffer memory). +- Global totals include both all running instances and the daemon process itself (`daemon_cpu_percent`, `daemon_memory_bytes`). ### 5.9 Instance lifecycle (frontend) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b6120cc..475c461 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -88,7 +88,7 @@ The sidebar shows a pinned **Main Workspace** item at the top (purple accent), f - On server startup, stale persisted `running` records are reconciled to `stopped` because in-memory stdin/stdout bindings cannot be resumed after process restart. The same startup pass also calls `Manager.PurgeOrphanLogFiles()` to remove dead `.log` files left behind by pre-buffer versions; missing or empty `logs/` directories are not an error. - **Rename**: `PATCH /api/instances` updates an instance's display name (`name` field). The rename takes effect immediately in the UI and persists to `state.json`. - **Tab ordering**: `PATCH /api/instances/reorder` persists per-worktree tab order to `state.json` (`tab_order` map + array order in `State.Instances`). Uses **optimistic locking** — the client sends the `version` observed from `GET /api/instances`. If the state has been modified since (e.g., another user started an instance), the server returns HTTP 409 Conflict and the client refreshes and retries. -- **Resource monitoring**: A clickable transport status bar in the bottom-right of the workspace opens a resource monitor modal. The modal shows per-instance CPU%, memory RSS, and connection type (WebSocket/SSE) grouped by worktree, with subtotals and a global summary. Data is fetched via `GET /api/instances/stats` (1-second polling when open, stops when closed). CPU% uses delta calculation from `process.Times()` with a per-PID baseline stored in the `Collector` struct. +- **Resource monitoring**: A clickable transport status bar in the bottom-right of the workspace opens a resource monitor modal. The modal shows per-instance CPU%, memory RSS, ring buffer usage (actual / capacity), and connection type (WebSocket/SSE) grouped by worktree, with subtotals and a global summary. The global totals include the mw daemon process itself (`daemon_cpu_percent`, `daemon_memory_bytes`). Data is fetched via `GET /api/instances/stats` (1-second polling when open, stops when closed). CPU% uses delta calculation from `process.Times()` with a per-PID baseline stored in the `Collector` struct. The UI includes a disclaimer that grandchild processes spawned inside instances are not individually tracked. - **Browser close protection**: The frontend registers a `beforeunload` event handler that unconditionally triggers a browser-native confirmation dialog on any page close/refresh/navigation attempt. This is purely a client-side UX safeguard — backend instances are unaffected and continue running. ### 4.1 Instance log buffer (in-memory) diff --git a/internal/app/app.go b/internal/app/app.go index ab42ce4..49d465e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -1572,12 +1572,14 @@ func (s *Server) handleInstanceStats(w http.ResponseWriter, r *http.Request) { continue } flat = append(flat, monitor.InputInstance{ - ID: inst.ID, - Name: inst.Name, - WorktreeID: inst.WorktreeID, - WorktreeName: inst.WorktreeName, - PID: inst.PID, - Status: inst.Status, + ID: inst.ID, + Name: inst.Name, + WorktreeID: inst.WorktreeID, + WorktreeName: inst.WorktreeName, + PID: inst.PID, + Status: inst.Status, + BufferCapBytes: s.instanceMgr.BufferCapBytesFor(inst.ID), + BufferUsedBytes: s.instanceMgr.BufferUsedBytesFor(inst.ID), }) } diff --git a/internal/instance/manager.go b/internal/instance/manager.go index 66302e3..a813cea 100644 --- a/internal/instance/manager.go +++ b/internal/instance/manager.go @@ -971,6 +971,38 @@ func (m *Manager) dropBufferLocked(id string) { delete(m.buffers, id) } +// BufferCapBytesFor returns the ring buffer capacity in bytes for a given +// instance, or 0 if the instance has no active buffer. +func (m *Manager) BufferCapBytesFor(id string) int64 { + m.stateMu.Lock() + defer m.stateMu.Unlock() + p, ok := m.buffers[id] + if !ok || p == nil { + return 0 + } + rb := p.Load() + if rb == nil { + return 0 + } + return rb.CapBytes() +} + +// BufferUsedBytesFor returns the ring buffer actual usage in bytes for a +// given instance, or 0 if the instance has no active buffer. +func (m *Manager) BufferUsedBytesFor(id string) int64 { + m.stateMu.Lock() + defer m.stateMu.Unlock() + p, ok := m.buffers[id] + if !ok || p == nil { + return 0 + } + rb := p.Load() + if rb == nil { + return 0 + } + return rb.BytesUsed() +} + func sanitizedEnv(in map[string]string) map[string]string { if len(in) == 0 { return nil diff --git a/internal/monitor/collector.go b/internal/monitor/collector.go index 1cbd3ac..257a497 100644 --- a/internal/monitor/collector.go +++ b/internal/monitor/collector.go @@ -1,6 +1,7 @@ package monitor import ( + "os" "sync" "time" @@ -17,15 +18,110 @@ type Collector struct { // a fully aggregated Stats response. instances is the list of all instances // (both running and stopped). connTypes maps instance ID to its current // transport ("websocket", "sse", or ""). +type procSample struct { + user, system float64 + hasTimes bool + rss uint64 +} + +func sampleProcess(pid int32) procSample { + var s procSample + p, err := process.NewProcess(pid) + if err != nil { + return s + } + if times, err := p.Times(); err == nil { + s.user = times.User + s.system = times.System + s.hasTimes = true + } + if mem, err := p.MemoryInfo(); err == nil { + s.rss = mem.RSS + } + return s +} + +func cpuDelta(prev cpuSnapshot, sample procSample, now time.Time) float64 { + if !sample.hasTimes { + return 0 + } + elapsed := now.Sub(prev.Times).Seconds() + if elapsed <= 0 { + return 0 + } + deltaUser := sample.user - prev.User + deltaSystem := sample.system - prev.System + cpu := (deltaUser + deltaSystem) / elapsed * 100 + if cpu < 0 { + return 0 + } + return cpu +} + func (c *Collector) Collect(instances []InputInstance, connTypes map[string]string) Stats { now := time.Now() + // Phase 1: collect all process data outside the lock (system calls). + daemonPID := os.Getpid() + daemonSample := sampleProcess(int32(daemonPID)) + + instanceSamples := make(map[int]procSample, len(instances)) + for _, inst := range instances { + if inst.Status != "running" || inst.PID <= 0 { + continue + } + instanceSamples[inst.PID] = sampleProcess(int32(inst.PID)) + } + + // Phase 2: compute CPU% and update snapshots under the lock. c.mu.Lock() if c.snapshots == nil { c.snapshots = make(map[int]cpuSnapshot) } - defer c.mu.Unlock() + daemonCPU := 0.0 + if daemonSample.hasTimes { + if prev, ok := c.snapshots[daemonPID]; ok { + daemonCPU = cpuDelta(prev, daemonSample, now) + } + c.snapshots[daemonPID] = cpuSnapshot{ + Times: now, + User: daemonSample.user, + System: daemonSample.system, + } + } + + instanceCPUs := make(map[int]float64, len(instanceSamples)) + for pid, sample := range instanceSamples { + cpuPct := 0.0 + if sample.hasTimes { + if prev, ok := c.snapshots[pid]; ok { + cpuPct = cpuDelta(prev, sample, now) + } + c.snapshots[pid] = cpuSnapshot{ + Times: now, + User: sample.user, + System: sample.system, + } + } + instanceCPUs[pid] = cpuPct + } + + livePIDs := make(map[int]struct{}, len(instances)+1) + livePIDs[daemonPID] = struct{}{} + for _, inst := range instances { + if inst.Status == "running" && inst.PID > 0 { + livePIDs[inst.PID] = struct{}{} + } + } + for pid := range c.snapshots { + if _, ok := livePIDs[pid]; !ok { + delete(c.snapshots, pid) + } + } + c.mu.Unlock() + + // Phase 3: build result outside the lock (no more snapshot access). var instancesOut []InstanceStat worktreeMap := make(map[string]*WorktreeStat) var globalTotalCPU float64 @@ -40,34 +136,11 @@ func (c *Collector) Collect(instances []InputInstance, connTypes map[string]stri cpuPercent := 0.0 memRSS := uint64(0) - if inst.Status == "running" && inst.PID > 0 { - p, err := process.NewProcess(int32(inst.PID)) - if err == nil { - times, err := p.Times() - if err == nil { - prev, ok := c.snapshots[inst.PID] - if ok { - elapsed := now.Sub(prev.Times).Seconds() - if elapsed > 0 { - deltaUser := times.User - prev.User - deltaSystem := times.System - prev.System - cpuPercent = (deltaUser + deltaSystem) / elapsed * 100 - if cpuPercent < 0 { - cpuPercent = 0 - } - } - } - c.snapshots[inst.PID] = cpuSnapshot{ - Times: now, - User: times.User, - System: times.System, - } - } - - mem, err := p.MemoryInfo() - if err == nil { - memRSS = mem.RSS + if sample, ok := instanceSamples[inst.PID]; ok { + memRSS = sample.rss + if cp, ok := instanceCPUs[inst.PID]; ok { + cpuPercent = cp } } } @@ -79,15 +152,17 @@ func (c *Collector) Collect(instances []InputInstance, connTypes map[string]stri } instancesOut = append(instancesOut, InstanceStat{ - ID: inst.ID, - Name: inst.Name, - WorktreeID: inst.WorktreeID, - WorktreeName: inst.WorktreeName, - PID: inst.PID, - Status: inst.Status, - CPUPercent: cpuPercent, - MemoryRSSBytes: memRSS, - ConnectionType: connType, + ID: inst.ID, + Name: inst.Name, + WorktreeID: inst.WorktreeID, + WorktreeName: inst.WorktreeName, + PID: inst.PID, + Status: inst.Status, + CPUPercent: cpuPercent, + MemoryRSSBytes: memRSS, + MemoryBufferBytes: inst.BufferUsedBytes, + MemoryBufferCapBytes: inst.BufferCapBytes, + ConnectionType: connType, }) if inst.Status == "running" { @@ -111,26 +186,15 @@ func (c *Collector) Collect(instances []InputInstance, connTypes map[string]stri worktreesOut = append(worktreesOut, *ws) } - // Prune stale snapshots: keep only PIDs that are currently running. - livePIDs := make(map[int]struct{}, len(instances)) - for _, inst := range instances { - if inst.Status == "running" && inst.PID > 0 { - livePIDs[inst.PID] = struct{}{} - } - } - for pid := range c.snapshots { - if _, ok := livePIDs[pid]; !ok { - delete(c.snapshots, pid) - } - } - return Stats{ Instances: instancesOut, Worktrees: worktreesOut, Global: GlobalStat{ - TotalCPU: globalTotalCPU, - TotalMemory: globalTotalMem, - InstanceCount: globalCount, + TotalCPU: globalTotalCPU + daemonCPU, + TotalMemory: globalTotalMem + daemonSample.rss, + InstanceCount: globalCount, + DaemonCPUPercent: daemonCPU, + DaemonMemoryBytes: daemonSample.rss, }, } } diff --git a/internal/monitor/stats.go b/internal/monitor/stats.go index 97054c1..88ddb67 100644 --- a/internal/monitor/stats.go +++ b/internal/monitor/stats.go @@ -19,9 +19,11 @@ type InstanceStat struct { WorktreeName string `json:"worktree_name"` PID int `json:"pid"` Status string `json:"status"` - CPUPercent float64 `json:"cpu_percent"` - MemoryRSSBytes uint64 `json:"memory_rss_bytes"` - ConnectionType string `json:"connection_type"` + CPUPercent float64 `json:"cpu_percent"` + MemoryRSSBytes uint64 `json:"memory_rss_bytes"` + MemoryBufferBytes int64 `json:"memory_buffer_bytes"` + MemoryBufferCapBytes int64 `json:"memory_buffer_cap_bytes"` + ConnectionType string `json:"connection_type"` } // WorktreeStat aggregates resource usage for instances within a single worktree. @@ -35,9 +37,11 @@ type WorktreeStat struct { // GlobalStat aggregates resource usage across all worktrees. type GlobalStat struct { - TotalCPU float64 `json:"total_cpu"` - TotalMemory uint64 `json:"total_memory"` - InstanceCount int `json:"instance_count"` + TotalCPU float64 `json:"total_cpu"` + TotalMemory uint64 `json:"total_memory"` + InstanceCount int `json:"instance_count"` + DaemonCPUPercent float64 `json:"daemon_cpu_percent"` + DaemonMemoryBytes uint64 `json:"daemon_memory_bytes"` } // cpuSnapshot stores per-PID CPU timing for delta-based CPU% calculation. @@ -49,10 +53,12 @@ type cpuSnapshot struct { // InputInstance is the minimal instance info needed for resource collection. type InputInstance struct { - ID string - Name string - WorktreeID string - WorktreeName string - PID int - Status string + ID string + Name string + WorktreeID string + WorktreeName string + PID int + Status string + BufferCapBytes int64 + BufferUsedBytes int64 } diff --git a/internal/ui/static/index.html b/internal/ui/static/index.html index d124853..8b837c3 100644 --- a/internal/ui/static/index.html +++ b/internal/ui/static/index.html @@ -3341,11 +3341,17 @@ ${escapeHTML(inst.status)} ${escapeHTML(connLabel)} ${inst.cpu_percent.toFixed(1)}% - ${formatBytes(inst.memory_rss_bytes)} + ${formatBytes((inst.memory_rss_bytes || 0) + (inst.memory_buffer_bytes || 0))}${buildBufferNote(inst)} `; } } + if (data.global) { + html += 'mw daemon' + (data.global.daemon_cpu_percent || 0).toFixed(1) + '%' + formatBytes(data.global.daemon_memory_bytes || 0) + ''; + } tbody.innerHTML = html; + const discRow = document.createElement('tr'); + discRow.innerHTML = '实例里运行的孙进程程序不做统计'; + tbody.appendChild(discRow); } function formatBytes(bytes) { @@ -3356,6 +3362,13 @@ return (bytes / (1024 * 1024 * 1024)).toFixed(2) + " GB"; } + function buildBufferNote(inst) { + const used = inst.memory_buffer_bytes || 0; + const cap = inst.memory_buffer_cap_bytes || 0; + if (cap <= 0) return ""; + return "  buf " + formatBytes(used) + "/" + formatBytes(cap); + } + function escapeHTML(s) { if (s === null || s === undefined) return ""; return String(s) From 2befdd1f6b5b5069628098bcc41bc0dcce3f7c5b Mon Sep 17 00:00:00 2001 From: Letian Lin Date: Sat, 20 Jun 2026 12:03:00 +0800 Subject: [PATCH 3/5] feat(ui): translate Resource Monitor subprocess disclaimer to English --- internal/ui/static/index.html | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/ui/static/index.html b/internal/ui/static/index.html index 8b837c3..bb4f475 100644 --- a/internal/ui/static/index.html +++ b/internal/ui/static/index.html @@ -3350,7 +3350,7 @@ } tbody.innerHTML = html; const discRow = document.createElement('tr'); - discRow.innerHTML = '实例里运行的孙进程程序不做统计'; + discRow.innerHTML = 'Child processes spawned by instances are not counted'; tbody.appendChild(discRow); } From 99349d722bc3b7aa30ae017d602139257bbf596f Mon Sep 17 00:00:00 2001 From: Letian Lin Date: Sat, 20 Jun 2026 12:08:22 +0800 Subject: [PATCH 4/5] perf(monitor): reduce resource stats polling interval from 1s to 2s Halves the per-second process.NewProcess() syscall storm when the resource monitor modal is open, reducing daemon CPU overhead. --- internal/ui/static/index.html | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/ui/static/index.html b/internal/ui/static/index.html index bb4f475..6c93fde 100644 --- a/internal/ui/static/index.html +++ b/internal/ui/static/index.html @@ -3246,7 +3246,7 @@ const dlg = document.getElementById("modal-monitor"); dlg.showModal(); fetchMonitorStats(); - monitorTimer = setInterval(fetchMonitorStats, 1000); + monitorTimer = setInterval(fetchMonitorStats, 2000); } function closeMonitorModal() { From b5916e9303c855c39f9be0a1952d026bb13210da Mon Sep 17 00:00:00 2001 From: Letian Lin Date: Sat, 20 Jun 2026 12:10:19 +0800 Subject: [PATCH 5/5] style(monitor): fix gofmt alignment in InstanceStat struct fields --- internal/monitor/stats.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/internal/monitor/stats.go b/internal/monitor/stats.go index 88ddb67..f8d1cba 100644 --- a/internal/monitor/stats.go +++ b/internal/monitor/stats.go @@ -13,17 +13,17 @@ type Stats struct { // InstanceStat describes the resource usage and connection status of a single instance. type InstanceStat struct { - ID string `json:"id"` - Name string `json:"name"` - WorktreeID string `json:"worktree_id"` - WorktreeName string `json:"worktree_name"` - PID int `json:"pid"` - Status string `json:"status"` - CPUPercent float64 `json:"cpu_percent"` - MemoryRSSBytes uint64 `json:"memory_rss_bytes"` - MemoryBufferBytes int64 `json:"memory_buffer_bytes"` - MemoryBufferCapBytes int64 `json:"memory_buffer_cap_bytes"` - ConnectionType string `json:"connection_type"` + ID string `json:"id"` + Name string `json:"name"` + WorktreeID string `json:"worktree_id"` + WorktreeName string `json:"worktree_name"` + PID int `json:"pid"` + Status string `json:"status"` + CPUPercent float64 `json:"cpu_percent"` + MemoryRSSBytes uint64 `json:"memory_rss_bytes"` + MemoryBufferBytes int64 `json:"memory_buffer_bytes"` + MemoryBufferCapBytes int64 `json:"memory_buffer_cap_bytes"` + ConnectionType string `json:"connection_type"` } // WorktreeStat aggregates resource usage for instances within a single worktree.