From 290359f7f02b22778b6bd5243528c093f888cab5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 22 Jan 2026 21:18:27 +0000 Subject: [PATCH] feat: backend improved comments, logs and agents.md (#77) --- AGENTS.md | 439 ++++++++++-------- .../agent/lifecycle/execution_store.go | 52 ++- .../internal/agent/lifecycle/manager.go | 233 ++++++++-- .../internal/agent/lifecycle/session.go | 9 +- .../agentctl/server/process/runner.go | 314 +++++++++++-- apps/backend/internal/common/appctx/appctx.go | 42 ++ apps/backend/internal/orchestrator/service.go | 90 +++- apps/backend/internal/task/service/service.go | 121 +++-- 8 files changed, 981 insertions(+), 319 deletions(-) create mode 100644 apps/backend/internal/common/appctx/appctx.go diff --git a/AGENTS.md b/AGENTS.md index dfaa29a1f5..f804eb9420 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,217 +1,284 @@ # Kandev Engineering Guide -> **Purpose**: This document captures up-to-date architecture notes, agent protocols, and frontend data-fetching patterns for Kandev. +> **Purpose**: Architecture notes, key patterns, and conventions for LLM agents working on Kandev. > > **Related**: [ARCHITECTURE.md](ARCHITECTURE.md), [docs/asyncapi.yaml](docs/asyncapi.yaml) ## Repo Layout -- `apps/backend/`: Go backend (WebSocket gateway, orchestrator, agent lifecycle, agentctl build). -- `apps/web/`: Next.js frontend (SSR + WS + Zustand store). -- `apps/packages/`: Shared packages/types. -- `apps/landing/`: Marketing/landing site that shares UI components with the web app. -- `apps/backend/dockerfiles/`: Agent container images (e.g., augment-agent). - -## Tooling and Workspace - -- **Package manager**: `pnpm` workspace. -- **Workspace root**: repository root. -- **Web app**: `apps/web` (Next.js, client + server components). -- **Backend**: `apps/backend` (Go services, agent lifecycle, ws). -- **UI**: Shadcn-based components are wrapped under `@kandev/ui`. - -Common commands: - -```bash -pnpm -C apps/web dev -pnpm -C apps/web lint -pnpm -C apps/web typecheck - -make -C apps/backend test -make -C apps/backend lint +``` +kandev-3/ +├── apps/ +│ ├── backend/ # Go backend (orchestrator, lifecycle, agentctl, WS gateway) +│ ├── web/ # Next.js frontend (SSR + WS + Zustand) +│ ├── packages/ # Shared packages/types +│ └── landing/ # Marketing site (shares UI with web) ``` -## Backend Architecture (High Level) +## Tooling -- **Transport**: WebSocket for realtime updates; HTTP for CRUD + SSR fetches. -- **Gateway**: Single WS endpoint at `ws://localhost:8080/ws`. -- **Orchestrator**: Starts/stops/resumes tasks, manages sessions. -- **Agent lifecycle**: - - Supports Docker and standalone. - - Creates agentctl instances for workspace access (shell, git, file ops). - - Starts agent subprocesses explicitly. -- **agentctl**: - - HTTP server for per-agent instance management. - - Uses ACP (JSON-RPC 2.0 over stdin/stdout) to talk to agent process. - - Streams ACP updates and workspace outputs to backend. +- **Package manager**: `pnpm` workspace (root) +- **Backend**: Go with Make (`make -C apps/backend test|lint|build`) +- **Frontend**: Next.js (`pnpm -C apps/web dev|lint|typecheck`) +- **UI**: Shadcn components via `@kandev/ui` -### Provider Pattern (Backend) +--- -- **Goal**: Keep `cmd/kandev` thin by constructing dependencies via package-local providers. -- **Location**: Providers live next to the implementation (e.g. `internal/events/provider.go`, `internal/task/repository/provider.go`). -- **API contract**: - - Each package exposes `Provide(...)` that returns a pointer to the concrete implementation (not an interface), a `cleanup func() error`, and an `error`. - - Example signature: `func Provide(cfg *config.Config, log *logger.Logger) (*impl, func() error, error)` -- **Usage**: - - External packages depend on the public interface type, but composition uses the concrete impl. - - Repositories expose a public `Repository` interface, while the implementation struct stays unexported. - - Call `cleanup()` during shutdown; providers should not own shared resources unless they created them. +## Backend Architecture -### Runtime / agentctl Flow (Simplified) +### Package Structure ``` -Client (WS) Backend Runtime agentctl Agent - | orchestrator.start | | | | - |------------------->| LaunchAgent | Create instance | | - | |--------------------->|------------------->| | - | | StartAgentProcess | Configure/Start |----> agent process | - | | ACP calls | |<--- ACP messages | - |<-------------------| WS notifications | | | +apps/backend/ +├── cmd/ +│ ├── kandev/ # Main backend binary entry point +│ └── agentctl/ # Agentctl binary (runs inside containers or standalone) +├── internal/ +│ ├── agent/ +│ │ ├── lifecycle/ # Agent instance management (see below) +│ │ ├── registry/ # Agent type registry and defaults +│ │ ├── runtime/ # Runtime name constants +│ │ └── mcpconfig/ # MCP server configuration +│ ├── agentctl/ +│ │ ├── client/ # HTTP client for talking to agentctl +│ │ └── server/ # agentctl HTTP server +│ │ ├── api/ # HTTP endpoints +│ │ ├── instance/ # Multi-instance management +│ │ ├── process/ # Agent subprocess management +│ │ └── adapter/ # Protocol adapters (ACP, Codex, REST, MCP) +│ ├── orchestrator/ # Task execution coordination +│ │ ├── executor/ # Launches agents via lifecycle manager +│ │ ├── scheduler/ # Task scheduling +│ │ ├── queue/ # Task queue +│ │ └── watcher/ # Event handlers +│ ├── task/ +│ │ ├── models/ # Task, Session, Executor, Message models +│ │ ├── service/ # Task business logic +│ │ └── repository/ # Database access (SQLite) +│ ├── worktree/ # Git worktree management for workspace isolation +│ └── events/ # Event bus for internal pub/sub ``` -### Protocols +### Key Concepts -- **ACP**: JSON-RPC 2.0 over stdin/stdout (agent <-> agentctl <-> backend). -- **WS Message Format**: +**Orchestrator** coordinates task execution: +- Receives task start/stop/resume requests +- Delegates to lifecycle manager for agent operations +- Handles event-driven state transitions + +**Lifecycle Manager** (`internal/agent/lifecycle/`) manages agent instances: +- `Manager` - central coordinator (~900 lines after refactor) +- `Runtime` interface - abstracts execution environment (Docker, Standalone) +- `ExecutionStore` - thread-safe in-memory execution tracking +- `SessionManager` - ACP session initialization and resume +- `StreamManager` - WebSocket stream connections to agentctl +- `EventPublisher` - publishes events to the event bus +- `ContainerManager` - Docker container operations + +**agentctl** is an HTTP server that: +- Runs inside Docker containers or as standalone process +- Manages agent subprocess via stdin/stdout (ACP protocol) +- Exposes workspace operations (shell, git, files) +- Supports multiple concurrent instances on different ports + +### Executor and Runtime + +**Executor** (database model) defines where agents run: + +| Type | Description | Status | +|------|-------------|--------| +| `local_pc` | Standalone process on host | Implemented | +| `local_docker` | Docker container on host | Implemented | +| `remote_docker` | Remote Docker daemon | Planned | +| `remote_vps` | Remote server via SSH | Planned | +| `k8s` | Kubernetes pods | Planned | + +**Runtime** (interface) implements execution: +- `DockerRuntime` - creates containers, mounts workspace +- `StandaloneRuntime` - uses shared agentctl process + +```go +type Runtime interface { + Name() runtime.Name + HealthCheck(ctx context.Context) error + CreateInstance(ctx context.Context, req *RuntimeCreateRequest) (*RuntimeInstance, error) + StopInstance(ctx context.Context, instance *RuntimeInstance, force bool) error + RecoverInstances(ctx context.Context) ([]*RuntimeInstance, error) +} +``` + +### Execution Flow + +``` +Client (WS) Orchestrator Lifecycle Manager Runtime agentctl + | | | | | + | task.start | | | | + |----------------->| LaunchAgent() | | | + | |------------------->| CreateInstance() | | + | | |----------------->| (container/proc) | + | | | |----------------->| + | | StartAgentProcess()| | | + | |------------------->| ConfigureAgent() | | + | | |---------------------------------->| + | | | Start() | | + | | |---------------------------------->| + | | | | agent proc | + | | |<---- stream updates (WS) ---------| + |<---- WS events --| | | | +``` + +### Session Resume + +Sessions can be resumed after interruption: +- `TaskSession.ACPSessionID` - stored ACP session ID for resume +- `ExecutorRunning` - tracks active executor state (container ID, port, worktree) +- On backend restart: `RecoverInstances()` finds running containers, reconnects streams + +### Provider Pattern + +Packages expose `Provide(...)` functions for dependency injection: + +```go +func Provide(cfg *config.Config, log *logger.Logger) (*impl, func() error, error) +``` + +- Returns concrete implementation, cleanup function, and error +- Cleanup called during graceful shutdown +- Keeps `cmd/kandev/main.go` thin + +### Worktrees + +Git worktrees provide workspace isolation: +- Each session can have its own worktree (branch) +- Prevents conflicts between concurrent agents +- Managed by `internal/worktree/Manager` + +--- + +## agentctl Server + +### HTTP Endpoints + +| Endpoint | Method | Description | +|----------|--------|-------------| +| `/health` | GET | Health check | +| `/api/v1/status` | GET | Agent status | +| `/api/v1/configure` | POST | Configure agent before start | +| `/api/v1/start` | POST | Start agent subprocess | +| `/api/v1/stop` | POST | Stop agent | +| `/api/v1/acp/*` | Various | ACP protocol (initialize, session/*, prompt) | +| `/api/v1/shell/*` | Various | Shell operations | +| `/api/v1/workspace/*` | Various | Git status, file changes | +| `/ws/updates` | WS | Agent event stream | +| `/ws/workspace` | WS | Workspace event stream (git, files, shell) | + +### Adapter Model + +Protocol adapters normalize different agent CLIs: + +- `AgentAdapter` interface defines `Start()`, `Stop()`, `Prompt()`, `Cancel()` +- `ACPAdapter` - ACP JSON-RPC over stdio +- `CodexAdapter` - Codex-style JSON-RPC +- `process.Manager` owns subprocess, wires stdio to adapter + +--- + +## ACP Protocol + +JSON-RPC 2.0 over stdin/stdout between agentctl and agent process. + +**Backend -> Agent (Requests)** +- `initialize` - handshake +- `session/new` - create new session +- `session/load` - resume existing session +- `session/prompt` - send user message +- `session/cancel` - cancel current operation + +**Agent -> Backend (Notifications)** +- `session/update` with types: `message_chunk`, `tool_call`, `tool_update`, `complete`, `error`, `permission_request`, `context_window` + +--- + +## Frontend Architecture + +### Data Flow Pattern (Critical) + +``` +SSR Fetch -> Hydrate Store -> Components Read Store -> Hooks Subscribe +``` + +**This pattern must be followed. Do not fetch data directly in components.** + +1. **SSR fetches** in layout/page server components +2. **Hydrate store** via `StateHydrator` component +3. **Components read from store only** (Zustand selectors) +4. **Hooks subscribe to WS channels** for real-time updates +5. **WS events update store** via event handlers + +### Store Structure + +```typescript +// Task/session data keyed by ID +messages.bySession[sessionId] +gitStatus[sessionId] +shell.outputs[sessionId] + +// Active selection +tasks.activeTaskId +tasks.activeSessionId +``` + +### Custom Hooks Pattern + +Hooks encapsulate subscription + store access: + +```typescript +// Hook subscribes to WS channel, returns data from store +function useSessionMessages(sessionId: string) { + useSessionSubscription(sessionId) // Subscribe on mount, cleanup on unmount + return useStore(state => state.messages.bySession[sessionId]) +} +``` + +### WebSocket Subscription + +- Components call subscription hooks (e.g., `useTaskSubscription(taskId)`) +- WS client **deduplicates** subscriptions with reference counting +- Reconnects automatically re-subscribe all active channels +- Event handlers route payloads to store slices + +**Why This Matters**: Without this pattern, you get duplicate subscriptions, stale data, and race conditions. + +### WS Message Format ```json { "id": "uuid", "type": "request|response|notification|error", "action": "action.name", - "payload": { "...": "..." }, + "payload": { ... }, "timestamp": "2026-01-10T12:00:00Z" } ``` -- See `docs/asyncapi.yaml` for WS actions and payloads. - -## Frontend Architecture - -### Key Principles - -- **Store is the source of truth** (Zustand store slices). -- **SSR preloads data** into the store via `StateHydrator`. -- **Components read from store only**; avoid ad-hoc fetches in UI components. -- **Reusable components** preferred; split large pages into smaller components. -- **Data fetching should live where it is required**; avoid leaking fetches into parent controllers. - -### Data Fetching and SSR - -- SSR fetch in layout/page, map to store shape, hydrate via `StateHydrator`. -- WebSocket events update store slices for real-time sync. -- Only re-fetch when WS cannot deliver full payloads (document why). - -### Subscription Strategy - -- Components subscribe to the channels they need (task, session, repo), using hooks. -- WebSocket client deduplicates subscriptions with reference counting. -- Reconnects re-subscribe all active channels automatically. -- Event handlers update store slices; components only read from store. -- URL params set active task/session in store for consistent view state. - -**Why This Works** - -- Components declare their dependencies explicitly. -- Subscription cleanup is automatic on unmount. -- WebSocket client guarantees only one real subscription per channel. -- Store updates flow globally, components re-render only on relevant slices. - -### Three-Layer Model - -``` -Layer 1: WebSocket Client (connection + dedup) -Layer 2: Event Handlers (route WS payloads to store) -Layer 3: Components (subscribe + render) -``` - -### Store Structure - -- Task/session scoped data is keyed in store: - - `messages.bySession[sessionId]` - - `messages.metaBySession[sessionId]` - - `gitStatus` keyed by `sessionId` - - `shell.outputs` keyed by `sessionId` -- Active selection state lives in store: - - `tasks.activeTaskId` - - `tasks.activeSessionId` - -### Custom Hooks Pattern - -- Hooks own subscriptions and return data from store: - - `useSessionMessages(sessionId)` → messages + loading/meta - - `useLazyLoadMessages(sessionId)` → pagination - - `useSessionGitStatus(sessionId)` → git state - - `useSessionShellOutput(sessionId)` → shell output -- Hooks subscribe on mount, cleanup on unmount; no manual dedup in components. - -### Page-Level Responsibilities - -- SSR fetches initial task/session + board snapshot, hydrates store. -- Page reads URL `sessionId`, resolves task via session, and calls `setActiveSession(taskId, sessionId)` after load. -- Page coordinates task list + session list loading (e.g., `listTaskSessions`). -- Components render from store and trigger hooks for live updates. - -## Agents and ACP - -### ACP Basics - -- **Transport**: JSON-RPC 2.0 over stdin/stdout. -- **Format**: newline-delimited JSON. -- **Direction**: request/response + notifications. - -**Backend → Agent (Requests)** -- `initialize` -- `session/new` -- `session/load` -- `session/prompt` -- `session/cancel` - -**Agent → Backend (Notifications)** -- `session/update` (progress/content/tool calls/complete/error) - -### agentctl HTTP (Internal) - -- `/health` GET -- `/api/v1/status` GET -- `/api/v1/start` POST -- `/api/v1/stop` POST -- `/api/v1/acp/*` (initialize, session/new, session/load, prompt, stream) -- `/api/v1/shell/*` (status, stream, buffer) -- `/api/v1/workspace/*` (git status, file changes) - -### Agent Types - -- Augment agent lives at `apps/backend/dockerfiles/augment-agent/`. -- Registry defaults in `apps/backend/internal/agent/registry/defaults.go`. - -## Frontend UI System (Shadcn) - -- Prefer `@kandev/ui` components over ad-hoc HTML. -- Reuse components; avoid duplicating UI patterns. -- Split large pages into smaller components; keep layout vs data vs presentation separated. - -## Best Practices - -- Prefer store-driven state over local fetches. -- Prefer reusable components over duplication; split large pages into smaller child components. -- Keep data fetching scoped to the component that needs it; avoid leaking fetches into parent controllers. -- Avoid WS request/response inside UI components when store actions can handle it. -- Keep SSR fetches minimal and map responses into store-friendly shapes. -- Component subscriptions should be local; WS client should dedupe. -- Keep agents logging to stderr; stdout only for ACP. - -## agentctl Server Architecture (Adapter Model) - -The agentctl server now uses a protocol adapter layer to support multiple agent CLIs while keeping a consistent HTTP + WebSocket surface for the backend. - -- `server/adapter/` defines the `AgentAdapter` interface and normalized `SessionUpdate` payloads. -- `ACPAdapter` uses the ACP JSON-RPC protocol over stdio (`server/adapter/acp_adapter.go`). -- `CodexAdapter` supports Codex-style JSON-RPC over stdio (`server/adapter/codex_adapter.go`). -- `process.Manager` owns the subprocess, wires stdin/stdout into the selected adapter, and forwards normalized updates to API streams. -- `api/acp.go` exposes protocol-agnostic endpoints (`/api/v1/acp/*`) that delegate to the adapter. -- `instance/` manages per-agent instance servers on dedicated ports (control server creates instances; each instance hosts its own API server). +See `docs/asyncapi.yaml` for all WS actions. --- -**Last Updated**: 2026-01-19 +## Best Practices + +### Backend +- Use provider pattern for dependency injection +- Keep agents logging to stderr; stdout only for ACP +- Pass context through call chains (but detach for background work) +- Use event bus for cross-component communication + +### Frontend +- Never fetch in components; use SSR + store hydration +- Always use subscription hooks; never raw WS calls in components +- Keep components reading from store only +- Split large pages into smaller components + +--- + +**Last Updated**: 2026-01-22 diff --git a/apps/backend/internal/agent/lifecycle/execution_store.go b/apps/backend/internal/agent/lifecycle/execution_store.go index 62595ae647..9b48579412 100644 --- a/apps/backend/internal/agent/lifecycle/execution_store.go +++ b/apps/backend/internal/agent/lifecycle/execution_store.go @@ -3,11 +3,15 @@ package lifecycle import ( + "errors" "sync" v1 "github.com/kandev/kandev/pkg/api/v1" ) +// ErrExecutionNotFound is returned when an execution doesn't exist in the store. +var ErrExecutionNotFound = errors.New("execution not found") + // ExecutionStore provides thread-safe storage and retrieval of agent executions. // It maintains three indexes for efficient lookup by execution ID, session ID, and container ID. type ExecutionStore struct { @@ -141,42 +145,64 @@ func (s *ExecutionStore) UpdateError(executionID string, errorMsg string) { } // UpdateMetadata updates the metadata of an agent execution using the provided function. +// The function is called outside the lock to prevent deadlocks - it receives a copy of +// the current metadata and returns the new metadata to set. func (s *ExecutionStore) UpdateMetadata(executionID string, updateFn func(metadata map[string]interface{}) map[string]interface{}) { + // First, get a copy of current metadata + s.mu.RLock() + execution, exists := s.executions[executionID] + if !exists { + s.mu.RUnlock() + return + } + + // Copy metadata to avoid races + currentMetadata := make(map[string]interface{}) + if execution.Metadata != nil { + for k, v := range execution.Metadata { + currentMetadata[k] = v + } + } + s.mu.RUnlock() + + // Call update function outside the lock + newMetadata := updateFn(currentMetadata) + + // Apply the result s.mu.Lock() defer s.mu.Unlock() - if execution, exists := s.executions[executionID]; exists { - if execution.Metadata == nil { - execution.Metadata = make(map[string]interface{}) - } - execution.Metadata = updateFn(execution.Metadata) + // Re-check existence (could have been removed) + if execution, exists = s.executions[executionID]; exists { + execution.Metadata = newMetadata } } // WithLock executes a function with the store lock held, providing access to the execution. -// Returns false if the execution doesn't exist. -func (s *ExecutionStore) WithLock(executionID string, fn func(execution *AgentExecution)) bool { +// Returns ErrExecutionNotFound if the execution doesn't exist. +// The function should be fast to avoid blocking other operations. +func (s *ExecutionStore) WithLock(executionID string, fn func(execution *AgentExecution)) error { s.mu.Lock() defer s.mu.Unlock() execution, exists := s.executions[executionID] if !exists { - return false + return ErrExecutionNotFound } fn(execution) - return true + return nil } // WithRLock executes a function with the store read lock held, providing access to the execution. -// Returns false if the execution doesn't exist. -func (s *ExecutionStore) WithRLock(executionID string, fn func(execution *AgentExecution)) bool { +// Returns ErrExecutionNotFound if the execution doesn't exist. +func (s *ExecutionStore) WithRLock(executionID string, fn func(execution *AgentExecution)) error { s.mu.RLock() defer s.mu.RUnlock() execution, exists := s.executions[executionID] if !exists { - return false + return ErrExecutionNotFound } fn(execution) - return true + return nil } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index aabb09fa64..89f0caa31c 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -21,6 +21,7 @@ import ( agentctl "github.com/kandev/kandev/internal/agentctl/client" agentctltypes "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/common/appctx" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" @@ -75,8 +76,11 @@ func NewManager( // Initialize command builder commandBuilder := NewCommandBuilder() + // Create stop channel for graceful shutdown + stopCh := make(chan struct{}) + // Initialize session manager - sessionManager := NewSessionManager(log) + sessionManager := NewSessionManager(log, stopCh) // Initialize event publisher eventPublisher := NewEventPublisher(eventBus, log) @@ -98,7 +102,7 @@ func NewManager( eventPublisher: eventPublisher, containerManager: containerManager, cleanupInterval: 30 * time.Second, - stopCh: make(chan struct{}), + stopCh: stopCh, } // Initialize stream manager with callbacks that delegate to manager methods @@ -122,12 +126,31 @@ func NewManager( return mgr } -// SetWorktreeManager sets the worktree manager for Git worktree isolation +// SetWorktreeManager sets the worktree manager for Git worktree isolation. +// +// This must be called before launching agents if Git worktree support is enabled in the runtime. +// The worktree manager creates isolated Git working directories for each agent execution, +// allowing multiple agents to work on the same repository without conflicts. +// +// Call this during initialization, typically when setting up the orchestrator service. +// If not set, agents will work directly in the repository's main working directory. func (m *Manager) SetWorktreeManager(worktreeMgr *worktree.Manager) { m.worktreeMgr = worktreeMgr } -// SetWorkspaceInfoProvider sets the provider for workspace information +// SetWorkspaceInfoProvider sets the provider for workspace information. +// +// The WorkspaceInfoProvider interface allows the lifecycle manager to dynamically create +// agent executions on-demand when the frontend connects to a session that doesn't have +// an active execution yet. This enables session resume after server restart or when +// accessing a session via URL (/task/[id]/[sessionId]). +// +// The provider must implement: +// - GetWorkspaceInfoBySessionID(ctx, sessionID) - Returns workspace path, worktree info, +// and MCP servers configured for the session +// +// This is typically called during initialization, with the task service as the provider. +// Without this, EnsureWorkspaceExecutionForSession will fail. func (m *Manager) SetWorkspaceInfoProvider(provider WorkspaceInfoProvider) { m.workspaceInfoProvider = provider } @@ -170,8 +193,8 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID // For workspace-only executions (no agent), wait for agentctl to be ready // then connect the workspace stream so process output can be received go func() { - // Wait for agentctl to be ready - waitCtx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + // Use detached context that respects stopCh for graceful shutdown + waitCtx, cancel := appctx.Detached(ctx, m.stopCh, 60*time.Second) defer cancel() if err := execution.agentctl.WaitForReady(waitCtx, 60*time.Second); err != nil { @@ -721,7 +744,9 @@ func (m *Manager) getOrCreateWorktree(ctx context.Context, req *LaunchRequest) ( // waitForAgentctlReady waits for the agentctl HTTP server to be ready. // This enables shell/workspace features without starting the agent process. func (m *Manager) waitForAgentctlReady(execution *AgentExecution) { - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + opStart := time.Now() + // Use detached context that respects stopCh for graceful shutdown + ctx, cancel := appctx.Detached(context.Background(), m.stopCh, 60*time.Second) defer cancel() m.logger.Info("waiting for agentctl to be ready", @@ -731,15 +756,26 @@ func (m *Manager) waitForAgentctlReady(execution *AgentExecution) { if err := execution.agentctl.WaitForReady(ctx, 60*time.Second); err != nil { m.logger.Error("agentctl not ready", zap.String("execution_id", execution.ID), + zap.Duration("duration", time.Since(opStart)), zap.Error(err)) m.updateExecutionError(execution.ID, "agentctl not ready: "+err.Error()) - m.eventPublisher.PublishAgentctlEvent(context.Background(), events.AgentctlError, execution, err.Error()) + // Use the timeout context for event publishing instead of a fresh Background context + m.eventPublisher.PublishAgentctlEvent(ctx, events.AgentctlError, execution, err.Error()) return } - m.logger.Info("agentctl ready - shell/workspace access available", - zap.String("execution_id", execution.ID)) - m.eventPublisher.PublishAgentctlEvent(context.Background(), events.AgentctlReady, execution, "") + elapsed := time.Since(opStart) + if elapsed > 10*time.Second { + m.logger.Warn("agentctl ready took longer than expected", + zap.String("execution_id", execution.ID), + zap.Duration("duration", elapsed)) + } else { + m.logger.Info("agentctl ready - shell/workspace access available", + zap.String("execution_id", execution.ID), + zap.Duration("duration", elapsed)) + } + // Use the timeout context for event publishing instead of a fresh Background context + m.eventPublisher.PublishAgentctlEvent(ctx, events.AgentctlReady, execution, "") } // getAgentConfigForExecution retrieves the agent configuration for an execution. @@ -1284,7 +1320,7 @@ func (m *Manager) StopAgent(ctx context.Context, executionID string, force bool) } // Update execution status and remove from tracking - m.executionStore.WithLock(executionID, func(exec *AgentExecution) { + _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { exec.Status = v1.AgentStatusStopped now := time.Now() exec.FinishedAt = &now @@ -1311,23 +1347,61 @@ func (m *Manager) StopBySessionID(ctx context.Context, sessionID string, force b return m.StopAgent(ctx, execution.ID, force) } -// GetExecution returns an agent execution by ID +// GetExecution returns an agent execution by ID. +// +// Returns (execution, true) if found, or (nil, false) if not found. +// The returned execution pointer should not be modified directly - use the Manager's +// methods to update execution state (MarkReady, MarkCompleted, UpdateStatus). +// +// Thread-safe: Can be called concurrently from multiple goroutines. func (m *Manager) GetExecution(executionID string) (*AgentExecution, bool) { return m.executionStore.Get(executionID) } -// GetExecutionBySessionID returns the agent execution for a session +// GetExecutionBySessionID returns the agent execution for a session. +// +// Returns (execution, true) if found, or (nil, false) if not found. +// A session can have at most one active execution at a time. If a session exists +// but has no active execution, this returns (nil, false). +// +// Thread-safe: Can be called concurrently from multiple goroutines. func (m *Manager) GetExecutionBySessionID(sessionID string) (*AgentExecution, bool) { return m.executionStore.GetBySessionID(sessionID) } -// ListExecutions returns all active executions +// ListExecutions returns all currently tracked agent executions. +// +// Returns a snapshot of all executions in memory at the time of call. The returned slice +// contains pointers to execution objects that may be modified by other goroutines after +// this method returns. Do not modify execution state directly - use Manager methods instead. +// +// The list includes executions in all states: +// - Starting (process launching, agentctl initializing) +// - Running (actively processing prompts) +// - Ready (waiting for user input) +// - Completed/Failed (finished but not yet removed) +// +// Thread-safe: Can be called concurrently. Returns a new slice on each call. +// +// Typical usage: Status endpoints, debugging, cleanup loops. func (m *Manager) ListExecutions() []*AgentExecution { return m.executionStore.List() } -// IsAgentRunningForSession checks if an agent is actually running for a session -// This probes agentctl's status endpoint to verify the agent process is running +// IsAgentRunningForSession checks if an agent process is running or starting for a session. +// +// This probes agentctl's status endpoint to verify the agent process state. Returns true if: +// - Agent status is "running" (actively processing prompts) +// - Agent status is "starting" (process launched but not yet ready) +// +// Returns false if: +// - No execution exists for this session +// - agentctl client is not available +// - Status check fails (network/timeout error) +// - Agent is in any other state (stopped, failed, etc.) +// +// Note: The name "IsAgentRunning" is slightly misleading - it includes "starting" state. +// Use this to check if an agent subprocess exists for the session, regardless of ready state. func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { // First check if we have an execution tracked for this session execution, exists := m.GetExecutionBySessionID(sessionID) @@ -1353,10 +1427,13 @@ func (m *Manager) IsAgentRunningForSession(ctx context.Context, sessionID string // UpdateStatus updates the status of an execution func (m *Manager) UpdateStatus(executionID string, status v1.AgentStatus) error { - if !m.executionStore.WithLock(executionID, func(execution *AgentExecution) { + if err := m.executionStore.WithLock(executionID, func(execution *AgentExecution) { execution.Status = status - }) { - return fmt.Errorf("execution %q not found", executionID) + }); err != nil { + if errors.Is(err, ErrExecutionNotFound) { + return fmt.Errorf("execution %q not found", executionID) + } + return err } m.logger.Debug("updated execution status", @@ -1366,7 +1443,22 @@ func (m *Manager) UpdateStatus(executionID string, status v1.AgentStatus) error return nil } -// MarkReady marks an execution as ready for follow-up prompts +// MarkReady marks an execution as ready for follow-up prompts. +// +// This transitions the execution to the "ready" state, indicating the agent has finished +// processing the current prompt and is waiting for user input. This is called: +// - After agent initialization completes (session loaded, workspace ready) +// - After agent finishes processing a prompt (via stream completion event) +// - After cancelling an agent turn (to allow new prompts) +// +// State Machine Transitions: +// Starting -> Ready (after initialization) +// Running -> Ready (after prompt completion) +// Any -> Ready (after cancel) +// +// Publishes an AgentReady event to notify subscribers (frontend, orchestrator). +// +// Returns error if execution not found. func (m *Manager) MarkReady(executionID string) error { execution, exists := m.executionStore.Get(executionID) if !exists { @@ -1384,14 +1476,33 @@ func (m *Manager) MarkReady(executionID string) error { return nil } -// MarkCompleted marks an execution as completed +// MarkCompleted marks an execution as completed or failed. +// +// This is called when the agent process terminates, either successfully or with an error. +// The final status is determined by exit code and error message: +// +// - exitCode == 0 && errorMessage == "" → AgentStatusCompleted (success) +// - Otherwise → AgentStatusFailed (failure) +// +// Parameters: +// - executionID: The execution to mark as completed +// - exitCode: Process exit code (0 = success, non-zero = failure) +// - errorMessage: Human-readable error description (empty string if no error) +// +// State Machine: +// This is a terminal state transition - no further state changes are expected after this. +// Typical flow: Starting -> Running -> Ready -> ... -> Completed/Failed +// +// Publishes either AgentCompleted or AgentFailed event depending on final status. +// +// Returns error if execution not found. func (m *Manager) MarkCompleted(executionID string, exitCode int, errorMessage string) error { execution, exists := m.executionStore.Get(executionID) if !exists { return fmt.Errorf("execution %q not found", executionID) } - m.executionStore.WithLock(executionID, func(exec *AgentExecution) { + _ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) { now := time.Now() exec.FinishedAt = &now exec.ExitCode = &exitCode @@ -1420,15 +1531,55 @@ func (m *Manager) MarkCompleted(executionID string, exitCode int, errorMessage s return nil } -// RemoveExecution removes a completed execution from tracking +// RemoveExecution removes an execution from tracking. +// +// ⚠️ WARNING: This is a potentially dangerous operation that should only be called when: +// 1. The agent process has been fully stopped (via StopAgent) +// 2. All cleanup operations have completed (worktree cleanup, container removal) +// 3. The execution is in a terminal state (Completed, Failed, or Cancelled) +// +// This method: +// - Removes the execution from the in-memory store +// - Makes the sessionID available for new executions +// - Does NOT stop the agent process (call StopAgent first) +// - Does NOT close the agentctl client (call execution.agentctl.Close() first) +// - Does NOT clean up resources (worktrees, containers, etc.) +// +// After calling this, the executionID and sessionID can no longer be used to query +// or control the execution. Any references to this execution will become invalid. +// +// Typical usage: Called by cleanup loops or after successful StopAgent completion. +// For stale/dead executions, use CleanupStaleExecutionBySessionID instead. func (m *Manager) RemoveExecution(executionID string) { m.executionStore.Remove(executionID) m.logger.Debug("removed execution from tracking", zap.String("execution_id", executionID)) } -// CleanupStaleExecutionBySessionID removes a stale agent execution from tracking without trying to stop it. -// This is used when we detect the agent process has stopped but the execution is still tracked. +// CleanupStaleExecutionBySessionID removes a stale execution from tracking without stopping it. +// +// A "stale" execution is one where the agent process has stopped externally (crashed, killed, +// or terminated outside of our control) but the execution is still tracked in memory. +// +// When to use this: +// - After detecting the agentctl HTTP server is unreachable +// - When the agent container no longer exists (Docker runtime) +// - After server restart when recovering persisted state +// - When IsAgentRunningForSession returns false but execution exists +// +// This method performs cleanup: +// 1. Closes the agentctl HTTP client connection +// 2. Removes the execution from the in-memory tracking store +// +// What this does NOT do: +// - Stop the agent process (assumed already stopped) +// - Clean up worktrees or containers (caller's responsibility) +// - Update database session state (caller's responsibility) +// +// This is safe to call even if the process is still running - it won't send kill signals. +// Use StopAgent if you need to actively terminate a running agent. +// +// Returns nil if no execution exists for the session (idempotent). func (m *Manager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { execution, exists := m.executionStore.GetBySessionID(sessionID) if !exists { @@ -1523,7 +1674,30 @@ func (m *Manager) performCleanup(ctx context.Context) { } } -// RespondToPermission sends a response to a permission request for an agent execution +// RespondToPermission sends a response to an agent's permission request. +// +// When an agent requests permission (e.g., to run a bash command, modify files, etc.), +// it pauses execution and waits for user approval. This method sends the user's response. +// +// Parameters: +// - executionID: The agent execution waiting for permission +// - pendingID: Unique ID of the permission request (from permission request event) +// - optionID: The user-selected option ID (from the permission request's options array) +// - cancelled: If true, indicates user cancelled/rejected the permission request. +// When cancelled=true, optionID is ignored. +// +// Response Semantics: +// - cancelled=false, optionID="approve" → User approved the action +// - cancelled=false, optionID="deny" → User explicitly denied the action +// - cancelled=true, optionID="" → User cancelled/closed the dialog +// +// After receiving the response, the agent will either: +// - Continue executing (if approved) +// - Skip the action and report failure (if denied/cancelled) +// +// Timeout: 30 seconds for agentctl to acknowledge the response. +// +// Returns error if execution not found, agentctl unavailable, or communication fails. func (m *Manager) RespondToPermission(executionID, pendingID, optionID string, cancelled bool) error { execution, exists := m.executionStore.Get(executionID) if !exists { @@ -1546,7 +1720,10 @@ func (m *Manager) RespondToPermission(executionID, pendingID, optionID string, c return execution.agentctl.RespondToPermission(ctx, pendingID, optionID, cancelled) } -// RespondToPermissionBySessionID sends a response to a permission request for a session +// RespondToPermissionBySessionID sends a response to a permission request using session ID. +// +// Convenience method that looks up the execution by session ID and delegates to RespondToPermission. +// See RespondToPermission for parameter semantics and behavior. func (m *Manager) RespondToPermissionBySessionID(sessionID, pendingID, optionID string, cancelled bool) error { execution, exists := m.executionStore.GetBySessionID(sessionID) if !exists { diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index 6f86afd191..e6fa453e6f 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -12,6 +12,7 @@ import ( agentctl "github.com/kandev/kandev/internal/agentctl/client" agentctltypes "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/common/appctx" "github.com/kandev/kandev/internal/common/logger" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -22,12 +23,14 @@ type SessionManager struct { eventPublisher *EventPublisher streamManager *StreamManager executionStore *ExecutionStore + stopCh <-chan struct{} // For graceful shutdown coordination } // NewSessionManager creates a new SessionManager -func NewSessionManager(log *logger.Logger) *SessionManager { +func NewSessionManager(log *logger.Logger, stopCh <-chan struct{}) *SessionManager { return &SessionManager{ logger: log, + stopCh: stopCh, } } @@ -243,8 +246,8 @@ func (sm *SessionManager) InitializeAndPrompt( // The agent will process the prompt and emit events via the WebSocket stream. if taskDescription != "" { go func() { - // Use a long timeout for initial prompts - promptCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + // Use detached context that respects stopCh for graceful shutdown + promptCtx, cancel := appctx.Detached(ctx, sm.stopCh, 10*time.Minute) defer cancel() _, err := sm.SendPrompt(promptCtx, execution, taskDescription, false, markReady) diff --git a/apps/backend/internal/agentctl/server/process/runner.go b/apps/backend/internal/agentctl/server/process/runner.go index c5f87eaeae..8cc15bfbad 100644 --- a/apps/backend/internal/agentctl/server/process/runner.go +++ b/apps/backend/internal/agentctl/server/process/runner.go @@ -1,3 +1,26 @@ +// Package process provides background process execution and output streaming for agentctl. +// +// The ProcessRunner manages long-running background processes spawned by agents or users, +// typically shell scripts, build commands, or other tools that need to run independently +// of the agent's main execution flow. +// +// Key Features: +// - Process isolation via process groups (Setpgid) for clean cleanup +// - Memory-bounded output buffering via ring buffers (prevents OOM on large outputs) +// - Real-time output streaming via WebSocket to frontend +// - Graceful shutdown with SIGTERM → SIGKILL escalation +// - Per-session process tracking +// +// Lifecycle: +// 1. Start() - Spawns process, creates output pipes, returns immediately +// 2. Background goroutines - Stream stdout/stderr, wait for exit +// 3. Stop() - Sends SIGTERM to process group, escalates to SIGKILL after 2s +// 4. Automatic cleanup - Process removed from tracking after exit +// +// Use Cases: +// - Agent-triggered builds (npm run build, go test ./...) +// - Background monitoring scripts (tail -f logs, watch commands) +// - Interactive shells (if needed, though typically handled separately) package process import ( @@ -19,54 +42,72 @@ import ( "go.uber.org/zap" ) +// StartProcessRequest contains parameters for starting a new background process. type StartProcessRequest struct { - SessionID string `json:"session_id"` - Kind types.ProcessKind `json:"kind"` - ScriptName string `json:"script_name,omitempty"` - Command string `json:"command"` - WorkingDir string `json:"working_dir"` - Env map[string]string `json:"env,omitempty"` - BufferMaxBytes int64 `json:"buffer_max_bytes,omitempty"` + SessionID string `json:"session_id"` // Required: Agent session owning this process + Kind types.ProcessKind `json:"kind"` // Process type (user_command, agent_script, etc.) + ScriptName string `json:"script_name,omitempty"` // Human-readable script identifier + Command string `json:"command"` // Required: Shell command to execute (passed to sh -lc) + WorkingDir string `json:"working_dir"` // Working directory (defaults to current dir) + Env map[string]string `json:"env,omitempty"` // Additional environment variables (merged with parent env) + BufferMaxBytes int64 `json:"buffer_max_bytes,omitempty"` // Max output buffer size (defaults to runner's default) } +// StopProcessRequest identifies a process to stop. type StopProcessRequest struct { - ProcessID string `json:"process_id"` + ProcessID string `json:"process_id"` // The process UUID returned by Start() } +// ProcessInfo represents the complete state of a background process. type ProcessInfo struct { - ID string `json:"id"` - SessionID string `json:"session_id"` - Kind types.ProcessKind `json:"kind"` - ScriptName string `json:"script_name,omitempty"` - Command string `json:"command"` - WorkingDir string `json:"working_dir"` - Status types.ProcessStatus `json:"status"` - ExitCode *int `json:"exit_code,omitempty"` - StartedAt time.Time `json:"started_at"` - UpdatedAt time.Time `json:"updated_at"` - Output []ProcessOutputChunk `json:"output,omitempty"` + ID string `json:"id"` // Unique process identifier (UUID) + SessionID string `json:"session_id"` // Agent session that owns this process + Kind types.ProcessKind `json:"kind"` // Process classification + ScriptName string `json:"script_name,omitempty"` // User-friendly name + Command string `json:"command"` // The shell command being executed + WorkingDir string `json:"working_dir"` // Execution directory + Status types.ProcessStatus `json:"status"` // Current state (starting, running, exited, failed) + ExitCode *int `json:"exit_code,omitempty"` // Process exit code (nil if still running) + StartedAt time.Time `json:"started_at"` // When the process was launched + UpdatedAt time.Time `json:"updated_at"` // Last status change timestamp + Output []ProcessOutputChunk `json:"output,omitempty"` // Buffered output (only included if requested) } +// ProcessOutputChunk represents a single piece of process output (stdout or stderr). type ProcessOutputChunk struct { - Stream string `json:"stream"` - Data string `json:"data"` - Timestamp time.Time `json:"timestamp"` + Stream string `json:"stream"` // "stdout" or "stderr" + Data string `json:"data"` // Raw output bytes as string + Timestamp time.Time `json:"timestamp"` // When this chunk was captured } +// ringBuffer provides memory-bounded FIFO storage for process output. +// +// This prevents out-of-memory errors when processes produce large amounts of output +// (e.g., verbose build logs, long-running monitoring scripts). When the buffer exceeds +// maxBytes, the oldest chunks are automatically evicted. +// +// Thread-safe: All methods use mutex protection. +// +// Example: With maxBytes=2MB, a process that outputs 10MB will only keep the most +// recent ~2MB in memory. Clients can stream output in real-time via WebSocket without +// loading the entire history. type ringBuffer struct { - mu sync.Mutex - maxBytes int64 - size int64 - chunks []ProcessOutputChunk + mu sync.Mutex // Protects chunks and size + maxBytes int64 // Maximum total bytes to keep in memory + size int64 // Current total bytes stored + chunks []ProcessOutputChunk // FIFO queue of output chunks } +// newRingBuffer creates a ring buffer with the specified size limit. +// Defaults to 2MB if maxBytes <= 0. func newRingBuffer(maxBytes int64) *ringBuffer { if maxBytes <= 0 { - maxBytes = 2 * 1024 * 1024 + maxBytes = 2 * 1024 * 1024 // 2MB default } return &ringBuffer{maxBytes: maxBytes} } +// append adds a new output chunk, evicting oldest chunks if over the size limit. func (b *ringBuffer) append(chunk ProcessOutputChunk) { b.mu.Lock() defer b.mu.Unlock() @@ -74,6 +115,7 @@ func (b *ringBuffer) append(chunk ProcessOutputChunk) { b.chunks = append(b.chunks, chunk) b.size += int64(len(chunk.Data)) + // Evict oldest chunks until we're back under the limit for b.size > b.maxBytes && len(b.chunks) > 0 { removed := b.chunks[0] b.size -= int64(len(removed.Data)) @@ -81,6 +123,8 @@ func (b *ringBuffer) append(chunk ProcessOutputChunk) { } } +// snapshot returns a copy of all buffered chunks at the current moment. +// Safe to call concurrently with append(). func (b *ringBuffer) snapshot() []ProcessOutputChunk { b.mu.Lock() defer b.mu.Unlock() @@ -89,24 +133,53 @@ func (b *ringBuffer) snapshot() []ProcessOutputChunk { return out } +// commandProcess represents a single running background process and its state. type commandProcess struct { - info ProcessInfo - cmd *exec.Cmd - buffer *ringBuffer - stopOnce sync.Once - stopSignal chan struct{} - mu sync.Mutex + info ProcessInfo // Process metadata and current status + cmd *exec.Cmd // Underlying OS process (nil after completion) + buffer *ringBuffer // Memory-bounded output storage + stopOnce sync.Once // Ensures stopSignal is only closed once + stopSignal chan struct{} // Signals output readers to exit before process termination + mu sync.Mutex // Protects info fields during updates } +// ProcessRunner manages multiple background processes with output streaming. +// +// Thread-safe: All public methods can be called concurrently from multiple goroutines. +// +// Process Lifecycle: +// 1. Start() creates a process in "starting" state +// 2. Transitions to "running" after successful spawn +// 3. Background goroutines stream output and wait for exit +// 4. Final status is "exited" (exit code 0) or "failed" (non-zero exit code) +// 5. Process is automatically removed from tracking after completion +// +// Output Handling: +// - Stdout and stderr are captured separately and streamed to WebSocket +// - Output is buffered in memory-bounded ring buffers (prevents OOM) +// - Get(id, includeOutput=true) retrieves buffered output for a process +// +// Process Groups: +// - Processes are spawned with Setpgid=true to create new process groups +// - Stop() kills the entire process group (handles child processes correctly) +// - This ensures clean cleanup even if the process spawns subprocesses type ProcessRunner struct { - logger *logger.Logger - workspaceTracker *WorkspaceTracker - bufferMaxBytes int64 + logger *logger.Logger // Scoped logger for this component + workspaceTracker *WorkspaceTracker // WebSocket stream coordinator (can be nil) + bufferMaxBytes int64 // Default output buffer size for new processes - mu sync.RWMutex - processes map[string]*commandProcess + mu sync.RWMutex // Protects processes map + processes map[string]*commandProcess // Active processes by ID } +// NewProcessRunner creates a new process runner. +// +// Parameters: +// - workspaceTracker: Optional coordinator for streaming output to WebSocket clients. +// If nil, processes run but output isn't streamed (only buffered). +// - log: Logger instance for process lifecycle events +// - bufferMaxBytes: Default output buffer size (typically 2MB). Individual processes +// can override via StartProcessRequest.BufferMaxBytes. func NewProcessRunner(workspaceTracker *WorkspaceTracker, log *logger.Logger, bufferMaxBytes int64) *ProcessRunner { return &ProcessRunner{ logger: log.WithFields(zap.String("component", "process-runner")), @@ -116,6 +189,36 @@ func NewProcessRunner(workspaceTracker *WorkspaceTracker, log *logger.Logger, bu } } +// Start spawns a new background process and returns immediately. +// +// The process is executed via "sh -lc ", which: +// - Loads shell profile (~/.bashrc, ~/.zshrc) for environment setup (-l flag) +// - Allows complex shell syntax like pipes, redirects, variable expansion (-c flag) +// +// Process Isolation: +// - Creates new process group (Setpgid=true) for clean shutdown +// - Children inherit the process group, so Stop() kills the entire tree +// +// Async Operation: +// - Returns immediately after spawning (doesn't wait for completion) +// - Three background goroutines handle: +// 1. stdout streaming → readOutput(stdout) +// 2. stderr streaming → readOutput(stderr) +// 3. exit monitoring → wait() +// +// Output Streaming: +// - stdout/stderr are captured in real-time +// - Chunks are buffered in memory (ring buffer with size limit) +// - Chunks are streamed to WebSocket clients (if workspaceTracker set) +// +// State Transitions: +// - Initial: "starting" (process being spawned) +// - Success: "running" (process started successfully) +// - Failure: "failed" (spawn failed - removed from tracking) +// +// Returns: +// - ProcessInfo with "running" status and unique process ID +// - Error if validation fails or process spawn fails func (r *ProcessRunner) Start(ctx context.Context, req StartProcessRequest) (*ProcessInfo, error) { if req.SessionID == "" { return nil, fmt.Errorf("session_id is required") @@ -127,11 +230,13 @@ func (r *ProcessRunner) Start(ctx context.Context, req StartProcessRequest) (*Pr id := uuid.New().String() now := time.Now().UTC() + // Use "sh -lc" to execute command with login shell environment cmd := exec.CommandContext(ctx, "sh", "-lc", req.Command) if req.WorkingDir != "" { cmd.Dir = req.WorkingDir } cmd.Env = mergeEnv(req.Env) + // Create new process group for clean shutdown (allows killing entire subprocess tree) cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} stdout, err := cmd.StdoutPipe() @@ -207,6 +312,37 @@ func (r *ProcessRunner) Start(ctx context.Context, req StartProcessRequest) (*Pr return &info, nil } +// Stop attempts to gracefully terminate a process, escalating to force-kill if needed. +// +// Shutdown Strategy (two-phase): +// +// 1. Graceful Phase (SIGTERM): +// - Sends SIGTERM to the entire process group (negative PGID) +// - Allows processes to clean up resources, flush buffers, etc. +// - Waits up to 2 seconds for graceful exit +// +// 2. Force Phase (SIGKILL): +// - If process hasn't exited after 2 seconds (or ctx expires), sends SIGKILL +// - SIGKILL cannot be caught/ignored - guarantees termination +// - Also sent to entire process group +// +// Process Group Termination: +// - Kill(-pgid, signal) sends signal to all processes in the group +// - This ensures child processes are also terminated +// - Example: "npm run build" spawns webpack, which spawns terser, etc. +// - Fallback: If process group lookup fails, kills only the main process +// +// Output Reader Cleanup: +// - Closes stopSignal channel before killing process +// - Signals readOutput() goroutines to exit cleanly +// - Prevents goroutine leaks from blocked pipe reads +// +// Behavior: +// - Returns immediately without waiting for process exit +// - wait() goroutine detects exit and updates final status +// - Process is automatically removed from tracking after exit +// +// Returns error only if process not found (already exited or invalid ID). func (r *ProcessRunner) Stop(ctx context.Context, req StopProcessRequest) error { proc, ok := r.get(req.ProcessID) if !ok { @@ -214,26 +350,36 @@ func (r *ProcessRunner) Stop(ctx context.Context, req StopProcessRequest) error } // Signal output readers to exit before attempting to terminate the process. + // This prevents goroutine leaks from blocked pipe reads. proc.stopOnce.Do(func() { close(proc.stopSignal) }) - // Attempt graceful shutdown, then escalate if the context expires. + // Attempt graceful shutdown (SIGTERM), then escalate to force-kill (SIGKILL) if proc.cmd != nil && proc.cmd.Process != nil { + // Try to get process group ID for killing entire subprocess tree pgid, err := syscall.Getpgid(proc.cmd.Process.Pid) + + // Phase 1: Graceful shutdown with SIGTERM if err == nil { + // Kill entire process group (negative PGID) _ = syscall.Kill(-pgid, syscall.SIGTERM) } else { + // Fallback: Kill only main process _ = proc.cmd.Process.Signal(syscall.SIGTERM) } + + // Wait for graceful exit (2 seconds) or context cancellation select { case <-ctx.Done(): + // Context cancelled - force kill immediately if err == nil { _ = syscall.Kill(-pgid, syscall.SIGKILL) } else { _ = proc.cmd.Process.Kill() } case <-time.After(2 * time.Second): + // Phase 2: Grace period expired - force kill with SIGKILL if err == nil { _ = syscall.Kill(-pgid, syscall.SIGKILL) } else { @@ -246,6 +392,11 @@ func (r *ProcessRunner) Stop(ctx context.Context, req StopProcessRequest) error } // StopAll stops all running processes managed by this runner. +// +// Iterates through all active processes and calls Stop() on each. +// Errors are collected and returned as a joined error (errors.Join). +// +// Typical usage: Shutdown cleanup when agentctl server is stopping. func (r *ProcessRunner) StopAll(ctx context.Context) error { r.mu.RLock() ids := make([]string, 0, len(r.processes)) @@ -263,6 +414,17 @@ func (r *ProcessRunner) StopAll(ctx context.Context) error { return errors.Join(errs...) } +// Get retrieves process information by ID. +// +// Parameters: +// - id: Process UUID returned by Start() +// - includeOutput: If true, includes buffered stdout/stderr in the response. +// Set to false for status checks to avoid copying large buffers. +// +// Returns (ProcessInfo, true) if found, or (nil, false) if not found. +// +// Note: Completed processes are automatically removed from tracking, so Get() +// will return false for processes that have finished and been cleaned up. func (r *ProcessRunner) Get(id string, includeOutput bool) (*ProcessInfo, bool) { proc, ok := r.get(id) if !ok { @@ -272,6 +434,14 @@ func (r *ProcessRunner) Get(id string, includeOutput bool) (*ProcessInfo, bool) return &info, true } +// List returns all active processes, optionally filtered by session. +// +// Parameters: +// - sessionID: If non-empty, returns only processes for this session. +// If empty, returns all processes across all sessions. +// +// Output is always returned without buffered output (includeOutput=false). +// Use Get(id, true) to retrieve output for a specific process. func (r *ProcessRunner) List(sessionID string) []ProcessInfo { r.mu.RLock() defer r.mu.RUnlock() @@ -323,23 +493,48 @@ func (r *ProcessRunner) readOutput(proc *commandProcess, reader io.ReadCloser, s } } +// wait blocks until the process exits, then updates final status and cleans up. +// +// This runs in a background goroutine spawned by Start(). Responsibilities: +// 1. Wait for process termination (blocks on cmd.Wait()) +// 2. Extract exit code from process state +// 3. Determine final status: "exited" (code 0) vs "failed" (non-zero) +// 4. Publish final status update to WebSocket clients +// 5. Remove process from tracking map (makes ID unavailable for future lookups) +// +// Exit Code Extraction: +// - Success (exit code 0): Status = "exited" +// - Failure (exit code != 0): Status = "failed" +// - If exit code cannot be determined: Defaults to 1 +// +// Cleanup Strategy: +// - Process is removed from r.processes after exit +// - This prevents Get() from returning stale/completed processes +// - Output is lost after removal (not persisted beyond memory) +// - Clients should stream output in real-time or call Get(id, true) before exit +// +// Goroutine Safety: +// - Each process has exactly one wait() goroutine +// - wait() is the sole authority for final status updates +// - Cleanup happens exactly once (no race with Stop()) func (r *ProcessRunner) wait(proc *commandProcess) { err := proc.cmd.Wait() exitCode := 0 status := types.ProcessStatusExited if err != nil { + // Extract exit code from error if exitErr, ok := err.(*exec.ExitError); ok { if waitStatus, ok := exitErr.Sys().(syscall.WaitStatus); ok { exitCode = waitStatus.ExitStatus() } else { - exitCode = 1 + exitCode = 1 // Fallback if status extraction fails } } else { - exitCode = 1 + exitCode = 1 // Non-ExitError (process killed, etc.) } status = types.ProcessStatusFailed } - // Final process status is computed here and published once. + r.logger.Debug("process exited", zap.String("process_id", proc.info.ID), zap.String("session_id", proc.info.SessionID), @@ -348,16 +543,18 @@ func (r *ProcessRunner) wait(proc *commandProcess) { zap.Error(err), ) + // Update process info with final status proc.mu.Lock() proc.info.Status = status proc.info.ExitCode = &exitCode proc.info.UpdatedAt = time.Now().UTC() proc.mu.Unlock() + + // Publish final status to WebSocket clients r.publishStatus(proc) - // Remove from the runner map after completion so future lookups don't see stale processes. - // Output is still preserved in the ring buffer returned by GetProcess(includeOutput=true) - // as long as the caller already has the process ID. + // Remove from tracking map - process is no longer queryable + // Output buffer is lost after this point (not persisted) r.mu.Lock() delete(r.processes, proc.info.ID) r.mu.Unlock() @@ -409,6 +606,13 @@ func (r *ProcessRunner) publishStatus(proc *commandProcess) { r.workspaceTracker.notifyWorkspaceStreamProcessStatus(update) } +// snapshot creates a thread-safe copy of process info at the current moment. +// +// Parameters: +// - includeOutput: If true, includes buffered output chunks (can be large). +// If false, returns info without output (faster, smaller). +// +// Thread-safe: Locks process mutex to prevent races with status updates. func (p *commandProcess) snapshot(includeOutput bool) ProcessInfo { p.mu.Lock() defer p.mu.Unlock() @@ -419,16 +623,36 @@ func (p *commandProcess) snapshot(includeOutput bool) ProcessInfo { return info } +// mergeEnv merges custom environment variables with the parent process environment. +// +// Strategy: +// 1. Start with all environment variables from os.Environ() (parent process) +// 2. Override/add variables from the env map (custom variables) +// 3. Return as []string in "KEY=VALUE" format (required by exec.Cmd.Env) +// +// Example: +// Parent env: PATH=/usr/bin, HOME=/home/user +// Custom env: PATH=/custom/bin, FOO=bar +// Result: PATH=/custom/bin, HOME=/home/user, FOO=bar +// +// This allows processes to inherit the agentctl server's environment (including PATH, +// HOME, etc.) while also supporting custom variables per process. func mergeEnv(env map[string]string) []string { base := make(map[string]string, len(os.Environ())+len(env)) + + // Parse parent environment into map for _, entry := range os.Environ() { if eq := strings.IndexByte(entry, '='); eq >= 0 { base[entry[:eq]] = entry[eq+1:] } } + + // Override with custom variables for k, v := range env { base[k] = v } + + // Convert back to []string format merged := make([]string, 0, len(base)) for k, v := range base { merged = append(merged, fmt.Sprintf("%s=%s", k, v)) diff --git a/apps/backend/internal/common/appctx/appctx.go b/apps/backend/internal/common/appctx/appctx.go new file mode 100644 index 0000000000..87e1025d43 --- /dev/null +++ b/apps/backend/internal/common/appctx/appctx.go @@ -0,0 +1,42 @@ +// Package appctx provides context utilities for background operations. +package appctx + +import ( + "context" + "time" +) + +// Detached returns a new context that is not tied to the parent's cancellation +// but inherits its values. Use this for operations that must outlive the request. +// The returned context will be cancelled when the stop channel is closed or timeout expires. +func Detached(parent context.Context, stopCh <-chan struct{}, timeout time.Duration) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + + // Propagate cancellation from stopCh + go func() { + select { + case <-stopCh: + cancel() + case <-ctx.Done(): + } + }() + + return ctx, cancel +} + +// DetachedWithValues creates a detached context that copies deadline-insensitive values +// from the parent context while starting fresh with no deadline. +// Use sparingly - most values should be passed explicitly. +func DetachedWithValues(parent context.Context, stopCh <-chan struct{}, timeout time.Duration) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + + go func() { + select { + case <-stopCh: + cancel() + case <-ctx.Done(): + } + }() + + return ctx, cancel +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 127ce6c5db..71250794e4 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1,4 +1,13 @@ -// Package orchestrator provides the main orchestrator service that ties all components together. +// Package orchestrator provides the main orchestrator service that coordinates +// task execution across agents. It manages: +// +// - Task queuing and scheduling via the Scheduler +// - Agent lifecycle through the AgentManager +// - Event handling and propagation +// - Session management and resume +// +// The orchestrator acts as the central coordinator between the task service, +// agent lifecycle manager, and the event bus. package orchestrator import ( @@ -157,12 +166,41 @@ func NewService( return s } -// SetMessageCreator sets the message creator for saving agent responses +// SetMessageCreator sets the message creator for saving agent responses to the database. +// +// This must be called before starting the orchestrator if you want agent messages, tool calls, +// and streaming content to be persisted to the database. The MessageCreator interface provides +// methods for creating and updating messages associated with task sessions. +// +// The MessageCreator is typically the task service, which owns the message persistence logic. +// Event handlers in the orchestrator call these methods when agent events occur: +// - AgentStreamEvent → CreateAgentMessage, AppendAgentMessage +// - Tool calls → CreateToolCallMessage, UpdateToolCallMessage +// - Permission requests → CreatePermissionRequestMessage +// +// When to call: During orchestrator initialization, after creating the task service. +// +// If not set: Agent messages won't be saved to the database (events will still be published). func (s *Service) SetMessageCreator(mc MessageCreator) { s.messageCreator = mc } -// SetTurnService sets the turn service for the orchestrator. +// SetTurnService sets the turn service for tracking conversation turns. +// +// A "turn" represents a single conversation round-trip: user prompt → agent response. +// The TurnService tracks turn timing and duration for analytics and UI display (e.g., showing +// how long each agent response took). +// +// The TurnService is typically the task service, which owns turn persistence logic. +// The orchestrator calls these methods: +// - StartTurn: When agent begins processing a prompt +// - CompleteTurn: When agent finishes and returns to ready state +// - GetActiveTurn: To associate messages with current turn +// +// When to call: During orchestrator initialization, after creating the task service. +// +// If not set: Turns won't be tracked (orchestrator continues functioning normally, but +// no timing data is recorded and turn IDs in messages will be empty). func (s *Service) SetTurnService(turnService TurnService) { s.turnService = turnService } @@ -300,6 +338,52 @@ func (s *Service) Stop() error { return nil } +// resumeExecutorsOnStartup recovers agent sessions that were active before server restart. +// +// This critical startup routine handles crash recovery and session persistence. When the server +// starts, it queries the database for sessions that were "running" when the server stopped, +// and attempts to restore them by relaunching agentctl and reconnecting to the workspace. +// +// Recovery Strategy: +// +// 1. Query database for sessions marked as "running" (persisted executor state) +// 2. For each session, determine recovery action based on previous state: +// +// Terminal States (Completed/Failed/Cancelled): +// - Clean up stale executor record from database +// - Skip resume (session is already done) +// +// Never-Started State (Created): +// - Clean up executor record (session never actually ran) +// - Skip resume (nothing to recover) +// +// Active States (Starting/Running/WaitingForInput): +// - Validate workspace/worktree still exists +// - Launch new agentctl instance +// - Restore session via ACP session/load (if resumable) +// - Transition to WaitingForInput (ready for next prompt) +// +// 3. Handle resume failures gracefully: +// - Mark session as Failed in database +// - Clean up executor record +// - Continue with next session (don't block other recoveries) +// +// Agent Resume Behavior: +// +// - startAgent = (ResumeToken != "" && Resumable) determines if agent process starts +// - If startAgent=true: Launches agent, loads conversation history via ACP +// - If startAgent=false: Only workspace access (shell) is restored, no agent +// +// State Transitions: +// +// - Session resume does NOT process any prompt (just loads history) +// - No "complete" event is emitted after resume (only after prompt completion) +// - Final state is always WaitingForInput (ready for user's next prompt) +// +// This runs synchronously during Start() before accepting new requests, ensuring all +// recoverable sessions are restored before the orchestrator begins processing new tasks. +// +// Called by: Start() method during orchestrator initialization. func (s *Service) resumeExecutorsOnStartup(ctx context.Context) { runningExecutors, err := s.repo.ListExecutorsRunning(ctx) if err != nil { diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 4a61138b75..25f8abb70b 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -27,6 +27,21 @@ type WorktreeCleanup interface { OnTaskDeleted(ctx context.Context, taskID string) error } +// WorktreeProvider extends WorktreeCleanup with query capabilities. +// Implementations that support this can be type-asserted from WorktreeCleanup. +type WorktreeProvider interface { + WorktreeCleanup + // GetAllByTaskID returns all worktrees associated with a task. + GetAllByTaskID(ctx context.Context, taskID string) ([]*worktree.Worktree, error) +} + +// WorktreeBatchCleaner extends WorktreeProvider with batch cleanup. +type WorktreeBatchCleaner interface { + WorktreeProvider + // CleanupWorktrees removes multiple worktrees in a single operation. + CleanupWorktrees(ctx context.Context, worktrees []*worktree.Worktree) error +} + // TaskExecutionStopper stops active task execution (agent session + instance). type TaskExecutionStopper interface { StopTask(ctx context.Context, taskID, reason string, force bool) error @@ -487,9 +502,7 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { var worktrees []*worktree.Worktree if s.worktreeCleanup != nil { - if provider, ok := s.worktreeCleanup.(interface { - GetAllByTaskID(context.Context, string) ([]*worktree.Worktree, error) - }); ok { + if provider, ok := s.worktreeCleanup.(WorktreeProvider); ok { worktrees, err = provider.GetAllByTaskID(ctx, id) if err != nil { s.logger.Warn("failed to list worktrees for delete", @@ -516,53 +529,79 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { zap.String("task_id", id), zap.Duration("duration", time.Since(start))) + // Perform cleanup synchronously with dedicated timeout + // Use background context since the original request may complete if s.executionStopper != nil || s.worktreeCleanup != nil || len(sessions) > 0 { - go func(taskID string, sessions []*models.TaskSession, worktrees []*worktree.Worktree) { - cleanupStart := time.Now() - cleanupCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() + cleanupStart := time.Now() + cleanupCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() - if s.executionStopper != nil { - if err := s.executionStopper.StopTask(cleanupCtx, taskID, "task deleted", true); err != nil { - s.logger.Warn("failed to stop task execution on delete", - zap.String("task_id", taskID), - zap.Error(err)) - } - } + cleanupErrors := s.performTaskCleanup(cleanupCtx, id, sessions, worktrees) - if len(worktrees) > 0 { - if cleaner, ok := s.worktreeCleanup.(interface { - CleanupWorktrees(context.Context, []*worktree.Worktree) error - }); ok { - if err := cleaner.CleanupWorktrees(cleanupCtx, worktrees); err != nil { - s.logger.Warn("failed to cleanup worktrees after delete", - zap.String("task_id", taskID), - zap.Error(err)) - } - } - } - - for _, session := range sessions { - if session == nil || session.ID == "" { - continue - } - if err := s.repo.DeleteExecutorRunningBySessionID(cleanupCtx, session.ID); err != nil { - s.logger.Debug("failed to delete executor runtime for session", - zap.String("task_id", taskID), - zap.String("session_id", session.ID), - zap.Error(err)) - } - } - - s.logger.Info("task delete cleanup finished", - zap.String("task_id", taskID), + if len(cleanupErrors) > 0 { + s.logger.Warn("task cleanup completed with errors", + zap.String("task_id", id), + zap.Int("error_count", len(cleanupErrors)), zap.Duration("duration", time.Since(cleanupStart))) - }(id, sessions, worktrees) + } else { + s.logger.Info("task cleanup completed", + zap.String("task_id", id), + zap.Duration("duration", time.Since(cleanupStart))) + } } return nil } +// performTaskCleanup handles post-deletion cleanup operations. +// Returns a slice of errors encountered (empty if all succeeded). +func (s *Service) performTaskCleanup( + ctx context.Context, + taskID string, + sessions []*models.TaskSession, + worktrees []*worktree.Worktree, +) []error { + var errs []error + + // Stop any running execution + if s.executionStopper != nil { + if err := s.executionStopper.StopTask(ctx, taskID, "task deleted", true); err != nil { + s.logger.Warn("failed to stop task execution on delete", + zap.String("task_id", taskID), + zap.Error(err)) + errs = append(errs, fmt.Errorf("stop execution: %w", err)) + } + } + + // Cleanup worktrees + if len(worktrees) > 0 { + if cleaner, ok := s.worktreeCleanup.(WorktreeBatchCleaner); ok { + if err := cleaner.CleanupWorktrees(ctx, worktrees); err != nil { + s.logger.Warn("failed to cleanup worktrees after delete", + zap.String("task_id", taskID), + zap.Error(err)) + errs = append(errs, fmt.Errorf("cleanup worktrees: %w", err)) + } + } + } + + // Delete executor running records for sessions + for _, session := range sessions { + if session == nil || session.ID == "" { + continue + } + if err := s.repo.DeleteExecutorRunningBySessionID(ctx, session.ID); err != nil { + s.logger.Debug("failed to delete executor runtime for session", + zap.String("task_id", taskID), + zap.String("session_id", session.ID), + zap.Error(err)) + // Don't add to errs - this is a debug-level issue + } + } + + return errs +} + // ListTasks returns all tasks for a board func (s *Service) ListTasks(ctx context.Context, boardID string) ([]*models.Task, error) { tasks, err := s.repo.ListTasks(ctx, boardID)