From aebb0c0c761e51d30bf8988af8b7d31fa5011848 Mon Sep 17 00:00:00 2001
From: jcfs <joao.salavisa@gmail.com>
Date: Sat, 10 Jan 2026 10:51:11 +0000
Subject: [PATCH] Implement bidirectional ACP communication with agents

- Add ACP session manager for JSON-RPC 2.0 over stdin/stdout
- Create jsonrpc package with Client for bidirectional communication
- Update agent.sh to run auggie --acp mode
- Add Docker client methods for container stdin/stdout attachment
- Update lifecycle manager to use ACP instead of log streaming
- Add new API endpoints: POST /prompt, POST /cancel, GET /session
- Update AGENTS.md with ACP protocol documentation
- Update OpenAPI spec with new endpoints and schemas

This enables:
- Sending prompts to running agents
- Cancelling agent operations
- Session management (create, load, resume)
- Real-time progress updates via session/update notifications
---
 AGENTS.md                                   | 147 ++++++++-
 backend/cmd/agent-manager/main.go           |   4 +-
 backend/cmd/kandev/main.go                  |  13 +-
 backend/dockerfiles/augment-agent/agent.sh  | 200 ++----------
 backend/internal/agent/acp/session.go       | 330 ++++++++++++++++++++
 backend/internal/agent/api/handlers.go      | 130 ++++++++
 backend/internal/agent/api/handlers_test.go |   1 +
 backend/internal/agent/api/requests.go      |  30 ++
 backend/internal/agent/api/router.go        |   8 +-
 backend/internal/agent/docker/client.go     | 110 +++++++
 backend/internal/agent/lifecycle/manager.go | 101 ++++--
 backend/pkg/acp/jsonrpc/client.go           | 203 ++++++++++++
 backend/pkg/acp/jsonrpc/types.go            | 155 +++++++++
 backend/pkg/acp/protocol/message.go         |  20 +-
 backend/pkg/acp/protocol/types.go           |  64 +++-
 docs/openapi.yaml                           | 119 +++++++
 16 files changed, 1402 insertions(+), 233 deletions(-)
 create mode 100644 backend/internal/agent/acp/session.go
 create mode 100644 backend/pkg/acp/jsonrpc/client.go
 create mode 100644 backend/pkg/acp/jsonrpc/types.go

diff --git a/AGENTS.md b/AGENTS.md
index 0832f8c265..f0bba3977e 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -15,20 +15,25 @@ Kandev uses Docker containers to run AI agents that execute tasks from the Kanba
 ## Architecture
 
 ```
-┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
-│   HTTP Client   │────▶│  Kandev Backend  │────▶│ Docker Container│
-│  (Frontend/CLI) │     │   (Port 8080)    │     │   (AI Agent)    │
-└─────────────────┘     └────────┬─────────┘     └────────┬────────┘
-                                 │                        │
-                                 │◀───────────────────────┘
-                                 │    ACP Messages (stdout)
-                                 ▼
+┌─────────────────┐     ┌──────────────────┐     ┌─────────────────────────────────┐
+│   HTTP Client   │────▶│  Kandev Backend  │────▶│      Docker Container           │
+│  (Frontend/CLI) │     │   (Port 8080)    │     │  ┌─────────────────────────┐    │
+└─────────────────┘     └────────┬─────────┘     │  │  auggie --acp           │    │
+                                 │               │  │  (ACP Mode)             │    │
+                                 │◀─────────────▶│  └─────────────────────────┘    │
+                                 │   JSON-RPC    │                                 │
+                                 │  stdin/stdout └─────────────────────────────────┘
+                                 │
                         ┌──────────────────┐
                         │   Event Bus      │
                         │ (In-Memory)      │
                         └──────────────────┘
 ```
 
+**Bidirectional Communication:**
+- Backend → Agent: `session/prompt`, `session/cancel`
+- Agent → Backend: `session/update` notifications
+
 ## Available Agent Types
 
 ### Augment Agent (`augment-agent`)
@@ -59,9 +64,73 @@ The primary AI coding agent powered by Auggie CLI.
 
 ## Agent Communication Protocol (ACP)
 
-Agents communicate by writing JSON messages to stdout. Each line must be a complete JSON object.
+Kandev uses **ACP (Agent Client Protocol)** for bidirectional communication with agents. ACP is based on JSON-RPC 2.0 over stdin/stdout.
 
-### Message Format
+### ACP Mode
+
+Agents run in ACP mode using `auggie --acp`. This enables:
+- **Bidirectional communication**: Send prompts and receive updates
+- **Session management**: Create, load, and resume sessions
+- **Cancellation**: Cancel operations in progress
+
+### JSON-RPC 2.0 Protocol
+
+#### Client → Agent (Requests)
+
+**Initialize Connection:**
+```json
+{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {
+  "protocolVersion": "1.0",
+  "clientInfo": {"name": "kandev", "version": "0.1.0"},
+  "capabilities": {"streaming": true}
+}}
+```
+
+**Create New Session:**
+```json
+{"jsonrpc": "2.0", "id": 2, "method": "session/new", "params": {}}
+```
+
+**Resume Existing Session:**
+```json
+{"jsonrpc": "2.0", "id": 3, "method": "session/load", "params": {"sessionId": "abc123"}}
+```
+
+**Send Prompt:**
+```json
+{"jsonrpc": "2.0", "id": 4, "method": "session/prompt", "params": {"message": "Fix the bug in main.go"}}
+```
+
+**Cancel Operation (notification, no response):**
+```json
+{"jsonrpc": "2.0", "method": "session/cancel", "params": {"reason": "user requested"}}
+```
+
+#### Agent → Client (Notifications)
+
+**Session Update (progress, content, tool calls):**
+```json
+{"jsonrpc": "2.0", "method": "session/update", "params": {
+  "type": "content",
+  "data": {"text": "I'm analyzing the code..."}
+}}
+```
+
+```json
+{"jsonrpc": "2.0", "method": "session/update", "params": {
+  "type": "toolCall",
+  "data": {"toolName": "file_edit", "status": "running"}
+}}
+```
+
+```json
+{"jsonrpc": "2.0", "method": "session/update", "params": {
+  "type": "complete",
+  "data": {"sessionId": "abc123", "success": true}
+}}
+```
+
+### Legacy Message Format (for reference)
 
 ```json
 {
@@ -73,7 +142,7 @@ Agents communicate by writing JSON messages to stdout. Each line must be a compl
 }
 ```
 
-### Message Types
+### Legacy Message Types
 
 #### `progress` - Progress Updates
 ```json
@@ -501,6 +570,62 @@ Response:
 GET /agents/types/:typeId
 ```
 
+### Send Prompt to Agent (ACP)
+```
+POST /agents/:instanceId/prompt
+```
+
+Request:
+```json
+{
+  "message": "Fix the bug in main.go"
+}
+```
+
+Response: `200 OK`
+```json
+{
+  "success": true,
+  "message": "Prompt sent successfully"
+}
+```
+
+### Cancel Agent Operation (ACP)
+```
+POST /agents/:instanceId/cancel
+```
+
+Request:
+```json
+{
+  "reason": "User requested cancellation"
+}
+```
+
+Response: `200 OK`
+```json
+{
+  "success": true,
+  "message": "Cancel request sent"
+}
+```
+
+### Get Agent Session (ACP)
+```
+GET /agents/:instanceId/session
+```
+
+Response: `200 OK`
+```json
+{
+  "instance_id": "uuid",
+  "task_id": "uuid",
+  "session_id": "abc123",
+  "status": "running",
+  "created_at": "2026-01-10T12:00:00Z"
+}
+```
+
 ---
 
 ## Orchestrator API
diff --git a/backend/cmd/agent-manager/main.go b/backend/cmd/agent-manager/main.go
index 27690e2308..1a313d3e58 100644
--- a/backend/cmd/agent-manager/main.go
+++ b/backend/cmd/agent-manager/main.go
@@ -106,10 +106,10 @@ func main() {
 
 	// 11. Register API routes
 	v1 := router.Group("/api/v1/agents")
-	api.SetupRoutes(v1, lifecycleMgr, reg, dockerClient, log)
+	api.SetupRoutes(v1, lifecycleMgr, reg, dockerClient, nil, log)
 
 	// Health check endpoint at root level
-	handler := api.NewHandler(lifecycleMgr, reg, dockerClient, log)
+	handler := api.NewHandler(lifecycleMgr, reg, dockerClient, nil, log)
 	router.GET("/health", handler.HealthCheck)
 
 	// 12. Create HTTP server
diff --git a/backend/cmd/kandev/main.go b/backend/cmd/kandev/main.go
index 5bf676fddc..31055c06d9 100644
--- a/backend/cmd/kandev/main.go
+++ b/backend/cmd/kandev/main.go
@@ -28,8 +28,8 @@ import (
 	"github.com/kandev/kandev/internal/agent/lifecycle"
 	"github.com/kandev/kandev/internal/agent/registry"
 
-	// Agent streaming
-	agentstreaming "github.com/kandev/kandev/internal/agent/streaming"
+	// Agent ACP
+	agentacp "github.com/kandev/kandev/internal/agent/acp"
 
 	// Orchestrator packages
 	"github.com/kandev/kandev/internal/orchestrator"
@@ -132,6 +132,7 @@ func main() {
 	// ============================================
 	var lifecycleMgr *lifecycle.Manager
 	var agentRegistry *registry.Registry
+	var acpSessionMgr *agentacp.SessionManager
 
 	if dockerClient != nil {
 		log.Info("Initializing Agent Manager...")
@@ -149,12 +150,12 @@ func main() {
 		// Note: credsMgr available for future use when integrating with lifecycle manager
 		_ = credsMgr
 
-		// Streaming Manager for ACP message capture
-		streamMgr := agentstreaming.NewManager(eventBus, log)
+		// ACP Session Manager for bidirectional communication
+		acpSessionMgr = agentacp.NewSessionManager(eventBus, log)
 
 		// Lifecycle Manager
 		lifecycleMgr = lifecycle.NewManager(dockerClient, agentRegistry, eventBus, log)
-		lifecycleMgr.SetStreamingManager(streamMgr)
+		lifecycleMgr.SetACPManager(acpSessionMgr)
 
 		if err := lifecycleMgr.Start(ctx); err != nil {
 			log.Fatal("Failed to start lifecycle manager", zap.Error(err))
@@ -252,7 +253,7 @@ func main() {
 
 	// Agent Manager routes (if available)
 	if lifecycleMgr != nil && agentRegistry != nil && dockerClient != nil {
-		agentapi.SetupRoutes(v1Group, lifecycleMgr, agentRegistry, dockerClient, log)
+		agentapi.SetupRoutes(v1Group, lifecycleMgr, agentRegistry, dockerClient, acpSessionMgr, log)
 	}
 
 	// Health check
diff --git a/backend/dockerfiles/augment-agent/agent.sh b/backend/dockerfiles/augment-agent/agent.sh
index 4841e06609..b3126e8a41 100644
--- a/backend/dockerfiles/augment-agent/agent.sh
+++ b/backend/dockerfiles/augment-agent/agent.sh
@@ -1,187 +1,31 @@
 #!/bin/bash
-# Augment Agent - Integrates with Auggie CLI (Augment Code)
-# Outputs ACP (Agent Communication Protocol) messages to stdout
+# Augment Agent - Runs in ACP (Agent Client Protocol) mode
+# Uses JSON-RPC 2.0 over stdin/stdout for bidirectional communication
 
 set -e
 
-# Common variables for ACP messages
-AGENT_ID="${KANDEV_INSTANCE_ID:-unknown}"
-TASK_ID="${KANDEV_TASK_ID:-${TASK_ID:-unknown}}"
+# Check if auggie is available
+if ! command -v auggie &> /dev/null; then
+    echo '{"jsonrpc":"2.0","method":"error","params":{"message":"Auggie CLI not found"}}' >&2
+    exit 1
+fi
 
-# Helper function to output ACP messages
-acp_message() {
-    local type="$1"
-    local content="$2"
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
+# Check for authentication
+if [ -z "$AUGMENT_SESSION_AUTH" ]; then
+    echo '{"jsonrpc":"2.0","method":"error","params":{"message":"AUGMENT_SESSION_AUTH required"}}' >&2
+    exit 1
+fi
 
-    jq -n -c \
-        --arg type "$type" \
-        --arg agentId "$AGENT_ID" \
-        --arg taskId "$TASK_ID" \
-        --arg timestamp "$timestamp" \
-        --arg content "$content" \
-        '{type: $type, agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {content: $content}}'
-}
+# Export for auggie to use
+export AUGMENT_SESSION_AUTH
 
-acp_progress() {
-    local progress="$1"
-    local message="$2"
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
+# Build auggie command
+AUGGIE_ARGS="--acp"
 
-    jq -n -c \
-        --arg agentId "$AGENT_ID" \
-        --arg taskId "$TASK_ID" \
-        --arg timestamp "$timestamp" \
-        --arg message "$message" \
-        --argjson progress "$progress" \
-        '{type: "progress", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {progress: $progress, message: $message}}'
-}
-
-acp_complete() {
-    local result="$1"
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
-
-    jq -n -c \
-        --arg agentId "$AGENT_ID" \
-        --arg taskId "$TASK_ID" \
-        --arg timestamp "$timestamp" \
-        --arg result "$result" \
-        '{type: "result", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {status: "completed", summary: $result}}'
-}
-
-acp_error() {
-    local error="$1"
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
-
-    jq -n -c \
-        --arg agentId "$AGENT_ID" \
-        --arg taskId "$TASK_ID" \
-        --arg timestamp "$timestamp" \
-        --arg error "$error" \
-        '{type: "error", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {error: $error}}'
-}
-
-# Session info message to store session_id for resumption
-acp_session_info() {
-    local session_id="$1"
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
-
-    jq -n -c \
-        --arg agentId "$AGENT_ID" \
-        --arg taskId "$TASK_ID" \
-        --arg timestamp "$timestamp" \
-        --arg sessionId "$session_id" \
-        '{type: "session_info", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {session_id: $sessionId}}'
-}
-
-# Main agent logic
-main() {
-    acp_message "thinking" "Starting Auggie agent for task: ${TASK_TITLE:-No title}"
-    acp_progress 5 "Initializing Auggie CLI..."
-
-    # Check if auggie is available
-    if ! command -v auggie &> /dev/null; then
-        acp_error "Auggie CLI not found. Please ensure @augmentcode/auggie is installed."
-        exit 1
-    fi
-
-    # Check for authentication
-    # AUGMENT_SESSION_AUTH should contain the full session JSON (same format as ~/.augment/session.json)
-    if [ -z "$AUGMENT_SESSION_AUTH" ]; then
-        acp_error "AUGMENT_SESSION_AUTH environment variable is required for authentication. It should contain the full session JSON."
-        exit 1
-    fi
-
-    # Export for auggie to use
-    export AUGMENT_SESSION_AUTH
-
-    # Check for task description
-    if [ -z "$TASK_DESCRIPTION" ]; then
-        acp_error "TASK_DESCRIPTION environment variable is required."
-        exit 1
-    fi
-
-    # Check if we're resuming a session
-    if [ -n "$AUGGIE_SESSION_ID" ]; then
-        acp_progress 10 "Resuming Auggie session: $AUGGIE_SESSION_ID"
-        acp_message "thinking" "Continuing previous session with: $TASK_DESCRIPTION"
-    else
-        acp_progress 10 "Starting new Auggie session..."
-        acp_message "thinking" "Task: $TASK_DESCRIPTION"
-    fi
-
-    # Build auggie command
-    # --print: One-shot mode, output and exit
-    # --output-format json: Structured JSON output
-    AUGGIE_CMD="auggie --print --output-format json"
-
-    # Add workspace root if we're in a workspace
-    if [ -d "/workspace" ] && [ "$(ls -A /workspace 2>/dev/null)" ]; then
-        AUGGIE_CMD="$AUGGIE_CMD --workspace-root /workspace"
-    fi
-
-    # Resume session if session ID is provided
-    if [ -n "$AUGGIE_SESSION_ID" ]; then
-        AUGGIE_CMD="$AUGGIE_CMD --resume $AUGGIE_SESSION_ID"
-    fi
-
-    # Add the instruction
-    AUGGIE_CMD="$AUGGIE_CMD --instruction \"$TASK_DESCRIPTION\""
-
-    acp_progress 20 "Executing Auggie..."
-
-    # Run auggie and capture output
-    # The output will be JSON when using --output-format json
-    set +e
-    AUGGIE_OUTPUT=$(eval $AUGGIE_CMD 2>&1)
-    AUGGIE_EXIT_CODE=$?
-    set -e
-
-    acp_progress 90 "Processing Auggie output..."
-
-    # Check if auggie succeeded
-    if [ $AUGGIE_EXIT_CODE -ne 0 ]; then
-        acp_error "Auggie failed with exit code $AUGGIE_EXIT_CODE: $AUGGIE_OUTPUT"
-        exit $AUGGIE_EXIT_CODE
-    fi
-
-    # Extract session_id from auggie output for future resumption
-    local session_id=""
-    if echo "$AUGGIE_OUTPUT" | jq -e '.session_id' >/dev/null 2>&1; then
-        session_id=$(echo "$AUGGIE_OUTPUT" | jq -r '.session_id')
-    fi
-
-    # Output the session_id as a separate ACP message so the orchestrator can store it
-    if [ -n "$session_id" ]; then
-        acp_session_info "$session_id"
-    fi
-
-    # Output the auggie result as an ACP message
-    local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
-
-    # Try to parse as JSON, if it fails, treat as plain text
-    if echo "$AUGGIE_OUTPUT" | jq -e . >/dev/null 2>&1; then
-        # Valid JSON output from auggie - wrap in ACP format
-        jq -n -c \
-            --arg agentId "$AGENT_ID" \
-            --arg taskId "$TASK_ID" \
-            --arg timestamp "$timestamp" \
-            --argjson output "$AUGGIE_OUTPUT" \
-            '{type: "auggie_result", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {output: $output}}'
-    else
-        # Plain text output
-        jq -n -c \
-            --arg agentId "$AGENT_ID" \
-            --arg taskId "$TASK_ID" \
-            --arg timestamp "$timestamp" \
-            --arg output "$AUGGIE_OUTPUT" \
-            '{type: "auggie_result", agent_id: $agentId, task_id: $taskId, timestamp: $timestamp, data: {output: $output}}'
-    fi
-
-    acp_progress 100 "Task completed"
-    acp_complete "Auggie agent finished successfully"
-}
-
-# Run main and catch errors
-main "$@" || acp_error "Agent failed: $?"
+# Add workspace root if we're in a workspace with files
+if [ -d "/workspace" ] && [ "$(ls -A /workspace 2>/dev/null)" ]; then
+    AUGGIE_ARGS="$AUGGIE_ARGS --workspace-root /workspace"
+fi
 
+# Execute auggie in ACP mode - stdin/stdout will be used for JSON-RPC communication
+exec auggie $AUGGIE_ARGS
diff --git a/backend/internal/agent/acp/session.go b/backend/internal/agent/acp/session.go
new file mode 100644
index 0000000000..f88c64525b
--- /dev/null
+++ b/backend/internal/agent/acp/session.go
@@ -0,0 +1,330 @@
+// Package acp provides ACP (Agent Client Protocol) session management
+package acp
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/pkg/acp/jsonrpc"
+	"go.uber.org/zap"
+)
+
+// Session represents an active ACP session with an agent
+type Session struct {
+	InstanceID  string
+	TaskID      string
+	SessionID   string // ACP session ID (from agent)
+	Client      *jsonrpc.Client
+	Stdin       io.WriteCloser
+	Stdout      io.Reader
+	CreatedAt   time.Time
+	Status      string // initializing, ready, prompting, complete, error
+	mu          sync.RWMutex
+}
+
+// UpdateHandler is called when the agent sends session updates
+type UpdateHandler func(instanceID, taskID string, updateType string, data json.RawMessage)
+
+// SessionManager manages ACP sessions for multiple agents
+type SessionManager struct {
+	sessions map[string]*Session // by instance ID
+	mu       sync.RWMutex
+	eventBus bus.EventBus
+	logger   *logger.Logger
+	
+	// Handler for session updates
+	updateHandler UpdateHandler
+}
+
+// NewSessionManager creates a new session manager
+func NewSessionManager(eventBus bus.EventBus, log *logger.Logger) *SessionManager {
+	return &SessionManager{
+		sessions: make(map[string]*Session),
+		eventBus: eventBus,
+		logger:   log.WithFields(zap.String("component", "acp-session-manager")),
+	}
+}
+
+// SetUpdateHandler sets the handler for session updates
+func (m *SessionManager) SetUpdateHandler(handler UpdateHandler) {
+	m.updateHandler = handler
+}
+
+// CreateSession creates a new ACP session for an agent instance
+func (m *SessionManager) CreateSession(ctx context.Context, instanceID, taskID string, stdin io.WriteCloser, stdout io.Reader) error {
+	m.logger.Info("Creating ACP session",
+		zap.String("instance_id", instanceID),
+		zap.String("task_id", taskID))
+
+	// Create JSON-RPC client
+	client := jsonrpc.NewClient(stdin, stdout, m.logger)
+
+	session := &Session{
+		InstanceID: instanceID,
+		TaskID:     taskID,
+		Client:     client,
+		Stdin:      stdin,
+		Stdout:     stdout,
+		CreatedAt:  time.Now(),
+		Status:     "initializing",
+	}
+
+	// Set notification handler
+	client.SetNotificationHandler(func(method string, params json.RawMessage) {
+		m.handleNotification(session, method, params)
+	})
+
+	// Start the client read loop
+	client.Start(ctx)
+
+	// Store session
+	m.mu.Lock()
+	m.sessions[instanceID] = session
+	m.mu.Unlock()
+
+	return nil
+}
+
+// Initialize performs the ACP initialize handshake
+func (m *SessionManager) Initialize(ctx context.Context, instanceID string) error {
+	session, err := m.getSession(instanceID)
+	if err != nil {
+		return err
+	}
+
+	m.logger.Info("Initializing ACP session", zap.String("instance_id", instanceID))
+
+	params := jsonrpc.InitializeParams{
+		ProtocolVersion: "1.0",
+		ClientInfo: jsonrpc.ClientInfo{
+			Name:    "kandev",
+			Version: "0.1.0",
+		},
+		Capabilities: jsonrpc.ClientCapabilities{
+			Streaming: true,
+		},
+	}
+
+	resp, err := session.Client.Call(ctx, jsonrpc.MethodInitialize, params)
+	if err != nil {
+		return fmt.Errorf("initialize failed: %w", err)
+	}
+
+	if resp.Error != nil {
+		return fmt.Errorf("initialize error: %s (code %d)", resp.Error.Message, resp.Error.Code)
+	}
+
+	session.mu.Lock()
+	session.Status = "ready"
+	session.mu.Unlock()
+
+	m.logger.Info("ACP session initialized", zap.String("instance_id", instanceID))
+	return nil
+}
+
+// NewSession creates a new ACP session (session/new)
+func (m *SessionManager) NewSession(ctx context.Context, instanceID string) (string, error) {
+	session, err := m.getSession(instanceID)
+	if err != nil {
+		return "", err
+	}
+
+	m.logger.Info("Creating new ACP session", zap.String("instance_id", instanceID))
+
+	resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionNew, jsonrpc.SessionNewParams{})
+	if err != nil {
+		return "", fmt.Errorf("session/new failed: %w", err)
+	}
+
+	if resp.Error != nil {
+		return "", fmt.Errorf("session/new error: %s", resp.Error.Message)
+	}
+
+	var result jsonrpc.SessionNewResult
+	if err := json.Unmarshal(resp.Result, &result); err != nil {
+		return "", fmt.Errorf("failed to parse session/new result: %w", err)
+	}
+
+	session.mu.Lock()
+	session.SessionID = result.SessionID
+	session.mu.Unlock()
+
+	m.logger.Info("ACP session created",
+		zap.String("instance_id", instanceID),
+		zap.String("session_id", result.SessionID))
+
+	return result.SessionID, nil
+}
+
+// LoadSession resumes an existing ACP session (session/load)
+func (m *SessionManager) LoadSession(ctx context.Context, instanceID, sessionID string) error {
+	session, err := m.getSession(instanceID)
+	if err != nil {
+		return err
+	}
+
+	m.logger.Info("Loading ACP session",
+		zap.String("instance_id", instanceID),
+		zap.String("session_id", sessionID))
+
+	params := jsonrpc.SessionLoadParams{
+		SessionID: sessionID,
+	}
+
+	resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionLoad, params)
+	if err != nil {
+		return fmt.Errorf("session/load failed: %w", err)
+	}
+
+	if resp.Error != nil {
+		return fmt.Errorf("session/load error: %s", resp.Error.Message)
+	}
+
+	session.mu.Lock()
+	session.SessionID = sessionID
+	session.mu.Unlock()
+
+	m.logger.Info("ACP session loaded", zap.String("instance_id", instanceID))
+	return nil
+}
+
+// Prompt sends a prompt to the agent (session/prompt)
+func (m *SessionManager) Prompt(ctx context.Context, instanceID, message string) error {
+	session, err := m.getSession(instanceID)
+	if err != nil {
+		return err
+	}
+
+	m.logger.Info("Sending prompt to agent",
+		zap.String("instance_id", instanceID),
+		zap.String("message_length", fmt.Sprintf("%d", len(message))))
+
+	session.mu.Lock()
+	session.Status = "prompting"
+	session.mu.Unlock()
+
+	params := jsonrpc.SessionPromptParams{
+		Message: message,
+	}
+
+	resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionPrompt, params)
+	if err != nil {
+		return fmt.Errorf("session/prompt failed: %w", err)
+	}
+
+	if resp.Error != nil {
+		return fmt.Errorf("session/prompt error: %s", resp.Error.Message)
+	}
+
+	return nil
+}
+
+// Cancel cancels the current operation (session/cancel)
+func (m *SessionManager) Cancel(ctx context.Context, instanceID, reason string) error {
+	session, err := m.getSession(instanceID)
+	if err != nil {
+		return err
+	}
+
+	m.logger.Info("Cancelling agent operation",
+		zap.String("instance_id", instanceID),
+		zap.String("reason", reason))
+
+	params := jsonrpc.SessionCancelParams{
+		Reason: reason,
+	}
+
+	// Cancel is a notification, not a request
+	return session.Client.Notify(jsonrpc.MethodSessionCancel, params)
+}
+
+// CloseSession closes an ACP session
+func (m *SessionManager) CloseSession(instanceID string) error {
+	m.mu.Lock()
+	session, exists := m.sessions[instanceID]
+	if exists {
+		delete(m.sessions, instanceID)
+	}
+	m.mu.Unlock()
+
+	if !exists {
+		return fmt.Errorf("session not found: %s", instanceID)
+	}
+
+	m.logger.Info("Closing ACP session", zap.String("instance_id", instanceID))
+
+	session.Client.Stop()
+	if session.Stdin != nil {
+		session.Stdin.Close()
+	}
+
+	return nil
+}
+
+// GetSession returns a session by instance ID
+func (m *SessionManager) GetSession(instanceID string) (*Session, bool) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	session, exists := m.sessions[instanceID]
+	return session, exists
+}
+
+// getSession returns a session or error if not found
+func (m *SessionManager) getSession(instanceID string) (*Session, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	session, exists := m.sessions[instanceID]
+	if !exists {
+		return nil, fmt.Errorf("session not found: %s", instanceID)
+	}
+	return session, nil
+}
+
+// handleNotification handles incoming notifications from the agent
+func (m *SessionManager) handleNotification(session *Session, method string, params json.RawMessage) {
+	m.logger.Debug("Received notification",
+		zap.String("instance_id", session.InstanceID),
+		zap.String("method", method))
+
+	switch method {
+	case jsonrpc.NotificationSessionUpdate:
+		var update jsonrpc.SessionUpdate
+		if err := json.Unmarshal(params, &update); err != nil {
+			m.logger.Error("Failed to parse session update", zap.Error(err))
+			return
+		}
+
+		// Handle completion
+		if update.Type == "complete" {
+			session.mu.Lock()
+			session.Status = "complete"
+			session.mu.Unlock()
+		}
+
+		// Forward to update handler
+		if m.updateHandler != nil {
+			m.updateHandler(session.InstanceID, session.TaskID, update.Type, update.Data)
+		}
+
+	default:
+		m.logger.Warn("Unknown notification method", zap.String("method", method))
+	}
+}
+
+// GetSessionID returns the ACP session ID for an instance
+func (m *SessionManager) GetSessionID(instanceID string) (string, bool) {
+	session, exists := m.GetSession(instanceID)
+	if !exists {
+		return "", false
+	}
+	session.mu.RLock()
+	defer session.mu.RUnlock()
+	return session.SessionID, session.SessionID != ""
+}
+
diff --git a/backend/internal/agent/api/handlers.go b/backend/internal/agent/api/handlers.go
index 69606fda9b..067443627a 100644
--- a/backend/internal/agent/api/handlers.go
+++ b/backend/internal/agent/api/handlers.go
@@ -22,11 +22,19 @@ type DockerClient interface {
 	GetContainerLogs(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error)
 }
 
+// ACPManager interface for sending commands to agents
+type ACPManager interface {
+	Prompt(ctx context.Context, instanceID, message string) error
+	Cancel(ctx context.Context, instanceID, reason string) error
+	GetSessionID(instanceID string) (string, bool)
+}
+
 // Handler contains HTTP handlers for the agent manager API
 type Handler struct {
 	lifecycle *lifecycle.Manager
 	registry  *registry.Registry
 	docker    DockerClient
+	acp       ACPManager
 	logger    *logger.Logger
 }
 
@@ -35,12 +43,14 @@ func NewHandler(
 	lm *lifecycle.Manager,
 	reg *registry.Registry,
 	docker DockerClient,
+	acp ACPManager,
 	log *logger.Logger,
 ) *Handler {
 	return &Handler{
 		lifecycle: lm,
 		registry:  reg,
 		docker:    docker,
+		acp:       acp,
 		logger:    log.WithFields(zap.String("component", "agent-api")),
 	}
 }
@@ -276,6 +286,126 @@ func (h *Handler) HealthCheck(c *gin.Context) {
 	})
 }
 
+// SendPrompt sends a prompt to an agent
+// POST /api/v1/agents/:instanceId/prompt
+func (h *Handler) SendPrompt(c *gin.Context) {
+	instanceID := c.Param("instanceId")
+	if instanceID == "" {
+		appErr := errors.BadRequest("instanceId is required")
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	// Check if agent exists
+	_, found := h.lifecycle.GetInstance(instanceID)
+	if !found {
+		appErr := errors.NotFound("agent instance", instanceID)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	var req SendPromptRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		appErr := errors.BadRequest("invalid request body: " + err.Error())
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	if h.acp == nil {
+		appErr := errors.InternalError("ACP manager not available", nil)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	if err := h.acp.Prompt(c.Request.Context(), instanceID, req.Message); err != nil {
+		h.logger.Error("failed to send prompt to agent",
+			zap.String("instance_id", instanceID),
+			zap.Error(err))
+		appErr := errors.InternalError("failed to send prompt", err)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	sessionID, _ := h.acp.GetSessionID(instanceID)
+	c.JSON(http.StatusOK, SendPromptResponse{
+		Success:   true,
+		SessionID: sessionID,
+		Message:   "prompt sent successfully",
+	})
+}
+
+// CancelAgent cancels an agent's current operation
+// POST /api/v1/agents/:instanceId/cancel
+func (h *Handler) CancelAgent(c *gin.Context) {
+	instanceID := c.Param("instanceId")
+	if instanceID == "" {
+		appErr := errors.BadRequest("instanceId is required")
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	// Check if agent exists
+	_, found := h.lifecycle.GetInstance(instanceID)
+	if !found {
+		appErr := errors.NotFound("agent instance", instanceID)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	var req CancelAgentRequest
+	_ = c.ShouldBindJSON(&req) // Optional body
+
+	if h.acp == nil {
+		appErr := errors.InternalError("ACP manager not available", nil)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	if err := h.acp.Cancel(c.Request.Context(), instanceID, req.Reason); err != nil {
+		h.logger.Error("failed to cancel agent operation",
+			zap.String("instance_id", instanceID),
+			zap.Error(err))
+		appErr := errors.InternalError("failed to cancel operation", err)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	c.JSON(http.StatusOK, CancelAgentResponse{
+		Success: true,
+		Message: "cancel request sent",
+	})
+}
+
+// GetSession returns the ACP session info for an agent
+// GET /api/v1/agents/:instanceId/session
+func (h *Handler) GetSession(c *gin.Context) {
+	instanceID := c.Param("instanceId")
+	if instanceID == "" {
+		appErr := errors.BadRequest("instanceId is required")
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	// Check if agent exists
+	instance, found := h.lifecycle.GetInstance(instanceID)
+	if !found {
+		appErr := errors.NotFound("agent instance", instanceID)
+		c.JSON(appErr.HTTPStatus, appErr)
+		return
+	}
+
+	sessionID := ""
+	if h.acp != nil {
+		sessionID, _ = h.acp.GetSessionID(instanceID)
+	}
+
+	c.JSON(http.StatusOK, GetSessionResponse{
+		InstanceID: instanceID,
+		SessionID:  sessionID,
+		Status:     string(instance.Status),
+	})
+}
+
 // instanceToResponse converts an AgentInstance to an AgentInstanceResponse
 func instanceToResponse(instance *lifecycle.AgentInstance) AgentInstanceResponse {
 	return AgentInstanceResponse{
diff --git a/backend/internal/agent/api/handlers_test.go b/backend/internal/agent/api/handlers_test.go
index e4d00af22b..d20e91cb90 100644
--- a/backend/internal/agent/api/handlers_test.go
+++ b/backend/internal/agent/api/handlers_test.go
@@ -116,6 +116,7 @@ func setupTestRouter(
 	handler := &Handler{
 		registry: reg,
 		docker:   mockDocker,
+		acp:      nil,
 		logger:   log,
 	}
 
diff --git a/backend/internal/agent/api/requests.go b/backend/internal/agent/api/requests.go
index e6f8c85d01..9c05daf602 100644
--- a/backend/internal/agent/api/requests.go
+++ b/backend/internal/agent/api/requests.go
@@ -75,3 +75,33 @@ type HealthResponse struct {
 	Timestamp time.Time `json:"timestamp"`
 }
 
+// SendPromptRequest for sending a prompt to an agent
+type SendPromptRequest struct {
+	Message string `json:"message" binding:"required"`
+}
+
+// SendPromptResponse for prompt acknowledgment
+type SendPromptResponse struct {
+	Success   bool   `json:"success"`
+	SessionID string `json:"session_id,omitempty"`
+	Message   string `json:"message,omitempty"`
+}
+
+// CancelAgentRequest for cancelling agent operation
+type CancelAgentRequest struct {
+	Reason string `json:"reason,omitempty"`
+}
+
+// CancelAgentResponse for cancel acknowledgment
+type CancelAgentResponse struct {
+	Success bool   `json:"success"`
+	Message string `json:"message,omitempty"`
+}
+
+// GetSessionResponse for getting session info
+type GetSessionResponse struct {
+	InstanceID string `json:"instance_id"`
+	SessionID  string `json:"session_id"`
+	Status     string `json:"status"`
+}
+
diff --git a/backend/internal/agent/api/router.go b/backend/internal/agent/api/router.go
index 5472237f73..f5108d950e 100644
--- a/backend/internal/agent/api/router.go
+++ b/backend/internal/agent/api/router.go
@@ -15,9 +15,10 @@ func SetupRoutes(
 	lm *lifecycle.Manager,
 	reg *registry.Registry,
 	docker DockerClient,
+	acp ACPManager,
 	log *logger.Logger,
 ) {
-	handler := NewHandler(lm, reg, docker, log)
+	handler := NewHandler(lm, reg, docker, acp, log)
 
 	// Agent instance endpoints under /api/v1/agents
 	agents := router.Group("/agents")
@@ -36,6 +37,11 @@ func SetupRoutes(
 		agents.GET("/:instanceId/status", handler.GetAgentStatus)
 		agents.GET("/:instanceId/logs", handler.GetAgentLogs)
 		agents.DELETE("/:instanceId", handler.StopAgent)
+
+		// ACP communication endpoints
+		agents.POST("/:instanceId/prompt", handler.SendPrompt)
+		agents.POST("/:instanceId/cancel", handler.CancelAgent)
+		agents.GET("/:instanceId/session", handler.GetSession)
 	}
 }
 
diff --git a/backend/internal/agent/docker/client.go b/backend/internal/agent/docker/client.go
index 806d004703..c071ad0376 100644
--- a/backend/internal/agent/docker/client.go
+++ b/backend/internal/agent/docker/client.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 	"time"
 
 	"github.com/docker/docker/api/types/container"
@@ -391,3 +392,112 @@ func (c *Client) Ping(ctx context.Context) error {
 	return nil
 }
 
+// AttachResult contains the streams for container I/O
+type AttachResult struct {
+	Stdin  io.WriteCloser
+	Stdout io.Reader
+	Stderr io.Reader
+	Conn   net.Conn
+}
+
+// CreateContainerInteractive creates a container with stdin attached for interactive use
+func (c *Client) CreateContainerInteractive(ctx context.Context, cfg ContainerConfig) (string, error) {
+	c.logger.Info("Creating interactive container",
+		zap.String("name", cfg.Name),
+		zap.String("image", cfg.Image),
+	)
+
+	// Build mounts
+	mounts := make([]mount.Mount, 0, len(cfg.Mounts))
+	for _, m := range cfg.Mounts {
+		mounts = append(mounts, mount.Mount{
+			Type:     mount.TypeBind,
+			Source:   m.Source,
+			Target:   m.Target,
+			ReadOnly: m.ReadOnly,
+		})
+	}
+
+	// Container configuration with stdin attached
+	containerCfg := &container.Config{
+		Image:        cfg.Image,
+		Cmd:          cfg.Cmd,
+		Env:          cfg.Env,
+		WorkingDir:   cfg.WorkingDir,
+		Labels:       cfg.Labels,
+		OpenStdin:    true,
+		StdinOnce:    false,
+		AttachStdin:  true,
+		AttachStdout: true,
+		AttachStderr: true,
+		Tty:          false, // Important: no TTY for JSON-RPC
+	}
+
+	// Host configuration
+	hostCfg := &container.HostConfig{
+		Mounts:      mounts,
+		NetworkMode: container.NetworkMode(cfg.NetworkMode),
+		AutoRemove:  cfg.AutoRemove,
+		Resources: container.Resources{
+			Memory:   cfg.Memory,
+			CPUQuota: cfg.CPUQuota,
+		},
+	}
+
+	resp, err := c.cli.ContainerCreate(ctx, containerCfg, hostCfg, nil, nil, cfg.Name)
+	if err != nil {
+		c.logger.Error("Failed to create interactive container",
+			zap.String("name", cfg.Name),
+			zap.Error(err),
+		)
+		return "", fmt.Errorf("failed to create interactive container %s: %w", cfg.Name, err)
+	}
+
+	c.logger.Info("Interactive container created", zap.String("id", resp.ID), zap.String("name", cfg.Name))
+	return resp.ID, nil
+}
+
+// AttachContainer attaches to a container's stdin, stdout, and stderr
+func (c *Client) AttachContainer(ctx context.Context, containerID string) (*AttachResult, error) {
+	c.logger.Info("Attaching to container", zap.String("container_id", containerID))
+
+	opts := container.AttachOptions{
+		Stream: true,
+		Stdin:  true,
+		Stdout: true,
+		Stderr: true,
+	}
+
+	resp, err := c.cli.ContainerAttach(ctx, containerID, opts)
+	if err != nil {
+		c.logger.Error("Failed to attach to container", zap.String("container_id", containerID), zap.Error(err))
+		return nil, fmt.Errorf("failed to attach to container %s: %w", containerID, err)
+	}
+
+	// Create a pipe for stdin
+	stdinReader, stdinWriter := io.Pipe()
+
+	// Start goroutine to copy from pipe to container
+	go func() {
+		io.Copy(resp.Conn, stdinReader)
+	}()
+
+	c.logger.Info("Attached to container", zap.String("container_id", containerID))
+
+	return &AttachResult{
+		Stdin:  stdinWriter,
+		Stdout: resp.Reader, // This is a multiplexed stream (stdout + stderr)
+		Conn:   resp.Conn,
+	}, nil
+}
+
+// Close closes the attach result
+func (a *AttachResult) Close() error {
+	if a.Stdin != nil {
+		a.Stdin.Close()
+	}
+	if a.Conn != nil {
+		a.Conn.Close()
+	}
+	return nil
+}
diff --git a/backend/internal/agent/lifecycle/manager.go b/backend/internal/agent/lifecycle/manager.go
index 3492be6bb2..bc637b244f 100644
--- a/backend/internal/agent/lifecycle/manager.go
+++ b/backend/internal/agent/lifecycle/manager.go
@@ -15,6 +15,8 @@ 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/docker"
 	"github.com/kandev/kandev/internal/agent/registry"
 	"github.com/kandev/kandev/internal/common/logger"
@@ -47,19 +49,25 @@ type LaunchRequest struct {
 	Metadata      map[string]interface{}
 }
 
-// StreamingManager interface for ACP message streaming
-type StreamingManager interface {
-	StartStreaming(ctx context.Context, instanceID, taskID string, reader io.ReadCloser) error
-	StopStreaming(instanceID string) error
+// 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 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)
 }
 
 // Manager manages agent instance lifecycles
 type Manager struct {
-	docker    *docker.Client
-	registry  *registry.Registry
-	eventBus  bus.EventBus
-	streamMgr StreamingManager
-	logger    *logger.Logger
+	docker   *docker.Client
+	registry *registry.Registry
+	eventBus bus.EventBus
+	acpMgr   ACPManager
+	logger   *logger.Logger
 
 	// Track active instances
 	instances   map[string]*AgentInstance // by instance ID
@@ -93,9 +101,9 @@ func NewManager(
 	}
 }
 
-// SetStreamingManager sets the streaming manager for ACP message capture
-func (m *Manager) SetStreamingManager(streamMgr StreamingManager) {
-	m.streamMgr = streamMgr
+// SetACPManager sets the ACP session manager
+func (m *Manager) SetACPManager(acpMgr ACPManager) {
+	m.acpMgr = acpMgr
 }
 
 // Start starts the lifecycle manager background tasks
@@ -149,8 +157,8 @@ 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
-	containerID, err := m.docker.CreateContainer(ctx, containerConfig)
+	// 5. Create and start the container WITH stdin attached for ACP
+	containerID, err := m.docker.CreateContainerInteractive(ctx, containerConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create container: %w", err)
 	}
@@ -183,25 +191,61 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentInstanc
 	// 7. Publish agent.started event
 	m.publishEvent(ctx, events.AgentStarted, instance)
 
-	// 8. Start streaming ACP messages from the container
-	// Use a background context for streaming since the request context will be cancelled
-	// when the HTTP request completes, but we need streaming to continue
-	if m.streamMgr != nil {
-		streamCtx := context.Background()
-		logsReader, err := m.docker.GetContainerLogs(streamCtx, containerID, true, "all")
+	// 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 get container logs for streaming",
+			m.logger.Warn("failed to attach to container for ACP",
 				zap.String("container_id", containerID),
 				zap.Error(err))
 		} else {
-			if err := m.streamMgr.StartStreaming(streamCtx, instanceID, req.TaskID, logsReader); err != nil {
-				m.logger.Warn("failed to start streaming",
+			// 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 {
-				m.logger.Info("started ACP streaming for agent",
-					zap.String("instance_id", instanceID),
-					zap.String("task_id", req.TaskID))
+				// 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
+					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)
+							} else {
+								sessionID = existingSessionID
+							}
+						} else {
+							sessionID, _ = m.acpMgr.NewSession(context.Background(), instanceID)
+						}
+					} else {
+						sessionID, _ = m.acpMgr.NewSession(context.Background(), instanceID)
+					}
+
+					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 taskDesc, ok := req.Env["TASK_DESCRIPTION"]; ok && taskDesc != "" {
+						if err := m.acpMgr.Prompt(context.Background(), instanceID, taskDesc); err != nil {
+							m.logger.Warn("failed to send initial prompt",
+								zap.String("instance_id", instanceID),
+								zap.Error(err))
+						}
+					}
+				}
 			}
 		}
 	}
@@ -328,6 +372,11 @@ 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/pkg/acp/jsonrpc/client.go b/backend/pkg/acp/jsonrpc/client.go
new file mode 100644
index 0000000000..56299bf8f7
--- /dev/null
+++ b/backend/pkg/acp/jsonrpc/client.go
@@ -0,0 +1,203 @@
+package jsonrpc
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"sync"
+	"sync/atomic"
+
+	"github.com/kandev/kandev/internal/common/logger"
+	"go.uber.org/zap"
+)
+
+// Client handles JSON-RPC 2.0 communication over stdin/stdout streams
+type Client struct {
+	stdin  io.Writer
+	stdout io.Reader
+
+	requestID atomic.Int64
+	pending   map[interface{}]chan *Response
+	mu        sync.Mutex
+
+	// Notification handler
+	onNotification func(method string, params json.RawMessage)
+
+	logger *logger.Logger
+	done   chan struct{}
+}
+
+// NewClient creates a new JSON-RPC client
+func NewClient(stdin io.Writer, stdout io.Reader, log *logger.Logger) *Client {
+	return &Client{
+		stdin:   stdin,
+		stdout:  stdout,
+		pending: make(map[interface{}]chan *Response),
+		logger:  log.WithFields(zap.String("component", "jsonrpc-client")),
+		done:    make(chan struct{}),
+	}
+}
+
+// SetNotificationHandler sets the handler for incoming notifications
+func (c *Client) SetNotificationHandler(handler func(method string, params json.RawMessage)) {
+	c.onNotification = handler
+}
+
+// Start begins reading responses from stdout
+func (c *Client) Start(ctx context.Context) {
+	go c.readLoop(ctx)
+}
+
+// Stop stops the client
+func (c *Client) Stop() {
+	close(c.done)
+}
+
+// Call sends a request and waits for a response
+func (c *Client) Call(ctx context.Context, method string, params interface{}) (*Response, error) {
+	id := c.requestID.Add(1)
+
+	var paramsJSON json.RawMessage
+	if params != nil {
+		var err error
+		paramsJSON, err = json.Marshal(params)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal params: %w", err)
+		}
+	}
+
+	req := &Request{
+		JSONRPC: "2.0",
+		ID:      id,
+		Method:  method,
+		Params:  paramsJSON,
+	}
+
+	// Create response channel
+	respCh := make(chan *Response, 1)
+	c.mu.Lock()
+	c.pending[id] = respCh
+	c.mu.Unlock()
+
+	defer func() {
+		c.mu.Lock()
+		delete(c.pending, id)
+		c.mu.Unlock()
+	}()
+
+	// Send request
+	if err := c.send(req); err != nil {
+		return nil, err
+	}
+
+	// Wait for response
+	select {
+	case resp := <-respCh:
+		return resp, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-c.done:
+		return nil, fmt.Errorf("client closed")
+	}
+}
+
+// Notify sends a notification (no response expected)
+func (c *Client) Notify(method string, params interface{}) error {
+	var paramsJSON json.RawMessage
+	if params != nil {
+		var err error
+		paramsJSON, err = json.Marshal(params)
+		if err != nil {
+			return fmt.Errorf("failed to marshal params: %w", err)
+		}
+	}
+
+	notif := &Notification{
+		JSONRPC: "2.0",
+		Method:  method,
+		Params:  paramsJSON,
+	}
+
+	return c.send(notif)
+}
+
+func (c *Client) send(msg interface{}) error {
+	data, err := json.Marshal(msg)
+	if err != nil {
+		return fmt.Errorf("failed to marshal message: %w", err)
+	}
+
+	data = append(data, '\n')
+	_, err = c.stdin.Write(data)
+	if err != nil {
+		return fmt.Errorf("failed to write message: %w", err)
+	}
+
+	c.logger.Debug("sent message", zap.String("data", string(data)))
+	return nil
+}
+
+func (c *Client) readLoop(ctx context.Context) {
+	scanner := bufio.NewScanner(c.stdout)
+	// Increase buffer size for large messages
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, 1024*1024)
+
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-c.done:
+			return
+		default:
+		}
+
+		line := scanner.Bytes()
+		if len(line) == 0 {
+			continue
+		}
+
+		c.logger.Debug("received message", zap.String("data", string(line)))
+
+		// Try to parse as response first
+		var resp Response
+		if err := json.Unmarshal(line, &resp); err == nil && resp.ID != nil {
+			c.handleResponse(&resp)
+			continue
+		}
+
+		// Try to parse as notification
+		var notif Notification
+		if err := json.Unmarshal(line, &notif); err == nil && notif.Method != "" {
+			c.handleNotification(&notif)
+			continue
+		}
+
+		c.logger.Warn("received unknown message format", zap.String("data", string(line)))
+	}
+
+	if err := scanner.Err(); err != nil {
+		c.logger.Error("read loop error", zap.Error(err))
+	}
+}
+
+func (c *Client) handleResponse(resp *Response) {
+	c.mu.Lock()
+	ch, ok := c.pending[resp.ID]
+	c.mu.Unlock()
+
+	if ok {
+		ch <- resp
+	} else {
+		c.logger.Warn("received response for unknown request", zap.Any("id", resp.ID))
+	}
+}
+
+func (c *Client) handleNotification(notif *Notification) {
+	if c.onNotification != nil {
+		c.onNotification(notif.Method, notif.Params)
+	}
+}
+
diff --git a/backend/pkg/acp/jsonrpc/types.go b/backend/pkg/acp/jsonrpc/types.go
new file mode 100644
index 0000000000..981462de3f
--- /dev/null
+++ b/backend/pkg/acp/jsonrpc/types.go
@@ -0,0 +1,155 @@
+// Package jsonrpc implements JSON-RPC 2.0 protocol for ACP (Agent Client Protocol)
+package jsonrpc
+
+import "encoding/json"
+
+// Request represents a JSON-RPC 2.0 request
+type Request struct {
+	JSONRPC string          `json:"jsonrpc"` // Always "2.0"
+	ID      interface{}     `json:"id,omitempty"` // Request ID (int or string), omit for notifications
+	Method  string          `json:"method"`
+	Params  json.RawMessage `json:"params,omitempty"`
+}
+
+// Response represents a JSON-RPC 2.0 response
+type Response struct {
+	JSONRPC string          `json:"jsonrpc"` // Always "2.0"
+	ID      interface{}     `json:"id"`
+	Result  json.RawMessage `json:"result,omitempty"`
+	Error   *Error          `json:"error,omitempty"`
+}
+
+// Error represents a JSON-RPC 2.0 error
+type Error struct {
+	Code    int             `json:"code"`
+	Message string          `json:"message"`
+	Data    json.RawMessage `json:"data,omitempty"`
+}
+
+// Notification represents a JSON-RPC 2.0 notification (no ID, no response expected)
+type Notification struct {
+	JSONRPC string          `json:"jsonrpc"`
+	Method  string          `json:"method"`
+	Params  json.RawMessage `json:"params,omitempty"`
+}
+
+// Standard JSON-RPC error codes
+const (
+	ParseError     = -32700
+	InvalidRequest = -32600
+	MethodNotFound = -32601
+	InvalidParams  = -32602
+	InternalError  = -32603
+)
+
+// ACP Methods
+const (
+	// Client -> Agent methods
+	MethodInitialize     = "initialize"
+	MethodSessionNew     = "session/new"
+	MethodSessionPrompt  = "session/prompt"
+	MethodSessionLoad    = "session/load"
+	MethodSessionCancel  = "session/cancel"
+	MethodAuthenticate   = "authenticate"
+
+	// Agent -> Client notifications
+	NotificationSessionUpdate = "session/update"
+)
+
+// InitializeParams for initialize method
+type InitializeParams struct {
+	ProtocolVersion string              `json:"protocolVersion"`
+	ClientInfo      ClientInfo          `json:"clientInfo"`
+	Capabilities    ClientCapabilities  `json:"capabilities,omitempty"`
+}
+
+// ClientInfo identifies the client
+type ClientInfo struct {
+	Name    string `json:"name"`
+	Version string `json:"version"`
+}
+
+// ClientCapabilities describes what the client supports
+type ClientCapabilities struct {
+	Streaming bool `json:"streaming,omitempty"`
+}
+
+// InitializeResult from initialize method
+type InitializeResult struct {
+	ProtocolVersion string             `json:"protocolVersion"`
+	ServerInfo      ServerInfo         `json:"serverInfo"`
+	Capabilities    ServerCapabilities `json:"capabilities,omitempty"`
+}
+
+// ServerInfo identifies the server (agent)
+type ServerInfo struct {
+	Name    string `json:"name"`
+	Version string `json:"version"`
+}
+
+// ServerCapabilities describes what the server supports
+type ServerCapabilities struct {
+	ToolsProvider bool `json:"toolsProvider,omitempty"`
+}
+
+// SessionNewParams for session/new method
+type SessionNewParams struct {
+	// Empty for now, may include workspace config later
+}
+
+// SessionNewResult from session/new method
+type SessionNewResult struct {
+	SessionID string `json:"sessionId"`
+}
+
+// SessionPromptParams for session/prompt method
+type SessionPromptParams struct {
+	Message string `json:"message"`
+}
+
+// SessionPromptResult from session/prompt method
+type SessionPromptResult struct {
+	// Result is empty, updates come via notifications
+}
+
+// SessionLoadParams for session/load method (resume session)
+type SessionLoadParams struct {
+	SessionID string `json:"sessionId"`
+}
+
+// SessionLoadResult from session/load method
+type SessionLoadResult struct {
+	SessionID string `json:"sessionId"`
+	Restored  bool   `json:"restored"`
+}
+
+// SessionCancelParams for session/cancel notification
+type SessionCancelParams struct {
+	Reason string `json:"reason,omitempty"`
+}
+
+// SessionUpdate notification from agent
+type SessionUpdate struct {
+	Type string          `json:"type"` // content, toolCall, thinking, error, complete
+	Data json.RawMessage `json:"data,omitempty"`
+}
+
+// SessionUpdateContent for type="content"
+type SessionUpdateContent struct {
+	Text string `json:"text"`
+}
+
+// SessionUpdateToolCall for type="toolCall"
+type SessionUpdateToolCall struct {
+	ToolName string          `json:"toolName"`
+	Args     json.RawMessage `json:"args,omitempty"`
+	Status   string          `json:"status"` // pending, running, complete, error
+	Result   string          `json:"result,omitempty"`
+}
+
+// SessionUpdateComplete for type="complete"
+type SessionUpdateComplete struct {
+	SessionID string `json:"sessionId"`
+	Success   bool   `json:"success"`
+}
+
diff --git a/backend/pkg/acp/protocol/message.go b/backend/pkg/acp/protocol/message.go
index cf00e14101..77e258ba88 100644
--- a/backend/pkg/acp/protocol/message.go
+++ b/backend/pkg/acp/protocol/message.go
@@ -9,13 +9,19 @@ import (
 type MessageType string
 
 const (
-	MessageTypeProgress  MessageType = "progress"
-	MessageTypeLog       MessageType = "log"
-	MessageTypeResult    MessageType = "result"
-	MessageTypeError     MessageType = "error"
-	MessageTypeStatus    MessageType = "status"
-	MessageTypeHeartbeat MessageType = "heartbeat"
-	MessageTypeControl   MessageType = "control" // For commands sent to agents
+	// Agent → Backend message types
+	MessageTypeProgress      MessageType = "progress"
+	MessageTypeLog           MessageType = "log"
+	MessageTypeResult        MessageType = "result"
+	MessageTypeError         MessageType = "error"
+	MessageTypeStatus        MessageType = "status"
+	MessageTypeHeartbeat     MessageType = "heartbeat"
+	MessageTypeInputRequired MessageType = "input_required" // Agent requests input from user
+	MessageTypeSessionInfo   MessageType = "session_info"   // Agent reports session information
+
+	// Backend → Agent message types
+	MessageTypeControl       MessageType = "control"        // Control commands (pause, resume, stop)
+	MessageTypeInputResponse MessageType = "input_response" // Response to input_required
 )
 
 // Message represents an ACP protocol message
diff --git a/backend/pkg/acp/protocol/types.go b/backend/pkg/acp/protocol/types.go
index 64641a4e28..22acf2e3c8 100644
--- a/backend/pkg/acp/protocol/types.go
+++ b/backend/pkg/acp/protocol/types.go
@@ -43,9 +43,69 @@ type StatusData struct {
 	Message string `json:"message,omitempty"`
 }
 
-// ControlData represents control commands for agents
+// ControlData represents control commands for agents (Backend → Agent)
 type ControlData struct {
-	Action string `json:"action"` // pause, resume, stop
+	Action string `json:"action"` // pause, resume, stop, cancel
 	Reason string `json:"reason,omitempty"`
 }
 
+// InputRequiredData represents a request for user input (Agent → Backend)
+type InputRequiredData struct {
+	PromptID    string   `json:"prompt_id"`              // Unique ID for this prompt
+	Prompt      string   `json:"prompt"`                 // Question/prompt for the user
+	InputType   string   `json:"input_type"`             // text, choice, confirm, file
+	Options     []string `json:"options,omitempty"`      // For choice type
+	Default     string   `json:"default,omitempty"`      // Default value
+	Required    bool     `json:"required"`               // Whether input is required
+	Timeout     int      `json:"timeout,omitempty"`      // Timeout in seconds (0 = no timeout)
+	Context     string   `json:"context,omitempty"`      // Additional context for the prompt
+	Validation  string   `json:"validation,omitempty"`   // Validation pattern (regex)
+}
+
+// InputResponseData represents user's response to input request (Backend → Agent)
+type InputResponseData struct {
+	PromptID  string `json:"prompt_id"`           // ID of the prompt being responded to
+	Response  string `json:"response"`            // User's response
+	Cancelled bool   `json:"cancelled,omitempty"` // User cancelled the input
+}
+
+// SessionInfoData represents session information from the agent
+type SessionInfoData struct {
+	SessionID   string                 `json:"session_id"`
+	Resumable   bool                   `json:"resumable"`
+	StartedAt   string                 `json:"started_at,omitempty"`
+	Metadata    map[string]interface{} `json:"metadata,omitempty"`
+}
+
+// AgentCapabilities represents what the agent can do
+type AgentCapabilities struct {
+	SupportsInput   bool     `json:"supports_input"`   // Can handle input_required
+	SupportsControl bool     `json:"supports_control"` // Can handle control commands
+	SupportsPause   bool     `json:"supports_pause"`   // Can be paused/resumed
+	Features        []string `json:"features,omitempty"`
+}
+
+// ACPUpdateData represents session update data from ACP protocol
+type ACPUpdateData struct {
+	Type string                 `json:"type"` // content, toolCall, thinking, error, complete
+	Data map[string]interface{} `json:"data,omitempty"`
+}
+
+// ACPContentData represents content update from ACP
+type ACPContentData struct {
+	Text string `json:"text"`
+}
+
+// ACPToolCallData represents tool call update from ACP
+type ACPToolCallData struct {
+	ToolName string                 `json:"tool_name"`
+	Args     map[string]interface{} `json:"args,omitempty"`
+	Status   string                 `json:"status"` // pending, running, complete, error
+	Result   string                 `json:"result,omitempty"`
+}
+
+// ACPCompleteData represents completion update from ACP
+type ACPCompleteData struct {
+	SessionID string `json:"session_id"`
+	Success   bool   `json:"success"`
+}
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index c0a56e95c7..0005c70d05 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -445,6 +445,75 @@ paths:
         '404':
           $ref: '#/components/responses/NotFound'
 
+  /agents/{instanceId}/prompt:
+    parameters:
+      - $ref: '#/components/parameters/instanceId'
+    post:
+      tags: [Agents]
+      summary: Send prompt to agent (ACP)
+      description: Send a message/prompt to a running agent via ACP protocol
+      operationId: sendAgentPrompt
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/SendPromptRequest'
+      responses:
+        '200':
+          description: Prompt sent successfully
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/SendPromptResponse'
+        '404':
+          $ref: '#/components/responses/NotFound'
+        '500':
+          $ref: '#/components/responses/InternalError'
+
+  /agents/{instanceId}/cancel:
+    parameters:
+      - $ref: '#/components/parameters/instanceId'
+    post:
+      tags: [Agents]
+      summary: Cancel agent operation (ACP)
+      description: Send a cancellation request to a running agent via ACP protocol
+      operationId: cancelAgent
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CancelAgentRequest'
+      responses:
+        '200':
+          description: Cancel request sent
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CancelAgentResponse'
+        '404':
+          $ref: '#/components/responses/NotFound'
+        '500':
+          $ref: '#/components/responses/InternalError'
+
+  /agents/{instanceId}/session:
+    parameters:
+      - $ref: '#/components/parameters/instanceId'
+    get:
+      tags: [Agents]
+      summary: Get agent ACP session
+      description: Get the current ACP session information for an agent
+      operationId: getAgentSession
+      responses:
+        '200':
+          description: Session information
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/GetSessionResponse'
+        '404':
+          $ref: '#/components/responses/NotFound'
+
   # ============ ORCHESTRATOR ============
   /orchestrator/status:
     get:
@@ -965,6 +1034,56 @@ components:
         reason:
           type: string
 
+    SendPromptRequest:
+      type: object
+      properties:
+        message:
+          type: string
+          description: The prompt message to send to the agent
+      required: [message]
+
+    SendPromptResponse:
+      type: object
+      properties:
+        success:
+          type: boolean
+        message:
+          type: string
+      required: [success]
+
+    CancelAgentRequest:
+      type: object
+      properties:
+        reason:
+          type: string
+          description: Reason for cancellation
+
+    CancelAgentResponse:
+      type: object
+      properties:
+        success:
+          type: boolean
+        message:
+          type: string
+      required: [success]
+
+    GetSessionResponse:
+      type: object
+      properties:
+        instance_id:
+          type: string
+        task_id:
+          type: string
+        session_id:
+          type: string
+        status:
+          type: string
+          enum: [initializing, ready, running, completed, failed]
+        created_at:
+          type: string
+          format: date-time
+      required: [instance_id, status, created_at]
+
     AgentsListResponse:
       type: object
       properties: