dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

feat: backend improved comments, logs and agents.md (#77)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-01-22 21:18:27 +00:00
committed by GitHub
parent 752e6b4655
commit 290359f7f0
8 changed files with 981 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+253 -186
View File
@@ -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
apps/backend/internal/agent/lifecycle/execution_store.go
+39 -13
View File
@@ -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
}
apps/backend/internal/agent/lifecycle/manager.go
+205 -28
View File
@@ -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 {
apps/backend/internal/agent/lifecycle/session.go
+6 -3
View File
@@ -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)
apps/backend/internal/agentctl/server/process/runner.go
+269 -45
View File
@@ -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 <command>", 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))
apps/backend/internal/common/appctx/appctx.go
+42
View File
@@ -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
}
apps/backend/internal/orchestrator/service.go
+87 -3
View File
@@ -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 {
apps/backend/internal/task/service/service.go
+80 -41
View File
@@ -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)