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

feat: remove premature executor deletion

Browse Source
This commit is contained in:
Carlos Florêncio
2026-01-17 19:42:44 +00:00
parent fb042e4c59
commit dce2c6fe40
8 changed files with 93 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/agent/lifecycle/runtime_standalone.go
+32
View File
@@ -3,6 +3,7 @@ package lifecycle
import (
"context"
"fmt"
"time"
"go.uber.org/zap"
@@ -37,7 +38,38 @@ func (r *StandaloneRuntime) HealthCheck(ctx context.Context) error {
return r.ctl.Health(ctx)
}
func (r *StandaloneRuntime) waitForReady(ctx context.Context) error {
if err := r.ctl.Health(ctx); err == nil {
return nil
}
waitCtx := ctx
if _, hasDeadline := ctx.Deadline(); !hasDeadline {
var cancel context.CancelFunc
waitCtx, cancel = context.WithTimeout(ctx, 10*time.Second)
defer cancel()
}
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-waitCtx.Done():
return fmt.Errorf("agentctl not ready: %w", waitCtx.Err())
case <-ticker.C:
if err := r.ctl.Health(waitCtx); err == nil {
return nil
}
}
}
}
func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRequest) (*RuntimeInstance, error) {
if err := r.waitForReady(ctx); err != nil {
return nil, err
}
// Build environment variables
env := req.Env
if env == nil {
apps/backend/internal/gateway/websocket/task_notifications.go
+2 -1
View File
@@ -50,6 +50,7 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub *
b.subscribe(eventBus, events.EnvironmentCreated, ws.ActionEnvironmentCreated)
b.subscribe(eventBus, events.EnvironmentUpdated, ws.ActionEnvironmentUpdated)
b.subscribe(eventBus, events.EnvironmentDeleted, ws.ActionEnvironmentDeleted)
b.subscribe(eventBus, events.TaskSessionStateChanged, ws.ActionSessionStateChanged)
b.subscribe(eventBus, events.MessageAdded, ws.ActionSessionMessageAdded)
b.subscribe(eventBus, events.MessageUpdated, ws.ActionSessionMessageUpdated)
b.subscribe(eventBus, events.AgentctlStarting, ws.ActionSessionAgentctlStarting)
@@ -81,7 +82,7 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action
return nil
}
switch action {
case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError:
case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError, ws.ActionSessionStateChanged:
if sessionID, ok := event.Data["session_id"].(string); ok && sessionID != "" {
b.hub.BroadcastToSession(sessionID, msg)
return nil
apps/backend/internal/orchestrator/executor/executor.go
+12 -30
View File
@@ -862,6 +862,7 @@ func (e *Executor) RespondToPermission(ctx context.Context, sessionID, pendingID
// GetExecution returns the current execution state for a task
func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
ctx := context.Background()
const startupGracePeriod = 30 * time.Second
e.mu.RLock()
execution, exists := e.executions[taskID]
@@ -870,23 +871,9 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
if exists {
// Verify the agent is actually running by probing agentctl
if execution.SessionID == "" || !e.agentManager.IsAgentRunningForSession(ctx, execution.SessionID) {
// Agent stopped - clean up in-memory state
e.mu.Lock()
delete(e.executions, taskID)
e.mu.Unlock()
// Also update DB if we have a session
if execution.SessionID != "" {
_ = e.repo.UpdateTaskSessionState(ctx, execution.SessionID, models.TaskSessionStateCancelled, "agent process stopped")
}
// Clean up the stale execution from the agent manager
if execution.SessionID != "" {
if err := e.agentManager.CleanupStaleExecutionBySessionID(ctx, execution.SessionID); err != nil {
e.logger.Warn("failed to cleanup stale agent execution",
zap.String("session_id", execution.SessionID),
zap.Error(err))
}
if time.Since(execution.StartedAt) < startupGracePeriod {
execCopy := *execution
return &execCopy, true
}
return nil, false
}
@@ -905,19 +892,14 @@ func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) {
// Verify the agent is actually running by probing the lifecycle manager
// This handles the case where backend was restarted and DB has stale "running" sessions
if session.ID == "" || !e.agentManager.IsAgentRunningForSession(ctx, session.ID) {
// Agent is not running - mark the session as stopped in DB and return false
e.logger.Info("stale agent session detected - agent not running",
zap.String("task_id", taskID),
zap.String("session_id", session.ID))
_ = e.repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateCancelled, "agent not running after backend restart")
// Clean up the stale execution from the agent manager
if session.ID != "" {
if err := e.agentManager.CleanupStaleExecutionBySessionID(ctx, session.ID); err != nil {
e.logger.Warn("failed to cleanup stale agent execution",
zap.String("session_id", session.ID),
zap.Error(err))
}
if (session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning) &&
time.Since(session.UpdatedAt) < startupGracePeriod {
execution = FromTaskSession(session)
e.mu.Lock()
e.executions[taskID] = execution
e.mu.Unlock()
execCopy := *execution
return &execCopy, true
}
return nil, false
}
apps/backend/internal/orchestrator/service.go
-1
View File
@@ -486,7 +486,6 @@ func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID st
resp.IsAgentRunning = false
resp.IsResumable = false
resp.NeedsResume = false
resp.Error = "session has no resume token"
return resp, nil
}
apps/web/app/task/[id]/page-client.tsx
+4 -2
View File
@@ -69,13 +69,15 @@ export default function TaskPage({
const initialSessionId = sessionId ?? taskSessionId ?? null;
const effectiveSessionId = activeSessionId ?? initialSessionId;
const storeSessionState = useAppStore((state) =>
effectiveSessionId ? state.taskSessions.items[effectiveSessionId]?.state ?? null : null
);
const agentctlStatus = useSessionAgentctl(effectiveSessionId);
// Session resumption hook - handles auto-resume on page reload
const {
resumptionState,
error: resumptionError,
taskSessionState: resumedSessionState,
worktreePath: resumedWorktreePath,
worktreeBranch: resumedWorktreeBranch,
} = useSessionResumption(task?.id ?? null, effectiveSessionId);
@@ -86,7 +88,7 @@ export default function TaskPage({
// Both hooks read taskSessionState from the store, so prefer resumedSessionState as it's always
// updated when we have a sessionId from URL. For worktree, use resumed values when resuming/resumed.
const taskSessionState = resumedSessionState ?? agentSessionState;
const taskSessionState = storeSessionState ?? agentSessionState;
const worktreePath = effectiveSessionId ? (resumedWorktreePath ?? agentWorktreePath) : agentWorktreePath;
const worktreeBranch = effectiveSessionId ? (resumedWorktreeBranch ?? agentWorktreeBranch) : agentWorktreeBranch;
apps/web/components/task/shell-terminal.tsx
+38 -12
View File
@@ -16,6 +16,7 @@ export function ShellTerminal() {
const lastOutputLengthRef = useRef(0);
const subscriptionIdRef = useRef(0);
const onDataDisposableRef = useRef<{ dispose: () => void } | null>(null);
const retryTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const storeApi = useAppStoreApi();
const sessionId = useAppStore((state) => state.tasks.activeSessionId);
@@ -139,21 +140,46 @@ export function ShellTerminal() {
const client = getWebSocketClient();
if (!client) return;
client
.request<{ success: boolean; buffer?: string }>('shell.subscribe', { session_id: sessionId })
.then((response) => {
if (subscriptionIdRef.current !== currentSubscriptionId) return;
if (response.buffer) {
storeApi.getState().appendShellOutput(sessionId, response.buffer);
}
})
.catch((err) => {
if (subscriptionIdRef.current !== currentSubscriptionId) return;
console.error('Failed to subscribe to shell:', err);
});
if (retryTimeoutRef.current) {
clearTimeout(retryTimeoutRef.current);
retryTimeoutRef.current = null;
}
let cancelled = false;
const attemptSubscribe = () => {
client
.request<{ success: boolean; buffer?: string }>('shell.subscribe', { session_id: sessionId })
.then((response) => {
if (cancelled || subscriptionIdRef.current !== currentSubscriptionId) return;
if (response.buffer) {
storeApi.getState().appendShellOutput(sessionId, response.buffer);
}
})
.catch((err) => {
if (cancelled || subscriptionIdRef.current !== currentSubscriptionId) return;
const message = err instanceof Error ? err.message : String(err);
if (message.includes('no agent running')) {
retryTimeoutRef.current = setTimeout(() => {
if (!cancelled && subscriptionIdRef.current === currentSubscriptionId) {
attemptSubscribe();
}
}, 1000);
return;
}
console.error('Failed to subscribe to shell:', err);
});
};
attemptSubscribe();
return () => {
subscriptionIdRef.current += 1;
cancelled = true;
if (retryTimeoutRef.current) {
clearTimeout(retryTimeoutRef.current);
retryTimeoutRef.current = null;
}
};
}, [taskId, sessionId, storeApi, canSubscribe]);
apps/web/components/task/task-session-switcher.tsx
+3 -2
View File
@@ -170,12 +170,13 @@ export function TaskSessionSwitcher({
<CommandItem
key={session.id}
value={session.id}
data-selected={isActive}
onSelect={() => {
if (!taskId) return;
onSelectSession(taskId, session.id);
}}
className="group flex w-full cursor-pointer items-center gap-3 px-2 py-1.5 rounded-sm text-left transition-colors data-[selected=true]:bg-muted"
className={`group flex w-full cursor-pointer items-center gap-3 px-2 py-1.5 rounded-sm text-left transition-colors ${
isActive ? 'bg-muted' : 'hover:bg-muted/30'
}`}
>
<span className="text-xs font-medium text-muted-foreground w-8 shrink-0">
{formatSessionLabel(index, sortedSessions.length)}
apps/web/lib/http/index.ts
+2 -1
View File
@@ -35,7 +35,8 @@ function resolveUrl(pathOrUrl: string, baseUrl: string) {
async function fetchJson<T>(pathOrUrl: string, options?: ApiRequestOptions): Promise<T> {
const baseUrl = options?.baseUrl ?? getBackendConfig().apiBaseUrl;
const response = await fetch(resolveUrl(pathOrUrl, baseUrl), {
const url = resolveUrl(pathOrUrl, baseUrl);
const response = await fetch(url, {
...options?.init,
cache: options?.cache,
headers: {