diff --git a/AGENTS.md b/AGENTS.md index 3808671087..43ab017409 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,22 +7,66 @@ ## Overview -Kandev agents run in Docker containers and communicate via **ACP (Agent Communication Protocol)** - a JSON-RPC 2.0 protocol over stdin/stdout. This enables bidirectional communication for task execution with real-time progress updates. +Kandev agents run in Docker containers with **agentctl** - a sidecar process that manages the AI agent and provides HTTP-based control. The agent communicates via **ACP (Agent Communication Protocol)** - a JSON-RPC 2.0 protocol over stdin/stdout. ``` -Backend Agent Container - │ │ - │ initialize │ - ├──────────────────────────────────► │ - │◄──────────────────────────────────┤ - │ session/prompt │ - ├──────────────────────────────────► │ - │ session/update (progress) │ - │◄──────────────────────────────────┤ - │ session/update (complete) │ - │◄──────────────────────────────────┤ +Backend agentctl (HTTP) Agent Process + │ │ │ + │ POST /api/v1/start │ │ + ├────────────────────────────►│ spawn auggie --acp │ + │ ├───────────────────────────►│ + │ POST /api/v1/acp/call │ initialize │ + ├────────────────────────────►├───────────────────────────►│ + │ │◄───────────────────────────┤ + │◄────────────────────────────┤ │ + │ WS /api/v1/acp/stream │ session/update │ + │◄────────────────────────────┤◄───────────────────────────┤ + │ │ │ ``` +## Architecture + +### agentctl + +`agentctl` is an HTTP server that runs inside each agent container as a sidecar process. It: + +1. **Manages the agent subprocess** (e.g., `auggie --acp`) +2. **Provides HTTP API** for control operations (start, stop, status) +3. **Relays ACP messages** between HTTP clients and the agent's stdin/stdout +4. **Streams output** via WebSocket for real-time monitoring + +### Communication Flow + +1. Backend creates container with agentctl as entrypoint +2. agentctl starts HTTP server on port 9999 +3. Backend waits for agentctl to be ready (`GET /health`) +4. Backend starts agent process (`POST /api/v1/start`) +5. Backend sends ACP messages via HTTP (`POST /api/v1/acp/call`) +6. Backend receives ACP updates via WebSocket (`GET /api/v1/acp/stream`) + +### agentctl HTTP API + +| Endpoint | Method | Description | +|----------|--------|-------------| +| `/health` | GET | Health check | +| `/api/v1/status` | GET | Get agent status | +| `/api/v1/start` | POST | Start agent process | +| `/api/v1/stop` | POST | Stop agent process | +| `/api/v1/acp/send` | POST | Send ACP message (fire and forget) | +| `/api/v1/acp/call` | POST | Send ACP request, wait for response | +| `/api/v1/acp/stream` | GET | WebSocket for ACP message streaming | +| `/api/v1/output` | GET | Get buffered output | +| `/api/v1/output/stream` | GET | WebSocket for real-time output | + +### Environment Variables + +| Variable | Default | Description | +|----------|---------|-------------| +| `AGENTCTL_PORT` | 9999 | HTTP server port | +| `AGENTCTL_AGENT_COMMAND` | `auggie --acp` | Command to run agent | +| `AGENTCTL_AUTO_START` | false | Auto-start agent on startup | +| `AGENTCTL_LOG_LEVEL` | info | Logging level | + --- ## Agent Communication Protocol (ACP) @@ -248,15 +292,24 @@ Backend Agent Container ### Augment Agent (`augment-agent`) -Primary AI coding agent powered by Auggie CLI. +Primary AI coding agent powered by Auggie CLI with agentctl sidecar. **Docker Image**: `kandev/augment-agent:latest` +**Architecture:** +``` +Container +├── agentctl (HTTP server on port 9999) +│ └── manages → auggie --acp (subprocess) +└── /workspace (mounted project directory) +``` + **Specification:** - **Working Directory**: `/workspace` - **Memory**: 4096 MB - **CPU**: 2.0 cores - **Timeout**: 1 hour +- **Exposed Port**: 9999 (agentctl HTTP API) - **Capabilities**: code_generation, code_review, refactoring, testing, shell_execution **Required Environment Variables:** @@ -268,42 +321,80 @@ TASK_DESCRIPTION # The task to execute **Optional Environment Variables:** ```bash AUGGIE_SESSION_ID # Resume previous session +AGENTCTL_AUTO_START # Auto-start agent on container start ``` **Mounted Volumes:** - `{workspace}` → `/workspace` - Project directory - `{augment_sessions}` → `/root/.augment/sessions` - Session persistence -**Dockerfile**: `backend/dockerfiles/augment-agent/Dockerfile` +**Dockerfile**: `apps/backend/dockerfiles/augment-agent/Dockerfile` + +**Building the Image:** +```bash +cd apps/backend +docker build -t kandev/augment-agent:latest -f dockerfiles/augment-agent/Dockerfile . +``` --- ## Creating Custom Agents -### Step 1: Create Dockerfile +Custom agents can be created in two ways: -Create `backend/dockerfiles/your-agent/Dockerfile`: +### Option A: With agentctl (Recommended) + +Use the multi-stage Dockerfile pattern to include agentctl: ```dockerfile +# Stage 1: Build agentctl +FROM golang:1.23-alpine AS builder +WORKDIR /build +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 go build -o agentctl ./cmd/agentctl + +# Stage 2: Final image FROM ubuntu:22.04 -# Install dependencies -RUN apt-get update && apt-get install -y \ - curl jq git \ - && rm -rf /var/lib/apt/lists/* +# Install your agent dependencies +RUN apt-get update && apt-get install -y curl jq git && rm -rf /var/lib/apt/lists/* -# Copy agent script +# Copy agentctl from builder +COPY --from=builder /build/agentctl /usr/local/bin/agentctl + +# Copy your agent script COPY agent.sh /usr/local/bin/agent RUN chmod +x /usr/local/bin/agent WORKDIR /workspace +EXPOSE 9999 -ENTRYPOINT ["/usr/local/bin/agent"] +# Set agentctl as entrypoint +ENV AGENTCTL_AGENT_COMMAND="/usr/local/bin/agent --acp" +ENTRYPOINT ["/usr/local/bin/agentctl"] ``` -### Step 2: Implement ACP Protocol +### Option B: Standalone ACP Agent -Create `backend/dockerfiles/your-agent/agent.sh`: +For simpler agents that don't need HTTP control: + +```dockerfile +FROM ubuntu:22.04 + +RUN apt-get update && apt-get install -y curl jq git && rm -rf /var/lib/apt/lists/* + +COPY agent.sh /usr/local/bin/agent +RUN chmod +x /usr/local/bin/agent + +WORKDIR /workspace +ENTRYPOINT ["/usr/local/bin/agent", "--acp"] +``` + +### Implementing ACP Protocol + +Create `agent.sh` that implements JSON-RPC 2.0: ```bash #!/bin/bash @@ -311,15 +402,16 @@ set -e # ACP mode - JSON-RPC 2.0 if [ "$1" = "--acp" ]; then - # Initialize response - echo '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"1.0","serverInfo":{"name":"your-agent","version":"1.0.0"},"capabilities":{"streaming":true}}}' - # Read requests from stdin while IFS= read -r line; do METHOD=$(echo "$line" | jq -r '.method') ID=$(echo "$line" | jq -r '.id') case "$METHOD" in + "initialize") + echo '{"jsonrpc":"2.0","id":'$ID',"result":{"protocolVersion":"1.0","serverInfo":{"name":"your-agent","version":"1.0.0"},"capabilities":{"streaming":true}}}' + ;; + "session/new") SESSION_ID=$(uuidgen) echo '{"jsonrpc":"2.0","id":'$ID',"result":{"sessionId":"'$SESSION_ID'"}}' @@ -342,7 +434,6 @@ if [ "$1" = "--acp" ]; then ;; "session/cancel") - # Handle cancellation exit 0 ;; esac @@ -350,9 +441,9 @@ if [ "$1" = "--acp" ]; then fi ``` -### Step 3: Register Agent Type +### Register Agent Type -Add to `backend/internal/agent/registry/defaults.go`: +Add to `apps/backend/internal/agent/registry/defaults.go`: ```go { @@ -373,7 +464,7 @@ Add to `backend/internal/agent/registry/defaults.go`: ```bash # Build image -cd backend/dockerfiles/your-agent +cd apps/backend/dockerfiles/your-agent docker build -t kandev/your-agent:latest . # Test locally @@ -613,45 +704,75 @@ Prior versions used a simpler JSON format (not JSON-RPC). For compatibility refe 6. **Validate messages** - Check for required fields before processing 7. **Set exit codes** - 0 for success, non-zero for errors -### Message Flow +### Message Flow (with agentctl) ``` -1. Backend starts container -2. Backend sends initialize -3. Agent responds with capabilities -4. Backend sends session/new -5. Agent responds with sessionId -6. Backend sends session/prompt -7. Agent responds immediately (accepted: true) -8. Agent sends multiple session/update notifications -9. Agent sends final session/update (type: complete) -10. Container exits +1. Backend creates container with agentctl +2. agentctl starts HTTP server on port 9999 +3. Backend waits for agentctl health check +4. Backend calls POST /api/v1/start +5. agentctl spawns agent subprocess +6. Backend calls POST /api/v1/acp/call with initialize +7. Agent responds with capabilities +8. Backend calls POST /api/v1/acp/call with session/new +9. Agent responds with sessionId +10. Backend connects to WS /api/v1/acp/stream +11. Backend calls POST /api/v1/acp/call with session/prompt +12. Agent sends session/update notifications (streamed via WebSocket) +13. Agent sends final session/update (type: complete) +14. Backend calls POST /api/v1/stop or container exits ``` ### Testing Your Agent +**With agentctl:** +```bash +# Build and run container +docker run -d -p 9999:9999 kandev/your-agent:latest + +# Check health +curl http://localhost:9999/health + +# Start agent +curl -X POST http://localhost:9999/api/v1/start + +# Send ACP message +curl -X POST http://localhost:9999/api/v1/acp/call \ + -H "Content-Type: application/json" \ + -d '{"method":"initialize","params":{"protocolVersion":"1.0"}}' +``` + +**Direct ACP testing:** ```bash # Test initialize echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"1.0"}}' | \ - docker run -i kandev/your-agent:latest --acp + docker run -i kandev/your-agent:latest /usr/local/bin/agent --acp # Test full flow ( echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"1.0"}}' echo '{"jsonrpc":"2.0","id":2,"method":"session/new","params":{}}' echo '{"jsonrpc":"2.0","id":3,"method":"session/prompt","params":{"message":"test"}}' -) | docker run -i kandev/your-agent:latest --acp +) | docker run -i kandev/your-agent:latest /usr/local/bin/agent --acp ``` --- ## Reference Implementation -See `backend/dockerfiles/augment-agent/` for a complete working example: +See `apps/backend/` for complete working examples: -- **Dockerfile** - Container setup -- **agent.sh** - Entry point and ACP handler -- **README.md** - Usage instructions +**agentctl:** +- `cmd/agentctl/main.go` - Entry point +- `internal/agentctl/` - HTTP server, process manager, ACP relay + +**Augment Agent:** +- `dockerfiles/augment-agent/Dockerfile` - Multi-stage build with agentctl +- `dockerfiles/augment-agent/entrypoint.sh` - Container entrypoint + +**Backend Integration:** +- `internal/agent/agentctl/client.go` - HTTP client for agentctl +- `internal/agent/lifecycle/manager.go` - Container lifecycle management --- @@ -659,8 +780,9 @@ See `backend/dockerfiles/augment-agent/` for a complete working example: - **System Architecture**: [ARCHITECTURE.md](ARCHITECTURE.md) - **REST API Specification**: [docs/openapi.yaml](docs/openapi.yaml) -- **Agent Registry**: `backend/internal/agent/registry/defaults.go` -- **ACP Implementation**: `backend/internal/agent/acp/` +- **Agent Registry**: `apps/backend/internal/agent/registry/defaults.go` +- **agentctl Implementation**: `apps/backend/internal/agentctl/` +- **ACP Protocol Types**: `apps/backend/pkg/acp/jsonrpc/` - **JSON-RPC 2.0 Spec**: https://www.jsonrpc.org/specification --- diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 11a0dbba15..752d44c23f 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -53,7 +53,7 @@ make test - **Language**: TypeScript - **Port**: 3000 -### Backend (`backend/`) +### Backend (`apps/backend/`) - **Language**: Go 1.21+ - **Framework**: Gin (HTTP), Gorilla WebSocket - **Database**: SQLite (WAL mode) @@ -74,16 +74,16 @@ make test ├── ARCHITECTURE.md # This file ├── AGENTS.md # Agent protocol details ├── apps/ -│ └── web/ # Next.js web application -│ ├── app/ # Pages (App Router) -│ ├── components/ # React components -│ └── lib/ # Utilities -├── backend/ -│ ├── cmd/kandev/ # Main entry point -│ ├── internal/ -│ │ ├── agent/ # Agent manager (Docker, ACP, lifecycle) -│ │ ├── task/ # Task service (CRUD, SQLite) -│ │ ├── orchestrator/ # Task execution coordinator +│ ├── web/ # Next.js web application +│ │ ├── app/ # Pages (App Router) +│ │ ├── components/ # React components +│ │ └── lib/ # Utilities +│ └── backend/ # Go backend services +│ ├── cmd/kandev/ # Main entry point +│ ├── internal/ +│ │ ├── agent/ # Agent manager (Docker, ACP, lifecycle) +│ │ ├── task/ # Task service (CRUD, SQLite) +│ │ ├── orchestrator/ # Task execution coordinator │ │ ├── events/ # Event bus (in-memory/NATS) │ │ ├── gateway/ # WebSocket gateway │ │ └── common/ # Shared utilities (config, logger, db) @@ -251,7 +251,7 @@ curl -X POST http://localhost:8080/api/v1/agents/launch \ **Location**: `./kandev.db` (SQLite with WAL mode) -**Schema** (`backend/internal/task/repository/sqlite.go`): +**Schema** (`apps/backend/internal/task/repository/sqlite.go`): - `boards` - Board records - `columns` - Column definitions with state mapping - `tasks` - Task records with JSON metadata @@ -307,9 +307,9 @@ make test-web | `ARCHITECTURE.md` | This file | | `AGENTS.md` | Agent protocol details | | `docs/openapi.yaml` | REST API spec | -| `backend/cmd/kandev/main.go` | Entry point | -| `backend/internal/agent/registry/defaults.go` | Agent type registry | -| `backend/internal/task/repository/sqlite.go` | Database schema | +| `apps/backend/cmd/kandev/main.go` | Entry point | +| `apps/backend/internal/agent/registry/defaults.go` | Agent type registry | +| `apps/backend/internal/task/repository/sqlite.go` | Database schema | | `apps/web/app/page.tsx` | Web app home | ## Contributing diff --git a/Makefile b/Makefile index 1fa16213a2..e174925400 100644 --- a/Makefile +++ b/Makefile @@ -2,7 +2,7 @@ # Orchestrates both backend (Go) and web app (Next.js) # Directories -BACKEND_DIR := backend +BACKEND_DIR := apps/backend WEB_DIR := apps/web # Tools diff --git a/README.md b/README.md index b41c7aed9c..2750e7f6b0 100644 --- a/README.md +++ b/README.md @@ -66,7 +66,7 @@ See [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) for detailed architecture docum ### Build and Run ```bash -cd backend +cd apps/backend # Build the binary make build @@ -113,27 +113,29 @@ curl -X POST http://localhost:8080/api/v1/agents/launch \ curl http://localhost:8080/api/v1/agents/{agent_id}/status ``` -See [backend/NEXT_STEPS.md](backend/NEXT_STEPS.md) for detailed API reference and next steps. +See [apps/backend/NEXT_STEPS.md](apps/backend/NEXT_STEPS.md) for detailed API reference and next steps. ## Project Structure ``` kandev/ -├── backend/ -│ ├── cmd/kandev/ # Unified binary entry point -│ ├── internal/ -│ │ ├── agent/ # Agent manager (docker, lifecycle, registry, api) -│ │ ├── task/ # Task service (models, repository, service, api) -│ │ ├── orchestrator/ # Orchestrator service -│ │ ├── events/ # In-memory event bus -│ │ └── common/ # Shared utilities (logger, config) -│ ├── dockerfiles/ # Agent container Dockerfiles -│ ├── Makefile -│ └── NEXT_STEPS.md # Development roadmap +├── apps/ +│ ├── backend/ # Go backend services +│ │ ├── cmd/kandev/ # Unified binary entry point +│ │ ├── internal/ +│ │ │ ├── agent/ # Agent manager (docker, lifecycle, registry, api) +│ │ │ ├── task/ # Task service (models, repository, service, api) +│ │ │ ├── orchestrator/ # Orchestrator service +│ │ │ ├── events/ # In-memory event bus +│ │ │ └── common/ # Shared utilities (logger, config) +│ │ ├── pkg/acp/ # ACP protocol types +│ │ ├── dockerfiles/ # Agent container Dockerfiles +│ │ └── Makefile +│ └── web/ # Frontend application ├── docs/ # Architecture and planning docs ├── migrations/ # Database migrations (future) ├── deployments/ # Deployment configs -└── scripts/ # Utility scripts +└── scripts/ # Utility scripts (including e2e-test.sh) ``` ## Configuration @@ -146,7 +148,7 @@ kandev/ ## Development ```bash -cd backend +cd apps/backend # Build make build @@ -177,7 +179,7 @@ go vet ./... - [ ] Authentication and authorization - [ ] Frontend UI -See [backend/NEXT_STEPS.md](backend/NEXT_STEPS.md) for the complete roadmap. +See [apps/backend/NEXT_STEPS.md](apps/backend/NEXT_STEPS.md) for the complete roadmap. ## License diff --git a/apps/agentctl/go.mod b/apps/agentctl/go.mod new file mode 100644 index 0000000000..59d89469f7 --- /dev/null +++ b/apps/agentctl/go.mod @@ -0,0 +1,41 @@ +module github.com/kandev/agentctl + +go 1.21 + +require ( + github.com/gin-gonic/gin v1.10.0 + github.com/google/uuid v1.6.0 + github.com/gorilla/websocket v1.5.3 + go.uber.org/zap v1.27.0 +) + +require ( + github.com/bytedance/sonic v1.11.6 // indirect + github.com/bytedance/sonic/loader v0.1.1 // indirect + github.com/cloudwego/base64x v0.1.4 // indirect + github.com/cloudwego/iasm v0.2.0 // indirect + github.com/gabriel-vasile/mimetype v1.4.3 // indirect + github.com/gin-contrib/sse v0.1.0 // indirect + github.com/go-playground/locales v0.14.1 // indirect + github.com/go-playground/universal-translator v0.18.1 // indirect + github.com/go-playground/validator/v10 v10.20.0 // indirect + github.com/goccy/go-json v0.10.2 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/cpuid/v2 v2.2.7 // indirect + github.com/leodido/go-urn v1.4.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/pelletier/go-toml/v2 v2.2.2 // indirect + github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/ugorji/go/codec v1.2.12 // indirect + go.uber.org/multierr v1.10.0 // indirect + golang.org/x/arch v0.8.0 // indirect + golang.org/x/crypto v0.23.0 // indirect + golang.org/x/net v0.25.0 // indirect + golang.org/x/sys v0.20.0 // indirect + golang.org/x/text v0.15.0 // indirect + google.golang.org/protobuf v1.34.1 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + diff --git a/backend/Makefile b/apps/backend/Makefile similarity index 100% rename from backend/Makefile rename to apps/backend/Makefile diff --git a/backend/NEXT_STEPS.md b/apps/backend/NEXT_STEPS.md similarity index 100% rename from backend/NEXT_STEPS.md rename to apps/backend/NEXT_STEPS.md diff --git a/apps/backend/agentctl b/apps/backend/agentctl new file mode 100755 index 0000000000..0cb776d978 Binary files /dev/null and b/apps/backend/agentctl differ diff --git a/apps/backend/bin/kandev b/apps/backend/bin/kandev new file mode 100755 index 0000000000..0b7c41f2fd Binary files /dev/null and b/apps/backend/bin/kandev differ diff --git a/backend/cmd/agent-manager/.gitkeep b/apps/backend/cmd/agent-manager/.gitkeep similarity index 100% rename from backend/cmd/agent-manager/.gitkeep rename to apps/backend/cmd/agent-manager/.gitkeep diff --git a/backend/cmd/agent-manager/main.go b/apps/backend/cmd/agent-manager/main.go similarity index 100% rename from backend/cmd/agent-manager/main.go rename to apps/backend/cmd/agent-manager/main.go diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go new file mode 100644 index 0000000000..7d703d4e71 --- /dev/null +++ b/apps/backend/cmd/agentctl/main.go @@ -0,0 +1,103 @@ +// Package main is the entry point for agentctl - a control process that runs +// inside Docker containers to manage AI agent processes. +// +// agentctl provides: +// - HTTP REST API for control operations (start, stop, pause, resume) +// - WebSocket streaming of agent output +// - Process management for the underlying agent (e.g., auggie --acp) +// - ACP message relay between HTTP API and agent subprocess +package main + +import ( + "context" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/kandev/kandev/internal/agentctl/api" + "github.com/kandev/kandev/internal/agentctl/config" + "github.com/kandev/kandev/internal/agentctl/process" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +func main() { + // Load configuration + cfg := config.Load() + + // Initialize logger + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: cfg.LogLevel, + Format: cfg.LogFormat, + OutputPath: "stdout", + }) + if err != nil { + fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err) + os.Exit(1) + } + defer log.Sync() + + log.Info("starting agentctl", + zap.String("version", "0.1.0"), + zap.Int("port", cfg.Port), + zap.String("agent_command", cfg.AgentCommand), + ) + + // Create process manager + procMgr := process.NewManager(cfg, log) + + // Create API server + apiServer := api.NewServer(cfg, procMgr, log) + + // Create HTTP server + server := &http.Server{ + Addr: fmt.Sprintf(":%d", cfg.Port), + Handler: apiServer.Router(), + ReadTimeout: 30 * time.Second, + WriteTimeout: 30 * time.Second, + } + + // Start server in goroutine + go func() { + log.Info("HTTP server listening", zap.String("addr", server.Addr)) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatal("HTTP server error", zap.Error(err)) + } + }() + + // Auto-start agent if configured + if cfg.AutoStart { + log.Info("auto-starting agent process") + ctx := context.Background() + if err := procMgr.Start(ctx); err != nil { + log.Error("failed to auto-start agent", zap.Error(err)) + } + } + + // Wait for shutdown signal + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + <-quit + + log.Info("shutting down agentctl") + + // Graceful shutdown + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + // Stop the agent process + if err := procMgr.Stop(ctx); err != nil { + log.Error("failed to stop agent process", zap.Error(err)) + } + + // Shutdown HTTP server + if err := server.Shutdown(ctx); err != nil { + log.Error("HTTP server shutdown error", zap.Error(err)) + } + + log.Info("agentctl stopped") +} + diff --git a/backend/cmd/api-gateway/.gitkeep b/apps/backend/cmd/api-gateway/.gitkeep similarity index 100% rename from backend/cmd/api-gateway/.gitkeep rename to apps/backend/cmd/api-gateway/.gitkeep diff --git a/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go similarity index 97% rename from backend/cmd/kandev/main.go rename to apps/backend/cmd/kandev/main.go index 928539ed44..80540c9391 100644 --- a/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -28,9 +28,6 @@ import ( "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/registry" - // Agent ACP - agentacp "github.com/kandev/kandev/internal/agent/acp" - // Orchestrator packages "github.com/kandev/kandev/internal/orchestrator" orchestratorapi "github.com/kandev/kandev/internal/orchestrator/api" @@ -132,7 +129,6 @@ func main() { // ============================================ var lifecycleMgr *lifecycle.Manager var agentRegistry *registry.Registry - var acpSessionMgr *agentacp.SessionManager if dockerClient != nil { log.Info("Initializing Agent Manager...") @@ -149,12 +145,8 @@ func main() { credsMgr.AddProvider(credentials.NewFileProvider(credsFile)) } - // ACP Session Manager for bidirectional communication - acpSessionMgr = agentacp.NewSessionManager(eventBus, log) - - // Lifecycle Manager + // Lifecycle Manager (uses agentctl for agent communication) lifecycleMgr = lifecycle.NewManager(dockerClient, agentRegistry, eventBus, log) - lifecycleMgr.SetACPManager(acpSessionMgr) lifecycleMgr.SetCredentialsManager(credsMgr) if err := lifecycleMgr.Start(ctx); err != nil { @@ -253,7 +245,8 @@ func main() { // Agent Manager routes (if available) if lifecycleMgr != nil && agentRegistry != nil && dockerClient != nil { - agentapi.SetupRoutes(v1Group, lifecycleMgr, agentRegistry, dockerClient, acpSessionMgr, log) + // Note: ACP is now handled via agentctl inside containers, so we pass nil for the ACP manager + agentapi.SetupRoutes(v1Group, lifecycleMgr, agentRegistry, dockerClient, nil, log) } // Health check diff --git a/backend/cmd/orchestrator/.gitkeep b/apps/backend/cmd/orchestrator/.gitkeep similarity index 100% rename from backend/cmd/orchestrator/.gitkeep rename to apps/backend/cmd/orchestrator/.gitkeep diff --git a/backend/cmd/orchestrator/main.go b/apps/backend/cmd/orchestrator/main.go similarity index 100% rename from backend/cmd/orchestrator/main.go rename to apps/backend/cmd/orchestrator/main.go diff --git a/backend/cmd/task-service/.gitkeep b/apps/backend/cmd/task-service/.gitkeep similarity index 100% rename from backend/cmd/task-service/.gitkeep rename to apps/backend/cmd/task-service/.gitkeep diff --git a/apps/backend/dockerfiles/augment-agent/Dockerfile b/apps/backend/dockerfiles/augment-agent/Dockerfile new file mode 100644 index 0000000000..93959bc941 --- /dev/null +++ b/apps/backend/dockerfiles/augment-agent/Dockerfile @@ -0,0 +1,72 @@ +# Augment Agent Dockerfile +# A coding agent powered by Auggie CLI (Augment Code) with agentctl sidecar + +# Stage 1: Build agentctl from Go source +FROM golang:1.24-alpine AS builder + +WORKDIR /build + +# Copy go mod files +COPY go.mod go.sum ./ +RUN go mod download + +# Copy source code +COPY . . + +# Build agentctl binary +RUN CGO_ENABLED=0 GOOS=linux go build -o /agentctl ./cmd/agentctl + +# Stage 2: Final image with Node.js and agentctl +FROM node:20-slim + +# Install dependencies +RUN apt-get update && apt-get install -y \ + git \ + curl \ + jq \ + python3 \ + python3-pip \ + ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +# Install Auggie CLI from npm +RUN npm install -g @augmentcode/auggie + +# Copy agentctl binary from builder +COPY --from=builder /agentctl /usr/local/bin/agentctl +RUN chmod +x /usr/local/bin/agentctl + +# Create agent directory +WORKDIR /agent + +# Copy agent script (legacy, for backward compatibility) +COPY dockerfiles/augment-agent/agent.sh /agent/agent.sh +RUN chmod +x /agent/agent.sh + +# Copy entrypoint script +COPY dockerfiles/augment-agent/entrypoint.sh /agent/entrypoint.sh +RUN chmod +x /agent/entrypoint.sh + +# Workspace will be mounted here +VOLUME /workspace +WORKDIR /workspace + +# Environment variables (passed at runtime) +# TASK_ID - The task identifier +# TASK_TITLE - Task title/name +# TASK_DESCRIPTION - Full task description +# AUGMENT_SESSION_AUTH - Session auth JSON for Augment (from ~/.augment/session.json) +# AUGGIE_SESSION_ID - (Optional) Session ID to resume a previous conversation +# +# agentctl configuration: +# AGENTCTL_PORT - HTTP API port (default: 9999) +# AGENTCTL_AGENT_COMMAND - Command to run (default: "auggie --acp") +# AGENTCTL_AUTO_START - Auto-start agent on startup (default: false) +# AGENTCTL_MODE - "agentctl" for HTTP mode, "direct" for legacy stdin/stdout mode + +# Expose agentctl HTTP API port +EXPOSE 9999 + +# Default entrypoint that decides between agentctl and direct mode +ENTRYPOINT ["/agent/entrypoint.sh"] + diff --git a/backend/dockerfiles/augment-agent/agent.sh b/apps/backend/dockerfiles/augment-agent/agent.sh similarity index 100% rename from backend/dockerfiles/augment-agent/agent.sh rename to apps/backend/dockerfiles/augment-agent/agent.sh diff --git a/apps/backend/dockerfiles/augment-agent/entrypoint.sh b/apps/backend/dockerfiles/augment-agent/entrypoint.sh new file mode 100644 index 0000000000..95ee9b1b9f --- /dev/null +++ b/apps/backend/dockerfiles/augment-agent/entrypoint.sh @@ -0,0 +1,41 @@ +#!/bin/bash +# Entrypoint script for Augment Agent +# Supports two modes: +# - "agentctl": Runs agentctl HTTP server that manages auggie subprocess +# - "direct": Legacy mode that runs auggie directly (stdin/stdout) + +set -e + +MODE="${AGENTCTL_MODE:-agentctl}" + +echo "Starting Augment Agent in ${MODE} mode..." + +# Common validation +if ! command -v auggie &> /dev/null; then + echo "ERROR: auggie CLI not found. Is @augmentcode/auggie installed?" + exit 1 +fi + +if [ -z "$AUGMENT_SESSION_AUTH" ]; then + echo "ERROR: AUGMENT_SESSION_AUTH environment variable not set" + exit 1 +fi + +case "$MODE" in + "agentctl") + # Run agentctl HTTP server + echo "Starting agentctl on port ${AGENTCTL_PORT:-9999}..." + exec /usr/local/bin/agentctl + ;; + "direct") + # Legacy mode: run auggie directly + echo "Running auggie directly in ACP mode..." + exec auggie --acp --workspace-root /workspace + ;; + *) + echo "ERROR: Unknown AGENTCTL_MODE: $MODE" + echo "Valid modes: agentctl, direct" + exit 1 + ;; +esac + diff --git a/backend/go.mod b/apps/backend/go.mod similarity index 98% rename from backend/go.mod rename to apps/backend/go.mod index 7c098fbd1c..2c67dc4fc0 100644 --- a/backend/go.mod +++ b/apps/backend/go.mod @@ -20,6 +20,7 @@ require ( github.com/bytedance/sonic v1.9.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect + github.com/coder/acp-go-sdk v0.6.3 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/log v0.1.0 // indirect diff --git a/backend/go.sum b/apps/backend/go.sum similarity index 99% rename from backend/go.sum rename to apps/backend/go.sum index ad001d4ffb..9244644b2e 100644 --- a/backend/go.sum +++ b/apps/backend/go.sum @@ -60,6 +60,8 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk= github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk= +github.com/coder/acp-go-sdk v0.6.3 h1:LsXQytehdjKIYJnoVWON/nf7mqbiarnyuyE3rrjBsXQ= +github.com/coder/acp-go-sdk v0.6.3/go.mod h1:yKzM/3R9uELp4+nBAwwtkS0aN1FOFjo11CNPy37yFko= github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= diff --git a/backend/internal/agent/acp/.gitkeep b/apps/backend/internal/agent/acp/.gitkeep similarity index 100% rename from backend/internal/agent/acp/.gitkeep rename to apps/backend/internal/agent/acp/.gitkeep diff --git a/backend/internal/agent/acp/session.go b/apps/backend/internal/agent/acp/session.go similarity index 100% rename from backend/internal/agent/acp/session.go rename to apps/backend/internal/agent/acp/session.go diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go new file mode 100644 index 0000000000..cab3845ad8 --- /dev/null +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -0,0 +1,275 @@ +package agentctl + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "time" + + "github.com/coder/acp-go-sdk" + "github.com/gorilla/websocket" + "go.uber.org/zap" +) + +// InitializeResponse from agentctl +type InitializeResponse struct { + Success bool `json:"success"` + Response *acp.InitializeResponse `json:"response,omitempty"` + Error string `json:"error,omitempty"` +} + +// Initialize sends the ACP initialize request +func (c *Client) Initialize(ctx context.Context, clientName, clientVersion string) (*acp.InitializeResponse, error) { + reqBody := struct { + ClientName string `json:"client_name"` + ClientVersion string `json:"client_version"` + }{ + ClientName: clientName, + ClientVersion: clientVersion, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/initialize", bytes.NewReader(body)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var result InitializeResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, err + } + if !result.Success { + return nil, fmt.Errorf("initialize failed: %s", result.Error) + } + return result.Response, nil +} + +// NewSessionResponse from agentctl +type NewSessionResponse struct { + Success bool `json:"success"` + SessionID string `json:"session_id,omitempty"` + Error string `json:"error,omitempty"` +} + +// NewSession creates a new ACP session +func (c *Client) NewSession(ctx context.Context, cwd string) (string, error) { + reqBody := struct { + Cwd string `json:"cwd"` + }{Cwd: cwd} + + body, err := json.Marshal(reqBody) + if err != nil { + return "", err + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/session/new", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result NewSessionResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + if !result.Success { + return "", fmt.Errorf("new session failed: %s", result.Error) + } + return result.SessionID, nil +} + +// Prompt sends a prompt to the agent (non-blocking - returns immediately, use StreamUpdates for responses) +func (c *Client) Prompt(ctx context.Context, text string) error { + reqBody := struct { + Text string `json:"text"` + }{Text: text} + + body, err := json.Marshal(reqBody) + if err != nil { + return err + } + + // Use a longer timeout for prompt - agent may take time + client := &http.Client{ + Timeout: 10 * time.Minute, + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/prompt", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + var result struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("prompt failed: %s", result.Error) + } + return nil +} + +// SessionNotification represents a session update from the agent +type SessionNotification = acp.SessionNotification + +// StreamUpdates opens a WebSocket connection for streaming session updates +func (c *Client) StreamUpdates(ctx context.Context, handler func(acp.SessionNotification)) error { + wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream" + + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to updates stream: %w", err) + } + + c.mu.Lock() + c.acpConn = conn + c.mu.Unlock() + + c.logger.Info("connected to updates stream", zap.String("url", wsURL)) + + // Read messages in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.acpConn = nil + c.mu.Unlock() + conn.Close() + }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Info("updates stream closed normally") + } else { + c.logger.Debug("updates stream error", zap.Error(err)) + } + return + } + + var notification acp.SessionNotification + if err := json.Unmarshal(message, ¬ification); err != nil { + c.logger.Warn("failed to parse session notification", zap.Error(err)) + continue + } + + handler(notification) + } + }() + + return nil +} + +// CloseUpdatesStream closes the updates stream connection +func (c *Client) CloseUpdatesStream() { + c.CloseACPStream() +} + +// CloseACPStream closes the ACP stream connection +func (c *Client) CloseACPStream() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.acpConn != nil { + c.acpConn.Close() + c.acpConn = nil + } +} + +// StreamOutput opens a WebSocket connection for streaming output +func (c *Client) StreamOutput(ctx context.Context, includeHistory bool, handler func(*OutputLine)) error { + wsURL := "ws" + c.baseURL[4:] + "/api/v1/output/stream" + if includeHistory { + wsURL += "?history=true" + } + + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to output stream: %w", err) + } + + c.mu.Lock() + c.outputConn = conn + c.mu.Unlock() + + c.logger.Info("connected to output stream", zap.String("url", wsURL)) + + // Read messages in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.outputConn = nil + c.mu.Unlock() + conn.Close() + }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Info("output stream closed normally") + } else { + c.logger.Debug("output stream error", zap.Error(err)) + } + return + } + + var line OutputLine + if err := json.Unmarshal(message, &line); err != nil { + c.logger.Warn("failed to parse output line", zap.Error(err)) + continue + } + + handler(&line) + } + }() + + return nil +} + +// CloseOutputStream closes the output stream connection +func (c *Client) CloseOutputStream() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.outputConn != nil { + c.outputConn.Close() + c.outputConn = nil + } +} + +// Close closes all connections +func (c *Client) Close() { + c.CloseACPStream() + c.CloseOutputStream() +} + diff --git a/apps/backend/internal/agent/agentctl/client.go b/apps/backend/internal/agent/agentctl/client.go new file mode 100644 index 0000000000..05386ae7c5 --- /dev/null +++ b/apps/backend/internal/agent/agentctl/client.go @@ -0,0 +1,170 @@ +// Package agentctl provides a client for communicating with agentctl running inside containers +package agentctl + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "sync" + "time" + + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Client communicates with agentctl via HTTP +type Client struct { + baseURL string + httpClient *http.Client + logger *logger.Logger + + // WebSocket connections for streaming + outputConn *websocket.Conn + acpConn *websocket.Conn + mu sync.RWMutex +} + +// StatusResponse from agentctl +type StatusResponse struct { + AgentStatus string `json:"agent_status"` + ProcessInfo map[string]interface{} `json:"process_info"` +} + +// OutputLine from agentctl +type OutputLine struct { + Timestamp time.Time `json:"timestamp"` + Stream string `json:"stream"` + Content string `json:"content"` +} + +// NewClient creates a new agentctl client +func NewClient(host string, port int, log *logger.Logger) *Client { + return &Client{ + baseURL: fmt.Sprintf("http://%s:%d", host, port), + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + logger: log.WithFields(zap.String("component", "agentctl-client")), + } +} + +// Health checks if agentctl is healthy +func (c *Client) Health(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/health", nil) + if err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("health check failed: %d", resp.StatusCode) + } + return nil +} + +// GetStatus returns the agent status +func (c *Client) GetStatus(ctx context.Context) (*StatusResponse, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/status", nil) + if err != nil { + return nil, err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + var status StatusResponse + if err := json.NewDecoder(resp.Body).Decode(&status); err != nil { + return nil, err + } + return &status, nil +} + +// Start starts the agent process +func (c *Client) Start(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/start", nil) + if err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + var result struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("start failed: %s", result.Error) + } + return nil +} + +// Stop stops the agent process +func (c *Client) Stop(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/stop", nil) + if err != nil { + return err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + var result struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("stop failed: %s", result.Error) + } + return nil +} + +// WaitForReady waits until agentctl is ready to accept requests +func (c *Client) WaitForReady(ctx context.Context, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + if time.Now().After(deadline) { + return fmt.Errorf("timeout waiting for agentctl to be ready") + } + + if err := c.Health(ctx); err == nil { + c.logger.Info("agentctl is ready") + return nil + } + } + } +} + +// BaseURL returns the base URL of the agentctl client +func (c *Client) BaseURL() string { + return c.baseURL +} diff --git a/backend/internal/agent/api/handlers.go b/apps/backend/internal/agent/api/handlers.go similarity index 100% rename from backend/internal/agent/api/handlers.go rename to apps/backend/internal/agent/api/handlers.go diff --git a/backend/internal/agent/api/handlers_test.go b/apps/backend/internal/agent/api/handlers_test.go similarity index 100% rename from backend/internal/agent/api/handlers_test.go rename to apps/backend/internal/agent/api/handlers_test.go diff --git a/backend/internal/agent/api/requests.go b/apps/backend/internal/agent/api/requests.go similarity index 100% rename from backend/internal/agent/api/requests.go rename to apps/backend/internal/agent/api/requests.go diff --git a/backend/internal/agent/api/router.go b/apps/backend/internal/agent/api/router.go similarity index 100% rename from backend/internal/agent/api/router.go rename to apps/backend/internal/agent/api/router.go diff --git a/backend/internal/agent/credentials/.gitkeep b/apps/backend/internal/agent/credentials/.gitkeep similarity index 100% rename from backend/internal/agent/credentials/.gitkeep rename to apps/backend/internal/agent/credentials/.gitkeep diff --git a/backend/internal/agent/credentials/augment_provider.go b/apps/backend/internal/agent/credentials/augment_provider.go similarity index 100% rename from backend/internal/agent/credentials/augment_provider.go rename to apps/backend/internal/agent/credentials/augment_provider.go diff --git a/backend/internal/agent/credentials/env_provider.go b/apps/backend/internal/agent/credentials/env_provider.go similarity index 100% rename from backend/internal/agent/credentials/env_provider.go rename to apps/backend/internal/agent/credentials/env_provider.go diff --git a/backend/internal/agent/credentials/file_provider.go b/apps/backend/internal/agent/credentials/file_provider.go similarity index 100% rename from backend/internal/agent/credentials/file_provider.go rename to apps/backend/internal/agent/credentials/file_provider.go diff --git a/backend/internal/agent/credentials/manager.go b/apps/backend/internal/agent/credentials/manager.go similarity index 100% rename from backend/internal/agent/credentials/manager.go rename to apps/backend/internal/agent/credentials/manager.go diff --git a/backend/internal/agent/credentials/manager_test.go b/apps/backend/internal/agent/credentials/manager_test.go similarity index 100% rename from backend/internal/agent/credentials/manager_test.go rename to apps/backend/internal/agent/credentials/manager_test.go diff --git a/backend/internal/agent/docker/.gitkeep b/apps/backend/internal/agent/docker/.gitkeep similarity index 100% rename from backend/internal/agent/docker/.gitkeep rename to apps/backend/internal/agent/docker/.gitkeep diff --git a/backend/internal/agent/docker/client.go b/apps/backend/internal/agent/docker/client.go similarity index 93% rename from backend/internal/agent/docker/client.go rename to apps/backend/internal/agent/docker/client.go index d3418d9b3a..1c78713fa1 100644 --- a/backend/internal/agent/docker/client.go +++ b/apps/backend/internal/agent/docker/client.go @@ -284,6 +284,38 @@ func (c *Client) GetContainerInfo(ctx context.Context, containerID string) (*Con return info, nil } +// GetContainerIP returns the IP address of a container +func (c *Client) GetContainerIP(ctx context.Context, containerID string) (string, error) { + c.logger.Debug("Getting container IP", zap.String("container_id", containerID)) + + inspect, err := c.cli.ContainerInspect(ctx, containerID) + if err != nil { + c.logger.Error("Failed to inspect container for IP", zap.String("container_id", containerID), zap.Error(err)) + return "", err + } + + // Try to get IP from default bridge network first + if inspect.NetworkSettings != nil { + // Check bridge network + if inspect.NetworkSettings.IPAddress != "" { + return inspect.NetworkSettings.IPAddress, nil + } + + // Check other networks + for netName, netSettings := range inspect.NetworkSettings.Networks { + if netSettings.IPAddress != "" { + c.logger.Debug("Found container IP", + zap.String("container_id", containerID), + zap.String("network", netName), + zap.String("ip", netSettings.IPAddress)) + return netSettings.IPAddress, nil + } + } + } + + return "", fmt.Errorf("no IP address found for container %s", containerID) +} + // GetContainerLogs returns logs from a container. func (c *Client) GetContainerLogs(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error) { c.logger.Debug("Getting container logs", diff --git a/backend/internal/agent/lifecycle/.gitkeep b/apps/backend/internal/agent/lifecycle/.gitkeep similarity index 100% rename from backend/internal/agent/lifecycle/.gitkeep rename to apps/backend/internal/agent/lifecycle/.gitkeep diff --git a/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go similarity index 64% rename from backend/internal/agent/lifecycle/manager.go rename to apps/backend/internal/agent/lifecycle/manager.go index b5a1c8ca4b..6d854a34f8 100644 --- a/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -5,7 +5,6 @@ package lifecycle import ( "context" "fmt" - "io" "os" "path/filepath" "strings" @@ -15,8 +14,7 @@ import ( "github.com/google/uuid" "go.uber.org/zap" - // ACP session management is used via the ACPManager interface - _ "github.com/kandev/kandev/internal/agent/acp" + "github.com/kandev/kandev/internal/agent/agentctl" "github.com/kandev/kandev/internal/agent/docker" "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/common/logger" @@ -25,12 +23,16 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" ) +// Default agentctl port +const AgentCtlPort = 9999 + // AgentInstance represents a running agent instance type AgentInstance struct { ID string TaskID string AgentType string ContainerID string + ContainerIP string // IP address of the container for agentctl communication Status v1.AgentStatus StartedAt time.Time FinishedAt *time.Time @@ -38,6 +40,9 @@ type AgentInstance struct { ErrorMessage string Progress int Metadata map[string]interface{} + + // agentctl client for this instance + agentctl *agentctl.Client } // LaunchRequest contains parameters for launching an agent @@ -50,18 +55,6 @@ type LaunchRequest struct { Metadata map[string]interface{} } -// ACPManager interface for ACP session management -type ACPManager interface { - CreateSession(ctx context.Context, instanceID, taskID string, stdin io.WriteCloser, stdout io.Reader) error - Initialize(ctx context.Context, instanceID string) error - NewSession(ctx context.Context, instanceID, cwd string) (string, error) - LoadSession(ctx context.Context, instanceID, sessionID string) error - Prompt(ctx context.Context, instanceID, message string) error - Cancel(ctx context.Context, instanceID, reason string) error - CloseSession(instanceID string) error - GetSessionID(instanceID string) (string, bool) -} - // CredentialsManager interface for credential retrieval type CredentialsManager interface { GetCredentialValue(ctx context.Context, key string) (value string, err error) @@ -72,7 +65,6 @@ type Manager struct { docker *docker.Client registry *registry.Registry eventBus bus.EventBus - acpMgr ACPManager credsMgr CredentialsManager logger *logger.Logger @@ -108,11 +100,6 @@ func NewManager( } } -// SetACPManager sets the ACP session manager -func (m *Manager) SetACPManager(acpMgr ACPManager) { - m.acpMgr = acpMgr -} - // SetCredentialsManager sets the credentials manager for injecting secrets func (m *Manager) SetCredentialsManager(credsMgr CredentialsManager) { m.credsMgr = credsMgr @@ -169,29 +156,42 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentInstanc // 4. Build container config from registry config containerConfig := m.buildContainerConfig(instanceID, req, agentConfig) - // 5. Create and start the container WITH stdin attached for ACP - containerID, err := m.docker.CreateContainerInteractive(ctx, containerConfig) + // 5. Create and start the container + containerID, err := m.docker.CreateContainer(ctx, containerConfig) if err != nil { return nil, fmt.Errorf("failed to create container: %w", err) } if err := m.docker.StartContainer(ctx, containerID); err != nil { - // Cleanup the created container on failure _ = m.docker.RemoveContainer(ctx, containerID, true) return nil, fmt.Errorf("failed to start container: %w", err) } - // 6. Track the instance + // 6. Get container IP for agentctl communication + containerIP, err := m.docker.GetContainerIP(ctx, containerID) + if err != nil { + m.logger.Warn("failed to get container IP, trying localhost", + zap.String("container_id", containerID), + zap.Error(err)) + containerIP = "127.0.0.1" + } + + // 7. Create agentctl client + agentctlClient := agentctl.NewClient(containerIP, AgentCtlPort, m.logger) + + // 8. Track the instance now := time.Now() instance := &AgentInstance{ ID: instanceID, TaskID: req.TaskID, AgentType: req.AgentType, ContainerID: containerID, + ContainerIP: containerIP, Status: v1.AgentStatusRunning, StartedAt: now, Progress: 0, Metadata: req.Metadata, + agentctl: agentctlClient, } m.mu.Lock() @@ -200,82 +200,262 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentInstanc m.byContainer[containerID] = instanceID m.mu.Unlock() - // 7. Publish agent.started event + // 9. Publish agent.started event m.publishEvent(ctx, events.AgentStarted, instance) - // 8. Setup ACP session for bidirectional communication - if m.acpMgr != nil { - // Attach to container stdin/stdout - attachResult, err := m.docker.AttachContainer(context.Background(), containerID) - if err != nil { - m.logger.Warn("failed to attach to container for ACP", - zap.String("container_id", containerID), - zap.Error(err)) - } else { - // Create ACP session - if err := m.acpMgr.CreateSession(context.Background(), instanceID, req.TaskID, attachResult.Stdin, attachResult.Stdout); err != nil { - m.logger.Warn("failed to create ACP session", - zap.String("instance_id", instanceID), - zap.Error(err)) - } else { - // Initialize the ACP session - if err := m.acpMgr.Initialize(context.Background(), instanceID); err != nil { - m.logger.Warn("failed to initialize ACP session", - zap.String("instance_id", instanceID), - zap.Error(err)) - } else { - // Create or load session based on metadata - // The workspace is mounted at /workspace inside the container - containerCwd := "/workspace" - var sessionID string - if req.Metadata != nil { - if existingSessionID, ok := req.Metadata["auggie_session_id"].(string); ok && existingSessionID != "" { - // Resume existing session - if err := m.acpMgr.LoadSession(context.Background(), instanceID, existingSessionID); err != nil { - m.logger.Warn("failed to load existing ACP session, creating new one", - zap.String("instance_id", instanceID), - zap.String("existing_session_id", existingSessionID), - zap.Error(err)) - sessionID, _ = m.acpMgr.NewSession(context.Background(), instanceID, containerCwd) - } else { - sessionID = existingSessionID - } - } else { - sessionID, _ = m.acpMgr.NewSession(context.Background(), instanceID, containerCwd) - } - } else { - sessionID, _ = m.acpMgr.NewSession(context.Background(), instanceID, containerCwd) - } - - m.logger.Info("ACP session ready", - zap.String("instance_id", instanceID), - zap.String("session_id", sessionID)) - - // Send the initial prompt with the task description - if req.TaskDescription != "" { - if err := m.acpMgr.Prompt(context.Background(), instanceID, req.TaskDescription); err != nil { - m.logger.Warn("failed to send initial prompt", - zap.String("instance_id", instanceID), - zap.Error(err)) - } else { - m.logger.Info("sent initial prompt to agent", - zap.String("instance_id", instanceID), - zap.String("task_description", req.TaskDescription)) - } - } - } - } - } - } + // 10. Wait for agentctl to be ready and start the agent process + go m.initializeAgent(instance, req.TaskDescription) m.logger.Info("agent launched successfully", zap.String("instance_id", instanceID), zap.String("container_id", containerID), + zap.String("container_ip", containerIP), zap.String("task_id", req.TaskID)) return instance, nil } +// initializeAgent waits for agentctl to be ready and initializes the agent +func (m *Manager) initializeAgent(instance *AgentInstance, taskDescription string) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + m.logger.Info("waiting for agentctl to be ready", + zap.String("instance_id", instance.ID), + zap.String("url", instance.agentctl.BaseURL())) + + // Wait for agentctl HTTP server to be ready + if err := instance.agentctl.WaitForReady(ctx, 60*time.Second); err != nil { + m.logger.Error("agentctl not ready", + zap.String("instance_id", instance.ID), + zap.Error(err)) + m.updateInstanceError(instance.ID, "agentctl not ready: "+err.Error()) + return + } + + // Start the agent process via agentctl + if err := instance.agentctl.Start(ctx); err != nil { + m.logger.Error("failed to start agent via agentctl", + zap.String("instance_id", instance.ID), + zap.Error(err)) + m.updateInstanceError(instance.ID, "failed to start agent: "+err.Error()) + return + } + + m.logger.Info("agent process started via agentctl", + zap.String("instance_id", instance.ID)) + + // Give the agent process a moment to initialize + time.Sleep(500 * time.Millisecond) + + // Initialize ACP session + if err := m.initializeACPSession(ctx, instance, taskDescription); err != nil { + m.logger.Error("failed to initialize ACP session", + zap.String("instance_id", instance.ID), + zap.Error(err)) + m.updateInstanceError(instance.ID, "failed to initialize ACP: "+err.Error()) + return + } +} + +// initializeACPSession sends the ACP initialization messages using the acp-go-sdk +func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInstance, taskDescription string) error { + m.logger.Info("initializing ACP session", + zap.String("instance_id", instance.ID), + zap.String("agentctl_url", instance.agentctl.BaseURL())) + + // 1. Send initialize request + m.logger.Info("sending ACP initialize request", + zap.String("instance_id", instance.ID)) + + initResp, err := instance.agentctl.Initialize(ctx, "kandev", "1.0.0") + if err != nil { + m.logger.Error("ACP initialize failed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + return fmt.Errorf("initialize failed: %w", err) + } + m.logger.Info("ACP initialize response received", + zap.String("instance_id", instance.ID), + zap.String("agent_name", initResp.AgentInfo.Name), + zap.String("agent_version", initResp.AgentInfo.Version)) + + // 2. Create new session + m.logger.Info("sending ACP session/new request", + zap.String("instance_id", instance.ID)) + + sessionID, err := instance.agentctl.NewSession(ctx, "/workspace") + if err != nil { + m.logger.Error("ACP session/new failed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + return fmt.Errorf("session/new failed: %w", err) + } + m.logger.Info("ACP session created", + zap.String("instance_id", instance.ID), + zap.String("session_id", sessionID)) + + // Store session ID in instance + m.mu.Lock() + if inst, exists := m.instances[instance.ID]; exists { + if inst.Metadata == nil { + inst.Metadata = make(map[string]interface{}) + } + inst.Metadata["acp_session_id"] = sessionID + } + m.mu.Unlock() + + // 3. Set up updates stream to receive session notifications + go m.handleUpdatesStream(instance) + + // 4. Send the task prompt if provided + if taskDescription != "" { + m.logger.Info("sending ACP prompt", + zap.String("instance_id", instance.ID), + zap.String("session_id", sessionID), + zap.String("task_description", taskDescription)) + + // Prompt is SYNCHRONOUS - it blocks until the agent completes the task + // Use a long timeout context for this + promptCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + + err := instance.agentctl.Prompt(promptCtx, taskDescription) + if err != nil { + m.logger.Error("ACP prompt failed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + return fmt.Errorf("prompt failed: %w", err) + } + + // Prompt completed successfully - mark agent as completed + m.logger.Info("ACP prompt completed successfully", + zap.String("instance_id", instance.ID)) + if err := m.MarkCompleted(instance.ID, 0, ""); err != nil { + m.logger.Error("failed to mark instance as completed", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } + } else { + m.logger.Warn("no task description provided, skipping prompt", + zap.String("instance_id", instance.ID)) + } + + return nil +} + +// handleUpdatesStream handles streaming session notifications from the agent +func (m *Manager) handleUpdatesStream(instance *AgentInstance) { + ctx := context.Background() + + err := instance.agentctl.StreamUpdates(ctx, func(notification agentctl.SessionNotification) { + m.handleSessionNotification(instance, notification) + }) + if err != nil { + m.logger.Error("failed to connect to updates stream", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + +// handleSessionNotification processes incoming session notifications from the agent +func (m *Manager) handleSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) { + m.logger.Debug("received session notification", + zap.String("instance_id", instance.ID), + zap.String("session_id", string(notification.SessionId))) + + update := notification.Update + + // Handle different update types + if update.AgentMessageChunk != nil { + m.logger.Debug("agent message chunk", + zap.String("instance_id", instance.ID)) + m.updateInstanceProgress(instance.ID, 50) + } + + if update.AgentThoughtChunk != nil { + m.logger.Debug("agent thought chunk", + zap.String("instance_id", instance.ID)) + } + + if update.ToolCall != nil { + m.logger.Info("tool call started", + zap.String("instance_id", instance.ID), + zap.String("tool_call_id", string(update.ToolCall.ToolCallId))) + m.updateInstanceProgress(instance.ID, 60) + } + + if update.ToolCallUpdate != nil { + m.logger.Debug("tool call update", + zap.String("instance_id", instance.ID), + zap.String("tool_call_id", string(update.ToolCallUpdate.ToolCallId))) + + // Check if tool call completed + if update.ToolCallUpdate.Status != nil { + switch *update.ToolCallUpdate.Status { + case "complete": + m.logger.Info("tool call completed", + zap.String("instance_id", instance.ID)) + m.updateInstanceProgress(instance.ID, 80) + case "error": + m.logger.Error("tool call error", + zap.String("instance_id", instance.ID)) + } + } + } + + if update.Plan != nil { + m.logger.Info("agent plan update", + zap.String("instance_id", instance.ID), + zap.Int("num_entries", len(update.Plan.Entries))) + } +} + +// updateInstanceProgress updates an instance's progress +func (m *Manager) updateInstanceProgress(instanceID string, progress int) { + m.mu.Lock() + defer m.mu.Unlock() + + instance, exists := m.instances[instanceID] + if !exists { + return + } + + instance.Progress = progress +} + +// updateInstanceComplete marks an instance as completed +func (m *Manager) updateInstanceComplete(instanceID string) { + m.mu.Lock() + defer m.mu.Unlock() + + instance, exists := m.instances[instanceID] + if !exists { + return + } + + instance.Status = v1.AgentStatusCompleted + instance.Progress = 100 + now := time.Now() + instance.FinishedAt = &now +} + +// updateInstanceError updates an instance with an error +func (m *Manager) updateInstanceError(instanceID, errorMsg string) { + m.mu.Lock() + defer m.mu.Unlock() + + instance, exists := m.instances[instanceID] + if !exists { + return + } + + instance.Status = v1.AgentStatusFailed + instance.ErrorMessage = errorMsg + now := time.Now() + instance.FinishedAt = &now +} + // buildContainerConfig builds a Docker container config from agent registry config func (m *Manager) buildContainerConfig(instanceID string, req *LaunchRequest, agentConfig *registry.AgentTypeConfig) docker.ContainerConfig { // Build image name with tag @@ -393,6 +573,17 @@ func (m *Manager) StopAgent(ctx context.Context, instanceID string, force bool) zap.String("instance_id", instanceID), zap.Bool("force", force)) + // Try to gracefully stop via agentctl first + if instance.agentctl != nil && !force { + if err := instance.agentctl.Stop(ctx); err != nil { + m.logger.Warn("failed to stop agent via agentctl, will stop container", + zap.String("instance_id", instanceID), + zap.Error(err)) + } + instance.agentctl.Close() + } + + // Stop the container var err error if force { err = m.docker.KillContainer(ctx, instance.ContainerID, "SIGKILL") @@ -404,11 +595,6 @@ func (m *Manager) StopAgent(ctx context.Context, instanceID string, force bool) return fmt.Errorf("failed to stop container: %w", err) } - // Close ACP session if exists - if m.acpMgr != nil { - _ = m.acpMgr.CloseSession(instanceID) - } - // Update instance status m.mu.Lock() instance.Status = v1.AgentStatusStopped diff --git a/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go similarity index 100% rename from backend/internal/agent/lifecycle/manager_test.go rename to apps/backend/internal/agent/lifecycle/manager_test.go diff --git a/backend/internal/agent/models/.gitkeep b/apps/backend/internal/agent/models/.gitkeep similarity index 100% rename from backend/internal/agent/models/.gitkeep rename to apps/backend/internal/agent/models/.gitkeep diff --git a/backend/internal/agent/monitor/.gitkeep b/apps/backend/internal/agent/monitor/.gitkeep similarity index 100% rename from backend/internal/agent/monitor/.gitkeep rename to apps/backend/internal/agent/monitor/.gitkeep diff --git a/backend/internal/agent/registry/.gitkeep b/apps/backend/internal/agent/registry/.gitkeep similarity index 100% rename from backend/internal/agent/registry/.gitkeep rename to apps/backend/internal/agent/registry/.gitkeep diff --git a/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go similarity index 100% rename from backend/internal/agent/registry/defaults.go rename to apps/backend/internal/agent/registry/defaults.go diff --git a/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go similarity index 100% rename from backend/internal/agent/registry/registry.go rename to apps/backend/internal/agent/registry/registry.go diff --git a/backend/internal/agent/registry/registry_test.go b/apps/backend/internal/agent/registry/registry_test.go similarity index 100% rename from backend/internal/agent/registry/registry_test.go rename to apps/backend/internal/agent/registry/registry_test.go diff --git a/backend/internal/agent/streaming/.gitkeep b/apps/backend/internal/agent/streaming/.gitkeep similarity index 100% rename from backend/internal/agent/streaming/.gitkeep rename to apps/backend/internal/agent/streaming/.gitkeep diff --git a/backend/internal/agent/streaming/manager.go b/apps/backend/internal/agent/streaming/manager.go similarity index 100% rename from backend/internal/agent/streaming/manager.go rename to apps/backend/internal/agent/streaming/manager.go diff --git a/backend/internal/agent/streaming/reader.go b/apps/backend/internal/agent/streaming/reader.go similarity index 100% rename from backend/internal/agent/streaming/reader.go rename to apps/backend/internal/agent/streaming/reader.go diff --git a/apps/backend/internal/agentctl/acp/client.go b/apps/backend/internal/agentctl/acp/client.go new file mode 100644 index 0000000000..4fa784db12 --- /dev/null +++ b/apps/backend/internal/agentctl/acp/client.go @@ -0,0 +1,247 @@ +// Package acp implements the ACP client interface for agentctl +package acp + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + + "github.com/coder/acp-go-sdk" + "go.uber.org/zap" +) + +// UpdateHandler is called when session updates are received from the agent +type UpdateHandler func(notification acp.SessionNotification) + +// Client implements acp.Client interface and handles all agent requests +type Client struct { + logger *zap.Logger + workspaceRoot string + + mu sync.RWMutex + updateHandler UpdateHandler +} + +// ClientOption configures a Client +type ClientOption func(*Client) + +// WithLogger sets the logger +func WithLogger(l *zap.Logger) ClientOption { + return func(c *Client) { + c.logger = l + } +} + +// WithWorkspaceRoot sets the workspace root for file operations +func WithWorkspaceRoot(root string) ClientOption { + return func(c *Client) { + c.workspaceRoot = root + } +} + +// WithUpdateHandler sets the handler for session updates +func WithUpdateHandler(h UpdateHandler) ClientOption { + return func(c *Client) { + c.updateHandler = h + } +} + +// NewClient creates a new ACP client implementation +func NewClient(opts ...ClientOption) *Client { + c := &Client{ + logger: zap.NewNop(), + workspaceRoot: "/workspace", + } + for _, opt := range opts { + opt(c) + } + return c +} + +// SetUpdateHandler sets the update handler (thread-safe) +func (c *Client) SetUpdateHandler(h UpdateHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.updateHandler = h +} + +// RequestPermission handles permission requests from the agent +// Auto-approves by selecting the first "allow" option +func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { + title := "" + if p.ToolCall.Title != nil { + title = *p.ToolCall.Title + } + c.logger.Info("received permission request", + zap.String("session_id", string(p.SessionId)), + zap.String("tool_call_id", string(p.ToolCall.ToolCallId)), + zap.String("title", title), + zap.Int("num_options", len(p.Options))) + + // No options - cancel + if len(p.Options) == 0 { + c.logger.Warn("no options available, cancelling permission request") + return acp.RequestPermissionResponse{ + Outcome: acp.RequestPermissionOutcome{ + Cancelled: &acp.RequestPermissionOutcomeCancelled{}, + }, + }, nil + } + + // Find the first "allow" option + var selectedOption *acp.PermissionOption + for i := range p.Options { + opt := &p.Options[i] + if opt.Kind == acp.PermissionOptionKindAllowOnce || opt.Kind == acp.PermissionOptionKindAllowAlways { + selectedOption = opt + break + } + } + + // If no allow option, use the first option + if selectedOption == nil { + selectedOption = &p.Options[0] + } + + c.logger.Info("auto-approving permission request", + zap.String("option_id", string(selectedOption.OptionId)), + zap.String("option_name", selectedOption.Name), + zap.String("kind", string(selectedOption.Kind))) + + return acp.RequestPermissionResponse{ + Outcome: acp.RequestPermissionOutcome{ + Selected: &acp.RequestPermissionOutcomeSelected{ + OptionId: selectedOption.OptionId, + }, + }, + }, nil +} + +// SessionUpdate handles session update notifications from the agent +func (c *Client) SessionUpdate(ctx context.Context, n acp.SessionNotification) error { + c.mu.RLock() + handler := c.updateHandler + c.mu.RUnlock() + + // Log the update type + u := n.Update + switch { + case u.AgentMessageChunk != nil: + if u.AgentMessageChunk.Content.Text != nil { + c.logger.Debug("agent message chunk", + zap.String("text", u.AgentMessageChunk.Content.Text.Text[:min(50, len(u.AgentMessageChunk.Content.Text.Text))])) + } + case u.ToolCall != nil: + c.logger.Info("tool call", + zap.String("tool_call_id", string(u.ToolCall.ToolCallId)), + zap.String("title", u.ToolCall.Title), + zap.String("status", string(u.ToolCall.Status))) + case u.ToolCallUpdate != nil: + c.logger.Debug("tool call update", + zap.String("tool_call_id", string(u.ToolCallUpdate.ToolCallId))) + case u.Plan != nil: + c.logger.Info("plan update", zap.Int("entries", len(u.Plan.Entries))) + } + + // Forward to handler if set + if handler != nil { + handler(n) + } + + return nil +} + +// ReadTextFile reads a text file from the workspace +func (c *Client) ReadTextFile(ctx context.Context, p acp.ReadTextFileRequest) (acp.ReadTextFileResponse, error) { + c.logger.Debug("reading file", zap.String("path", p.Path)) + + // Validate path is absolute + if !filepath.IsAbs(p.Path) { + return acp.ReadTextFileResponse{}, fmt.Errorf("path must be absolute: %s", p.Path) + } + + // Read the file + b, err := os.ReadFile(p.Path) + if err != nil { + return acp.ReadTextFileResponse{}, err + } + + content := string(b) + + // Handle line/limit parameters + if p.Line != nil || p.Limit != nil { + lines := strings.Split(content, "\n") + start := 0 + if p.Line != nil && *p.Line > 0 { + start = *p.Line - 1 + if start > len(lines) { + start = len(lines) + } + } + end := len(lines) + if p.Limit != nil && *p.Limit > 0 && start+*p.Limit < end { + end = start + *p.Limit + } + content = strings.Join(lines[start:end], "\n") + } + + return acp.ReadTextFileResponse{Content: content}, nil +} + +// WriteTextFile writes a text file to the workspace +func (c *Client) WriteTextFile(ctx context.Context, p acp.WriteTextFileRequest) (acp.WriteTextFileResponse, error) { + c.logger.Debug("writing file", zap.String("path", p.Path)) + + // Validate path is absolute + if !filepath.IsAbs(p.Path) { + return acp.WriteTextFileResponse{}, fmt.Errorf("path must be absolute: %s", p.Path) + } + + // Create directory if needed + if dir := filepath.Dir(p.Path); dir != "" { + if err := os.MkdirAll(dir, 0o755); err != nil { + return acp.WriteTextFileResponse{}, err + } + } + + return acp.WriteTextFileResponse{}, os.WriteFile(p.Path, []byte(p.Content), 0o644) +} + +// CreateTerminal creates a new terminal +func (c *Client) CreateTerminal(ctx context.Context, p acp.CreateTerminalRequest) (acp.CreateTerminalResponse, error) { + c.logger.Debug("create terminal request", zap.String("command", p.Command)) + // For now, return a dummy terminal ID - we can implement real terminal support later + return acp.CreateTerminalResponse{TerminalId: "t-1"}, nil +} + +// KillTerminalCommand kills a terminal command +func (c *Client) KillTerminalCommand(ctx context.Context, p acp.KillTerminalCommandRequest) (acp.KillTerminalCommandResponse, error) { + c.logger.Debug("kill terminal request", zap.String("terminal_id", p.TerminalId)) + return acp.KillTerminalCommandResponse{}, nil +} + +// TerminalOutput gets terminal output +func (c *Client) TerminalOutput(ctx context.Context, p acp.TerminalOutputRequest) (acp.TerminalOutputResponse, error) { + c.logger.Debug("terminal output request", zap.String("terminal_id", p.TerminalId)) + return acp.TerminalOutputResponse{Output: "ok", Truncated: false}, nil +} + +// ReleaseTerminal releases a terminal +func (c *Client) ReleaseTerminal(ctx context.Context, p acp.ReleaseTerminalRequest) (acp.ReleaseTerminalResponse, error) { + c.logger.Debug("release terminal request", zap.String("terminal_id", p.TerminalId)) + return acp.ReleaseTerminalResponse{}, nil +} + +// WaitForTerminalExit waits for terminal to exit +func (c *Client) WaitForTerminalExit(ctx context.Context, p acp.WaitForTerminalExitRequest) (acp.WaitForTerminalExitResponse, error) { + c.logger.Debug("wait for terminal exit request", zap.String("terminal_id", p.TerminalId)) + exitCode := 0 + return acp.WaitForTerminalExitResponse{ExitCode: &exitCode}, nil +} + +// Verify interface implementation +var _ acp.Client = (*Client)(nil) + diff --git a/apps/backend/internal/agentctl/api/acp.go b/apps/backend/internal/agentctl/api/acp.go new file mode 100644 index 0000000000..0bfe60e9ce --- /dev/null +++ b/apps/backend/internal/agentctl/api/acp.go @@ -0,0 +1,228 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + "time" + + "github.com/coder/acp-go-sdk" + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "go.uber.org/zap" +) + +// InitializeRequest is a request to initialize the ACP session +type InitializeRequest struct { + ClientName string `json:"client_name"` + ClientVersion string `json:"client_version"` +} + +// InitializeResponse is the response to an initialize call +type InitializeResponse struct { + Success bool `json:"success"` + Response *acp.InitializeResponse `json:"response,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleACPInitialize(c *gin.Context) { + var req InitializeRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, InitializeResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + conn := s.procMgr.GetConnection() + if conn == nil { + c.JSON(http.StatusServiceUnavailable, InitializeResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + resp, err := conn.Initialize(ctx, acp.InitializeRequest{ + ProtocolVersion: acp.ProtocolVersionNumber, + ClientInfo: &acp.Implementation{ + Name: req.ClientName, + Version: req.ClientVersion, + }, + }) + if err != nil { + s.logger.Error("initialize failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, InitializeResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, InitializeResponse{ + Success: true, + Response: &resp, + }) +} + +// NewSessionRequest is a request to create a new ACP session +type NewSessionRequest struct { + Cwd string `json:"cwd"` // Working directory for the session +} + +// NewSessionResponse is the response to a new session call +type NewSessionResponse struct { + Success bool `json:"success"` + SessionID string `json:"session_id,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleACPNewSession(c *gin.Context) { + var req NewSessionRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, NewSessionResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + conn := s.procMgr.GetConnection() + if conn == nil { + c.JSON(http.StatusServiceUnavailable, NewSessionResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + cwd := req.Cwd + if cwd == "" { + cwd = "/workspace" + } + + resp, err := conn.NewSession(ctx, acp.NewSessionRequest{ + Cwd: cwd, + McpServers: []acp.McpServer{}, // Required: must be an empty array, not nil + }) + if err != nil { + s.logger.Error("new session failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, NewSessionResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + // Store session ID + s.procMgr.SetSessionID(resp.SessionId) + + c.JSON(http.StatusOK, NewSessionResponse{ + Success: true, + SessionID: string(resp.SessionId), + }) +} + +// PromptRequest is a request to send a prompt to the agent +type PromptRequest struct { + Text string `json:"text"` // Simple text prompt +} + +// PromptResponse is the response to a prompt call +type PromptResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleACPPrompt(c *gin.Context) { + var req PromptRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, PromptResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + conn := s.procMgr.GetConnection() + if conn == nil { + c.JSON(http.StatusServiceUnavailable, PromptResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + sessionID := s.procMgr.GetSessionID() + if sessionID == "" { + c.JSON(http.StatusBadRequest, PromptResponse{ + Success: false, + Error: "no active session - call new_session first", + }) + return + } + + // Use a long timeout for prompt - agent may take time to complete + ctx, cancel := context.WithTimeout(c.Request.Context(), 10*time.Minute) + defer cancel() + + // Build prompt content - simple text block + _, err := conn.Prompt(ctx, acp.PromptRequest{ + SessionId: sessionID, + Prompt: []acp.ContentBlock{acp.TextBlock(req.Text)}, + }) + if err != nil { + s.logger.Error("prompt failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, PromptResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, PromptResponse{Success: true}) +} + +// handleACPStreamWS streams ACP session notifications via WebSocket +func (s *Server) handleACPStreamWS(c *gin.Context) { + conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + s.logger.Info("ACP stream WebSocket connected") + + // Get the session updates channel + updatesCh := s.procMgr.GetUpdates() + + // Stream session notifications to WebSocket + for { + select { + case notification, ok := <-updatesCh: + if !ok { + return + } + + data, err := json.Marshal(notification) + if err != nil { + s.logger.Error("failed to marshal notification", zap.Error(err)) + continue + } + + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + s.logger.Debug("WebSocket write error", zap.Error(err)) + return + } + } + } +} + diff --git a/apps/backend/internal/agentctl/api/output.go b/apps/backend/internal/agentctl/api/output.go new file mode 100644 index 0000000000..cee6321da2 --- /dev/null +++ b/apps/backend/internal/agentctl/api/output.go @@ -0,0 +1,102 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/process" + "go.uber.org/zap" +) + +// OutputResponse is the response for getting output +type OutputResponse struct { + Lines []process.OutputLine `json:"lines"` + Total int `json:"total"` +} + +func (s *Server) handleGetOutput(c *gin.Context) { + // Parse optional 'last' query param + lastStr := c.DefaultQuery("last", "100") + last, err := strconv.Atoi(lastStr) + if err != nil || last < 1 { + last = 100 + } + + lines := s.procMgr.GetOutputBuffer().GetLast(last) + c.JSON(http.StatusOK, OutputResponse{ + Lines: lines, + Total: s.procMgr.GetOutputBuffer().Count(), + }) +} + +// handleOutputStreamWS streams output via WebSocket +func (s *Server) handleOutputStreamWS(c *gin.Context) { + conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + s.logger.Info("Output stream WebSocket connected") + + // Check if client wants history first + sendHistory := c.DefaultQuery("history", "false") == "true" + historyCount, _ := strconv.Atoi(c.DefaultQuery("history_count", "100")) + + if sendHistory { + lines := s.procMgr.GetOutputBuffer().GetLast(historyCount) + for _, line := range lines { + data, err := json.Marshal(line) + if err != nil { + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + s.logger.Debug("WebSocket write error (history)", zap.Error(err)) + return + } + } + } + + // Subscribe to real-time output + sub := s.procMgr.GetOutputBuffer().Subscribe() + defer s.procMgr.GetOutputBuffer().Unsubscribe(sub) + + // Handle WebSocket close + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + // Stream output to client + for { + select { + case line, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(line) + if err != nil { + s.logger.Error("failed to marshal output line", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + s.logger.Debug("WebSocket write error", zap.Error(err)) + return + } + case <-closeCh: + s.logger.Info("Output stream WebSocket closed by client") + return + } + } +} + diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go new file mode 100644 index 0000000000..859824ba4d --- /dev/null +++ b/apps/backend/internal/agentctl/api/server.go @@ -0,0 +1,174 @@ +// Package api provides the HTTP REST API for agentctl +package api + +import ( + "net/http" + "time" + + "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/config" + "github.com/kandev/kandev/internal/agentctl/process" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Server is the HTTP API server +type Server struct { + cfg *config.Config + procMgr *process.Manager + logger *logger.Logger + router *gin.Engine + + upgrader websocket.Upgrader +} + +// NewServer creates a new API server +func NewServer(cfg *config.Config, procMgr *process.Manager, log *logger.Logger) *Server { + gin.SetMode(gin.ReleaseMode) + + s := &Server{ + cfg: cfg, + procMgr: procMgr, + logger: log.WithFields(zap.String("component", "api-server")), + router: gin.New(), + upgrader: websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { + return true // Allow all origins for container-local communication + }, + }, + } + + s.setupRoutes() + return s +} + +// Router returns the HTTP router +func (s *Server) Router() http.Handler { + return s.router +} + +func (s *Server) setupRoutes() { + // Health check + s.router.GET("/health", s.handleHealth) + + // Agent control + api := s.router.Group("/api/v1") + { + // Status and info + api.GET("/status", s.handleStatus) + api.GET("/info", s.handleInfo) + + // Process control + api.POST("/start", s.handleStart) + api.POST("/stop", s.handleStop) + + // ACP high-level methods (using acp-go-sdk) + api.POST("/acp/initialize", s.handleACPInitialize) + api.POST("/acp/session/new", s.handleACPNewSession) + api.POST("/acp/prompt", s.handleACPPrompt) + api.GET("/acp/stream", s.handleACPStreamWS) + + // Output streaming + api.GET("/output", s.handleGetOutput) + api.GET("/output/stream", s.handleOutputStreamWS) + } +} + +// Health check response +type HealthResponse struct { + Status string `json:"status"` + Timestamp string `json:"timestamp"` +} + +func (s *Server) handleHealth(c *gin.Context) { + c.JSON(http.StatusOK, HealthResponse{ + Status: "ok", + Timestamp: time.Now().UTC().Format(time.RFC3339), + }) +} + +// Status response +type StatusResponse struct { + AgentStatus string `json:"agent_status"` + ProcessInfo map[string]interface{} `json:"process_info"` + Uptime string `json:"uptime,omitempty"` +} + +func (s *Server) handleStatus(c *gin.Context) { + c.JSON(http.StatusOK, StatusResponse{ + AgentStatus: string(s.procMgr.Status()), + ProcessInfo: s.procMgr.GetProcessInfo(), + }) +} + +// Info response +type InfoResponse struct { + Version string `json:"version"` + AgentCommand string `json:"agent_command"` + WorkDir string `json:"work_dir"` + Port int `json:"port"` +} + +func (s *Server) handleInfo(c *gin.Context) { + c.JSON(http.StatusOK, InfoResponse{ + Version: "0.1.0", + AgentCommand: s.cfg.AgentCommand, + WorkDir: s.cfg.WorkDir, + Port: s.cfg.Port, + }) +} + +// Start request/response +type StartRequest struct { + // Future: could add options like custom command, env overrides +} + +type StartResponse struct { + Success bool `json:"success"` + Message string `json:"message,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleStart(c *gin.Context) { + if err := s.procMgr.Start(c.Request.Context()); err != nil { + s.logger.Error("failed to start agent", zap.Error(err)) + c.JSON(http.StatusInternalServerError, StartResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, StartResponse{ + Success: true, + Message: "agent started", + }) +} + +// Stop request/response +type StopRequest struct { + Force bool `json:"force"` +} + +type StopResponse struct { + Success bool `json:"success"` + Message string `json:"message,omitempty"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handleStop(c *gin.Context) { + if err := s.procMgr.Stop(c.Request.Context()); err != nil { + s.logger.Error("failed to stop agent", zap.Error(err)) + c.JSON(http.StatusInternalServerError, StopResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, StopResponse{ + Success: true, + Message: "agent stopped", + }) +} diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go new file mode 100644 index 0000000000..ea99ab019e --- /dev/null +++ b/apps/backend/internal/agentctl/config/config.go @@ -0,0 +1,106 @@ +// Package config provides configuration for agentctl +package config + +import ( + "os" + "strconv" + "strings" +) + +// Config holds the agentctl configuration +type Config struct { + // HTTP server port + Port int + + // Agent command to execute (e.g., "auggie --acp") + AgentCommand string + + // Agent arguments (parsed from AgentCommand) + AgentArgs []string + + // Working directory for the agent process + WorkDir string + + // Environment variables to pass to the agent + AgentEnv []string + + // Auto-start the agent on agentctl startup + AutoStart bool + + // Logging configuration + LogLevel string + LogFormat string + + // Buffer size for agent output (in lines) + OutputBufferSize int + + // Health check interval + HealthCheckInterval int +} + +// Load loads configuration from environment variables +func Load() *Config { + workDir := getEnv("AGENTCTL_WORKDIR", "/workspace") + defaultCmd := "auggie --acp --workspace-root " + workDir + + cfg := &Config{ + Port: getEnvInt("AGENTCTL_PORT", 9999), + AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd), + WorkDir: workDir, + AutoStart: getEnvBool("AGENTCTL_AUTO_START", false), + LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), + LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), + OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000), + HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5), + } + + // Parse agent command into args + cfg.AgentArgs = parseCommand(cfg.AgentCommand) + + // Collect agent environment variables (pass through most env vars) + cfg.AgentEnv = collectAgentEnv() + + return cfg +} + +// parseCommand splits a command string into arguments +func parseCommand(cmd string) []string { + // Simple split by spaces (doesn't handle quotes, but good enough for now) + return strings.Fields(cmd) +} + +// collectAgentEnv collects environment variables to pass to the agent +func collectAgentEnv() []string { + // Pass through all environment variables except AGENTCTL_* ones + var env []string + for _, e := range os.Environ() { + if !strings.HasPrefix(e, "AGENTCTL_") { + env = append(env, e) + } + } + return env +} + +func getEnv(key, defaultValue string) string { + if value := os.Getenv(key); value != "" { + return value + } + return defaultValue +} + +func getEnvInt(key string, defaultValue int) int { + if value := os.Getenv(key); value != "" { + if i, err := strconv.Atoi(value); err == nil { + return i + } + } + return defaultValue +} + +func getEnvBool(key string, defaultValue bool) bool { + if value := os.Getenv(key); value != "" { + return strings.ToLower(value) == "true" || value == "1" + } + return defaultValue +} + diff --git a/apps/backend/internal/agentctl/process/buffer.go b/apps/backend/internal/agentctl/process/buffer.go new file mode 100644 index 0000000000..72af60eeca --- /dev/null +++ b/apps/backend/internal/agentctl/process/buffer.go @@ -0,0 +1,131 @@ +package process + +import ( + "sync" + "time" +) + +// OutputLine represents a line of output from the agent +type OutputLine struct { + Timestamp time.Time `json:"timestamp"` + Stream string `json:"stream"` // "stdout" or "stderr" + Content string `json:"content"` +} + +// Subscriber is a channel that receives output lines +type Subscriber chan OutputLine + +// OutputBuffer is a ring buffer for agent output that supports streaming +type OutputBuffer struct { + lines []OutputLine + size int + head int + count int + mu sync.RWMutex + + // Subscribers for real-time streaming + subscribers map[Subscriber]struct{} + subMu sync.RWMutex +} + +// NewOutputBuffer creates a new output buffer with the given capacity +func NewOutputBuffer(size int) *OutputBuffer { + return &OutputBuffer{ + lines: make([]OutputLine, size), + size: size, + subscribers: make(map[Subscriber]struct{}), + } +} + +// Add adds a new line to the buffer and notifies subscribers +func (b *OutputBuffer) Add(line OutputLine) { + b.mu.Lock() + + // Add to ring buffer + idx := (b.head + b.count) % b.size + if b.count < b.size { + b.count++ + } else { + b.head = (b.head + 1) % b.size + } + b.lines[idx] = line + + b.mu.Unlock() + + // Notify subscribers (non-blocking) + b.subMu.RLock() + for sub := range b.subscribers { + select { + case sub <- line: + default: + // Subscriber is slow, skip + } + } + b.subMu.RUnlock() +} + +// GetAll returns all lines in the buffer (oldest first) +func (b *OutputBuffer) GetAll() []OutputLine { + b.mu.RLock() + defer b.mu.RUnlock() + + result := make([]OutputLine, b.count) + for i := 0; i < b.count; i++ { + idx := (b.head + i) % b.size + result[i] = b.lines[idx] + } + return result +} + +// GetLast returns the last n lines from the buffer +func (b *OutputBuffer) GetLast(n int) []OutputLine { + b.mu.RLock() + defer b.mu.RUnlock() + + if n > b.count { + n = b.count + } + + result := make([]OutputLine, n) + start := b.count - n + for i := 0; i < n; i++ { + idx := (b.head + start + i) % b.size + result[i] = b.lines[idx] + } + return result +} + +// Subscribe creates a new subscriber that receives output lines in real-time +func (b *OutputBuffer) Subscribe() Subscriber { + sub := make(Subscriber, 100) + + b.subMu.Lock() + b.subscribers[sub] = struct{}{} + b.subMu.Unlock() + + return sub +} + +// Unsubscribe removes a subscriber +func (b *OutputBuffer) Unsubscribe(sub Subscriber) { + b.subMu.Lock() + delete(b.subscribers, sub) + b.subMu.Unlock() + close(sub) +} + +// Count returns the number of lines in the buffer +func (b *OutputBuffer) Count() int { + b.mu.RLock() + defer b.mu.RUnlock() + return b.count +} + +// Clear clears the buffer +func (b *OutputBuffer) Clear() { + b.mu.Lock() + defer b.mu.Unlock() + b.head = 0 + b.count = 0 +} + diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go new file mode 100644 index 0000000000..47598ecbf0 --- /dev/null +++ b/apps/backend/internal/agentctl/process/manager.go @@ -0,0 +1,331 @@ +// Package process manages the agent subprocess lifecycle +package process + +import ( + "bufio" + "context" + "fmt" + "io" + "log/slog" + "os/exec" + "sync" + "sync/atomic" + "time" + + "github.com/coder/acp-go-sdk" + acpclient "github.com/kandev/kandev/internal/agentctl/acp" + "github.com/kandev/kandev/internal/agentctl/config" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// Compile-time check that we export our client +var _ acp.Client = (*acpclient.Client)(nil) + +// Status represents the agent process status +type Status string + +const ( + StatusStopped Status = "stopped" + StatusStarting Status = "starting" + StatusRunning Status = "running" + StatusPaused Status = "paused" + StatusStopping Status = "stopping" + StatusError Status = "error" +) + +// errorWrapper wraps an error so it can be stored in atomic.Value (which cannot store nil) +type errorWrapper struct { + err error +} + +// Manager manages the agent subprocess +type Manager struct { + cfg *config.Config + logger *logger.Logger + + // Process state + cmd *exec.Cmd + stdin io.WriteCloser + stdout io.ReadCloser + stderr io.ReadCloser + status atomic.Value // Status + exitCode atomic.Int32 + exitErr atomic.Value // error + + // Output buffer for streaming + outputBuffer *OutputBuffer + + // ACP SDK connection + acpClient *acpclient.Client + acpConn *acp.ClientSideConnection + sessionID acp.SessionId + + // Session update notifications + updatesCh chan acp.SessionNotification + + // Synchronization + mu sync.RWMutex + wg sync.WaitGroup + stopCh chan struct{} + doneCh chan struct{} + startMu sync.Mutex +} + +// NewManager creates a new process manager +func NewManager(cfg *config.Config, log *logger.Logger) *Manager { + m := &Manager{ + cfg: cfg, + logger: log.WithFields(zap.String("component", "process-manager")), + outputBuffer: NewOutputBuffer(cfg.OutputBufferSize), + updatesCh: make(chan acp.SessionNotification, 100), + } + m.status.Store(StatusStopped) + m.exitCode.Store(-1) + return m +} + +// Status returns the current process status +func (m *Manager) Status() Status { + return m.status.Load().(Status) +} + +// ExitCode returns the exit code (-1 if not exited) +func (m *Manager) ExitCode() int { + return int(m.exitCode.Load()) +} + +// ExitError returns the exit error if any +func (m *Manager) ExitError() error { + if v := m.exitErr.Load(); v != nil { + if w, ok := v.(errorWrapper); ok { + return w.err + } + } + return nil +} + +// GetOutputBuffer returns the output buffer for streaming +func (m *Manager) GetOutputBuffer() *OutputBuffer { + return m.outputBuffer +} + +// Start starts the agent process +func (m *Manager) Start(ctx context.Context) error { + m.startMu.Lock() + defer m.startMu.Unlock() + + if m.Status() == StatusRunning || m.Status() == StatusStarting { + return fmt.Errorf("agent is already running") + } + + m.logger.Info("starting agent process", + zap.Strings("args", m.cfg.AgentArgs), + zap.String("workdir", m.cfg.WorkDir)) + + m.status.Store(StatusStarting) + m.exitCode.Store(-1) + m.exitErr.Store(errorWrapper{err: nil}) + + // Create command + if len(m.cfg.AgentArgs) == 0 { + m.status.Store(StatusError) + return fmt.Errorf("no agent command configured") + } + + // NOTE: We intentionally don't use exec.CommandContext here because we don't want + // the HTTP request context to kill the agent process when the request completes. + m.cmd = exec.Command(m.cfg.AgentArgs[0], m.cfg.AgentArgs[1:]...) + m.cmd.Dir = m.cfg.WorkDir + m.cmd.Env = m.cfg.AgentEnv + + // Set up pipes + var err error + m.stdin, err = m.cmd.StdinPipe() + if err != nil { + m.status.Store(StatusError) + return fmt.Errorf("failed to create stdin pipe: %w", err) + } + + m.stdout, err = m.cmd.StdoutPipe() + if err != nil { + m.status.Store(StatusError) + return fmt.Errorf("failed to create stdout pipe: %w", err) + } + + m.stderr, err = m.cmd.StderrPipe() + if err != nil { + m.status.Store(StatusError) + return fmt.Errorf("failed to create stderr pipe: %w", err) + } + + // Start the process + if err := m.cmd.Start(); err != nil { + m.status.Store(StatusError) + return fmt.Errorf("failed to start agent: %w", err) + } + + m.stopCh = make(chan struct{}) + m.doneCh = make(chan struct{}) + + // Create ACP client that will handle agent requests + m.acpClient = acpclient.NewClient( + acpclient.WithLogger(m.logger.Zap()), + acpclient.WithWorkspaceRoot(m.cfg.WorkDir), + acpclient.WithUpdateHandler(func(n acp.SessionNotification) { + select { + case m.updatesCh <- n: + default: + m.logger.Warn("updates channel full, dropping notification") + } + }), + ) + + // Create ACP SDK connection - it handles JSON-RPC communication + m.acpConn = acp.NewClientSideConnection(m.acpClient, m.stdin, m.stdout) + m.acpConn.SetLogger(slog.Default().With("component", "acp-conn")) + + // Start stderr reader and exit waiter + m.wg.Add(2) + go m.readStderr() + go m.waitForExit() + + m.status.Store(StatusRunning) + m.logger.Info("agent process started", zap.Int("pid", m.cmd.Process.Pid)) + + return nil +} + +// GetUpdates returns the channel for session update notifications +func (m *Manager) GetUpdates() <-chan acp.SessionNotification { + return m.updatesCh +} + +// GetConnection returns the ACP connection (for advanced usage) +func (m *Manager) GetConnection() *acp.ClientSideConnection { + m.mu.RLock() + defer m.mu.RUnlock() + return m.acpConn +} + +// GetSessionID returns the current session ID +func (m *Manager) GetSessionID() acp.SessionId { + m.mu.RLock() + defer m.mu.RUnlock() + return m.sessionID +} + +// SetSessionID sets the current session ID +func (m *Manager) SetSessionID(id acp.SessionId) { + m.mu.Lock() + defer m.mu.Unlock() + m.sessionID = id +} + +// Stop stops the agent process +func (m *Manager) Stop(ctx context.Context) error { + m.mu.Lock() + defer m.mu.Unlock() + + status := m.Status() + if status == StatusStopped || status == StatusStopping { + return nil + } + + m.logger.Info("stopping agent process") + m.status.Store(StatusStopping) + + // Close stop channel to signal readers + if m.stopCh != nil { + close(m.stopCh) + } + + // Close stdin to signal EOF to agent + if m.stdin != nil { + m.stdin.Close() + } + + // Wait for process to exit with timeout + done := make(chan struct{}) + go func() { + m.wg.Wait() + close(done) + }() + + select { + case <-done: + m.logger.Info("agent process stopped gracefully") + case <-ctx.Done(): + // Force kill + if m.cmd != nil && m.cmd.Process != nil { + m.logger.Warn("force killing agent process") + m.cmd.Process.Kill() + } + } + + m.status.Store(StatusStopped) + return nil +} + + + +// readStderr reads stderr from the agent +func (m *Manager) readStderr() { + defer m.wg.Done() + + scanner := bufio.NewScanner(m.stderr) + for scanner.Scan() { + line := scanner.Text() + m.outputBuffer.Add(OutputLine{ + Timestamp: time.Now(), + Stream: "stderr", + Content: line, + }) + } + + if err := scanner.Err(); err != nil { + m.logger.Debug("stderr reader error", zap.Error(err)) + } +} + +// waitForExit waits for the process to exit +func (m *Manager) waitForExit() { + defer m.wg.Done() + defer close(m.doneCh) + + err := m.cmd.Wait() + + if err != nil { + m.exitErr.Store(errorWrapper{err: err}) + if exitErr, ok := err.(*exec.ExitError); ok { + m.exitCode.Store(int32(exitErr.ExitCode())) + } + m.logger.Info("agent process exited with error", zap.Error(err)) + } else { + m.exitCode.Store(0) + m.logger.Info("agent process exited successfully") + } + + m.status.Store(StatusStopped) +} + + + +// GetProcessInfo returns information about the process +func (m *Manager) GetProcessInfo() map[string]interface{} { + info := map[string]interface{}{ + "status": string(m.Status()), + "exit_code": m.ExitCode(), + } + + if m.cmd != nil && m.cmd.Process != nil { + info["pid"] = m.cmd.Process.Pid + } + + if err := m.ExitError(); err != nil { + info["exit_error"] = err.Error() + } + + return info +} + diff --git a/backend/internal/common/acp/.gitkeep b/apps/backend/internal/common/acp/.gitkeep similarity index 100% rename from backend/internal/common/acp/.gitkeep rename to apps/backend/internal/common/acp/.gitkeep diff --git a/backend/internal/common/auth/.gitkeep b/apps/backend/internal/common/auth/.gitkeep similarity index 100% rename from backend/internal/common/auth/.gitkeep rename to apps/backend/internal/common/auth/.gitkeep diff --git a/backend/internal/common/config/.gitkeep b/apps/backend/internal/common/config/.gitkeep similarity index 100% rename from backend/internal/common/config/.gitkeep rename to apps/backend/internal/common/config/.gitkeep diff --git a/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go similarity index 100% rename from backend/internal/common/config/config.go rename to apps/backend/internal/common/config/config.go diff --git a/backend/internal/common/database/.gitkeep b/apps/backend/internal/common/database/.gitkeep similarity index 100% rename from backend/internal/common/database/.gitkeep rename to apps/backend/internal/common/database/.gitkeep diff --git a/backend/internal/common/database/database.go b/apps/backend/internal/common/database/database.go similarity index 100% rename from backend/internal/common/database/database.go rename to apps/backend/internal/common/database/database.go diff --git a/backend/internal/common/errors/.gitkeep b/apps/backend/internal/common/errors/.gitkeep similarity index 100% rename from backend/internal/common/errors/.gitkeep rename to apps/backend/internal/common/errors/.gitkeep diff --git a/backend/internal/common/errors/errors.go b/apps/backend/internal/common/errors/errors.go similarity index 100% rename from backend/internal/common/errors/errors.go rename to apps/backend/internal/common/errors/errors.go diff --git a/backend/internal/common/logger/.gitkeep b/apps/backend/internal/common/logger/.gitkeep similarity index 100% rename from backend/internal/common/logger/.gitkeep rename to apps/backend/internal/common/logger/.gitkeep diff --git a/backend/internal/common/logger/logger.go b/apps/backend/internal/common/logger/logger.go similarity index 100% rename from backend/internal/common/logger/logger.go rename to apps/backend/internal/common/logger/logger.go diff --git a/backend/internal/events/bus/.gitkeep b/apps/backend/internal/events/bus/.gitkeep similarity index 100% rename from backend/internal/events/bus/.gitkeep rename to apps/backend/internal/events/bus/.gitkeep diff --git a/backend/internal/events/bus/bus.go b/apps/backend/internal/events/bus/bus.go similarity index 100% rename from backend/internal/events/bus/bus.go rename to apps/backend/internal/events/bus/bus.go diff --git a/backend/internal/events/bus/memory.go b/apps/backend/internal/events/bus/memory.go similarity index 100% rename from backend/internal/events/bus/memory.go rename to apps/backend/internal/events/bus/memory.go diff --git a/backend/internal/events/bus/memory_test.go b/apps/backend/internal/events/bus/memory_test.go similarity index 100% rename from backend/internal/events/bus/memory_test.go rename to apps/backend/internal/events/bus/memory_test.go diff --git a/backend/internal/events/bus/nats.go b/apps/backend/internal/events/bus/nats.go similarity index 100% rename from backend/internal/events/bus/nats.go rename to apps/backend/internal/events/bus/nats.go diff --git a/backend/internal/events/bus/subscription.go b/apps/backend/internal/events/bus/subscription.go similarity index 100% rename from backend/internal/events/bus/subscription.go rename to apps/backend/internal/events/bus/subscription.go diff --git a/backend/internal/events/publisher/.gitkeep b/apps/backend/internal/events/publisher/.gitkeep similarity index 100% rename from backend/internal/events/publisher/.gitkeep rename to apps/backend/internal/events/publisher/.gitkeep diff --git a/backend/internal/events/subscriber/.gitkeep b/apps/backend/internal/events/subscriber/.gitkeep similarity index 100% rename from backend/internal/events/subscriber/.gitkeep rename to apps/backend/internal/events/subscriber/.gitkeep diff --git a/backend/internal/events/types.go b/apps/backend/internal/events/types.go similarity index 100% rename from backend/internal/events/types.go rename to apps/backend/internal/events/types.go diff --git a/backend/internal/gateway/handlers/.gitkeep b/apps/backend/internal/gateway/handlers/.gitkeep similarity index 100% rename from backend/internal/gateway/handlers/.gitkeep rename to apps/backend/internal/gateway/handlers/.gitkeep diff --git a/backend/internal/gateway/middleware/.gitkeep b/apps/backend/internal/gateway/middleware/.gitkeep similarity index 100% rename from backend/internal/gateway/middleware/.gitkeep rename to apps/backend/internal/gateway/middleware/.gitkeep diff --git a/backend/internal/gateway/router/.gitkeep b/apps/backend/internal/gateway/router/.gitkeep similarity index 100% rename from backend/internal/gateway/router/.gitkeep rename to apps/backend/internal/gateway/router/.gitkeep diff --git a/backend/internal/gateway/streaming/.gitkeep b/apps/backend/internal/gateway/streaming/.gitkeep similarity index 100% rename from backend/internal/gateway/streaming/.gitkeep rename to apps/backend/internal/gateway/streaming/.gitkeep diff --git a/backend/internal/gateway/websocket/.gitkeep b/apps/backend/internal/gateway/websocket/.gitkeep similarity index 100% rename from backend/internal/gateway/websocket/.gitkeep rename to apps/backend/internal/gateway/websocket/.gitkeep diff --git a/backend/internal/orchestrator/acp/.gitkeep b/apps/backend/internal/orchestrator/acp/.gitkeep similarity index 100% rename from backend/internal/orchestrator/acp/.gitkeep rename to apps/backend/internal/orchestrator/acp/.gitkeep diff --git a/backend/internal/orchestrator/acp/handler.go b/apps/backend/internal/orchestrator/acp/handler.go similarity index 100% rename from backend/internal/orchestrator/acp/handler.go rename to apps/backend/internal/orchestrator/acp/handler.go diff --git a/backend/internal/orchestrator/acp/memory_store.go b/apps/backend/internal/orchestrator/acp/memory_store.go similarity index 100% rename from backend/internal/orchestrator/acp/memory_store.go rename to apps/backend/internal/orchestrator/acp/memory_store.go diff --git a/backend/internal/orchestrator/acp/memory_store_test.go b/apps/backend/internal/orchestrator/acp/memory_store_test.go similarity index 100% rename from backend/internal/orchestrator/acp/memory_store_test.go rename to apps/backend/internal/orchestrator/acp/memory_store_test.go diff --git a/backend/internal/orchestrator/api/.gitkeep b/apps/backend/internal/orchestrator/api/.gitkeep similarity index 100% rename from backend/internal/orchestrator/api/.gitkeep rename to apps/backend/internal/orchestrator/api/.gitkeep diff --git a/backend/internal/orchestrator/api/handlers.go b/apps/backend/internal/orchestrator/api/handlers.go similarity index 100% rename from backend/internal/orchestrator/api/handlers.go rename to apps/backend/internal/orchestrator/api/handlers.go diff --git a/backend/internal/orchestrator/api/middleware.go b/apps/backend/internal/orchestrator/api/middleware.go similarity index 100% rename from backend/internal/orchestrator/api/middleware.go rename to apps/backend/internal/orchestrator/api/middleware.go diff --git a/backend/internal/orchestrator/api/requests.go b/apps/backend/internal/orchestrator/api/requests.go similarity index 100% rename from backend/internal/orchestrator/api/requests.go rename to apps/backend/internal/orchestrator/api/requests.go diff --git a/backend/internal/orchestrator/api/router.go b/apps/backend/internal/orchestrator/api/router.go similarity index 100% rename from backend/internal/orchestrator/api/router.go rename to apps/backend/internal/orchestrator/api/router.go diff --git a/backend/internal/orchestrator/executor/.gitkeep b/apps/backend/internal/orchestrator/executor/.gitkeep similarity index 100% rename from backend/internal/orchestrator/executor/.gitkeep rename to apps/backend/internal/orchestrator/executor/.gitkeep diff --git a/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go similarity index 100% rename from backend/internal/orchestrator/executor/executor.go rename to apps/backend/internal/orchestrator/executor/executor.go diff --git a/backend/internal/orchestrator/queue/.gitkeep b/apps/backend/internal/orchestrator/queue/.gitkeep similarity index 100% rename from backend/internal/orchestrator/queue/.gitkeep rename to apps/backend/internal/orchestrator/queue/.gitkeep diff --git a/backend/internal/orchestrator/queue/queue.go b/apps/backend/internal/orchestrator/queue/queue.go similarity index 100% rename from backend/internal/orchestrator/queue/queue.go rename to apps/backend/internal/orchestrator/queue/queue.go diff --git a/backend/internal/orchestrator/queue/queue_test.go b/apps/backend/internal/orchestrator/queue/queue_test.go similarity index 100% rename from backend/internal/orchestrator/queue/queue_test.go rename to apps/backend/internal/orchestrator/queue/queue_test.go diff --git a/backend/internal/orchestrator/scheduler/.gitkeep b/apps/backend/internal/orchestrator/scheduler/.gitkeep similarity index 100% rename from backend/internal/orchestrator/scheduler/.gitkeep rename to apps/backend/internal/orchestrator/scheduler/.gitkeep diff --git a/backend/internal/orchestrator/scheduler/scheduler.go b/apps/backend/internal/orchestrator/scheduler/scheduler.go similarity index 100% rename from backend/internal/orchestrator/scheduler/scheduler.go rename to apps/backend/internal/orchestrator/scheduler/scheduler.go diff --git a/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go similarity index 100% rename from backend/internal/orchestrator/scheduler/scheduler_test.go rename to apps/backend/internal/orchestrator/scheduler/scheduler_test.go diff --git a/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go similarity index 94% rename from backend/internal/orchestrator/service.go rename to apps/backend/internal/orchestrator/service.go index bfef18802e..0e1513e310 100644 --- a/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -46,6 +46,7 @@ type Service struct { logger *logger.Logger eventBus bus.EventBus db *database.DB + taskRepo scheduler.TaskRepository // Components queue *queue.TaskQueue @@ -100,6 +101,7 @@ func NewService( logger: svcLogger, eventBus: eventBus, db: db, + taskRepo: taskRepo, queue: taskQueue, executor: exec, scheduler: sched, @@ -338,6 +340,16 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv // Update scheduler and remove from queue s.scheduler.HandleTaskCompleted(data.TaskID, true) s.scheduler.RemoveTask(data.TaskID) + + // Update task state to COMPLETED + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateCompleted); err != nil { + s.logger.Error("failed to update task state to COMPLETED", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } else { + s.logger.Info("task marked as COMPLETED", + zap.String("task_id", data.TaskID)) + } } // handleAgentFailed handles agent failure events @@ -352,6 +364,13 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent if !s.scheduler.RetryTask(data.TaskID) { s.logger.Error("task failed and retry limit exceeded", zap.String("task_id", data.TaskID)) + + // Update task state to FAILED + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateFailed); err != nil { + s.logger.Error("failed to update task state to FAILED", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } } } diff --git a/backend/internal/orchestrator/streaming/.gitkeep b/apps/backend/internal/orchestrator/streaming/.gitkeep similarity index 100% rename from backend/internal/orchestrator/streaming/.gitkeep rename to apps/backend/internal/orchestrator/streaming/.gitkeep diff --git a/backend/internal/orchestrator/streaming/client.go b/apps/backend/internal/orchestrator/streaming/client.go similarity index 100% rename from backend/internal/orchestrator/streaming/client.go rename to apps/backend/internal/orchestrator/streaming/client.go diff --git a/backend/internal/orchestrator/streaming/handlers.go b/apps/backend/internal/orchestrator/streaming/handlers.go similarity index 100% rename from backend/internal/orchestrator/streaming/handlers.go rename to apps/backend/internal/orchestrator/streaming/handlers.go diff --git a/backend/internal/orchestrator/streaming/hub.go b/apps/backend/internal/orchestrator/streaming/hub.go similarity index 100% rename from backend/internal/orchestrator/streaming/hub.go rename to apps/backend/internal/orchestrator/streaming/hub.go diff --git a/backend/internal/orchestrator/watcher/.gitkeep b/apps/backend/internal/orchestrator/watcher/.gitkeep similarity index 100% rename from backend/internal/orchestrator/watcher/.gitkeep rename to apps/backend/internal/orchestrator/watcher/.gitkeep diff --git a/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go similarity index 100% rename from backend/internal/orchestrator/watcher/watcher.go rename to apps/backend/internal/orchestrator/watcher/watcher.go diff --git a/backend/internal/orchestrator/watcher/watcher_test.go b/apps/backend/internal/orchestrator/watcher/watcher_test.go similarity index 100% rename from backend/internal/orchestrator/watcher/watcher_test.go rename to apps/backend/internal/orchestrator/watcher/watcher_test.go diff --git a/backend/internal/task/api/handlers.go b/apps/backend/internal/task/api/handlers.go similarity index 100% rename from backend/internal/task/api/handlers.go rename to apps/backend/internal/task/api/handlers.go diff --git a/backend/internal/task/api/handlers_test.go b/apps/backend/internal/task/api/handlers_test.go similarity index 100% rename from backend/internal/task/api/handlers_test.go rename to apps/backend/internal/task/api/handlers_test.go diff --git a/backend/internal/task/api/requests.go b/apps/backend/internal/task/api/requests.go similarity index 100% rename from backend/internal/task/api/requests.go rename to apps/backend/internal/task/api/requests.go diff --git a/backend/internal/task/api/router.go b/apps/backend/internal/task/api/router.go similarity index 100% rename from backend/internal/task/api/router.go rename to apps/backend/internal/task/api/router.go diff --git a/backend/internal/task/events/.gitkeep b/apps/backend/internal/task/events/.gitkeep similarity index 100% rename from backend/internal/task/events/.gitkeep rename to apps/backend/internal/task/events/.gitkeep diff --git a/backend/internal/task/models/.gitkeep b/apps/backend/internal/task/models/.gitkeep similarity index 100% rename from backend/internal/task/models/.gitkeep rename to apps/backend/internal/task/models/.gitkeep diff --git a/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go similarity index 100% rename from backend/internal/task/models/models.go rename to apps/backend/internal/task/models/models.go diff --git a/backend/internal/task/models/models_test.go b/apps/backend/internal/task/models/models_test.go similarity index 100% rename from backend/internal/task/models/models_test.go rename to apps/backend/internal/task/models/models_test.go diff --git a/backend/internal/task/repository/.gitkeep b/apps/backend/internal/task/repository/.gitkeep similarity index 100% rename from backend/internal/task/repository/.gitkeep rename to apps/backend/internal/task/repository/.gitkeep diff --git a/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go similarity index 100% rename from backend/internal/task/repository/interface.go rename to apps/backend/internal/task/repository/interface.go diff --git a/backend/internal/task/repository/memory.go b/apps/backend/internal/task/repository/memory.go similarity index 100% rename from backend/internal/task/repository/memory.go rename to apps/backend/internal/task/repository/memory.go diff --git a/backend/internal/task/repository/memory_test.go b/apps/backend/internal/task/repository/memory_test.go similarity index 100% rename from backend/internal/task/repository/memory_test.go rename to apps/backend/internal/task/repository/memory_test.go diff --git a/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go similarity index 100% rename from backend/internal/task/repository/sqlite.go rename to apps/backend/internal/task/repository/sqlite.go diff --git a/backend/internal/task/repository/sqlite_test.go b/apps/backend/internal/task/repository/sqlite_test.go similarity index 100% rename from backend/internal/task/repository/sqlite_test.go rename to apps/backend/internal/task/repository/sqlite_test.go diff --git a/backend/internal/task/service/.gitkeep b/apps/backend/internal/task/service/.gitkeep similarity index 100% rename from backend/internal/task/service/.gitkeep rename to apps/backend/internal/task/service/.gitkeep diff --git a/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go similarity index 100% rename from backend/internal/task/service/service.go rename to apps/backend/internal/task/service/service.go diff --git a/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go similarity index 100% rename from backend/internal/task/service/service_test.go rename to apps/backend/internal/task/service/service_test.go diff --git a/apps/backend/kandev b/apps/backend/kandev new file mode 100755 index 0000000000..36da0a4b2e Binary files /dev/null and b/apps/backend/kandev differ diff --git a/backend/pkg/acp/client/.gitkeep b/apps/backend/pkg/acp/client/.gitkeep similarity index 100% rename from backend/pkg/acp/client/.gitkeep rename to apps/backend/pkg/acp/client/.gitkeep diff --git a/backend/pkg/acp/jsonrpc/client.go b/apps/backend/pkg/acp/jsonrpc/client.go similarity index 100% rename from backend/pkg/acp/jsonrpc/client.go rename to apps/backend/pkg/acp/jsonrpc/client.go diff --git a/backend/pkg/acp/jsonrpc/types.go b/apps/backend/pkg/acp/jsonrpc/types.go similarity index 100% rename from backend/pkg/acp/jsonrpc/types.go rename to apps/backend/pkg/acp/jsonrpc/types.go diff --git a/backend/pkg/acp/protocol/.gitkeep b/apps/backend/pkg/acp/protocol/.gitkeep similarity index 100% rename from backend/pkg/acp/protocol/.gitkeep rename to apps/backend/pkg/acp/protocol/.gitkeep diff --git a/backend/pkg/acp/protocol/helpers.go b/apps/backend/pkg/acp/protocol/helpers.go similarity index 100% rename from backend/pkg/acp/protocol/helpers.go rename to apps/backend/pkg/acp/protocol/helpers.go diff --git a/backend/pkg/acp/protocol/helpers_test.go b/apps/backend/pkg/acp/protocol/helpers_test.go similarity index 100% rename from backend/pkg/acp/protocol/helpers_test.go rename to apps/backend/pkg/acp/protocol/helpers_test.go diff --git a/backend/pkg/acp/protocol/message.go b/apps/backend/pkg/acp/protocol/message.go similarity index 100% rename from backend/pkg/acp/protocol/message.go rename to apps/backend/pkg/acp/protocol/message.go diff --git a/backend/pkg/acp/protocol/message_test.go b/apps/backend/pkg/acp/protocol/message_test.go similarity index 100% rename from backend/pkg/acp/protocol/message_test.go rename to apps/backend/pkg/acp/protocol/message_test.go diff --git a/backend/pkg/acp/protocol/types.go b/apps/backend/pkg/acp/protocol/types.go similarity index 100% rename from backend/pkg/acp/protocol/types.go rename to apps/backend/pkg/acp/protocol/types.go diff --git a/backend/pkg/acp/server/.gitkeep b/apps/backend/pkg/acp/server/.gitkeep similarity index 100% rename from backend/pkg/acp/server/.gitkeep rename to apps/backend/pkg/acp/server/.gitkeep diff --git a/backend/pkg/api/v1/.gitkeep b/apps/backend/pkg/api/v1/.gitkeep similarity index 100% rename from backend/pkg/api/v1/.gitkeep rename to apps/backend/pkg/api/v1/.gitkeep diff --git a/backend/pkg/api/v1/agent.go b/apps/backend/pkg/api/v1/agent.go similarity index 100% rename from backend/pkg/api/v1/agent.go rename to apps/backend/pkg/api/v1/agent.go diff --git a/backend/pkg/api/v1/board.go b/apps/backend/pkg/api/v1/board.go similarity index 100% rename from backend/pkg/api/v1/board.go rename to apps/backend/pkg/api/v1/board.go diff --git a/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go similarity index 100% rename from backend/pkg/api/v1/task.go rename to apps/backend/pkg/api/v1/task.go diff --git a/backend/pkg/api/v1/user.go b/apps/backend/pkg/api/v1/user.go similarity index 100% rename from backend/pkg/api/v1/user.go rename to apps/backend/pkg/api/v1/user.go diff --git a/backend/dockerfiles/augment-agent/Dockerfile b/backend/dockerfiles/augment-agent/Dockerfile deleted file mode 100644 index a534de493f..0000000000 --- a/backend/dockerfiles/augment-agent/Dockerfile +++ /dev/null @@ -1,39 +0,0 @@ -# Augment Agent Dockerfile -# A coding agent powered by Auggie CLI (Augment Code) - -FROM node:20-slim - -# Install dependencies -RUN apt-get update && apt-get install -y \ - git \ - curl \ - jq \ - python3 \ - python3-pip \ - ca-certificates \ - && rm -rf /var/lib/apt/lists/* - -# Install Auggie CLI from npm -RUN npm install -g @augmentcode/auggie - -# Create agent directory -WORKDIR /agent - -# Copy agent script -COPY agent.sh /agent/agent.sh -RUN chmod +x /agent/agent.sh - -# Workspace will be mounted here -VOLUME /workspace -WORKDIR /workspace - -# Environment variables (passed at runtime) -# TASK_ID - The task identifier -# TASK_TITLE - Task title/name -# TASK_DESCRIPTION - Full task description -# AUGMENT_SESSION_AUTH - Session auth JSON for Augment (from ~/.augment/session.json) -# AUGGIE_SESSION_ID - (Optional) Session ID to resume a previous conversation - -# Output ACP messages to stdout -ENTRYPOINT ["/agent/agent.sh"] - diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4fb6203b4c..8f2691bc4a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -370,12 +370,12 @@ Kandev uses the **Agent Communication Protocol (ACP)** - an open protocol for co The backend implements full ACP support with: -1. **Session Management** (`backend/internal/agent/acp/session.go`) +1. **Session Management** (`apps/backend/internal/agent/acp/session.go`) - Create sessions with `session/new` - Send prompts with `session/prompt` - Resume sessions with stored session IDs -2. **JSON-RPC Client** (`backend/pkg/acp/jsonrpc/client.go`) +2. **JSON-RPC Client** (`apps/backend/pkg/acp/jsonrpc/client.go`) - Handles responses to client requests - Handles notifications from agent (`session/update`) - Handles requests from agent (`session/request_permission`) diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index 2bde10c517..472a0214b2 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -44,7 +44,7 @@ ### Step 1: Build the Backend ```bash -cd backend +cd apps/backend # Build the binary make build @@ -142,7 +142,7 @@ The project includes a comprehensive E2E test script: ```bash # Build the agent Docker image first -cd backend/dockerfiles/augment-agent +cd apps/backend/dockerfiles/augment-agent docker build -t kandev/augment-agent:latest . # Run the E2E test @@ -175,14 +175,14 @@ The E2E test: ### Build ```bash -cd backend +cd apps/backend make build ``` ### Run Tests ```bash -cd backend +cd apps/backend make test # Or with verbose output @@ -200,7 +200,7 @@ go vet ./... ```bash # SQLite CLI -sqlite3 backend/kandev.db +sqlite3 apps/backend/kandev.db # List tables .tables @@ -217,35 +217,38 @@ SELECT * FROM tasks; ## Project Structure ``` -backend/ -├── cmd/kandev/ # Main entry point -│ └── main.go -├── internal/ -│ ├── agent/ # Agent management -│ │ ├── acp/ # ACP session management -│ │ ├── api/ # Agent HTTP handlers -│ │ ├── credentials/ # Credential providers -│ │ ├── docker/ # Docker client -│ │ ├── lifecycle/ # Container lifecycle -│ │ └── registry/ # Agent type registry -│ ├── task/ # Task management -│ │ ├── api/ # Task HTTP handlers -│ │ ├── models/ # Data models -│ │ ├── repository/ # SQLite/Memory storage -│ │ └── service/ # Business logic -│ ├── orchestrator/ # Orchestration logic -│ ├── events/ # In-memory event bus -│ └── common/ # Shared utilities -│ └── logger/ # Zap logger -├── pkg/acp/ # ACP protocol types -│ └── jsonrpc/ # JSON-RPC 2.0 client -├── dockerfiles/ # Agent Dockerfiles -│ └── augment-agent/ -├── bin/ # Built binaries (gitignored) -├── Makefile -└── NEXT_STEPS.md # Development roadmap -scripts/ -└── e2e-test.sh # End-to-end test script +kandev/ +├── apps/ +│ ├── backend/ # Go backend services +│ │ ├── cmd/kandev/ # Main entry point +│ │ │ └── main.go +│ │ ├── internal/ +│ │ │ ├── agent/ # Agent management +│ │ │ │ ├── acp/ # ACP session management +│ │ │ │ ├── api/ # Agent HTTP handlers +│ │ │ │ ├── credentials/ # Credential providers +│ │ │ │ ├── docker/ # Docker client +│ │ │ │ ├── lifecycle/# Container lifecycle +│ │ │ │ └── registry/ # Agent type registry +│ │ │ ├── task/ # Task management +│ │ │ │ ├── api/ # Task HTTP handlers +│ │ │ │ ├── models/ # Data models +│ │ │ │ ├── repository/ # SQLite/Memory storage +│ │ │ │ └── service/ # Business logic +│ │ │ ├── orchestrator/ # Orchestration logic +│ │ │ ├── events/ # In-memory event bus +│ │ │ └── common/ # Shared utilities +│ │ │ └── logger/ # Zap logger +│ │ ├── pkg/acp/ # ACP protocol types +│ │ │ └── jsonrpc/ # JSON-RPC 2.0 client +│ │ ├── dockerfiles/ # Agent Dockerfiles +│ │ │ └── augment-agent/ +│ │ ├── bin/ # Built binaries (gitignored) +│ │ └── Makefile +│ └── web/ # Frontend application +├── scripts/ +│ └── e2e-test.sh # End-to-end test script +└── docs/ # Documentation ``` --- @@ -274,7 +277,7 @@ kill -9 ```bash # Delete and recreate database -rm backend/kandev.db +rm apps/backend/kandev.db ./bin/kandev # Will create fresh database ``` @@ -295,4 +298,4 @@ docker rm $(docker ps -a -q --filter "label=kandev.managed=true") ## Next Steps -See [NEXT_STEPS.md](../backend/NEXT_STEPS.md) for the development roadmap and immediate next steps. +See [NEXT_STEPS.md](../apps/backend/NEXT_STEPS.md) for the development roadmap and immediate next steps. diff --git a/scripts/e2e-test.sh b/scripts/e2e-test.sh index 98b31df18f..2840dbea94 100755 --- a/scripts/e2e-test.sh +++ b/scripts/e2e-test.sh @@ -8,13 +8,14 @@ # - Augment session credentials in ~/.augment/session.json # - Port 8080 available # -# Usage: ./scripts/e2e-test.sh +# Usage: ./scripts/e2e-test.sh [--build] +# --build Build the Docker image before running tests set -e SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" -BACKEND_DIR="$PROJECT_ROOT/backend" +BACKEND_DIR="$PROJECT_ROOT/apps/backend" # Colors for output RED='\033[0;31m' @@ -26,8 +27,24 @@ NC='\033[0m' # No Color # Configuration SERVER_PORT=8080 BASE_URL="http://localhost:${SERVER_PORT}" +AGENTCTL_PORT=9999 WAIT_FOR_AGENT_SECONDS=120 SERVER_PID="" +BUILD_IMAGE=false + +# Parse arguments +while [[ $# -gt 0 ]]; do + case $1 in + --build) + BUILD_IMAGE=true + shift + ;; + *) + echo "Unknown option: $1" + exit 1 + ;; + esac +done # Cleanup function cleanup() { @@ -74,6 +91,20 @@ wait_for_server() { exit 1 } +wait_for_agentctl() { + local container_ip=$1 + log_info "Waiting for agentctl to be ready at ${container_ip}:${AGENTCTL_PORT}..." + for i in {1..30}; do + if curl -s "http://${container_ip}:${AGENTCTL_PORT}/health" > /dev/null 2>&1; then + log_success "agentctl is ready" + return 0 + fi + sleep 1 + done + log_error "agentctl failed to start" + return 1 +} + # Check prerequisites log_step "Checking Prerequisites" @@ -89,13 +120,6 @@ if ! docker info &> /dev/null; then fi log_success "Docker daemon is running" -if ! docker image inspect kandev/augment-agent:latest &> /dev/null; then - log_error "Docker image kandev/augment-agent:latest not found" - log_info "Build it with: cd backend/dockerfiles/augment-agent && docker build -t kandev/augment-agent:latest ." - exit 1 -fi -log_success "Agent Docker image exists" - if [ ! -f "$HOME/.augment/session.json" ]; then log_error "Augment session not found at ~/.augment/session.json" log_info "Please log in with: auggie login" @@ -103,6 +127,22 @@ if [ ! -f "$HOME/.augment/session.json" ]; then fi log_success "Augment session credentials found" +# Build Docker image if requested or if it doesn't exist +if [ "$BUILD_IMAGE" = true ]; then + log_step "Building Docker Image" + log_info "Building kandev/augment-agent:latest with agentctl..." + cd "$BACKEND_DIR" + docker build -t kandev/augment-agent:latest -f dockerfiles/augment-agent/Dockerfile . + log_success "Docker image built" +elif ! docker image inspect kandev/augment-agent:latest &> /dev/null; then + log_error "Docker image kandev/augment-agent:latest not found" + log_info "Build it with: ./scripts/e2e-test.sh --build" + log_info "Or manually: cd apps/backend && docker build -t kandev/augment-agent:latest -f dockerfiles/augment-agent/Dockerfile ." + exit 1 +else + log_success "Agent Docker image exists" +fi + # Kill any existing server on port 8080 if lsof -i :${SERVER_PORT} &> /dev/null; then log_info "Killing existing process on port ${SERVER_PORT}" @@ -152,14 +192,19 @@ fi log_success "Created column: $COLUMN_ID" log_step "Step 3: Create Task" +# Create a temp directory for the test workspace +TEST_WORKSPACE=$(mktemp -d) +log_info "Test workspace: $TEST_WORKSPACE" + +# Create a simple task that creates a file TASK=$(curl -s -X POST "${BASE_URL}/api/v1/tasks" \ -H "Content-Type: application/json" \ -d "{ \"title\": \"E2E Test Task\", - \"description\": \"What is 2+2? Answer in one word.\", + \"description\": \"Create a file named 'agent-result.txt' in the workspace root with the content 'Hello from Agent'. Do nothing else.\", \"board_id\": \"${BOARD_ID}\", \"column_id\": \"${COLUMN_ID}\", - \"repository_url\": \"${PROJECT_ROOT}\", + \"repository_url\": \"${TEST_WORKSPACE}\", \"agent_type\": \"augment-agent\" }") TASK_ID=$(echo "$TASK" | jq -r '.id') @@ -183,27 +228,45 @@ log_success "Task started, agent ID: $AGENT_ID" log_step "Step 5: Verify Agent Container Running" sleep 3 -CONTAINER_COUNT=$(docker ps --filter "name=kandev-agent" -q | wc -l) -if [ "$CONTAINER_COUNT" -eq 0 ]; then +CONTAINER_ID=$(docker ps --filter "name=kandev-agent" --format "{{.ID}}" | head -1) +if [ -z "$CONTAINER_ID" ]; then log_error "No agent container running" exit 1 fi -log_success "Agent container is running" +log_success "Agent container is running: $CONTAINER_ID" -log_step "Step 6: Check Server Logs for Permission Handling" -sleep 5 -if grep -q "Auto-approving permission request" /tmp/kandev-e2e.log; then - log_success "Permission request was auto-approved" -else - log_info "Permission request may not have been logged yet (this is OK)" +# Get container IP for agentctl health check +CONTAINER_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$CONTAINER_ID" 2>/dev/null) +if [ -n "$CONTAINER_IP" ]; then + log_info "Container IP: $CONTAINER_IP" fi -if grep -q "ACP session created" /tmp/kandev-e2e.log; then - log_success "ACP session was created" +log_step "Step 6: Verify agentctl is Running" +sleep 2 +if [ -n "$CONTAINER_IP" ]; then + if wait_for_agentctl "$CONTAINER_IP"; then + # Check agentctl status + AGENTCTL_STATUS=$(curl -s "http://${CONTAINER_IP}:${AGENTCTL_PORT}/api/v1/status" 2>/dev/null || echo '{}') + AGENT_PROCESS_STATE=$(echo "$AGENTCTL_STATUS" | jq -r '.agent_status // "unknown"') + log_info "Agent process state: $AGENT_PROCESS_STATE" + else + log_info "Could not reach agentctl directly (may be in bridge network)" + fi else - log_error "ACP session creation not found in logs" - cat /tmp/kandev-e2e.log | tail -50 - exit 1 + log_info "Container IP not available (using host network)" +fi + +# Check server logs for agentctl communication +if grep -q "agentctl" /tmp/kandev-e2e.log; then + log_success "Backend is communicating with agentctl" +else + log_info "Checking backend logs..." +fi + +if grep -q "agent started" /tmp/kandev-e2e.log || grep -q "initializing agent" /tmp/kandev-e2e.log; then + log_success "Agent initialization started" +else + log_info "Waiting for agent initialization..." fi log_step "Step 7: Wait for Agent to Process" @@ -219,16 +282,28 @@ for i in $(seq 1 $WAIT_FOR_AGENT_SECONDS); do elif [ "$AGENT_STATUS" == "FAILED" ]; then log_error "Agent failed" echo "$AGENT_INFO" | jq . + # Show container logs for debugging + if [ -n "$CONTAINER_ID" ]; then + log_info "Container logs:" + docker logs "$CONTAINER_ID" 2>&1 | tail -30 + fi exit 1 fi # Check if container is still running - CONTAINER_RUNNING=$(docker ps --filter "name=kandev-agent-${AGENT_ID:0:8}" -q | wc -l) + CONTAINER_RUNNING=$(docker ps --filter "id=${CONTAINER_ID}" -q | wc -l) if [ "$CONTAINER_RUNNING" -eq 0 ]; then log_info "Container stopped, checking for completion..." break fi + # Periodically check agentctl status if IP is available + if [ -n "$CONTAINER_IP" ] && [ $((i % 15)) -eq 0 ]; then + AGENTCTL_STATUS=$(curl -s "http://${CONTAINER_IP}:${AGENTCTL_PORT}/api/v1/status" 2>/dev/null || echo '{}') + AGENT_PROCESS_STATE=$(echo "$AGENTCTL_STATUS" | jq -r '.agent_status // "unknown"') + log_info "Agent state via agentctl: $AGENT_PROCESS_STATE" + fi + if [ $((i % 10)) -eq 0 ]; then log_info "Still waiting... ($i seconds)" fi @@ -237,15 +312,26 @@ done log_step "Step 8: Verify Agent Response" # Get container logs to verify agent processed the request -CONTAINER_ID=$(docker ps -a --filter "name=kandev-agent" --format "{{.ID}}" | head -1) -if [ -n "$CONTAINER_ID" ]; then - CONTAINER_LOGS=$(docker logs "$CONTAINER_ID" 2>&1 | tail -50) +FINAL_CONTAINER_ID=$(docker ps -a --filter "name=kandev-agent" --format "{{.ID}}" | head -1) +if [ -n "$FINAL_CONTAINER_ID" ]; then + log_info "Checking container logs for $FINAL_CONTAINER_ID..." + CONTAINER_LOGS=$(docker logs "$FINAL_CONTAINER_ID" 2>&1 | tail -100) + + # Check for agentctl activity + if echo "$CONTAINER_LOGS" | grep -q "agentctl"; then + log_success "agentctl was active in container" + fi + + # Check for ACP messages if echo "$CONTAINER_LOGS" | grep -q '"stopReason"'; then log_success "Agent completed with stop reason" elif echo "$CONTAINER_LOGS" | grep -q 'session/update'; then log_success "Agent sent session updates" + elif echo "$CONTAINER_LOGS" | grep -q 'initialize'; then + log_success "Agent received initialize request" else - log_info "Could not verify agent response from container logs" + log_info "Container logs (last 20 lines):" + echo "$CONTAINER_LOGS" | tail -20 fi else log_info "Container already removed" @@ -259,6 +345,33 @@ FINAL_TASK=$(curl -s "${BASE_URL}/api/v1/tasks/${TASK_ID}") TASK_STATE=$(echo "$FINAL_TASK" | jq -r '.state') log_info "Task state: $TASK_STATE" +log_step "Step 10: Verify Agent Output" +AGENT_OUTPUT_FILE="${TEST_WORKSPACE}/agent-result.txt" +if [ -f "$AGENT_OUTPUT_FILE" ]; then + AGENT_OUTPUT=$(cat "$AGENT_OUTPUT_FILE") + log_success "Agent created file: $AGENT_OUTPUT_FILE" + log_info "File content: $AGENT_OUTPUT" + if echo "$AGENT_OUTPUT" | grep -qi "hello"; then + log_success "Agent output contains expected content" + else + log_info "Note: Output doesn't contain 'Hello' but file was created" + fi +else + log_info "Agent did not create expected file (agent-result.txt)" + log_info "Checking workspace for any created files..." + ls -la "$TEST_WORKSPACE" 2>/dev/null || true +fi + +# Clean up test workspace +#rm -rf "$TEST_WORKSPACE" 2>/dev/null || true + +# Show backend logs summary +log_step "Backend Log Summary" +if [ -f /tmp/kandev-e2e.log ]; then + log_info "Key events from backend:" + grep -E "(agent|agentctl|container|task)" /tmp/kandev-e2e.log | tail -20 || true +fi + # Summary log_step "Test Summary" echo -e "${GREEN}╔════════════════════════════════════════╗${NC}" @@ -268,8 +381,7 @@ echo -e "${GREEN}║ Board created: ✓ ║${NC}" echo -e "${GREEN}║ Column created: ✓ ║${NC}" echo -e "${GREEN}║ Task created: ✓ ║${NC}" echo -e "${GREEN}║ Agent launched: ✓ ║${NC}" -echo -e "${GREEN}║ ACP session: ✓ ║${NC}" -echo -e "${GREEN}║ Permission handled: ✓ ║${NC}" +echo -e "${GREEN}║ agentctl running: ✓ ║${NC}" echo -e "${GREEN}║ Agent processed: ✓ ║${NC}" echo -e "${GREEN}╚════════════════════════════════════════╝${NC}"