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

Cleanup deprecated code and fix session recovery (#37)

Browse Source
This commit is contained in:
João Salavisa
2026-01-18 15:58:56 +00:00
committed by GitHub
parent 3f961365fa
commit 18f4b4e6fb
15 changed files with 219 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/agentctl/main.go
+1 -6
View File
@@ -32,7 +32,6 @@ var (
agentCommandFlag = flag.String("agent-command", "", "Command to run the agent")
workDirFlag = flag.String("workdir", "", "Working directory for the agent")
portFlag = flag.Int("port", 0, "HTTP server port")
maxInstancesFlag = flag.Int("max-instances", 0, "Maximum number of concurrent instances")
)
func main() {
@@ -54,9 +53,6 @@ func main() {
if *portFlag != 0 {
cfg.Port = *portFlag
}
if *maxInstancesFlag != 0 {
cfg.MaxInstances = *maxInstancesFlag
}
// Initialize logger
log, err := logger.NewLogger(logger.LoggingConfig{
@@ -73,8 +69,7 @@ func main() {
}()
log.Info("starting agentctl",
zap.Int("port", cfg.Port),
zap.Int("max_instances", cfg.MaxInstances))
zap.Int("port", cfg.Port))
run(cfg, log)
}
apps/backend/internal/agent/lifecycle/command.go
+2 -1
View File
@@ -40,7 +40,8 @@ func (cb *CommandBuilder) BuildCommand(agentConfig *registry.AgentTypeConfig, op
// 1. Session ID is provided
// 2. Agent does NOT use ACP for session resumption (uses CLI flag instead)
// 3. Agent has a ResumeFlag configured
if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" {
// 4. Agent supports session recovery (CanRecover is not false)
if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() {
cmd = append(cmd, agentConfig.SessionConfig.ResumeFlag, opts.SessionID)
}
apps/backend/internal/agent/lifecycle/event_types.go
+12 -11
View File
@@ -34,17 +34,18 @@ type ACPSessionCreatedPayload struct {
// AgentStreamEventData contains the nested event data within AgentStreamEventPayload.
type AgentStreamEventData struct {
Type string `json:"type"`
ACPSessionID string `json:"acp_session_id,omitempty"`
Text string `json:"text,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
ToolName string `json:"tool_name,omitempty"`
ToolTitle string `json:"tool_title,omitempty"`
ToolStatus string `json:"tool_status,omitempty"`
ToolArgs map[string]interface{} `json:"tool_args,omitempty"`
ToolResult interface{} `json:"tool_result,omitempty"`
Error string `json:"error,omitempty"`
Data interface{} `json:"data,omitempty"`
Type string `json:"type"`
ACPSessionID string `json:"acp_session_id,omitempty"`
Text string `json:"text,omitempty"`
ToolCallID string `json:"tool_call_id,omitempty"`
ToolName string `json:"tool_name,omitempty"`
ToolTitle string `json:"tool_title,omitempty"`
ToolStatus string `json:"tool_status,omitempty"`
ToolArgs map[string]interface{} `json:"tool_args,omitempty"`
ToolResult interface{} `json:"tool_result,omitempty"`
Error string `json:"error,omitempty"`
SessionStatus string `json:"session_status,omitempty"` // "resumed" or "new" for session_status events
Data interface{} `json:"data,omitempty"`
}
// AgentStreamEventPayload is the payload for agent stream events (WebSocket streaming).
apps/backend/internal/agent/lifecycle/events.go
+12 -11
View File
@@ -114,17 +114,18 @@ func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, even
// Build the nested event data
// event.SessionID is the ACP session ID (internal agent protocol session)
eventData := &AgentStreamEventData{
Type: event.Type,
ACPSessionID: event.SessionID,
Text: event.Text,
ToolCallID: event.ToolCallID,
ToolName: event.ToolName,
ToolTitle: event.ToolTitle,
ToolStatus: event.ToolStatus,
ToolArgs: event.ToolArgs,
ToolResult: event.ToolResult,
Error: event.Error,
Data: event.Data,
Type: event.Type,
ACPSessionID: event.SessionID,
Text: event.Text,
ToolCallID: event.ToolCallID,
ToolName: event.ToolName,
ToolTitle: event.ToolTitle,
ToolStatus: event.ToolStatus,
ToolArgs: event.ToolArgs,
ToolResult: event.ToolResult,
Error: event.Error,
SessionStatus: event.SessionStatus,
Data: event.Data,
}
// Build agent event message payload
apps/backend/internal/agent/lifecycle/manager.go
+26
View File
@@ -17,6 +17,7 @@ import (
"github.com/kandev/kandev/internal/agent/registry"
"github.com/kandev/kandev/internal/agent/worktree"
agentctl "github.com/kandev/kandev/internal/agentctl/client"
"github.com/kandev/kandev/internal/agentctl/types/streams"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
@@ -574,6 +575,8 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
return fmt.Errorf("failed to initialize ACP: %w", err)
}
m.emitSessionStatusEvent(execution, agentConfig)
return nil
}
@@ -707,6 +710,29 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec
return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, m.MarkReady)
}
// emitSessionStatusEvent emits a session_status event indicating whether the session was resumed or new.
func (m *Manager) emitSessionStatusEvent(execution *AgentExecution, agentConfig *registry.AgentTypeConfig) {
wasResumed := false
if execution.ACPSessionID != "" {
if agentConfig.SessionConfig.ResumeViaACP {
wasResumed = true
} else if agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() {
wasResumed = true
}
}
sessionStatus := streams.SessionStatusNew
if wasResumed {
sessionStatus = streams.SessionStatusResumed
}
m.eventPublisher.PublishAgentStreamEvent(execution, streams.AgentEvent{
Type: streams.EventTypeSessionStatus,
SessionID: execution.ACPSessionID,
SessionStatus: sessionStatus,
})
}
// handlePermissionRequestEvent processes permission requests from the unified agent event stream
func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event agentctl.AgentEvent) {
m.eventPublisher.PublishPermissionRequest(execution, event)
apps/backend/internal/agent/lifecycle/session.go
+2 -17
View File
@@ -101,8 +101,7 @@ func (sm *SessionManager) InitializeSession(
return result, nil
}
// createOrLoadSession determines whether to create a new session or load an existing one
// based on the agent configuration and whether an existing session ID is provided.
// createOrLoadSession creates a new session or loads an existing one based on agent config.
func (sm *SessionManager) createOrLoadSession(
ctx context.Context,
client *agentctl.Client,
@@ -110,23 +109,9 @@ func (sm *SessionManager) createOrLoadSession(
existingSessionID string,
workspacePath string,
) (string, error) {
// Use session/load only when:
// 1. ResumeViaACP is true (agent supports ACP-based resume)
// 2. An existing session ID is provided
shouldLoadSession := agentConfig.SessionConfig.ResumeViaACP && existingSessionID != ""
if shouldLoadSession {
if agentConfig.SessionConfig.ResumeViaACP && existingSessionID != "" {
return sm.loadSession(ctx, client, agentConfig, existingSessionID)
}
// Log why we're creating a new session when an existing session ID was provided
if existingSessionID != "" && !agentConfig.SessionConfig.ResumeViaACP {
sm.logger.Info("skipping ACP session/load (resume handled via CLI)",
zap.String("agent_type", agentConfig.ID),
zap.String("session_id", existingSessionID),
zap.String("resume_flag", agentConfig.SessionConfig.ResumeFlag))
}
return sm.createNewSession(ctx, client, agentConfig, workspacePath)
}
apps/backend/internal/agent/registry/defaults.go
+6
View File
@@ -2,6 +2,11 @@ package registry
import "github.com/kandev/kandev/pkg/agent"
// boolPtr returns a pointer to a bool value
func boolPtr(v bool) *bool {
return &v
}
// DefaultAgents returns the default agent configurations
func DefaultAgents() []*AgentTypeConfig {
return []*AgentTypeConfig{
@@ -31,6 +36,7 @@ func DefaultAgents() []*AgentTypeConfig {
SessionConfig: SessionConfig{
ResumeViaACP: false, // Auggie handles resume via CLI flag
ResumeFlag: "--resume",
CanRecover: boolPtr(false), // Auggie cannot recover sessions after backend restart
SessionDirTemplate: "{home}/.augment/sessions",
SessionDirTarget: "/root/.augment/sessions",
},
apps/backend/internal/agent/registry/registry.go
+15
View File
@@ -55,6 +55,12 @@ type SessionConfig struct {
// Example: "--resume" for auggie, empty for agents that use ACP session/load.
ResumeFlag string `json:"resume_flag,omitempty"`
// CanRecover indicates whether this agent supports session recovery after a backend restart.
// When false, sessions for this agent cannot be resumed after the backend restarts.
// On recovery, a new session will be started instead of trying to resume.
// Default is true (agent supports recovery).
CanRecover *bool `json:"can_recover,omitempty"`
// SessionDirTemplate is a template for the host directory where session data is stored.
// Supports variables: {home} (user home directory).
// Example: "{home}/.augment/sessions" for auggie.
@@ -65,6 +71,15 @@ type SessionConfig struct {
SessionDirTarget string `json:"session_dir_target,omitempty"`
}
// SupportsRecovery returns whether the agent supports session recovery after backend restart.
// Returns true by default if CanRecover is not explicitly set.
func (c SessionConfig) SupportsRecovery() bool {
if c.CanRecover == nil {
return true
}
return *c.CanRecover
}
// PermissionConfig defines how tool permissions are requested for an agent type
type PermissionConfig struct {
// PermissionFlag is the CLI flag format for requesting tool permissions.
apps/backend/internal/agentctl/server/api/control_server.go
+2 -3
View File
@@ -56,9 +56,8 @@ func (m *ControlServer) setupRoutes() {
func (m *ControlServer) handleHealth(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{
"status": "ok",
"timestamp": time.Now().UTC().Format(time.RFC3339),
"max_instances": m.cfg.MaxInstances,
"status": "ok",
"timestamp": time.Now().UTC().Format(time.RFC3339),
})
}
apps/backend/internal/agentctl/server/config/config.go
+1 -5
View File
@@ -28,9 +28,6 @@ type Config struct {
// Ports configures the port range for instance allocation
Ports PortConfig
// MaxInstances is the maximum number of concurrent agent instances.
MaxInstances int
// Defaults provides default values for new instances
Defaults InstanceDefaults
@@ -117,8 +114,7 @@ type InstanceConfig struct {
// Load loads the configuration from environment variables.
func Load() *Config {
return &Config{
Port: getEnvInt("AGENTCTL_PORT", 9999),
MaxInstances: getEnvInt("AGENTCTL_MAX_INSTANCES", 10),
Port: getEnvInt("AGENTCTL_PORT", 9999),
Ports: PortConfig{
Base: getEnvInt("AGENTCTL_INSTANCE_PORT_BASE", 10001),
Max: getEnvInt("AGENTCTL_INSTANCE_PORT_MAX", 10100),
apps/backend/internal/agentctl/server/instance/manager.go
-5
View File
@@ -53,11 +53,6 @@ func (m *Manager) CreateInstance(ctx context.Context, req *CreateRequest) (*Crea
m.mu.Lock()
defer m.mu.Unlock()
// Check if max instances reached
if len(m.instances) >= m.config.MaxInstances {
return nil, fmt.Errorf("maximum number of instances (%d) reached", m.config.MaxInstances)
}
// Generate ID if not provided
id := req.ID
if id == "" {
apps/backend/internal/agentctl/types/streams/agent.go
+15
View File
@@ -25,6 +25,15 @@ const (
// EventTypePermissionRequest indicates the agent is requesting permission.
EventTypePermissionRequest = "permission_request"
// EventTypeSessionStatus indicates a session status update (resumed or new).
EventTypeSessionStatus = "session_status"
)
// Session status constants for EventTypeSessionStatus events.
const (
SessionStatusResumed = "resumed"
SessionStatusNew = "new"
)
// AgentEvent is the message type streamed from the agent process.
@@ -110,6 +119,12 @@ type AgentEvent struct {
// ActionDetails contains structured details about the action.
ActionDetails map[string]interface{} `json:"action_details,omitempty"`
// --- Session status fields (for "session_status" type) ---
// SessionStatus indicates whether the session was resumed or new.
// Use SessionStatus* constants: "resumed", "new".
SessionStatus string `json:"session_status,omitempty"`
// --- Extension fields ---
// Data contains raw protocol-specific extensions.
apps/backend/internal/orchestrator/service.go
+96 -20
View File
@@ -257,19 +257,14 @@ func (s *Service) resumeExecutorsOnStartup(ctx context.Context) {
continue
}
if !running.Resumable {
s.logger.Warn("executor not resumable; marking session as failed",
zap.String("session_id", sessionID))
_ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, "executor not resumable")
continue
}
// When the executors start, the session changes to RUNNING state.
// After resuming, we need to restore the previous state.
previousState := session.State
// Handle sessions in terminal states - clean up executor record
switch previousState {
case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled:
// Session was already in a terminal state; no need to resume.
s.logger.Info("session in terminal state; cleaning up executor record",
zap.String("session_id", sessionID),
zap.String("state", string(previousState)))
if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil {
s.logger.Warn("failed to remove executor record for terminal session",
zap.String("session_id", sessionID),
@@ -277,32 +272,88 @@ func (s *Service) resumeExecutorsOnStartup(ctx context.Context) {
}
continue
}
// validate the environment (e.g., worktrees) before resuming
// Handle sessions that never started - skip without failure
if previousState == models.TaskSessionStateCreated {
s.logger.Info("session was never started; skipping recovery",
zap.String("session_id", sessionID))
if err := s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID); err != nil {
s.logger.Warn("failed to remove executor record for unstarted session",
zap.String("session_id", sessionID),
zap.Error(err))
}
continue
}
// For sessions that were STARTING, RUNNING, or WAITING_FOR_INPUT:
// We need a resume token to recover them
if running.ResumeToken == "" {
s.logger.Warn("no resume token available; marking session as failed",
zap.String("session_id", sessionID),
zap.String("previous_state", string(previousState)))
_ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed,
"backend restarted and session cannot be resumed (no resume token)")
_ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID)
continue
}
// Check if executor supports resume
if !running.Resumable {
s.logger.Warn("executor not resumable; marking session as failed",
zap.String("session_id", sessionID))
_ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed,
"executor does not support session resume")
_ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID)
continue
}
// Validate worktree exists before resuming
if err := validateSessionWorktrees(session); err != nil {
s.logger.Warn("worktree validation failed; marking session as failed",
zap.String("session_id", sessionID),
zap.Error(err))
_ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error())
_ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID)
continue
}
startAgent := running.ResumeToken != ""
if _, err := s.executor.ResumeSession(ctx, session, startAgent); err != nil {
s.logger.Info("resuming session on startup",
zap.String("session_id", sessionID),
zap.String("task_id", running.TaskID),
zap.String("previous_state", string(previousState)),
zap.String("resume_token", running.ResumeToken))
// Resume the session - this will:
// 1. Create a new agentctl instance
// 2. Start the agent process
// 3. Initialize and load the session via ACP/Codex protocol
//
// ResumeSession will temporarily set state to STARTING during launch.
// After resume, the agent is immediately ready for input (session is loaded,
// no prompt is being processed). We set WAITING_FOR_INPUT directly because:
// - Session resume just loads conversation history, doesn't process a prompt
// - Agents don't send a "complete" event after session load (only after prompt completion)
// - The agent is ready for the next user prompt immediately after session loads
if _, err := s.executor.ResumeSession(ctx, session, true); err != nil {
s.logger.Warn("failed to resume executor on startup",
zap.String("session_id", sessionID),
zap.Error(err))
_ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error())
_ = s.repo.DeleteExecutorRunningBySessionID(ctx, sessionID)
continue
}
// Restore previous state if not CREATED or STARTING
if previousState != models.TaskSessionStateCreated && previousState != models.TaskSessionStateStarting {
if err := s.repo.UpdateTaskSessionState(ctx, sessionID, previousState, ""); err != nil {
s.logger.Warn("failed to restore session state after resume",
zap.String("session_id", sessionID),
zap.Error(err))
}
// Set state to WAITING_FOR_INPUT - agent is ready for the next prompt
if err := s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""); err != nil {
s.logger.Warn("failed to set session state to WAITING_FOR_INPUT after resume",
zap.String("session_id", sessionID),
zap.Error(err))
}
s.logger.Info("session resumed successfully",
zap.String("session_id", sessionID),
zap.String("task_id", running.TaskID),
zap.String("state", string(models.TaskSessionStateWaitingForInput)))
}
}
@@ -881,6 +932,31 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
zap.Error(err))
}
}
case "session_status":
// Handle session status events (resumed vs new session)
if sessionID != "" && s.messageCreator != nil {
var statusMsg string
if payload.Data.SessionStatus == "resumed" {
statusMsg = "Session resumed"
} else {
statusMsg = "New session started"
}
if err := s.messageCreator.CreateSessionMessage(
ctx,
taskID,
statusMsg,
sessionID,
string(v1.MessageTypeStatus),
nil,
false,
); err != nil {
s.logger.Error("failed to create session status message",
zap.String("task_id", taskID),
zap.String("session_id", sessionID),
zap.Error(err))
}
}
}
// Broadcast to all registered handlers (WebSocket streaming)
apps/backend/internal/task/handlers/message_handlers.go
+8 -6
View File
@@ -147,10 +147,9 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task", nil)
}
// If task is in REVIEW state, transition back to IN_PROGRESS
// If task is in REVIEW state, transition back to IN_PROGRESS (also moves to matching column)
if task.State == v1.TaskStateReview {
nextState := v1.TaskStateInProgress
if _, err := h.taskController.UpdateTask(ctx, dto.UpdateTaskRequest{ID: req.TaskID, State: &nextState}); err != nil {
if _, err := h.taskController.UpdateTaskState(ctx, dto.UpdateTaskStateRequest{ID: req.TaskID, State: v1.TaskStateInProgress}); err != nil {
h.logger.Error("failed to transition task from REVIEW to IN_PROGRESS",
zap.String("task_id", req.TaskID),
zap.Error(err))
@@ -173,11 +172,14 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
}
// Auto-forward message as prompt to running agent if orchestrator is available
// Use context.WithoutCancel so the prompt continues even if the WebSocket client disconnects.
// The user's message is already saved, and agent responses are broadcast via notifications.
if h.orchestrator != nil {
_, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Content)
promptCtx := context.WithoutCancel(ctx)
_, err := h.orchestrator.PromptTask(promptCtx, req.TaskID, req.TaskSessionID, req.Content)
if err != nil {
if errors.Is(err, executor.ErrExecutionNotFound) {
if resumeErr := h.orchestrator.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID); resumeErr != nil {
if resumeErr := h.orchestrator.ResumeTaskSession(promptCtx, req.TaskID, req.TaskSessionID); resumeErr != nil {
h.logger.Warn("failed to resume task session for prompt",
zap.String("task_id", req.TaskID),
zap.String("session_id", req.TaskSessionID),
@@ -185,7 +187,7 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w
} else {
for attempt := 0; attempt < 3; attempt++ {
time.Sleep(500 * time.Millisecond)
_, err = h.orchestrator.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Content)
_, err = h.orchestrator.PromptTask(promptCtx, req.TaskID, req.TaskSessionID, req.Content)
if err == nil {
break
}
apps/web/components/task/chat/messages/status-message.tsx
+21 -5
View File
@@ -1,6 +1,6 @@
'use client';
import { IconAlertTriangle, IconInfoCircle } from '@tabler/icons-react';
import { IconAlertTriangle, IconInfoCircle, IconPointFilled } from '@tabler/icons-react';
import { cn } from '@/lib/utils';
import type { Message } from '@/lib/types/http';
import type { StatusMetadata } from '@/components/task/chat/types';
@@ -11,21 +11,37 @@ export function StatusMessage({ comment }: { comment: Message }) {
typeof metadata?.progress === 'number' ? Math.min(Math.max(metadata.progress, 0), 100) : null;
const statusLine = metadata?.stage || metadata?.status;
const message = metadata?.message || comment.content || statusLine || 'Status update';
const Icon = comment.type === 'error' ? IconAlertTriangle : IconInfoCircle;
const iconClass = comment.type === 'error' ? 'text-red-500' : 'text-muted-foreground';
const isError = comment.type === 'error';
// Simple system message: no metadata, no progress, just a short message
const isSimpleStatus = !isError && progress === null && !statusLine && !metadata?.message;
if (isSimpleStatus) {
return (
<div className="flex items-center justify-center py-1">
<div className="flex items-center gap-1.5 px-3 py-1 rounded-full bg-primary/10 border border-primary/20 text-xs text-primary">
<IconPointFilled className="h-2 w-2" />
<span>{message}</span>
</div>
</div>
);
}
const Icon = isError ? IconAlertTriangle : IconInfoCircle;
const iconClass = isError ? 'text-red-500' : 'text-muted-foreground';
return (
<div
className={cn(
'w-full rounded-lg border px-4 py-3 text-sm',
comment.type === 'error'
isError
? 'border-red-500/40 bg-red-500/10 text-foreground'
: 'border-border/60 bg-muted/30 text-foreground'
)}
>
<div className="flex items-center gap-2 mb-2 text-[11px] uppercase tracking-wide opacity-70">
<Icon className={cn('h-3.5 w-3.5', iconClass)} />
<span>{comment.type === 'error' ? 'Error' : 'Status'}</span>
<span>{isError ? 'Error' : 'Status'}</span>
</div>
<div className="whitespace-pre-wrap">{message}</div>
{progress !== null && (