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
This commit is contained in:
jcfs
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 != ""
|
||||
}
|
||||
|
||||
@@ -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{
|
||||
|
||||
@@ -116,6 +116,7 @@ func setupTestRouter(
|
||||
handler := &Handler{
|
||||
registry: reg,
|
||||
docker: mockDocker,
|
||||
acp: nil,
|
||||
logger: log,
|
||||
}
|
||||
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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, ¬if); err == nil && notif.Method != "" {
|
||||
c.handleNotification(¬if)
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user