feat: install agents on env preparation remote executors (#352)

This commit is contained in:
Carlos Florêncio
2026-03-02 22:47:21 +00:00
committed by GitHub
parent d9d4beea98
commit cd96c6f699
67 changed files with 2460 additions and 1194 deletions
+2
View File
@@ -118,6 +118,7 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch
WorktreeID: worktreeID,
WorktreePath: worktreePath,
WorktreeBranch: worktreeBranch,
Metadata: execution.Metadata,
}, nil
}
@@ -230,6 +231,7 @@ func (a *lifecycleAdapter) PollRemoteStatusForRecords(ctx context.Context, recor
Runtime: r.Runtime,
AgentExecutionID: r.AgentExecutionID,
ContainerID: r.ContainerID,
Metadata: r.Metadata,
}
}
a.mgr.PollRemoteStatusForRecords(ctx, lcRecords)
@@ -48,6 +48,11 @@ type Agent interface {
// RemoteAuth returns the auth methods this agent supports in remote environments.
// Returns nil if the agent has no remote auth configuration.
RemoteAuth() *RemoteAuth
// --- Installation ---
// InstallScript returns shell commands to pre-install the agent CLI in remote environments.
// Returns empty string if no installation is needed.
InstallScript() string
}
// InferenceAgent is an optional capability for agents that support one-shot LLM inference.
+11 -5
View File
@@ -14,6 +14,8 @@ var ampLogoLight []byte
//go:embed logos/amp_dark.svg
var ampLogoDark []byte
const ampPkg = "@sourcegraph/amp@latest"
var (
_ Agent = (*Amp)(nil)
_ PassthroughAgent = (*Amp)(nil)
@@ -32,7 +34,7 @@ func NewAmp() *Amp {
Supported: true,
Label: "CLI Passthrough",
Description: "Show terminal directly instead of chat interface",
PassthroughCmd: NewCommand("npx", "-y", "@sourcegraph/amp@latest"),
PassthroughCmd: NewCommand("npx", "-y", ampPkg),
ModelFlag: NewParam("-m", "{model}"),
IdleTimeout: 3 * time.Second,
BufferMaxBytes: DefaultBufferMaxBytes,
@@ -87,7 +89,7 @@ func (a *Amp) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *Amp) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "@sourcegraph/amp@latest", "--execute", "--stream-json", "--stream-json-input").
return Cmd("npx", "-y", ampPkg, "--execute", "--stream-json", "--stream-json-input").
Model(NewParam("-m", "{model}"), opts.Model).
Settings(ampPermSettings, opts.PermissionValues).
Build()
@@ -96,7 +98,7 @@ func (a *Amp) BuildCommand(opts CommandOptions) Command {
func (a *Amp) Runtime() *RuntimeConfig {
canRecover := true
return &RuntimeConfig{
Cmd: Cmd("npx", "-y", "@sourcegraph/amp@latest", "--execute", "--stream-json", "--stream-json-input").Build(),
Cmd: Cmd("npx", "-y", ampPkg, "--execute", "--stream-json", "--stream-json-input").Build(),
WorkingDir: "{workspace}",
Env: map[string]string{},
ResourceLimits: ResourceLimits{MemoryMB: 4096, CPUCores: 2.0, Timeout: time.Hour},
@@ -105,14 +107,18 @@ func (a *Amp) Runtime() *RuntimeConfig {
SessionConfig: SessionConfig{
CanRecover: &canRecover,
SessionDirTemplate: "{home}/.config/amp",
ForkSessionCmd: Cmd("npx", "-y", "@sourcegraph/amp@latest", "threads", "fork").Build(),
ContinueSessionCmd: Cmd("npx", "-y", "@sourcegraph/amp@latest", "threads", "continue", "--execute", "--stream-json", "--stream-json-input").Build(),
ForkSessionCmd: Cmd("npx", "-y", ampPkg, "threads", "fork").Build(),
ContinueSessionCmd: Cmd("npx", "-y", ampPkg, "threads", "continue", "--execute", "--stream-json", "--stream-json-input").Build(),
},
}
}
func (a *Amp) RemoteAuth() *RemoteAuth { return nil }
func (a *Amp) InstallScript() string {
return "npm install -g " + ampPkg
}
func (a *Amp) PermissionSettings() map[string]PermissionSetting {
return ampPermSettings
}
+8 -2
View File
@@ -17,6 +17,8 @@ var auggieLogoLight []byte
//go:embed logos/auggie_dark.svg
var auggieLogoDark []byte
const auggiePkg = "@augmentcode/auggie@0.16.2"
var (
_ Agent = (*Auggie)(nil)
_ PassthroughAgent = (*Auggie)(nil)
@@ -90,7 +92,7 @@ func (a *Auggie) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *Auggie) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "@augmentcode/auggie@0.16.2", "--acp").
return Cmd("npx", "-y", auggiePkg, "--acp").
Model(NewParam("--model", "{model}"), opts.Model).
Resume(NewParam("--resume"), opts.SessionID, false).
Permissions("--permission", auggiePermTools, opts).
@@ -103,7 +105,7 @@ func (a *Auggie) Runtime() *RuntimeConfig {
return &RuntimeConfig{
Image: "kandev/multi-agent",
Tag: "latest",
Cmd: Cmd("npx", "-y", "@augmentcode/auggie@0.16.2", "--acp").Build(),
Cmd: Cmd("npx", "-y", auggiePkg, "--acp").Build(),
WorkingDir: "/workspace",
RequiredEnv: []string{"AUGMENT_SESSION_AUTH"},
Env: map[string]string{},
@@ -144,6 +146,10 @@ func (a *Auggie) RemoteAuth() *RemoteAuth {
}
}
func (a *Auggie) InstallScript() string {
return "npm install -g " + auggiePkg
}
func (a *Auggie) PermissionSettings() map[string]PermissionSetting {
return auggiePermSettings
}
@@ -14,6 +14,8 @@ var claudeCodeLogoLight []byte
//go:embed logos/claude_code_dark.svg
var claudeCodeLogoDark []byte
const claudeCodePkg = "@anthropic-ai/claude-code@2.1.50"
var (
_ Agent = (*ClaudeCode)(nil)
_ PassthroughAgent = (*ClaudeCode)(nil)
@@ -79,7 +81,7 @@ func (a *ClaudeCode) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *ClaudeCode) BuildCommand(opts CommandOptions) Command {
b := Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.50",
b := Cmd("npx", "-y", claudeCodePkg,
"-p", "--output-format=stream-json", "--input-format=stream-json",
"--permission-prompt-tool=stdio", "--disallowedTools=AskUserQuestion",
"--setting-sources=user,project", "--verbose",
@@ -107,7 +109,7 @@ func (a *ClaudeCode) BuildCommand(opts CommandOptions) Command {
func (a *ClaudeCode) Runtime() *RuntimeConfig {
canRecover := true
return &RuntimeConfig{
Cmd: Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.50",
Cmd: Cmd("npx", "-y", claudeCodePkg,
"-p", "--output-format=stream-json", "--input-format=stream-json",
"--permission-prompt-tool=stdio", "--disallowedTools=AskUserQuestion",
"--setting-sources=user,project", "--verbose",
@@ -147,6 +149,10 @@ chmod 600 "${HOME}/.claude.json"`,
}
}
func (a *ClaudeCode) InstallScript() string {
return "npm install -g " + claudeCodePkg
}
func (a *ClaudeCode) PermissionSettings() map[string]PermissionSetting {
return claudeCodePermSettings
}
+8 -2
View File
@@ -14,6 +14,8 @@ var codexLogoLight []byte
//go:embed logos/codex_dark.svg
var codexLogoDark []byte
const codexPkg = "@openai/codex@0.104.0"
var (
_ Agent = (*Codex)(nil)
_ PassthroughAgent = (*Codex)(nil)
@@ -74,7 +76,7 @@ func (a *Codex) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *Codex) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "@openai/codex@0.104.0", "app-server").
return Cmd("npx", "-y", codexPkg, "app-server").
Model(NewParam("-c", "model=\"{model}\""), opts.Model).
Settings(codexPermSettings, opts.PermissionValues).
Build()
@@ -85,7 +87,7 @@ func (a *Codex) Runtime() *RuntimeConfig {
return &RuntimeConfig{
Image: "kandev/multi-agent",
Tag: "latest",
Cmd: Cmd("npx", "-y", "@openai/codex@0.104.0", "app-server").Build(),
Cmd: Cmd("npx", "-y", codexPkg, "app-server").Build(),
WorkingDir: "/workspace",
Env: map[string]string{},
Mounts: []MountTemplate{
@@ -122,6 +124,10 @@ func (a *Codex) RemoteAuth() *RemoteAuth {
}
}
func (a *Codex) InstallScript() string {
return "npm install -g " + codexPkg
}
func (a *Codex) PermissionSettings() map[string]PermissionSetting {
return codexPermSettings
}
@@ -14,6 +14,8 @@ var copilotLogoLight []byte
//go:embed logos/copilot_dark.svg
var copilotLogoDark []byte
const copilotPkg = "@github/copilot@0.0.406"
var (
_ Agent = (*Copilot)(nil)
_ PassthroughAgent = (*Copilot)(nil)
@@ -87,7 +89,7 @@ func (a *Copilot) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *Copilot) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "@github/copilot@0.0.406", "--server", "--log-level", "error").
return Cmd("npx", "-y", copilotPkg, "--server", "--log-level", "error").
Model(NewParam("--model", "{model}"), opts.Model).
Settings(copilotPermSettings, opts.PermissionValues).
Build()
@@ -96,7 +98,7 @@ func (a *Copilot) BuildCommand(opts CommandOptions) Command {
func (a *Copilot) Runtime() *RuntimeConfig {
canRecover := true
return &RuntimeConfig{
Cmd: Cmd("npx", "-y", "@github/copilot@0.0.406", "--server", "--log-level", "error").Build(),
Cmd: Cmd("npx", "-y", copilotPkg, "--server", "--log-level", "error").Build(),
WorkingDir: "{workspace}",
Env: map[string]string{},
ResourceLimits: ResourceLimits{MemoryMB: 4096, CPUCores: 2.0, Timeout: time.Hour},
@@ -113,6 +115,10 @@ func (a *Copilot) Runtime() *RuntimeConfig {
func (a *Copilot) RemoteAuth() *RemoteAuth { return nil }
func (a *Copilot) InstallScript() string {
return "npm install -g " + copilotPkg
}
func (a *Copilot) PermissionSettings() map[string]PermissionSetting {
return copilotPermSettings
}
+8 -2
View File
@@ -14,6 +14,8 @@ var geminiLogoLight []byte
//go:embed logos/gemini_dark.svg
var geminiLogoDark []byte
const geminiPkg = "@google/gemini-cli@0.25.2"
var (
_ Agent = (*Gemini)(nil)
_ PassthroughAgent = (*Gemini)(nil)
@@ -89,7 +91,7 @@ func (a *Gemini) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *Gemini) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "@google/gemini-cli@0.25.2", "--experimental-acp").
return Cmd("npx", "-y", geminiPkg, "--experimental-acp").
Model(NewParam("--model", "{model}"), opts.Model).
Settings(geminiPermSettings, opts.PermissionValues).
Build()
@@ -98,7 +100,7 @@ func (a *Gemini) BuildCommand(opts CommandOptions) Command {
func (a *Gemini) Runtime() *RuntimeConfig {
canRecover := false
return &RuntimeConfig{
Cmd: Cmd("npx", "-y", "@google/gemini-cli@0.25.2", "--experimental-acp").Build(),
Cmd: Cmd("npx", "-y", geminiPkg, "--experimental-acp").Build(),
WorkingDir: "{workspace}",
Env: map[string]string{},
ResourceLimits: ResourceLimits{MemoryMB: 4096, CPUCores: 2.0, Timeout: time.Hour},
@@ -131,6 +133,10 @@ func (a *Gemini) RemoteAuth() *RemoteAuth {
}
}
func (a *Gemini) InstallScript() string {
return "npm install -g " + geminiPkg
}
func (a *Gemini) PermissionSettings() map[string]PermissionSetting {
return geminiPermSettings
}
@@ -91,6 +91,8 @@ func (a *MockAgent) Runtime() *RuntimeConfig {
func (a *MockAgent) RemoteAuth() *RemoteAuth { return nil }
func (a *MockAgent) InstallScript() string { return "" }
func (a *MockAgent) PermissionSettings() map[string]PermissionSetting {
return mockPermSettings
}
@@ -15,6 +15,8 @@ var opencodeLogoLight []byte
//go:embed logos/opencode_dark.svg
var opencodeLogoDark []byte
const opencodePkg = "opencode-ai@1.1.59"
var (
_ Agent = (*OpenCode)(nil)
_ PassthroughAgent = (*OpenCode)(nil)
@@ -94,13 +96,13 @@ func (a *OpenCode) ListModels(ctx context.Context) (*ModelList, error) {
}
func (a *OpenCode) BuildCommand(opts CommandOptions) Command {
return Cmd("npx", "-y", "opencode-ai@1.1.59", "serve", "--hostname", "127.0.0.1", "--port", "0").Build()
return Cmd("npx", "-y", opencodePkg, "serve", "--hostname", "127.0.0.1", "--port", "0").Build()
}
func (a *OpenCode) Runtime() *RuntimeConfig {
canRecover := true
return &RuntimeConfig{
Cmd: Cmd("npx", "-y", "opencode-ai@1.1.59", "serve", "--hostname", "127.0.0.1", "--port", "0").Build(),
Cmd: Cmd("npx", "-y", opencodePkg, "serve", "--hostname", "127.0.0.1", "--port", "0").Build(),
WorkingDir: "{workspace}",
Env: map[string]string{},
ResourceLimits: ResourceLimits{MemoryMB: 4096, CPUCores: 2.0, Timeout: time.Hour},
@@ -115,6 +117,10 @@ func (a *OpenCode) Runtime() *RuntimeConfig {
func (a *OpenCode) RemoteAuth() *RemoteAuth { return nil }
func (a *OpenCode) InstallScript() string {
return "npm install -g " + opencodePkg
}
func (a *OpenCode) PermissionSettings() map[string]PermissionSetting {
return opencodePermSettings
}
@@ -81,6 +81,10 @@ func (a *OpenCodeACP) Runtime() *RuntimeConfig {
func (a *OpenCodeACP) RemoteAuth() *RemoteAuth { return nil }
func (a *OpenCodeACP) InstallScript() string {
return "npm install -g " + opencodePkg
}
func (a *OpenCodeACP) PermissionSettings() map[string]PermissionSetting {
return opencodePermSettings
}
@@ -116,6 +116,7 @@ func (a *TUIAgent) BuildCommand(_ CommandOptions) Command {
return a.cmd
}
func (a *TUIAgent) InstallScript() string { return "" }
func (a *TUIAgent) RemoteAuth() *RemoteAuth { return nil }
func (a *TUIAgent) PermissionSettings() map[string]PermissionSetting { return nil }
@@ -30,6 +30,7 @@ func (a *discoveryTestAgent) PermissionSettings() map[string]agents.PermissionSe
}
func (a *discoveryTestAgent) Runtime() *agents.RuntimeConfig { return nil }
func (a *discoveryTestAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *discoveryTestAgent) InstallScript() string { return "" }
func (a *discoveryTestAgent) BuildCommand(opts agents.CommandOptions) agents.Command {
return agents.NewCommand()
}
@@ -89,18 +89,6 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
return "", nil, fmt.Errorf("agentctl health check failed: %w", err)
}
// Convert MCP server configs
var mcpServers []agentctl.McpServerConfig
for _, mcp := range config.McpServers {
mcpServers = append(mcpServers, agentctl.McpServerConfig{
Name: mcp.Name,
URL: mcp.URL,
Type: mcp.Type,
Command: mcp.Command,
Args: mcp.Args,
})
}
// Create an instance via the control API (same flow as standalone mode)
agentType := ""
if config.AgentConfig != nil {
@@ -113,7 +101,7 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe
AgentType: agentType,
Env: config.Credentials,
AutoStart: false,
McpServers: mcpServers,
McpServers: config.McpServers,
SessionID: config.SessionID,
}
@@ -56,6 +56,9 @@ cd {{workspace.path}}
# ---- Repository setup (if configured) ----
{{repository.setup_script}}
# ---- Pre-install agent CLI(s) ----
{{kandev.agents.install}}
# ---- Install and start Kandev agent controller ----
{{kandev.agentctl.install}}
{{kandev.agentctl.start}}
@@ -85,6 +88,9 @@ cd {{workspace.path}}
# ---- Repository setup (if configured) ----
{{repository.setup_script}}
# ---- Pre-install agent CLI(s) ----
{{kandev.agents.install}}
# ---- Install and start Kandev agent controller ----
echo "Starting agent controller..."
{{kandev.agentctl.install}}
@@ -3,6 +3,7 @@ package lifecycle
import (
"context"
"strings"
"time"
"github.com/kandev/kandev/internal/agent/agents"
@@ -40,13 +41,8 @@ type ExecutorBackend interface {
}
// McpServerConfig holds configuration for an MCP server.
type McpServerConfig struct {
Name string `json:"name"`
URL string `json:"url,omitempty"`
Type string `json:"type,omitempty"`
Command string `json:"command,omitempty"`
Args []string `json:"args,omitempty"`
}
// Type alias for agentctl.McpServerConfig to avoid conversion boilerplate.
type McpServerConfig = agentctl.McpServerConfig
// Metadata keys for runtime-specific configuration
const (
@@ -66,6 +62,72 @@ const (
MetadataKeyGitUserEmail = "git_user_email"
)
// persistentMetadataKeys lists metadata keys carried forward from a previous
// ExecutorRunning record when a session is resumed. Keys not listed here
// (e.g., task_description, session_id) are treated as launch-time-only and
// are NOT copied on resume.
var persistentMetadataKeys = map[string]bool{
// Sprites runtime
"sprite_name": true,
"sprite_state": true,
"sprite_created_at": true,
"local_port": true,
"instance_port": true,
// Executor type marker
MetadataKeyIsRemote: true,
// Executor profile / auth config
MetadataKeyCleanupScript: true,
MetadataKeyRepoSetupScript: true,
MetadataKeyRemoteAuthHome: true,
MetadataKeyGitUserName: true,
MetadataKeyGitUserEmail: true,
"remote_credentials": true,
"remote_auth_secrets": true,
"executor_mcp_policy": true,
"sprites_network_policy_rules": true,
"executor_profile_id": true,
}
// persistentMetadataPrefixes lists key prefixes that should persist.
// Any key starting with one of these prefixes is carried forward.
var persistentMetadataPrefixes = []string{
"env_secret_id_", // Secret store UUIDs for profile env vars
}
// ShouldPersistMetadataKey returns true if the given metadata key should
// be carried forward when resuming a session from an ExecutorRunning record.
func ShouldPersistMetadataKey(key string) bool {
if persistentMetadataKeys[key] {
return true
}
for _, prefix := range persistentMetadataPrefixes {
if strings.HasPrefix(key, prefix) {
return true
}
}
return false
}
// FilterPersistentMetadata returns a copy of src containing only keys that
// should be carried forward across session resumes. Returns nil if no keys match.
func FilterPersistentMetadata(src map[string]interface{}) map[string]interface{} {
if src == nil {
return nil
}
filtered := make(map[string]interface{})
for k, v := range src {
if ShouldPersistMetadataKey(k) {
filtered[k] = v
}
}
if len(filtered) == 0 {
return nil
}
return filtered
}
// RemoteStatus describes runtime health/details for remote executors.
// It is intentionally generic so each executor can include extra details in Details.
type RemoteStatus struct {
@@ -0,0 +1,67 @@
package lifecycle
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestShouldPersistMetadataKey(t *testing.T) {
tests := []struct {
name string
key string
want bool
}{
{name: "exact match sprite_name", key: "sprite_name", want: true},
{name: "exact match is_remote", key: MetadataKeyIsRemote, want: true},
{name: "exact match cleanup_script", key: MetadataKeyCleanupScript, want: true},
{name: "exact match executor_profile_id", key: "executor_profile_id", want: true},
{name: "prefix env_secret_id_", key: "env_secret_id_SPRITES_API_TOKEN", want: true},
{name: "prefix env_secret_id_ another key", key: "env_secret_id_OPENAI_KEY", want: true},
{name: "not persistent task_description", key: "task_description", want: false},
{name: "not persistent session_id", key: "session_id", want: false},
{name: "not persistent empty", key: "", want: false},
{name: "not persistent arbitrary key", key: "some_random_key", want: false},
{name: "partial prefix match (no underscore)", key: "env_secret_id", want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := ShouldPersistMetadataKey(tt.key)
require.Equal(t, tt.want, got)
})
}
}
func TestFilterPersistentMetadata(t *testing.T) {
t.Run("nil input returns nil", func(t *testing.T) {
require.Nil(t, FilterPersistentMetadata(nil))
})
t.Run("empty map returns nil", func(t *testing.T) {
require.Nil(t, FilterPersistentMetadata(map[string]interface{}{}))
})
t.Run("no persistent keys returns nil", func(t *testing.T) {
src := map[string]interface{}{
"task_description": "do something",
"session_id": "abc",
}
require.Nil(t, FilterPersistentMetadata(src))
})
t.Run("filters to persistent keys only", func(t *testing.T) {
src := map[string]interface{}{
"sprite_name": "kandev-abc",
"task_description": "should be dropped",
"env_secret_id_SPRITES_API_TOKEN": "secret-123",
MetadataKeyIsRemote: true,
}
got := FilterPersistentMetadata(src)
require.NotNil(t, got)
require.Equal(t, "kandev-abc", got["sprite_name"])
require.Equal(t, "secret-123", got["env_secret_id_SPRITES_API_TOKEN"])
require.Equal(t, true, got[MetadataKeyIsRemote])
require.NotContains(t, got, "task_description")
})
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,236 @@
package lifecycle
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
sprites "github.com/superfly/sprites-go"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/remoteauth"
)
// uploadCredentials reads the remote_credentials metadata and uploads the selected
// credential files to the sprite. Also handles gh_cli_token auto-detect,
// secret-based auth via remote_auth_secrets, and agent auth setup scripts.
func (r *SpritesExecutor) uploadCredentials(
ctx context.Context,
sprite *sprites.Sprite,
req *ExecutorCreateRequest,
onOutput func(string),
) error {
catalog := r.buildRemoteAuthCatalog()
// Handle secret-based auth (e.g., GITHUB_TOKEN from a stored secret)
r.resolveAuthSecrets(ctx, req, catalog)
// Run auth setup scripts for env-type methods (e.g., Claude Code credential files)
r.runAuthSetupScripts(ctx, sprite, req, catalog, onOutput)
credsJSON, _ := req.Metadata["remote_credentials"].(string)
if credsJSON == "" {
return nil
}
var selectedMethodIDs []string
if err := json.Unmarshal([]byte(credsJSON), &selectedMethodIDs); err != nil {
return fmt.Errorf("failed to parse remote_credentials: %w", err)
}
// Handle gh_cli_token: detect locally and inject as env var
selectedMethodIDs = r.resolveGHToken(selectedMethodIDs, req)
if len(selectedMethodIDs) == 0 {
return nil
}
fileMethods := make([]remoteauth.Method, 0, len(selectedMethodIDs))
for _, methodID := range selectedMethodIDs {
method, ok := catalog.FindMethod(methodID)
if !ok {
r.logger.Warn("unknown remote auth method, skipping", zap.String("method_id", methodID))
continue
}
if method.Type != "files" {
continue
}
fileMethods = append(fileMethods, method)
}
if len(fileMethods) == 0 {
return nil
}
stepCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
uploader := &spriteFileUploader{sprite: sprite, runtime: r}
targetHomeDir, err := r.resolveRemoteAuthHomeDir(stepCtx, req, uploader)
if err != nil {
return err
}
return UploadCredentialFiles(stepCtx, uploader, fileMethods, targetHomeDir, r.logger)
}
// runAuthSetupScripts executes setup scripts for env-type auth methods whose env var
// is present in req.Env. This handles both secret-store-resolved and directly-injected env vars.
// The optional onOutput callback streams output to the caller (e.g., progress UI).
func (r *SpritesExecutor) runAuthSetupScripts(
ctx context.Context,
sprite *sprites.Sprite,
req *ExecutorCreateRequest,
catalog remoteauth.Catalog,
onOutput func(string),
) {
for _, spec := range catalog.Specs {
for _, method := range spec.Methods {
if method.Type != "env" || method.SetupScript == "" || method.EnvVar == "" {
continue
}
if req.Env[method.EnvVar] == "" {
continue
}
if onOutput != nil {
onOutput(fmt.Sprintf("Setting up %s credentials...", spec.DisplayName))
}
stepCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
cmd := sprite.CommandContext(stepCtx, "sh", "-c", method.SetupScript)
cmd.Env = r.buildSpriteEnv(req.Env)
out, err := cmd.CombinedOutput()
cancel()
if err != nil {
r.logger.Warn("auth setup script failed",
zap.String("method_id", method.MethodID),
zap.String("output", strings.TrimSpace(string(out))),
zap.Error(err))
if onOutput != nil {
onOutput(fmt.Sprintf("Warning: %s credential setup failed", spec.DisplayName))
}
} else {
r.logger.Debug("auth setup script completed",
zap.String("method_id", method.MethodID))
}
}
}
}
// resolveGHToken handles the gh_cli_token credential: detects the token locally
// and injects it as GITHUB_TOKEN in the request env. Returns filtered credential IDs.
func (r *SpritesExecutor) resolveGHToken(credentialIDs []string, req *ExecutorCreateRequest) []string {
if !containsID(credentialIDs, "gh_cli_token") {
return credentialIDs
}
token, err := DetectGHToken()
if err != nil {
r.logger.Warn("failed to detect gh token", zap.Error(err))
} else {
if req.Env == nil {
req.Env = make(map[string]string)
}
req.Env["GITHUB_TOKEN"] = token
r.logger.Debug("set GITHUB_TOKEN from local gh auth token")
}
return removeID(credentialIDs, "gh_cli_token")
}
// resolveAuthSecrets reads remote_auth_secrets from metadata and resolves secret values
// into environment variables (e.g., gh_cli secret -> GITHUB_TOKEN).
func (r *SpritesExecutor) resolveAuthSecrets(
ctx context.Context,
req *ExecutorCreateRequest,
catalog remoteauth.Catalog,
) {
authSecretsJSON, _ := req.Metadata["remote_auth_secrets"].(string)
if authSecretsJSON == "" {
return
}
var authSecrets map[string]string
if err := json.Unmarshal([]byte(authSecretsJSON), &authSecrets); err != nil {
r.logger.Warn("failed to parse remote_auth_secrets", zap.Error(err))
return
}
for methodID, secretID := range authSecrets {
if secretID == "" {
continue
}
method, ok := catalog.FindMethod(methodID)
if !ok || method.Type != "env" || method.EnvVar == "" {
continue
}
value, err := r.secretStore.Reveal(ctx, secretID)
if err != nil {
r.logger.Warn("failed to resolve auth secret",
zap.String("method_id", methodID),
zap.String("secret_id", secretID),
zap.Error(err))
continue
}
if req.Env == nil {
req.Env = make(map[string]string)
}
req.Env[method.EnvVar] = value
r.logger.Debug("set env from auth secret", zap.String("key", method.EnvVar), zap.String("method_id", methodID))
}
}
func (r *SpritesExecutor) buildRemoteAuthCatalog() remoteauth.Catalog {
if r.agentList == nil {
return remoteauth.BuildCatalog(nil)
}
return remoteauth.BuildCatalog(r.agentList.ListEnabled())
}
func (r *SpritesExecutor) resolveRemoteAuthHomeDir(
ctx context.Context,
req *ExecutorCreateRequest,
cmdRunner commandOutputRunner,
) (string, error) {
if req != nil && req.Metadata != nil {
if override, ok := req.Metadata[MetadataKeyRemoteAuthHome].(string); ok {
trimmed := strings.TrimSpace(override)
if trimmed != "" {
r.logger.Debug("using remote auth home override", zap.String("home_dir", trimmed))
return trimmed, nil
}
}
}
if cmdRunner == nil {
return "", fmt.Errorf("failed to resolve remote user home for credential upload: command runner unavailable")
}
out, err := cmdRunner.RunCommandOutput(ctx, "sh", "-lc", "printf %s \"$HOME\"")
if err != nil {
return "", fmt.Errorf("failed to resolve remote user home for credential upload: %w", err)
}
home := strings.TrimSpace(string(out))
if home == "" {
return "", fmt.Errorf("failed to resolve remote user home for credential upload: empty HOME")
}
r.logger.Debug("resolved remote auth home", zap.String("home_dir", home))
return home, nil
}
func containsID(ids []string, target string) bool {
for _, id := range ids {
if id == target {
return true
}
}
return false
}
func removeID(ids []string, target string) []string {
result := make([]string, 0, len(ids))
for _, id := range ids {
if id != target {
result = append(result, id)
}
}
return result
}
@@ -0,0 +1,138 @@
package lifecycle
import (
"context"
"errors"
"math/rand"
"os"
"regexp"
"strconv"
"strings"
"time"
sprites "github.com/superfly/sprites-go"
"go.uber.org/zap"
)
var uploadHTTPStatusRE = regexp.MustCompile(`(?i)\b(?:http|status)\s*:?\s*(\d{3})\b`)
func (r *SpritesExecutor) injectTokenIntoURL(remoteURL string, env map[string]string) string {
token := env["GITHUB_TOKEN"]
if token == "" {
return remoteURL
}
if converted := rewriteGitHubSSHToHTTPS(remoteURL); converted != "" {
remoteURL = converted
}
if strings.HasPrefix(remoteURL, "https://") {
return strings.Replace(remoteURL, "https://", "https://"+token+"@", 1)
}
return remoteURL
}
func rewriteGitHubSSHToHTTPS(remoteURL string) string {
const (
sshPrefixA = "git@github.com:"
sshPrefixB = "ssh://git@github.com/"
)
switch {
case strings.HasPrefix(remoteURL, sshPrefixA):
return "https://github.com/" + strings.TrimPrefix(remoteURL, sshPrefixA)
case strings.HasPrefix(remoteURL, sshPrefixB):
return "https://github.com/" + strings.TrimPrefix(remoteURL, sshPrefixB)
default:
return ""
}
}
func (r *SpritesExecutor) writeFileWithRetry(
ctx context.Context,
sprite *sprites.Sprite,
path string,
data []byte,
mode os.FileMode,
) error {
backoff := 700 * time.Millisecond
var lastErr error
for attempt := 1; attempt <= spriteUploadMaxRetries+1; attempt++ {
err := sprite.Filesystem().WriteFileContext(ctx, path, data, mode)
if err == nil {
return nil
}
lastErr = err
if attempt > spriteUploadMaxRetries || !isTransientUploadError(err) || ctx.Err() != nil {
break
}
jitter := time.Duration(rand.Intn(300)) * time.Millisecond
wait := backoff + jitter
r.logger.Warn("retrying sprite file upload after transient error",
zap.String("path", path),
zap.Int("attempt", attempt),
zap.Duration("retry_in", wait),
zap.Error(err))
time.Sleep(wait)
backoff *= 2
}
return lastErr
}
func isTransientUploadError(err error) bool {
if err == nil {
return false
}
if errors.Is(err, context.DeadlineExceeded) {
return true
}
msg := strings.ToLower(err.Error())
if status := extractUploadHTTPStatus(msg); status != 0 {
if status == 408 || status == 429 || status >= 500 {
return true
}
}
return strings.Contains(msg, "client.timeout exceeded while awaiting headers") ||
strings.Contains(msg, "request canceled") ||
strings.Contains(msg, "timeout") ||
strings.Contains(msg, "text file busy") ||
strings.Contains(msg, "connection reset") ||
strings.Contains(msg, "temporary")
}
func extractUploadHTTPStatus(msg string) int {
matches := uploadHTTPStatusRE.FindStringSubmatch(msg)
if len(matches) < 2 {
return 0
}
code, err := strconv.Atoi(matches[1])
if err != nil {
return 0
}
return code
}
func (r *SpritesExecutor) buildSpriteEnv(env map[string]string) []string {
result := make([]string, 0, len(env))
for k, v := range env {
result = append(result, k+"="+v)
}
return result
}
// newStepReporter creates a reporting function that calls OnProgress if non-nil.
func newStepReporter(onProgress PrepareProgressCallback) func(PrepareStep, int) {
return func(step PrepareStep, idx int) {
if onProgress != nil {
onProgress(step, idx, spritesTotalSteps)
}
}
}
// lastLines returns the last n lines of s.
func lastLines(s string, n int) string {
lines := strings.Split(s, "\n")
if len(lines) <= n {
return s
}
return strings.Join(lines[len(lines)-n:], "\n")
}
@@ -0,0 +1,129 @@
package lifecycle
import (
"encoding/json"
"sync"
"testing"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/agent/agents"
)
func TestExtractAgentID(t *testing.T) {
tests := []struct {
name string
methodID string
want string
}{
{name: "full method ID", methodID: "agent:claude-code:env:ANTHROPIC_API_KEY", want: "claude-code"},
{name: "files method", methodID: "agent:claude-code:files:0", want: "claude-code"},
{name: "codex env", methodID: "agent:codex:env:OPENAI_API_KEY", want: "codex"},
{name: "two parts only", methodID: "agent:gemini", want: "gemini"},
{name: "non-agent prefix", methodID: "system:some-key:value", want: ""},
{name: "empty string", methodID: "", want: ""},
{name: "single segment", methodID: "agent", want: ""},
{name: "no colon", methodID: "agentclaude", want: ""},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := extractAgentID(tt.methodID)
require.Equal(t, tt.want, got)
})
}
}
// stubAgent wraps MockAgent to allow custom ID and install script.
type stubAgent struct {
*agents.MockAgent
id string
installScript string
}
func (a *stubAgent) ID() string { return a.id }
func (a *stubAgent) InstallScript() string { return a.installScript }
type mockAgentLister struct {
agents []agents.Agent
}
func (m *mockAgentLister) ListEnabled() []agents.Agent { return m.agents }
func newSpritesExecutorWithAgents(ags []agents.Agent) *SpritesExecutor {
return &SpritesExecutor{
agentList: &mockAgentLister{agents: ags},
logger: newTestLogger(),
tokens: make(map[string]string),
proxies: make(map[string]*SpritesProxySession),
mu: sync.RWMutex{},
}
}
func TestCollectAgentInstallScripts(t *testing.T) {
claude := &stubAgent{MockAgent: agents.NewMockAgent(), id: "claude-code", installScript: "npm i -g claude"}
codex := &stubAgent{MockAgent: agents.NewMockAgent(), id: "codex", installScript: "pip install codex"}
gemini := &stubAgent{MockAgent: agents.NewMockAgent(), id: "gemini", installScript: "gem install gemini"}
t.Run("no agents and no metadata returns empty", func(t *testing.T) {
r := newSpritesExecutorWithAgents([]agents.Agent{claude, codex})
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
Metadata: map[string]interface{}{},
})
require.Empty(t, got)
})
t.Run("current task agent included", func(t *testing.T) {
r := newSpritesExecutorWithAgents([]agents.Agent{claude, codex})
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
AgentConfig: claude,
Metadata: map[string]interface{}{},
})
require.Equal(t, []string{"npm i -g claude"}, got)
})
t.Run("agents from remote_credentials metadata", func(t *testing.T) {
credsJSON, _ := json.Marshal([]string{"agent:codex:env:OPENAI_API_KEY"})
r := newSpritesExecutorWithAgents([]agents.Agent{claude, codex})
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
Metadata: map[string]interface{}{"remote_credentials": string(credsJSON)},
})
require.Equal(t, []string{"pip install codex"}, got)
})
t.Run("agents from remote_auth_secrets metadata", func(t *testing.T) {
secretsJSON, _ := json.Marshal(map[string]string{"agent:gemini:env:API_KEY": "secret-1"})
r := newSpritesExecutorWithAgents([]agents.Agent{claude, codex, gemini})
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
Metadata: map[string]interface{}{"remote_auth_secrets": string(secretsJSON)},
})
require.Equal(t, []string{"gem install gemini"}, got)
})
t.Run("deduplicates agents from multiple sources", func(t *testing.T) {
credsJSON, _ := json.Marshal([]string{"agent:claude-code:files:0"})
secretsJSON, _ := json.Marshal(map[string]string{"agent:claude-code:env:KEY": "s1"})
r := newSpritesExecutorWithAgents([]agents.Agent{claude})
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
AgentConfig: claude,
Metadata: map[string]interface{}{
"remote_credentials": string(credsJSON),
"remote_auth_secrets": string(secretsJSON),
},
})
require.Equal(t, []string{"npm i -g claude"}, got)
})
t.Run("nil agentList returns empty", func(t *testing.T) {
r := &SpritesExecutor{
logger: newTestLogger(),
tokens: make(map[string]string),
proxies: make(map[string]*SpritesProxySession),
}
got := r.collectAgentInstallScripts(&ExecutorCreateRequest{
AgentConfig: claude,
Metadata: map[string]interface{}{},
})
require.Empty(t, got)
})
}
@@ -0,0 +1,233 @@
package lifecycle
import (
"context"
"fmt"
"strings"
"time"
sprites "github.com/superfly/sprites-go"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/scriptengine"
spritesutil "github.com/kandev/kandev/internal/sprites"
)
func (r *SpritesExecutor) reconnectSprite(ctx context.Context, client *sprites.Client, name string) (*sprites.Sprite, error) {
stepCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
sprite, err := client.GetSprite(stepCtx, name)
if err != nil {
return nil, fmt.Errorf("failed to reconnect sprite %q: %w", name, err)
}
return sprite, nil
}
func (r *SpritesExecutor) StopInstance(ctx context.Context, instance *ExecutorInstance, _ bool) error {
if instance == nil {
return nil
}
spriteName := getMetadataString(instance.Metadata, "sprite_name")
if spriteName == "" {
return nil
}
r.mu.Lock()
if proxy, ok := r.proxies[instance.InstanceID]; ok {
r.closeProxySession(proxy)
delete(r.proxies, instance.InstanceID)
}
r.mu.Unlock()
r.mu.RLock()
token := r.tokens[instance.InstanceID]
r.mu.RUnlock()
if token == "" {
token = r.resolveTokenFromMetadata(ctx, instance)
}
if token == "" {
r.logger.Warn("no cached API token for sprite instance, cannot destroy",
zap.String("instance_id", instance.InstanceID))
return nil
}
client := sprites.New(token)
sprite := client.Sprite(spriteName)
r.runTerminalCleanupScript(ctx, sprite, instance)
if err := sprite.Destroy(); err != nil {
r.logger.Warn("failed to destroy sprite",
zap.String("sprite_name", spriteName),
zap.Error(err))
return fmt.Errorf("failed to destroy sprite: %w", err)
}
r.mu.Lock()
delete(r.tokens, instance.InstanceID)
r.mu.Unlock()
r.logger.Info("sprite destroyed", zap.String("sprite_name", spriteName))
return nil
}
func (r *SpritesExecutor) runTerminalCleanupScript(ctx context.Context, sprite *sprites.Sprite, instance *ExecutorInstance) {
if sprite == nil || instance == nil {
return
}
if !shouldRunExecutorCleanup(instance.StopReason) {
return
}
script := strings.TrimSpace(getMetadataString(instance.Metadata, MetadataKeyCleanupScript))
if script == "" {
return
}
resolver := scriptengine.NewResolver().
WithProvider(scriptengine.WorkspaceProvider(spritesWorkspacePath)).
WithProvider(scriptengine.AgentctlProvider(r.agentctlPort, spritesWorkspacePath)).
WithProvider(scriptengine.GitIdentityProvider(instance.Metadata)).
WithProvider(scriptengine.RepositoryProvider(
instance.Metadata,
nil,
getGitRemoteURL,
r.injectTokenIntoURL,
))
resolved := resolver.Resolve(script)
if strings.TrimSpace(resolved) == "" {
return
}
stepCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
defer cancel()
out, err := sprite.CommandContext(stepCtx, "sh", "-c", resolved).CombinedOutput()
if err != nil {
r.logger.Warn("cleanup script failed in sprite",
zap.String("instance_id", instance.InstanceID),
zap.String("reason", instance.StopReason),
zap.String("output", strings.TrimSpace(lastLines(string(out), spriteOutputMaxLines))),
zap.Error(err))
return
}
r.logger.Debug("cleanup script completed in sprite",
zap.String("instance_id", instance.InstanceID),
zap.String("reason", instance.StopReason))
}
func shouldRunExecutorCleanup(reason string) bool {
switch strings.ToLower(strings.TrimSpace(reason)) {
case "task archived", "task deleted", "session archived", "session deleted":
return true
default:
return false
}
}
func (r *SpritesExecutor) GetRemoteStatus(ctx context.Context, instance *ExecutorInstance) (*RemoteStatus, error) {
if instance == nil {
return nil, fmt.Errorf("instance is nil")
}
spriteName := strings.TrimSpace(getMetadataString(instance.Metadata, "sprite_name"))
if spriteName == "" {
return &RemoteStatus{
RuntimeName: string(r.Name()),
State: "unknown",
LastCheckedAt: time.Now().UTC(),
}, nil
}
r.mu.RLock()
token := r.tokens[instance.InstanceID]
r.mu.RUnlock()
if token == "" {
token = r.resolveTokenFromMetadata(ctx, instance)
}
if token == "" {
return &RemoteStatus{
RuntimeName: string(r.Name()),
RemoteName: spriteName,
State: "unknown",
LastCheckedAt: time.Now().UTC(),
}, nil
}
stepCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
client := sprites.New(token, sprites.WithDisableControl())
sprite, err := client.GetSprite(stepCtx, spriteName)
if err != nil {
return nil, err
}
return &RemoteStatus{
RuntimeName: string(r.Name()),
RemoteName: spriteName,
State: spritesutil.NormalizeSpriteStatus(sprite.Status),
CreatedAt: nonZeroTimePtr(sprite.CreatedAt),
LastCheckedAt: time.Now().UTC(),
}, nil
}
// resolveTokenFromMetadata resolves the Sprites API token from the secret store
// using the secret ID persisted in metadata. This handles the post-restart case
// where the in-memory token cache is empty. On success, the token is cached.
func (r *SpritesExecutor) resolveTokenFromMetadata(ctx context.Context, instance *ExecutorInstance) string {
if r.secretStore == nil || instance == nil {
return ""
}
secretID := getMetadataString(instance.Metadata, "env_secret_id_SPRITES_API_TOKEN")
if secretID == "" {
return ""
}
revealed, err := r.secretStore.Reveal(ctx, secretID)
if err != nil || revealed == "" {
return ""
}
r.mu.Lock()
r.tokens[instance.InstanceID] = revealed
r.mu.Unlock()
return revealed
}
func nonZeroTimePtr(t time.Time) *time.Time {
if t.IsZero() {
return nil
}
return &t
}
func (r *SpritesExecutor) cleanupOnFailure(_ context.Context, sprite *sprites.Sprite, instanceID string, destroySprite bool) {
if sprite == nil {
return
}
r.logger.Warn("cleaning up sprite after failure", zap.String("instance_id", instanceID))
r.mu.Lock()
if proxy, ok := r.proxies[instanceID]; ok {
r.closeProxySession(proxy)
delete(r.proxies, instanceID)
}
delete(r.tokens, instanceID)
r.mu.Unlock()
if !destroySprite {
r.logger.Info("preserving sprite during cleanup (reconnect flow)",
zap.String("instance_id", instanceID))
return
}
if err := sprite.Destroy(); err != nil {
r.logger.Warn("failed to destroy sprite during cleanup", zap.Error(err))
}
}
func (r *SpritesExecutor) closeProxySession(proxy *SpritesProxySession) {
if proxy == nil {
return
}
if proxy.cancel != nil {
proxy.cancel()
}
if proxy.proxySession != nil {
_ = proxy.proxySession.Close()
}
}
@@ -0,0 +1,117 @@
package lifecycle
import (
"context"
"fmt"
"sync"
"testing"
"github.com/kandev/kandev/internal/secrets"
)
// mockSecretStore implements secrets.SecretStore for testing resolveTokenFromMetadata.
type mockSecretStore struct {
store map[string]string
err error
}
var _ secrets.SecretStore = (*mockSecretStore)(nil)
func (m *mockSecretStore) Create(_ context.Context, _ *secrets.SecretWithValue) error { return nil }
func (m *mockSecretStore) Get(_ context.Context, _ string) (*secrets.Secret, error) { return nil, nil }
func (m *mockSecretStore) Reveal(_ context.Context, id string) (string, error) {
if m.err != nil {
return "", m.err
}
return m.store[id], nil
}
func (m *mockSecretStore) Update(_ context.Context, _ string, _ *secrets.UpdateSecretRequest) error {
return nil
}
func (m *mockSecretStore) Delete(_ context.Context, _ string) error { return nil }
func (m *mockSecretStore) List(_ context.Context) ([]*secrets.SecretListItem, error) { return nil, nil }
func (m *mockSecretStore) Close() error { return nil }
func newTestSpritesExecutor(store secrets.SecretStore) *SpritesExecutor {
return &SpritesExecutor{
secretStore: store,
logger: newTestLogger(),
tokens: make(map[string]string),
proxies: make(map[string]*SpritesProxySession),
mu: sync.RWMutex{},
}
}
func TestResolveTokenFromMetadata(t *testing.T) {
t.Run("nil secret store returns empty", func(t *testing.T) {
r := &SpritesExecutor{
tokens: make(map[string]string),
}
got := r.resolveTokenFromMetadata(context.Background(), &ExecutorInstance{
InstanceID: "inst-1",
Metadata: map[string]interface{}{"env_secret_id_SPRITES_API_TOKEN": "secret-1"},
})
if got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("nil instance returns empty", func(t *testing.T) {
r := newTestSpritesExecutor(&mockSecretStore{store: map[string]string{"s1": "tok"}})
got := r.resolveTokenFromMetadata(context.Background(), nil)
if got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("no secret ID in metadata returns empty", func(t *testing.T) {
r := newTestSpritesExecutor(&mockSecretStore{store: map[string]string{"s1": "tok"}})
got := r.resolveTokenFromMetadata(context.Background(), &ExecutorInstance{
InstanceID: "inst-1",
Metadata: map[string]interface{}{},
})
if got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("secret store error returns empty", func(t *testing.T) {
r := newTestSpritesExecutor(&mockSecretStore{err: fmt.Errorf("vault sealed")})
got := r.resolveTokenFromMetadata(context.Background(), &ExecutorInstance{
InstanceID: "inst-1",
Metadata: map[string]interface{}{"env_secret_id_SPRITES_API_TOKEN": "secret-1"},
})
if got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("secret store returns empty value", func(t *testing.T) {
r := newTestSpritesExecutor(&mockSecretStore{store: map[string]string{"secret-1": ""}})
got := r.resolveTokenFromMetadata(context.Background(), &ExecutorInstance{
InstanceID: "inst-1",
Metadata: map[string]interface{}{"env_secret_id_SPRITES_API_TOKEN": "secret-1"},
})
if got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("valid secret returns token and caches it", func(t *testing.T) {
r := newTestSpritesExecutor(&mockSecretStore{store: map[string]string{"secret-1": "my-token"}})
got := r.resolveTokenFromMetadata(context.Background(), &ExecutorInstance{
InstanceID: "inst-1",
Metadata: map[string]interface{}{"env_secret_id_SPRITES_API_TOKEN": "secret-1"},
})
if got != "my-token" {
t.Fatalf("expected my-token, got %q", got)
}
// Verify token is cached
r.mu.RLock()
cached := r.tokens["inst-1"]
r.mu.RUnlock()
if cached != "my-token" {
t.Fatalf("expected cached token my-token, got %q", cached)
}
})
}
@@ -0,0 +1,84 @@
package lifecycle
import (
"context"
"encoding/json"
"fmt"
"net"
sprites "github.com/superfly/sprites-go"
"go.uber.org/zap"
)
func (r *SpritesExecutor) setupPortForwarding(
_ context.Context,
sprite *sprites.Sprite,
spriteName, instanceID string,
remotePort int,
) (int, error) {
localPort, err := getFreePort()
if err != nil {
return 0, fmt.Errorf("failed to get free port: %w", err)
}
r.logger.Debug("setting up port forwarding",
zap.Int("local_port", localPort),
zap.Int("remote_port", remotePort))
proxyCtx, cancel := context.WithCancel(context.Background())
session, err := sprite.ProxyPort(proxyCtx, localPort, remotePort)
if err != nil {
cancel()
return 0, fmt.Errorf("port forwarding failed: %w", err)
}
r.mu.Lock()
r.proxies[instanceID] = &SpritesProxySession{
spriteName: spriteName,
localPort: localPort,
proxySession: session,
cancel: cancel,
}
r.mu.Unlock()
return localPort, nil
}
func (r *SpritesExecutor) applyNetworkPolicy(
ctx context.Context,
client *sprites.Client,
spriteName string,
req *ExecutorCreateRequest,
) error {
rulesJSON, _ := req.Metadata["sprites_network_policy_rules"].(string)
if rulesJSON == "" {
return nil
}
var rules []sprites.NetworkPolicyRule
if err := json.Unmarshal([]byte(rulesJSON), &rules); err != nil {
return fmt.Errorf("failed to parse network policy rules: %w", err)
}
if len(rules) == 0 {
return nil
}
policyCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
r.logger.Info("applying network policy from profile",
zap.String("sprite_name", spriteName),
zap.Int("rule_count", len(rules)))
return client.UpdateNetworkPolicy(policyCtx, spriteName, &sprites.NetworkPolicy{Rules: rules})
}
// getFreePort finds an available local port.
func getFreePort() (int, error) {
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
return 0, err
}
defer func() { _ = l.Close() }()
return l.Addr().(*net.TCPAddr).Port, nil
}
@@ -0,0 +1,352 @@
package lifecycle
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"os"
"strings"
"sync"
"time"
sprites "github.com/superfly/sprites-go"
"go.uber.org/zap"
agentctl "github.com/kandev/kandev/internal/agentctl/client"
"github.com/kandev/kandev/internal/scriptengine"
)
// createSprite creates a new sprite via the API (explicit POST, not lazy).
func (r *SpritesExecutor) createSprite(ctx context.Context, client *sprites.Client, name string) (*sprites.Sprite, error) {
stepCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
r.logger.Debug("creating sprite via API", zap.String("sprite", name))
sprite, err := client.CreateSprite(stepCtx, name, nil)
if err != nil {
return nil, fmt.Errorf("failed to create sprite: %w", err)
}
return sprite, nil
}
func (r *SpritesExecutor) uploadAgentctl(ctx context.Context, sprite *sprites.Sprite) error {
binaryPath, err := r.agentctlResolver.ResolveLinuxBinary()
if err != nil {
return fmt.Errorf("agentctl binary not found: %w", err)
}
data, err := os.ReadFile(binaryPath)
if err != nil {
return fmt.Errorf("failed to read agentctl binary: %w", err)
}
stepCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
r.logger.Debug("uploading agentctl binary", zap.Int("size_bytes", len(data)))
// Upload via sprites filesystem API (HTTP PUT, much faster than stdin pipe for large binaries).
if err := r.writeFileWithRetry(stepCtx, sprite, "/usr/local/bin/agentctl", data, 0o755); err != nil {
return fmt.Errorf("failed to upload agentctl: %w", err)
}
// Verify the binary is present and executable.
if _, err := sprite.CommandContext(stepCtx, "test", "-x", "/usr/local/bin/agentctl").Output(); err != nil {
return fmt.Errorf("agentctl verification failed: %w", err)
}
return nil
}
// runPrepareScript resolves the prepare script with scriptengine and executes it,
// streaming stdout/stderr output through the onOutput callback.
func (r *SpritesExecutor) runPrepareScript(
ctx context.Context,
sprite *sprites.Sprite,
req *ExecutorCreateRequest,
onOutput func(string),
) error {
script := r.resolvePrepareScript(req)
if script == "" {
r.logger.Debug("no prepare script configured, skipping")
return nil
}
stepCtx, cancel := context.WithTimeout(ctx, spriteStepTimeout)
defer cancel()
r.logger.Debug("running prepare script")
cmd := sprite.CommandContext(stepCtx, "sh", "-c", script)
cmd.Env = r.buildSpriteEnv(req.Env)
stdout, err := cmd.StdoutPipe()
if err != nil {
return fmt.Errorf("failed to get stdout pipe: %w", err)
}
stderr, err := cmd.StderrPipe()
if err != nil {
return fmt.Errorf("failed to get stderr pipe: %w", err)
}
if err := cmd.Start(); err != nil {
return fmt.Errorf("failed to start prepare script: %w", err)
}
// Stream stdout and stderr concurrently (io.MultiReader is sequential,
// which blocks stderr until stdout EOF — bad for git progress output).
var outputBuf strings.Builder
var outputMu sync.Mutex
emitOutput := func(chunk []byte) {
outputMu.Lock()
outputBuf.Write(chunk)
if onOutput != nil {
onOutput(lastLines(outputBuf.String(), spriteOutputMaxLines))
}
outputMu.Unlock()
}
var wg sync.WaitGroup
readStream := func(rd io.Reader) {
defer wg.Done()
buf := make([]byte, 4096)
for {
n, readErr := rd.Read(buf)
if n > 0 {
emitOutput(buf[:n])
}
if readErr != nil {
break
}
}
}
wg.Add(2)
go readStream(stdout)
go readStream(stderr)
waitErr := cmd.Wait()
wg.Wait() // ensure all output is collected
if waitErr != nil {
return fmt.Errorf("prepare script failed: %w\n%s", waitErr, lastLines(outputBuf.String(), spriteOutputMaxLines))
}
return nil
}
// resolvePrepareScript builds the resolved prepare script using scriptengine.
func (r *SpritesExecutor) resolvePrepareScript(req *ExecutorCreateRequest) string {
script := getMetadataString(req.Metadata, MetadataKeySetupScript)
if script == "" {
script = DefaultPrepareScript("sprites")
}
if script == "" {
return ""
}
installScripts := r.collectAgentInstallScripts(req)
resolver := scriptengine.NewResolver().
WithProvider(scriptengine.WorkspaceProvider(spritesWorkspacePath)).
WithProvider(scriptengine.AgentctlProvider(r.agentctlPort, spritesWorkspacePath)).
WithProvider(scriptengine.GitIdentityProvider(req.Metadata)).
WithProvider(scriptengine.AgentInstallProvider(installScripts)).
WithProvider(scriptengine.RepositoryProvider(
req.Metadata,
req.Env,
getGitRemoteURL,
r.injectTokenIntoURL,
))
return resolver.Resolve(script)
}
// collectAgentInstallScripts extracts agent IDs from the executor profile metadata
// and the current task's agent config, then collects their install scripts.
func (r *SpritesExecutor) collectAgentInstallScripts(req *ExecutorCreateRequest) []string {
agentIDs := map[string]bool{}
// Always include the current task's agent.
if req.AgentConfig != nil {
agentIDs[req.AgentConfig.ID()] = true
}
// Extract agent IDs from remote_credentials (e.g., "agent:claude-code:files:0").
if credsJSON, _ := req.Metadata["remote_credentials"].(string); credsJSON != "" {
var methodIDs []string
if json.Unmarshal([]byte(credsJSON), &methodIDs) == nil {
for _, id := range methodIDs {
if agentID := extractAgentID(id); agentID != "" {
agentIDs[agentID] = true
}
}
}
}
// Extract agent IDs from remote_auth_secrets (e.g., "agent:codex:env:OPENAI_API_KEY").
if secretsJSON, _ := req.Metadata["remote_auth_secrets"].(string); secretsJSON != "" {
var secretsMap map[string]string
if json.Unmarshal([]byte(secretsJSON), &secretsMap) == nil {
for methodID := range secretsMap {
if agentID := extractAgentID(methodID); agentID != "" {
agentIDs[agentID] = true
}
}
}
}
// Look up install scripts from the agent list.
var scripts []string
if r.agentList != nil {
for _, ag := range r.agentList.ListEnabled() {
if agentIDs[ag.ID()] {
scripts = append(scripts, ag.InstallScript())
}
}
}
return scripts
}
// extractAgentID extracts the agent ID from method IDs like "agent:claude-code:env:TOKEN".
func extractAgentID(methodID string) string {
parts := strings.SplitN(methodID, ":", 3)
if len(parts) >= 2 && parts[0] == "agent" {
return parts[1]
}
return ""
}
// createAgentInstance creates a per-instance server on the agentctl control server
// running inside the sprite. Returns the port of the per-instance server.
func (r *SpritesExecutor) createAgentInstance(
ctx context.Context,
sprite *sprites.Sprite,
req *ExecutorCreateRequest,
) (int, error) {
instanceReq := agentctl.CreateInstanceRequest{
ID: req.InstanceID,
WorkspacePath: spritesWorkspacePath,
SessionID: req.SessionID,
Protocol: req.Protocol,
AgentType: agentTypeFromReq(req),
McpServers: req.McpServers,
}
reqJSON, err := json.Marshal(instanceReq)
if err != nil {
return 0, fmt.Errorf("failed to marshal instance request: %w", err)
}
stepCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
cmd := sprite.CommandContext(stepCtx, "sh", "-c",
fmt.Sprintf("curl -sf -X POST http://localhost:%d/api/v1/instances -H 'Content-Type: application/json' -d @-",
r.agentctlPort))
cmd.Stdin = bytes.NewReader(reqJSON)
out, err := cmd.Output()
if err != nil {
return 0, fmt.Errorf("failed to create agent instance: %w", err)
}
var resp agentctl.CreateInstanceResponse
if err := json.Unmarshal(out, &resp); err != nil {
return 0, fmt.Errorf("failed to parse instance response: %w (output: %s)", err, string(out))
}
r.logger.Debug("agent instance created inside sprite",
zap.String("instance_id", resp.ID),
zap.Int("port", resp.Port))
return resp.Port, nil
}
// getExistingInstancePort queries agentctl inside the sprite to check if a
// previously created instance is still running. Returns its port if found.
func (r *SpritesExecutor) getExistingInstancePort(
ctx context.Context,
sprite *sprites.Sprite,
instanceID string,
) (int, error) {
if instanceID == "" {
return 0, fmt.Errorf("no instance ID")
}
checkCmd := fmt.Sprintf(
"curl -sf http://localhost:%d/api/v1/instances/%s",
r.agentctlPort, instanceID)
stepCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel()
out, err := sprite.CommandContext(stepCtx, "sh", "-c", checkCmd).Output()
if err != nil {
return 0, fmt.Errorf("instance %s not found: %w", instanceID, err)
}
var resp struct {
Port int `json:"port"`
}
if err := json.Unmarshal(out, &resp); err != nil || resp.Port == 0 {
return 0, fmt.Errorf("failed to parse instance response: %v", err)
}
r.logger.Info("reusing existing agent instance",
zap.String("instance_id", instanceID),
zap.Int("port", resp.Port))
return resp.Port, nil
}
// isAgentSubprocessRunning checks whether the agent subprocess (e.g., Claude Code)
// is still alive inside an existing agentctl instance by querying the status endpoint.
func (r *SpritesExecutor) isAgentSubprocessRunning(
ctx context.Context,
sprite *sprites.Sprite,
instancePort int,
) bool {
statusCmd := fmt.Sprintf("curl -sf http://localhost:%d/api/v1/status", instancePort)
stepCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
out, err := sprite.CommandContext(stepCtx, "sh", "-c", statusCmd).Output()
if err != nil {
return false
}
var status struct {
AgentStatus string `json:"agent_status"`
}
if json.Unmarshal(out, &status) != nil {
return false
}
alive := status.AgentStatus == "running" || status.AgentStatus == "ready"
r.logger.Debug("checked agent subprocess status",
zap.Int("instance_port", instancePort),
zap.String("agent_status", status.AgentStatus),
zap.Bool("alive", alive))
return alive
}
func agentTypeFromReq(req *ExecutorCreateRequest) string {
if req.AgentConfig != nil {
return req.AgentConfig.ID()
}
return ""
}
func (r *SpritesExecutor) waitForHealth(ctx context.Context, sprite *sprites.Sprite) error {
deadline := time.Now().Add(spriteHealthTimeout)
healthURL := fmt.Sprintf("http://localhost:%d/health", r.agentctlPort)
for time.Now().Before(deadline) {
checkCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
out, err := sprite.CommandContext(checkCtx, "curl", "-sf", healthURL).Output()
cancel()
if err == nil && len(out) > 0 {
r.logger.Debug("agentctl is healthy in sprite")
return nil
}
time.Sleep(spriteHealthRetryWait)
}
return fmt.Errorf("agentctl did not become healthy within %v", spriteHealthTimeout)
}
@@ -81,18 +81,6 @@ func (r *StandaloneExecutor) CreateInstance(ctx context.Context, req *ExecutorCr
env["KANDEV_TASK_ID"] = req.TaskID
env["KANDEV_SESSION_ID"] = req.SessionID
// Convert MCP server configs
var mcpServers []agentctl.McpServerConfig
for _, mcp := range req.McpServers {
mcpServers = append(mcpServers, agentctl.McpServerConfig{
Name: mcp.Name,
URL: mcp.URL,
Type: mcp.Type,
Command: mcp.Command,
Args: mcp.Args,
})
}
// Create instance via control API
// Agent command is NOT set - workspace access only. Agent is started explicitly via agentctl client.
agentType := ""
@@ -107,7 +95,7 @@ func (r *StandaloneExecutor) CreateInstance(ctx context.Context, req *ExecutorCr
AgentType: agentType,
Env: env,
AutoStart: false,
McpServers: mcpServers,
McpServers: req.McpServers,
SessionID: req.SessionID,
}
@@ -13,6 +13,7 @@ import (
agentctltypes "github.com/kandev/kandev/internal/agentctl/types"
"github.com/kandev/kandev/internal/agentctl/types/streams"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
@@ -303,6 +304,36 @@ func (m *Manager) GetExecutionBySessionID(sessionID string) (*AgentExecution, bo
return m.executionStore.GetBySessionID(sessionID)
}
// IsRemoteSession checks whether a session is associated with a remote executor
// (e.g., sprites). It first checks the in-memory execution store, then falls back
// to the database via WorkspaceInfoProvider. This is useful when the execution
// hasn't been recreated yet after a backend restart.
func (m *Manager) IsRemoteSession(ctx context.Context, sessionID string) bool {
// Check in-memory execution first (fast path).
if execution, exists := m.executionStore.GetBySessionID(sessionID); exists {
if execution.RuntimeName == string(executor.NameSprites) {
return true
}
if execution.Metadata != nil {
if isRemote, ok := execution.Metadata[MetadataKeyIsRemote].(bool); ok && isRemote {
return true
}
}
return false
}
// Fall back to database records (post-restart, execution not yet recreated).
if m.workspaceInfoProvider == nil {
return false
}
info, err := m.workspaceInfoProvider.GetWorkspaceInfoForSession(ctx, "", sessionID)
if err != nil || info == nil {
return false
}
return info.RuntimeName == string(executor.NameSprites) || info.ExecutorType == string(models.ExecutorTypeSprites) ||
info.ExecutorType == string(models.ExecutorTypeRemoteDocker)
}
// GetAvailableCommandsForSession returns the available slash commands for a session.
// Returns nil if the session doesn't exist or has no commands stored.
//
@@ -3,6 +3,7 @@ package lifecycle
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"slices"
@@ -11,7 +12,11 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/agent/executor"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/task/models"
v1 "github.com/kandev/kandev/pkg/api/v1"
ws "github.com/kandev/kandev/pkg/websocket"
)
@@ -332,3 +337,113 @@ func TestManager_RestartAgentProcess_SessionInitFailure(t *testing.T) {
}
}
}
// --- IsRemoteSession tests ---
type mockWorkspaceInfoProvider struct {
infos map[string]*WorkspaceInfo
err error
}
func (m *mockWorkspaceInfoProvider) GetWorkspaceInfoForSession(_ context.Context, _, sessionID string) (*WorkspaceInfo, error) {
if m.err != nil {
return nil, m.err
}
return m.infos[sessionID], nil
}
func TestIsRemoteSession(t *testing.T) {
t.Run("in-memory execution with sprites runtime", func(t *testing.T) {
store := NewExecutionStore()
store.Add(&AgentExecution{
ID: "exec-1",
SessionID: "session-1",
RuntimeName: string(executor.NameSprites),
Status: v1.AgentStatusRunning,
})
mgr := &Manager{executionStore: store}
require.True(t, mgr.IsRemoteSession(context.Background(), "session-1"))
})
t.Run("in-memory execution with is_remote metadata", func(t *testing.T) {
store := NewExecutionStore()
store.Add(&AgentExecution{
ID: "exec-2",
SessionID: "session-2",
RuntimeName: string(executor.NameStandalone),
Status: v1.AgentStatusRunning,
Metadata: map[string]interface{}{MetadataKeyIsRemote: true},
})
mgr := &Manager{executionStore: store}
require.True(t, mgr.IsRemoteSession(context.Background(), "session-2"))
})
t.Run("in-memory execution with local runtime", func(t *testing.T) {
store := NewExecutionStore()
store.Add(&AgentExecution{
ID: "exec-3",
SessionID: "session-3",
RuntimeName: string(executor.NameStandalone),
Status: v1.AgentStatusRunning,
})
mgr := &Manager{executionStore: store}
require.False(t, mgr.IsRemoteSession(context.Background(), "session-3"))
})
t.Run("not in memory, DB returns sprites executor type", func(t *testing.T) {
store := NewExecutionStore()
provider := &mockWorkspaceInfoProvider{
infos: map[string]*WorkspaceInfo{
"session-4": {ExecutorType: string(models.ExecutorTypeSprites)},
},
}
mgr := &Manager{executionStore: store, workspaceInfoProvider: provider}
require.True(t, mgr.IsRemoteSession(context.Background(), "session-4"))
})
t.Run("not in memory, DB returns remote_docker executor type", func(t *testing.T) {
store := NewExecutionStore()
provider := &mockWorkspaceInfoProvider{
infos: map[string]*WorkspaceInfo{
"session-5": {ExecutorType: string(models.ExecutorTypeRemoteDocker)},
},
}
mgr := &Manager{executionStore: store, workspaceInfoProvider: provider}
require.True(t, mgr.IsRemoteSession(context.Background(), "session-5"))
})
t.Run("not in memory, DB returns sprites runtime name", func(t *testing.T) {
store := NewExecutionStore()
provider := &mockWorkspaceInfoProvider{
infos: map[string]*WorkspaceInfo{
"session-6": {RuntimeName: string(executor.NameSprites)},
},
}
mgr := &Manager{executionStore: store, workspaceInfoProvider: provider}
require.True(t, mgr.IsRemoteSession(context.Background(), "session-6"))
})
t.Run("not in memory, DB returns local type", func(t *testing.T) {
store := NewExecutionStore()
provider := &mockWorkspaceInfoProvider{
infos: map[string]*WorkspaceInfo{
"session-7": {ExecutorType: string(models.ExecutorTypeLocal)},
},
}
mgr := &Manager{executionStore: store, workspaceInfoProvider: provider}
require.False(t, mgr.IsRemoteSession(context.Background(), "session-7"))
})
t.Run("not in memory, DB error returns false", func(t *testing.T) {
store := NewExecutionStore()
provider := &mockWorkspaceInfoProvider{err: fmt.Errorf("db error")}
mgr := &Manager{executionStore: store, workspaceInfoProvider: provider}
require.False(t, mgr.IsRemoteSession(context.Background(), "session-8"))
})
t.Run("nil workspaceInfoProvider returns false", func(t *testing.T) {
store := NewExecutionStore()
mgr := &Manager{executionStore: store}
require.False(t, mgr.IsRemoteSession(context.Background(), "nonexistent"))
})
}
@@ -85,9 +85,15 @@ func (m *Manager) buildAgentCommand(req *LaunchRequest, profileInfo *AgentProfil
if req.ModelOverride != "" {
model = req.ModelOverride
}
// Only pass SessionID (for --resume flag) if the agent supports recovery.
// Agents with CanRecover=false (e.g. Auggie) use history context injection instead.
sessionID := req.ACPSessionID
if rt := agentConfig.Runtime(); rt != nil && !rt.SessionConfig.SupportsRecovery() {
sessionID = ""
}
cmdOpts := agents.CommandOptions{
Model: model,
SessionID: req.ACPSessionID,
SessionID: sessionID,
AutoApprove: autoApprove,
PermissionValues: permissionValues,
}
@@ -0,0 +1,73 @@
package lifecycle
import (
"strings"
"testing"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/agent/agents"
)
// resumeTestAgent is a minimal agent with a BuildCommand that respects the
// Resume helper (--resume <session_id>) so we can verify buildAgentCommand
// correctly gates the flag based on CanRecover.
type resumeTestAgent struct {
testAgent
canRecover *bool
}
func (a *resumeTestAgent) BuildCommand(opts agents.CommandOptions) agents.Command {
return agents.Cmd("test-agent", "--acp").
Resume(agents.NewParam("--resume"), opts.SessionID, false).
Build()
}
func (a *resumeTestAgent) Runtime() *agents.RuntimeConfig {
return &agents.RuntimeConfig{
Cmd: agents.NewCommand("test-agent", "--acp"),
SessionConfig: agents.SessionConfig{
CanRecover: a.canRecover,
},
}
}
func TestBuildAgentCommand_ResumeFlag(t *testing.T) {
mgr := newTestManager()
canRecoverTrue := true
canRecoverFalse := false
t.Run("CanRecover=true with ACPSessionID includes --resume", func(t *testing.T) {
ag := &resumeTestAgent{canRecover: &canRecoverTrue}
req := &LaunchRequest{ACPSessionID: "sess-123"}
cmds := mgr.buildAgentCommand(req, nil, ag)
require.Contains(t, cmds.initial, "--resume")
require.Contains(t, cmds.initial, "sess-123")
})
t.Run("CanRecover=false with ACPSessionID omits --resume", func(t *testing.T) {
ag := &resumeTestAgent{canRecover: &canRecoverFalse}
req := &LaunchRequest{ACPSessionID: "sess-123"}
cmds := mgr.buildAgentCommand(req, nil, ag)
require.False(t, strings.Contains(cmds.initial, "--resume"),
"expected no --resume flag, got: %s", cmds.initial)
require.False(t, strings.Contains(cmds.initial, "sess-123"),
"expected no session ID in command, got: %s", cmds.initial)
})
t.Run("CanRecover=true with empty ACPSessionID omits --resume", func(t *testing.T) {
ag := &resumeTestAgent{canRecover: &canRecoverTrue}
req := &LaunchRequest{ACPSessionID: ""}
cmds := mgr.buildAgentCommand(req, nil, ag)
require.False(t, strings.Contains(cmds.initial, "--resume"),
"expected no --resume flag when ACPSessionID is empty, got: %s", cmds.initial)
})
t.Run("CanRecover=nil (default true) with ACPSessionID includes --resume", func(t *testing.T) {
ag := &resumeTestAgent{canRecover: nil}
req := &LaunchRequest{ACPSessionID: "sess-456"}
cmds := mgr.buildAgentCommand(req, nil, ag)
require.Contains(t, cmds.initial, "--resume")
require.Contains(t, cmds.initial, "sess-456")
})
}
@@ -130,6 +130,7 @@ type RemoteStatusPollRecord struct {
Runtime string
AgentExecutionID string
ContainerID string
Metadata map[string]interface{}
}
// PollRemoteStatusForRecords performs a one-time remote status poll for executor records
@@ -157,6 +158,7 @@ func (m *Manager) PollRemoteStatusForRecords(ctx context.Context, records []Remo
SessionID: rec.SessionID,
RuntimeName: rec.Runtime,
ContainerID: rec.ContainerID,
Metadata: rec.Metadata,
}
status, statusErr := provider.GetRemoteStatus(ctx, instance)
if statusErr != nil {
@@ -36,7 +36,13 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
if execution.agentctl == nil {
return fmt.Errorf("execution %q has no agentctl client", executionID)
}
if execution.AgentCommand == "" {
// Check if we're reconnecting to an existing running agent process.
// When the existing process is still alive inside a remote executor (e.g., Sprites),
// we skip subprocess launch and go directly to ACP session initialization.
reuseExisting, _ := execution.Metadata["reuse_existing_process"].(bool)
if !reuseExisting && execution.AgentCommand == "" {
return fmt.Errorf("execution %q has no agent command configured", executionID)
}
@@ -47,25 +53,35 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
}
taskDescription := getTaskDescriptionFromMetadata(execution)
m.logger.Warn("StartAgentProcess: task description resolved",
zap.String("execution_id", executionID),
zap.String("task_id", execution.TaskID),
zap.Int("task_description_length", len(taskDescription)),
zap.String("agent_command", execution.AgentCommand),
zap.String("acp_session_id", execution.ACPSessionID))
approvalPolicy, agentDisplayName := m.resolveApprovalPolicyAndDisplayName(ctx, execution)
bootCommand, err := m.configureAndStartAgent(ctx, execution, taskDescription, approvalPolicy)
if err != nil {
return err
}
var bootCommand string
if reuseExisting {
// Agent subprocess is already running inside the remote executor.
// Skip configureAndStartAgent (which would spawn a conflicting subprocess)
// and go directly to ACP session initialization.
bootCommand = "reconnecting to running agent"
m.logger.Info("reusing existing agent process, skipping subprocess launch",
zap.String("execution_id", executionID),
zap.String("task_id", execution.TaskID))
} else {
m.logger.Info("StartAgentProcess: starting subprocess",
zap.String("execution_id", executionID),
zap.String("task_id", execution.TaskID),
zap.String("agent_command", execution.AgentCommand),
zap.String("acp_session_id", execution.ACPSessionID))
m.logger.Info("agent process started",
zap.String("execution_id", executionID),
zap.String("task_id", execution.TaskID),
zap.String("command", bootCommand))
var err error
bootCommand, err = m.configureAndStartAgent(ctx, execution, taskDescription, approvalPolicy)
if err != nil {
return err
}
m.logger.Info("agent process started",
zap.String("execution_id", executionID),
zap.String("task_id", execution.TaskID),
zap.String("command", bootCommand))
}
return m.initializeAgentSession(ctx, execution, bootCommand, agentDisplayName, taskDescription)
}
@@ -62,6 +62,7 @@ func (a *testAgent) Runtime() *agents.RuntimeConfig {
}
}
func (a *testAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *testAgent) InstallScript() string { return "" }
var (
_ agents.Agent = (*testAgent)(nil)
@@ -39,6 +39,7 @@ func (a *testAgent) PermissionSettings() map[string]agents.PermissionSetting {
}
func (a *testAgent) Runtime() *agents.RuntimeConfig { return a.runtime }
func (a *testAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *testAgent) InstallScript() string { return "" }
func newTestLogger() *logger.Logger {
log, _ := logger.NewLogger(logger.LoggingConfig{
@@ -40,6 +40,7 @@ func (a *availabilityTestAgent) PermissionSettings() map[string]agents.Permissio
}
func (a *availabilityTestAgent) Runtime() *agents.RuntimeConfig { return nil }
func (a *availabilityTestAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *availabilityTestAgent) InstallScript() string { return "" }
func (a *availabilityTestAgent) IsInstalled(ctx context.Context) (*agents.DiscoveryResult, error) {
a.mu.Lock()
@@ -98,6 +98,7 @@ func (a *testAgent) Runtime() *agents.RuntimeConfig {
return a.runtime
}
func (a *testAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *testAgent) InstallScript() string { return "" }
func newTestController(agentList map[string]agents.Agent) *Controller {
log, _ := logger.NewLogger(logger.LoggingConfig{
@@ -54,6 +54,7 @@ func (a *handlersTestAgent) PermissionSettings() map[string]agents.PermissionSet
}
func (a *handlersTestAgent) Runtime() *agents.RuntimeConfig { return nil }
func (a *handlersTestAgent) RemoteAuth() *agents.RemoteAuth { return nil }
func (a *handlersTestAgent) InstallScript() string { return "" }
func (a *handlersTestAgent) IsInstalled(ctx context.Context) (*agents.DiscoveryResult, error) {
return &agents.DiscoveryResult{Available: true}, nil
@@ -117,7 +117,7 @@ func (a *Adapter) Initialize(ctx context.Context) error {
"type": claudecode.MessageTypeControlRequest,
"request": map[string]any{"subtype": claudecode.SubtypeInitialize, "hooks": hooks},
})
initResp, err := a.client.Initialize(ctx, 60*time.Second, hooks)
initResp, err := a.client.Initialize(ctx, 180*time.Second, hooks)
initSpan.End()
if err != nil {
@@ -261,7 +261,7 @@ func (s *Server) handleWSInitialize(ctx context.Context, msg *ws.Message) *ws.Me
return resp
}
ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
ctx, cancel := context.WithTimeout(ctx, 180*time.Second)
defer cancel()
adapter := s.procMgr.GetAdapter()
@@ -13,11 +13,13 @@ import (
"strconv"
"strings"
"sync"
"time"
"github.com/gin-gonic/gin"
gorillaws "github.com/gorilla/websocket"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/executor"
"github.com/kandev/kandev/internal/agent/lifecycle"
agentctl "github.com/kandev/kandev/internal/agentctl/client"
"github.com/kandev/kandev/internal/agentctl/server/process"
@@ -176,7 +178,7 @@ func (h *TerminalHandler) HandleTerminalWS(c *gin.Context) {
c.JSON(http.StatusBadRequest, gin.H{"error": "terminalId required for shell mode"})
return
}
if h.shouldUseWorkspaceShell(sessionID) {
if h.shouldUseWorkspaceShell(c.Request.Context(), sessionID) {
h.handleRemoteUserShellWS(c, sessionID, terminalID)
return
}
@@ -198,25 +200,68 @@ func (h *TerminalHandler) HandleTerminalWS(c *gin.Context) {
}
}
func (h *TerminalHandler) shouldUseWorkspaceShell(sessionID string) bool {
func (h *TerminalHandler) shouldUseWorkspaceShell(ctx context.Context, sessionID string) bool {
// Check in-memory execution first (fast path when execution is already running).
execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID)
if !exists {
if exists {
if execution.RuntimeName == string(executor.NameSprites) {
return true
}
if execution.Metadata != nil {
if isRemote, ok := execution.Metadata[lifecycle.MetadataKeyIsRemote].(bool); ok && isRemote {
return true
}
}
return false
}
if execution.RuntimeName == "sprites" {
return true
// Execution not in memory — check DB records. After a backend restart the
// execution may not exist yet (resume in progress), but the session's
// executor type is persisted in the database.
return h.lifecycleMgr.IsRemoteSession(ctx, sessionID)
}
// waitForRemoteExecution polls the lifecycle manager for the session's execution,
// waiting up to 30 seconds for it to become available. After a backend restart the
// resume flow recreates the execution asynchronously; this bridges the gap so the
// terminal doesn't fall back to a local shell.
func (h *TerminalHandler) waitForRemoteExecution(ctx context.Context, sessionID string) (*lifecycle.AgentExecution, bool) {
// Fast path: execution already exists.
if execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID); exists {
return execution, true
}
if execution.Metadata == nil {
return false
h.logger.Info("waiting for remote execution to become available",
zap.String("session_id", sessionID))
ticker := time.NewTicker(500 * time.Millisecond)
defer ticker.Stop()
timer := time.NewTimer(30 * time.Second)
defer timer.Stop()
for {
select {
case <-ctx.Done():
return nil, false
case <-timer.C:
h.logger.Warn("timed out waiting for remote execution",
zap.String("session_id", sessionID))
return nil, false
case <-ticker.C:
if execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID); exists {
h.logger.Info("remote execution became available",
zap.String("session_id", sessionID),
zap.String("execution_id", execution.ID))
return execution, true
}
}
}
isRemote, ok := execution.Metadata[lifecycle.MetadataKeyIsRemote].(bool)
return ok && isRemote
}
func (h *TerminalHandler) handleRemoteUserShellWS(c *gin.Context, sessionID, terminalID string) {
execution, exists := h.lifecycleMgr.GetExecutionBySessionID(sessionID)
execution, exists := h.waitForRemoteExecution(c.Request.Context(), sessionID)
if !exists || execution.GetAgentCtlClient() == nil {
c.JSON(http.StatusNotFound, gin.H{"error": "execution not found"})
c.JSON(http.StatusNotFound, gin.H{"error": "remote execution not available"})
return
}
baseClient := execution.GetAgentCtlClient()
@@ -59,7 +59,9 @@ func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPS
// storeResumeToken stores an agent's session ID as the resume token for session recovery.
// This is called from handleACPSessionCreated, handleSessionStatusEvent, and handleCompleteStreamEvent.
// The optional lastMessageUUID is persisted alongside the token for --resume-session-at support.
// The resume token is only stored for agents that support native session resume (ACP session/load).
// The token is always stored. NativeSessionResume only gates ACP session/load vs session/new
// in session.go — agents without native resume (e.g., Claude Code) use the token for their
// own --resume CLI flag instead.
func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSessionID, lastMessageUUID string, preloadedSession ...*models.TaskSession) {
var session *models.TaskSession
if len(preloadedSession) > 0 && preloadedSession[0] != nil {
@@ -76,17 +78,6 @@ func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSe
}
}
// Clear the resume token for agents that don't support native session resume (ACP session/load).
// The ExecutorRunning record is still created (for worktree info etc.) but without a token,
// so the frontend won't attempt to auto-resume with a stale ACP session ID.
if session.AgentProfileID != "" {
profileInfo, profileErr := s.agentManager.ResolveAgentProfile(ctx, session.AgentProfileID)
if profileErr == nil && profileInfo != nil && !profileInfo.NativeSessionResume {
acpSessionID = ""
lastMessageUUID = ""
}
}
resumable := true
runtimeName := ""
if session.ExecutorID != "" {
@@ -96,6 +87,13 @@ func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSe
}
}
// Preserve existing metadata (e.g., sprite_name) from previous upsert by persistLaunchState.
// Without this, the full ON CONFLICT DO UPDATE in UpsertExecutorRunning would wipe metadata.
var existingMetadata map[string]interface{}
if existing, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID); err == nil && existing != nil {
existingMetadata = existing.Metadata
}
running := &models.ExecutorRunning{
ID: session.ID,
SessionID: session.ID,
@@ -108,6 +106,7 @@ func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSe
LastMessageUUID: lastMessageUUID,
AgentExecutionID: session.AgentExecutionID,
ContainerID: session.ContainerID,
Metadata: existingMetadata,
}
if len(session.Worktrees) > 0 {
running.WorktreeID = session.Worktrees[0].WorktreeID
@@ -511,6 +511,10 @@ func (s *Service) handleSessionModeEvent(ctx context.Context, payload *lifecycle
if sessionID == "" || s.eventBus == nil {
return
}
// Don't broadcast "default" mode — it's the baseline state, not worth sending.
if payload.Data.CurrentModeID == "default" {
return
}
eventPayload := lifecycle.SessionModeEventPayload{
TaskID: payload.TaskID,
SessionID: sessionID,
@@ -0,0 +1,95 @@
package orchestrator
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/events/bus"
)
// recordingEventBus records published events for assertions.
type recordingEventBus struct {
events []recordedEvent
}
type recordedEvent struct {
subject string
event *bus.Event
}
func (b *recordingEventBus) Publish(_ context.Context, subject string, event *bus.Event) error {
b.events = append(b.events, recordedEvent{subject: subject, event: event})
return nil
}
func (b *recordingEventBus) Subscribe(string, bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (b *recordingEventBus) QueueSubscribe(string, string, bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (b *recordingEventBus) Request(context.Context, string, *bus.Event, time.Duration) (*bus.Event, error) {
return nil, nil
}
func (b *recordingEventBus) Close() {}
func (b *recordingEventBus) IsConnected() bool { return true }
func TestHandleSessionModeEvent(t *testing.T) {
t.Run("publishes plan mode", func(t *testing.T) {
eb := &recordingEventBus{}
svc := &Service{logger: testLogger(), eventBus: eb}
svc.handleSessionModeEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
TaskID: "t1",
SessionID: "s1",
AgentID: "a1",
Data: &lifecycle.AgentStreamEventData{CurrentModeID: "plan"},
})
require.Len(t, eb.events, 1)
})
t.Run("skips default mode", func(t *testing.T) {
eb := &recordingEventBus{}
svc := &Service{logger: testLogger(), eventBus: eb}
svc.handleSessionModeEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
TaskID: "t1",
SessionID: "s1",
AgentID: "a1",
Data: &lifecycle.AgentStreamEventData{CurrentModeID: "default"},
})
require.Empty(t, eb.events)
})
t.Run("publishes empty mode (mode exit)", func(t *testing.T) {
eb := &recordingEventBus{}
svc := &Service{logger: testLogger(), eventBus: eb}
svc.handleSessionModeEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
TaskID: "t1",
SessionID: "s1",
AgentID: "a1",
Data: &lifecycle.AgentStreamEventData{CurrentModeID: ""},
})
require.Len(t, eb.events, 1)
})
t.Run("skips when session ID is empty", func(t *testing.T) {
eb := &recordingEventBus{}
svc := &Service{logger: testLogger(), eventBus: eb}
svc.handleSessionModeEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
TaskID: "t1",
SessionID: "",
Data: &lifecycle.AgentStreamEventData{CurrentModeID: "plan"},
})
require.Empty(t, eb.events)
})
}
@@ -137,6 +137,7 @@ type RemoteStatusPollRequest struct {
Runtime string
AgentExecutionID string
ContainerID string
Metadata map[string]interface{}
}
// AgentProfileInfo contains resolved profile information
@@ -200,6 +201,7 @@ type LaunchAgentResponse struct {
WorktreeID string
WorktreePath string
WorktreeBranch string
Metadata map[string]interface{}
}
// TaskExecution tracks an active task execution
@@ -95,10 +95,14 @@ func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID str
WorktreeID: resp.WorktreeID,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
Metadata: resp.Metadata,
}
if existingRunning != nil {
running.ResumeToken = existingRunning.ResumeToken
running.LastMessageUUID = existingRunning.LastMessageUUID
if running.Metadata == nil {
running.Metadata = lifecycle.FilterPersistentMetadata(existingRunning.Metadata)
}
}
if err := e.repo.UpsertExecutorRunning(ctx, running); err != nil {
e.logger.Warn("failed to persist executor runtime after launch",
@@ -338,6 +342,20 @@ func (e *Executor) applyRunningRecordToResumeRequest(ctx context.Context, req *L
req.PreviousExecutionID = running.AgentExecutionID
}
// Carry forward only persistent metadata from the previous run.
// Keys not in lifecycle.ShouldPersistMetadataKey() are launch-time-only
// and are intentionally NOT carried forward (e.g., task_description).
if running.Metadata != nil {
if req.Metadata == nil {
req.Metadata = make(map[string]interface{})
}
for k, v := range running.Metadata {
if _, exists := req.Metadata[k]; !exists && lifecycle.ShouldPersistMetadataKey(k) {
req.Metadata[k] = v
}
}
}
if running.ResumeToken != "" && startAgent {
req.ACPSessionID = running.ResumeToken
// Clear TaskDescription so the agent doesn't receive an automatic prompt on resume.
@@ -346,7 +364,8 @@ func (e *Executor) applyRunningRecordToResumeRequest(ctx context.Context, req *L
req.TaskDescription = ""
e.logger.Info("found resume token for session resumption",
zap.String("task_id", task.ID),
zap.String("session_id", session.ID))
zap.String("session_id", session.ID),
zap.Bool("has_resume_token", running.ResumeToken != ""))
} else if startAgent && session.State == models.TaskSessionStateWaitingForInput {
// Fresh-start resume (no resume token): don't auto-prompt with the task description.
req.TaskDescription = ""
@@ -452,10 +471,14 @@ func (e *Executor) persistResumeState(ctx context.Context, taskID string, sessio
WorktreeID: resp.WorktreeID,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
Metadata: resp.Metadata,
}
if existingRunning != nil {
running.ResumeToken = existingRunning.ResumeToken
running.LastMessageUUID = existingRunning.LastMessageUUID
if running.Metadata == nil {
running.Metadata = lifecycle.FilterPersistentMetadata(existingRunning.Metadata)
}
}
if err := e.repo.UpsertExecutorRunning(ctx, running); err != nil {
e.logger.Warn("failed to persist executor runtime after resume",
@@ -168,6 +168,13 @@ func (e *Executor) applyProfile(ctx context.Context, profileID string, cfg *exec
cfg.SetupScript = profile.PrepareScript
cfg.CleanupScript = profile.CleanupScript
cfg.ProfileEnv = e.resolveProfileEnvVars(ctx, profile.EnvVars)
// Persist secret store IDs in metadata so runtimes can resolve tokens after restart
// (e.g., SpritesExecutor needs SPRITES_API_TOKEN to poll remote status).
for _, ev := range profile.EnvVars {
if ev.SecretID != "" {
metadata["env_secret_id_"+ev.Key] = ev.SecretID
}
}
if profile.CleanupScript != "" {
metadata[lifecycle.MetadataKeyCleanupScript] = profile.CleanupScript
}
@@ -541,6 +541,7 @@ func (s *Service) reconcileSessionsOnStartup(ctx context.Context) {
Runtime: running.Runtime,
AgentExecutionID: running.AgentExecutionID,
ContainerID: running.ContainerID,
Metadata: running.Metadata,
})
}
s.reconcileOneSessionOnStartup(ctx, running)
@@ -100,6 +100,12 @@ var DefaultPlaceholders = []PlaceholderInfo{
Example: "main",
ExecutorTypes: []string{"worktree"},
},
{
Key: "kandev.agents.install",
Description: "Pre-installs agent CLIs globally for all agents configured on the profile",
Example: "npm install -g @anthropic-ai/claude-code@2.1.50",
ExecutorTypes: []string{"local_docker", "remote_docker", "sprites"},
},
{
Key: "kandev.agentctl.install",
Description: "Expands to full agentctl binary upload and install commands",
@@ -145,6 +145,24 @@ func repoNameFromPath(repoPath string) string {
return repoPath
}
// AgentInstallProvider returns a placeholder with combined install scripts for multiple agents.
func AgentInstallProvider(installScripts []string) PlaceholderProvider {
return func() map[string]string {
seen := map[string]bool{}
var lines []string
for _, s := range installScripts {
s = strings.TrimSpace(s)
if s != "" && !seen[s] {
seen[s] = true
lines = append(lines, s)
}
}
return map[string]string{
"kandev.agents.install": strings.Join(lines, "\n"),
}
}
}
func shellSingleQuote(value string) string {
return strings.ReplaceAll(value, "'", `'"'"'`)
}
@@ -1,6 +1,42 @@
package scriptengine
import "testing"
import (
"testing"
)
func TestAgentInstallProvider(t *testing.T) {
t.Run("empty scripts produce empty placeholder", func(t *testing.T) {
provider := AgentInstallProvider(nil)
vars := provider()
if got := vars["kandev.agents.install"]; got != "" {
t.Fatalf("expected empty, got %q", got)
}
})
t.Run("deduplicates scripts", func(t *testing.T) {
scripts := []string{
"npm install -g @anthropic-ai/claude-code@2.1.50",
"npm install -g @openai/codex@0.104.0",
"npm install -g @anthropic-ai/claude-code@2.1.50",
}
provider := AgentInstallProvider(scripts)
vars := provider()
want := "npm install -g @anthropic-ai/claude-code@2.1.50\nnpm install -g @openai/codex@0.104.0"
if got := vars["kandev.agents.install"]; got != want {
t.Fatalf("got %q, want %q", got, want)
}
})
t.Run("trims whitespace and skips empty", func(t *testing.T) {
scripts := []string{" npm install -g foo ", "", " ", "npm install -g bar"}
provider := AgentInstallProvider(scripts)
vars := provider()
want := "npm install -g foo\nnpm install -g bar"
if got := vars["kandev.agents.install"]; got != want {
t.Fatalf("got %q, want %q", got, want)
}
})
}
func TestRepositoryProvider_UsesRepositorySetupScriptKey(t *testing.T) {
provider := RepositoryProvider(map[string]any{
+11 -1
View File
@@ -318,7 +318,7 @@ func (h *Handler) listInstances(ctx context.Context, secretID string) ([]*Sprite
createdAt := s.CreatedAt.Format(time.RFC3339)
result = append(result, &SpritesInstance{
Name: s.Name,
HealthStatus: s.Status,
HealthStatus: NormalizeSpriteStatus(s.Status),
CreatedAt: createdAt,
UptimeSeconds: computeUptime(createdAt),
})
@@ -475,3 +475,13 @@ func computeUptime(createdAt string) int64 {
}
return int64(time.Since(t).Seconds())
}
// NormalizeSpriteStatus lowercases and trims the raw Sprites API status.
// The frontend handles display mapping for each state (running, cold, stopped, etc.).
func NormalizeSpriteStatus(status string) string {
s := strings.ToLower(strings.TrimSpace(status))
if s == "" {
return "unknown"
}
return s
}
@@ -0,0 +1,30 @@
package sprites
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestNormalizeSpriteStatus(t *testing.T) {
tests := []struct {
name string
input string
want string
}{
{name: "running", input: "Running", want: "running"},
{name: "cold", input: "Cold", want: "cold"},
{name: "stopped", input: "STOPPED", want: "stopped"},
{name: "with whitespace", input: " Running ", want: "running"},
{name: "empty string", input: "", want: "unknown"},
{name: "whitespace only", input: " ", want: "unknown"},
{name: "already lowercase", input: "starting", want: "starting"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := NormalizeSpriteStatus(tt.input)
require.Equal(t, tt.want, got)
})
}
}
+22 -21
View File
@@ -383,27 +383,28 @@ type Executor struct {
// ExecutorRunning tracks an active executor instance for a session.
type ExecutorRunning struct {
ID string `json:"id"`
SessionID string `json:"session_id"`
TaskID string `json:"task_id"`
ExecutorID string `json:"executor_id"`
Runtime string `json:"runtime,omitempty"`
Status string `json:"status"`
Resumable bool `json:"resumable"`
ResumeToken string `json:"resume_token,omitempty"`
LastMessageUUID string `json:"last_message_uuid,omitempty"`
AgentExecutionID string `json:"agent_execution_id,omitempty"`
ContainerID string `json:"container_id,omitempty"`
AgentctlURL string `json:"agentctl_url,omitempty"`
AgentctlPort int `json:"agentctl_port,omitempty"`
PID int `json:"pid,omitempty"`
WorktreeID string `json:"worktree_id,omitempty"`
WorktreePath string `json:"worktree_path,omitempty"`
WorktreeBranch string `json:"worktree_branch,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
LastSeenAt *time.Time `json:"last_seen_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
ID string `json:"id"`
SessionID string `json:"session_id"`
TaskID string `json:"task_id"`
ExecutorID string `json:"executor_id"`
Runtime string `json:"runtime,omitempty"`
Status string `json:"status"`
Resumable bool `json:"resumable"`
ResumeToken string `json:"resume_token,omitempty"`
LastMessageUUID string `json:"last_message_uuid,omitempty"`
AgentExecutionID string `json:"agent_execution_id,omitempty"`
ContainerID string `json:"container_id,omitempty"`
AgentctlURL string `json:"agentctl_url,omitempty"`
AgentctlPort int `json:"agentctl_port,omitempty"`
PID int `json:"pid,omitempty"`
WorktreeID string `json:"worktree_id,omitempty"`
WorktreePath string `json:"worktree_path,omitempty"`
WorktreeBranch string `json:"worktree_branch,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
LastSeenAt *time.Time `json:"last_seen_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// ProfileEnvVar represents an environment variable for an executor profile.
@@ -128,6 +128,8 @@ func (r *Repository) migrateTaskSessions() error {
func (r *Repository) runMigrations() error {
// Add last_message_uuid column to executors_running (ignore error if already exists)
_, _ = r.db.Exec(`ALTER TABLE executors_running ADD COLUMN last_message_uuid TEXT DEFAULT ''`)
// Add metadata column to executors_running (ignore error if already exists)
_, _ = r.db.Exec(`ALTER TABLE executors_running ADD COLUMN metadata TEXT DEFAULT '{}'`)
return nil
}
@@ -187,6 +189,7 @@ func (r *Repository) initInfraSchema() error {
worktree_branch TEXT DEFAULT '',
last_seen_at TIMESTAMP,
error_message TEXT DEFAULT '',
metadata TEXT DEFAULT '{}',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL
);
@@ -152,13 +152,22 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models.
}
running.UpdatedAt = now
metadataJSON := "{}"
if running.Metadata != nil {
b, err := json.Marshal(running.Metadata)
if err != nil {
return fmt.Errorf("failed to serialize executor running metadata: %w", err)
}
metadataJSON = string(b)
}
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
INSERT INTO executors_running (
id, session_id, task_id, executor_id, runtime, status, resumable, resume_token,
last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid,
worktree_id, worktree_path, worktree_branch, last_seen_at, error_message,
worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id) DO UPDATE SET
id = excluded.id,
task_id = excluded.task_id,
@@ -178,6 +187,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models.
worktree_branch = excluded.worktree_branch,
last_seen_at = excluded.last_seen_at,
error_message = excluded.error_message,
metadata = excluded.metadata,
updated_at = excluded.updated_at
`),
running.ID,
@@ -199,6 +209,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models.
running.WorktreeBranch,
running.LastSeenAt,
running.ErrorMessage,
metadataJSON,
running.CreatedAt,
running.UpdatedAt,
)
@@ -209,7 +220,7 @@ func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.Execut
rows, err := r.ro.QueryContext(ctx, `
SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token,
last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port,
worktree_id, worktree_path, worktree_branch, created_at, updated_at
worktree_id, worktree_path, worktree_branch, metadata, created_at, updated_at
FROM executors_running
ORDER BY updated_at DESC
`)
@@ -221,6 +232,7 @@ func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.Execut
var results []*models.ExecutorRunning
for rows.Next() {
running := &models.ExecutorRunning{}
var metadataJSON string
if scanErr := rows.Scan(
&running.ID,
&running.SessionID,
@@ -238,11 +250,17 @@ func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.Execut
&running.WorktreeID,
&running.WorktreePath,
&running.WorktreeBranch,
&metadataJSON,
&running.CreatedAt,
&running.UpdatedAt,
); scanErr != nil {
return nil, scanErr
}
if metadataJSON != "" && metadataJSON != "{}" {
if jsonErr := json.Unmarshal([]byte(metadataJSON), &running.Metadata); jsonErr != nil {
return nil, fmt.Errorf("failed to deserialize executor running metadata: %w", jsonErr)
}
}
results = append(results, running)
}
if err := rows.Err(); err != nil {
@@ -258,11 +276,12 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI
running := &models.ExecutorRunning{}
var resumable int
var lastSeen sql.NullTime
var metadataJSON string
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token,
last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid,
worktree_id, worktree_path, worktree_branch, last_seen_at, error_message,
worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, metadata,
created_at, updated_at
FROM executors_running
WHERE session_id = ?
@@ -286,6 +305,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI
&running.WorktreeBranch,
&lastSeen,
&running.ErrorMessage,
&metadataJSON,
&running.CreatedAt,
&running.UpdatedAt,
)
@@ -299,6 +319,11 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI
if lastSeen.Valid {
running.LastSeenAt = &lastSeen.Time
}
if metadataJSON != "" && metadataJSON != "{}" {
if jsonErr := json.Unmarshal([]byte(metadataJSON), &running.Metadata); jsonErr != nil {
return nil, fmt.Errorf("failed to deserialize executor running metadata: %w", jsonErr)
}
}
return running, nil
}
+12 -12
View File
@@ -369,16 +369,16 @@
@keyframes chat-input-running-plan {
0%,
100% {
border-color: color-mix(in oklab, oklch(0.6 0.05 260) 40%, var(--border));
border-color: color-mix(in oklab, oklch(0.65 0.2 300) 40%, var(--border));
box-shadow:
0 0 6px 1px color-mix(in oklab, oklch(0.6 0.05 260) 12%, transparent),
0 0 15px 3px color-mix(in oklab, oklch(0.6 0.05 260) 6%, transparent);
0 0 6px 1px color-mix(in oklab, oklch(0.65 0.2 300) 12%, transparent),
0 0 15px 3px color-mix(in oklab, oklch(0.65 0.2 300) 6%, transparent);
}
50% {
border-color: color-mix(in oklab, oklch(0.6 0.05 260) 70%, var(--border));
border-color: color-mix(in oklab, oklch(0.65 0.2 300) 70%, var(--border));
box-shadow:
0 0 20px 4px color-mix(in oklab, oklch(0.6 0.05 260) 20%, transparent),
0 0 50px 8px color-mix(in oklab, oklch(0.6 0.05 260) 10%, transparent);
0 0 20px 4px color-mix(in oklab, oklch(0.65 0.2 300) 20%, transparent),
0 0 50px 8px color-mix(in oklab, oklch(0.65 0.2 300) 10%, transparent);
}
}
@@ -389,16 +389,16 @@
@keyframes chat-input-starting {
0%,
100% {
border-color: color-mix(in oklab, oklch(0.7 0.12 85) 40%, var(--border));
border-color: color-mix(in oklab, var(--primary) 40%, var(--border));
box-shadow:
0 0 4px 1px color-mix(in oklab, oklch(0.7 0.12 85) 10%, transparent),
0 0 10px 2px color-mix(in oklab, oklch(0.7 0.12 85) 5%, transparent);
0 0 4px 1px color-mix(in oklab, var(--primary) 10%, transparent),
0 0 10px 2px color-mix(in oklab, var(--primary) 5%, transparent);
}
50% {
border-color: color-mix(in oklab, oklch(0.7 0.12 85) 60%, var(--border));
border-color: color-mix(in oklab, var(--primary) 60%, var(--border));
box-shadow:
0 0 12px 3px color-mix(in oklab, oklch(0.7 0.12 85) 15%, transparent),
0 0 30px 6px color-mix(in oklab, oklch(0.7 0.12 85) 8%, transparent);
0 0 12px 3px color-mix(in oklab, var(--primary) 15%, transparent),
0 0 30px 6px color-mix(in oklab, var(--primary) 8%, transparent);
}
}
@@ -1,6 +1,7 @@
"use client";
import { useEffect, useState } from "react";
import ReactMarkdown from "react-markdown";
import { IconLoader2 } from "@tabler/icons-react";
import { Badge } from "@kandev/ui/badge";
import { Card, CardContent, CardHeader, CardTitle, CardDescription } from "@kandev/ui/card";
@@ -372,13 +373,20 @@ function EnvOnlySection({
secrets: SecretListItem[];
}) {
return (
<InlineSecretSelect
secretId={secretId}
onSecretIdChange={onSecretIdChange}
secrets={secrets}
label={envMethod.env_var}
placeholder="Select or create a secret..."
/>
<>
{envMethod.setup_hint && (
<div className="markdown-body text-xs text-muted-foreground [&_p]:m-0">
<ReactMarkdown>{envMethod.setup_hint}</ReactMarkdown>
</div>
)}
<InlineSecretSelect
secretId={secretId}
onSecretIdChange={onSecretIdChange}
secrets={secrets}
label={envMethod.env_var}
placeholder="Select or create a secret..."
/>
</>
);
}
@@ -389,7 +397,9 @@ function GhTokenOption({ method, isSelected }: { method: RemoteAuthMethod; isSel
<div className="flex flex-col gap-0.5">
<span className="text-sm font-medium">{method.label ?? "Copy token from local CLI"}</span>
{method.setup_hint && (
<span className="text-xs text-muted-foreground">{method.setup_hint}</span>
<div className="markdown-body text-xs text-muted-foreground [&_p]:m-0">
<ReactMarkdown>{method.setup_hint}</ReactMarkdown>
</div>
)}
</div>
</label>
@@ -447,6 +457,11 @@ function EnvOption({
Set <code className="text-[11px] bg-muted px-1 rounded">{method.env_var}</code> via a
stored secret
</span>
{method.setup_hint && (
<div className="markdown-body text-xs text-muted-foreground [&_p]:m-0">
<ReactMarkdown>{method.setup_hint}</ReactMarkdown>
</div>
)}
</div>
</label>
{isSelected && (
@@ -291,17 +291,28 @@ function InstancesContent({
}
function HealthBadge({ status }: { status: string }) {
const label = status.charAt(0).toUpperCase() + status.slice(1);
switch (status) {
case "healthy":
case "running":
return (
<Badge variant="default" className="bg-green-600">
Healthy
{label}
</Badge>
);
case "unhealthy":
return <Badge variant="destructive">Unhealthy</Badge>;
case "cold":
return (
<Badge variant="secondary" className="bg-blue-600 text-white">
{label}
</Badge>
);
case "starting":
return <Badge variant="secondary">{label}</Badge>;
case "stopped":
case "stopping":
case "failed":
return <Badge variant="destructive">{label}</Badge>;
default:
return <Badge variant="secondary">Unknown</Badge>;
return <Badge variant="secondary">{label}</Badge>;
}
}
@@ -315,7 +315,7 @@ export function ChatInputArea({
);
return (
<div className="bg-card flex-shrink-0 px-2 pb-2 pt-1">
{agentMode && (
{agentMode && agentMode !== "default" && (
<div className="flex items-center gap-1.5 px-3 py-1 text-xs text-muted-foreground">
<IconBrain className="h-3 w-3" />
<span className="capitalize">{agentMode} mode</span>
@@ -217,7 +217,7 @@ export function ChatInputBody({
isStarting && !isAgentBusy && "chat-input-starting",
isAgentBusy && !planModeEnabled && "chat-input-running",
isAgentBusy && planModeEnabled && "chat-input-running-plan",
planModeEnabled && !isAgentBusy && "border-slate-400/50",
planModeEnabled && !isAgentBusy && "border-violet-400/50",
hasClarification && "border-sky-400/50",
showRequestChangesTooltip && "animate-pulse border-orange-500",
hasPendingComments && "border-amber-500/50",
@@ -94,7 +94,7 @@ function SubmitButton({
size="icon"
className={cn(
"h-7 w-7 rounded-full cursor-pointer",
planModeEnabled && "bg-slate-600 hover:bg-slate-500",
planModeEnabled && "bg-violet-600 hover:bg-violet-500",
)}
disabled={isDisabled}
onClick={onSubmit}
@@ -130,7 +130,7 @@ function PlanToggleButton({
size="sm"
className={cn(
"h-7 gap-1.5 px-2 hover:bg-muted/40 cursor-pointer",
planModeEnabled && planModeAvailable && "bg-slate-500/15 text-slate-400",
planModeEnabled && planModeAvailable && "bg-violet-500/15 text-violet-400",
)}
onClick={() => onPlanModeChange(!planModeEnabled)}
>
@@ -150,7 +150,7 @@ function ImplementPlanButton({ onClick }: { onClick: () => void }) {
type="button"
variant="ghost"
size="sm"
className="h-7 gap-1.5 px-2 cursor-pointer hover:bg-muted/40 text-slate-400"
className="h-7 gap-1.5 px-2 cursor-pointer hover:bg-muted/40 text-violet-400"
onClick={onClick}
>
<IconRocket className="h-4 w-4" />
@@ -11,10 +11,10 @@ type ResizeHandleProps = {
};
function getHandleColor(planModeEnabled?: boolean, isAgentBusy?: boolean, isStarting?: boolean) {
if (isStarting) return "bg-amber-400/60 hover:bg-amber-400";
if (isAgentBusy && planModeEnabled) return "bg-slate-400/60 hover:bg-slate-400";
if (isStarting) return "bg-primary/40 hover:bg-primary/70";
if (isAgentBusy && planModeEnabled) return "bg-violet-400/60 hover:bg-violet-400";
if (isAgentBusy) return "bg-primary/40 hover:bg-primary/70";
if (planModeEnabled) return "bg-slate-400/60 hover:bg-slate-400";
if (planModeEnabled) return "bg-violet-400/60 hover:bg-violet-400";
return "bg-border hover:bg-muted-foreground";
}
@@ -227,6 +227,7 @@ export const SessionMobileLayout = memo(function SessionMobileLayout({
style={{ paddingTop: "env(safe-area-inset-top, 0px)" }}
>
<SessionMobileTopBar
taskId={activeTaskId}
taskTitle={taskTitle}
sessionId={effectiveSessionId}
baseBranch={baseBranch}
@@ -2,16 +2,9 @@
import { memo, useState } from "react";
import Link from "next/link";
import {
IconArrowLeft,
IconMenu2,
IconGitBranch,
IconCheck,
IconCloud,
IconCloudOff,
} from "@tabler/icons-react";
import { IconArrowLeft, IconMenu2, IconGitBranch, IconCheck } from "@tabler/icons-react";
import { Button } from "@kandev/ui/button";
import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip";
import { RemoteCloudTooltip } from "@/components/task/remote-cloud-tooltip";
import { LineStat } from "@/components/diff-stat";
import { useSessionGitStatus } from "@/hooks/domains/session/use-session-git-status";
import { useSessionCommits } from "@/hooks/domains/session/use-session-commits";
@@ -24,6 +17,7 @@ import {
} from "./session-mobile-top-bar-git-controls";
type SessionMobileTopBarProps = {
taskId?: string | null;
taskTitle?: string;
sessionId?: string | null;
baseBranch?: string;
@@ -67,49 +61,8 @@ function MobileTaskTitle({
);
}
function RemoteExecutorCloudStatus({
isRemoteExecutor,
remoteExecutorName,
remoteExecutorType,
remoteState,
remoteCreatedAt,
remoteCheckedAt,
remoteStatusError,
}: {
isRemoteExecutor?: boolean;
remoteExecutorType?: string | null;
remoteExecutorName?: string | null;
remoteState?: string | null;
remoteCreatedAt?: string | null;
remoteCheckedAt?: string | null;
remoteStatusError?: string | null;
}) {
if (!isRemoteExecutor) return null;
return (
<Tooltip>
<TooltipTrigger asChild>
<span className="inline-flex items-center px-1 cursor-default">
{remoteStatusError ? (
<IconCloudOff className="h-4 w-4 text-destructive" />
) : (
<IconCloud className="h-4 w-4 text-muted-foreground" />
)}
</span>
</TooltipTrigger>
<TooltipContent side="bottom" className="space-y-0.5">
<div className="font-medium">{remoteExecutorName ?? remoteExecutorType ?? "Remote"}</div>
{remoteState && <div>State: {remoteState}</div>}
{remoteCreatedAt && <div>Created: {new Date(remoteCreatedAt).toLocaleString()}</div>}
{remoteCheckedAt && <div>Last check: {new Date(remoteCheckedAt).toLocaleString()}</div>}
{remoteStatusError && (
<div className="text-destructive">Status failed: {remoteStatusError}</div>
)}
</TooltipContent>
</Tooltip>
);
}
type MobileTopBarActionsProps = {
taskId?: string | null;
isRemoteExecutor?: boolean;
remoteExecutorType?: string | null;
remoteExecutorName?: string | null;
@@ -133,6 +86,7 @@ type MobileTopBarActionsProps = {
};
function MobileTopBarActions({
taskId,
isRemoteExecutor,
remoteExecutorType,
remoteExecutorName,
@@ -156,15 +110,21 @@ function MobileTopBarActions({
}: MobileTopBarActionsProps) {
return (
<div className="flex items-center gap-1">
<RemoteExecutorCloudStatus
isRemoteExecutor={isRemoteExecutor}
remoteExecutorType={remoteExecutorType}
remoteExecutorName={remoteExecutorName}
remoteState={remoteState}
remoteCreatedAt={remoteCreatedAt}
remoteCheckedAt={remoteCheckedAt}
remoteStatusError={remoteStatusError}
/>
{isRemoteExecutor && (
<RemoteCloudTooltip
taskId={taskId ?? ""}
sessionId={sessionId}
fallbackName={remoteExecutorName ?? remoteExecutorType}
iconClassName="h-4 w-4"
status={{
remote_name: remoteExecutorName ?? undefined,
remote_state: remoteState ?? undefined,
remote_created_at: remoteCreatedAt ?? undefined,
remote_checked_at: remoteCheckedAt ?? undefined,
remote_status_error: remoteStatusError ?? undefined,
}}
/>
)}
{showApproveButton && onApprove && (
<Button
size="sm"
@@ -195,6 +155,7 @@ function MobileTopBarActions({
}
export const SessionMobileTopBar = memo(function SessionMobileTopBar({
taskId,
taskTitle,
sessionId,
baseBranch,
@@ -250,6 +211,7 @@ export const SessionMobileTopBar = memo(function SessionMobileTopBar({
/>
</div>
<MobileTopBarActions
taskId={taskId}
isRemoteExecutor={isRemoteExecutor}
remoteExecutorType={remoteExecutorType}
remoteExecutorName={remoteExecutorName}
@@ -5,7 +5,7 @@ import { IconCloud, IconCloudOff } from "@tabler/icons-react";
import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip";
import { getWebSocketClient } from "@/lib/ws/connection";
type SessionStatusResponse = {
type RemoteStatusData = {
remote_name?: string;
remote_state?: string;
remote_created_at?: string;
@@ -18,8 +18,20 @@ type RemoteCloudTooltipProps = {
sessionId?: string | null;
fallbackName?: string | null;
iconClassName?: string;
/** When provided, uses this data directly instead of fetching via WS on hover. */
status?: RemoteStatusData | null;
};
const CONNECTED_THRESHOLD_MS = 2 * 60 * 1000;
function getCloudState(status: RemoteStatusData | null): "connected" | "error" | "stale" {
if (status?.remote_status_error) return "error";
if (!status?.remote_checked_at) return "stale";
const elapsed = Date.now() - new Date(status.remote_checked_at).getTime();
if (elapsed < CONNECTED_THRESHOLD_MS) return "connected";
return "stale";
}
function formatTimestamp(value?: string): string | null {
if (!value) return null;
const d = new Date(value);
@@ -56,55 +68,62 @@ function RemoteCloudStatusContent({
);
}
const CLOUD_STATE_CLASSES: Record<ReturnType<typeof getCloudState>, string> = {
error: "text-destructive",
connected: "text-emerald-500",
stale: "text-muted-foreground",
};
export function RemoteCloudTooltip({
taskId,
sessionId,
fallbackName,
iconClassName = "h-3.5 w-3.5",
status: externalStatus,
}: RemoteCloudTooltipProps) {
const hasExternalStatus = externalStatus !== undefined;
const [open, setOpen] = useState(false);
const [status, setStatus] = useState<SessionStatusResponse | null>(null);
const [fetchedStatus, setFetchedStatus] = useState<RemoteStatusData | null>(null);
const [fetchedSessionId, setFetchedSessionId] = useState<string | null>(null);
useEffect(() => {
if (hasExternalStatus) return;
if (!open || !sessionId || fetchedSessionId === sessionId) return;
const client = getWebSocketClient();
if (!client) return;
client
.request<SessionStatusResponse>(
.request<RemoteStatusData>(
"task.session.status",
{ task_id: taskId, session_id: sessionId },
10000,
)
.then((res) => setStatus(res))
.then((res) => setFetchedStatus(res))
.catch(() => {})
.finally(() => setFetchedSessionId(sessionId));
}, [open, fetchedSessionId, sessionId, taskId]);
}, [hasExternalStatus, open, fetchedSessionId, sessionId, taskId]);
const status = hasExternalStatus ? (externalStatus ?? null) : fetchedStatus;
const remoteName = status?.remote_name ?? fallbackName ?? "Remote executor";
const hasError = Boolean(status?.remote_status_error);
const loading = Boolean(open && sessionId && fetchedSessionId !== sessionId);
const Icon = hasError ? IconCloudOff : IconCloud;
const iconClass = hasError
? `${iconClassName} text-destructive`
: `${iconClassName} text-muted-foreground`;
const createdAt = formatTimestamp(status?.remote_created_at);
const checkedAt = formatTimestamp(status?.remote_checked_at);
const cloudState = getCloudState(status);
const loading = Boolean(
!hasExternalStatus && open && sessionId && fetchedSessionId !== sessionId,
);
const Icon = cloudState === "error" ? IconCloudOff : IconCloud;
return (
<Tooltip onOpenChange={setOpen}>
<TooltipTrigger asChild>
<span className="cursor-default">
<Icon className={iconClass} />
<Icon className={`${iconClassName} ${CLOUD_STATE_CLASSES[cloudState]}`} />
</span>
</TooltipTrigger>
<RemoteCloudStatusContent
remoteName={remoteName}
remoteState={status?.remote_state}
createdAt={createdAt}
checkedAt={checkedAt}
createdAt={formatTimestamp(status?.remote_created_at)}
checkedAt={formatTimestamp(status?.remote_checked_at)}
remoteStatusError={status?.remote_status_error}
loading={loading}
/>
+22 -53
View File
@@ -7,8 +7,6 @@ import {
IconCopy,
IconGitBranch,
IconCheck,
IconCloud,
IconCloudOff,
IconHome,
IconSettings,
} from "@tabler/icons-react";
@@ -32,6 +30,7 @@ import { DocumentControls } from "@/components/task/document/document-controls";
import { VcsSplitButton } from "@/components/vcs-split-button";
import { PRTopbarButton } from "@/components/github/pr-topbar-button";
import { WorkflowStepper, type WorkflowStepperStep } from "@/components/task/workflow-stepper";
import { RemoteCloudTooltip } from "@/components/task/remote-cloud-tooltip";
import { DEBUG_UI } from "@/lib/config";
type TaskTopBarProps = {
@@ -91,6 +90,8 @@ const TaskTopBar = memo(function TaskTopBar({
return (
<header className="@container/topbar grid grid-cols-[1fr_auto_1fr] items-center px-3 py-1 border-b border-border">
<TopBarLeft
taskId={taskId}
activeSessionId={activeSessionId}
taskTitle={taskTitle}
repositoryName={repositoryName}
displayBranch={displayBranch}
@@ -224,50 +225,10 @@ function BranchPathPopover({
);
}
function RemoteExecutorIndicator({
isRemoteExecutor,
remoteExecutorName,
remoteExecutorType,
remoteState,
remoteCreatedAt,
remoteCheckedAt,
remoteStatusError,
}: {
isRemoteExecutor?: boolean;
remoteExecutorName?: string | null;
remoteExecutorType?: string | null;
remoteState?: string | null;
remoteCreatedAt?: string | null;
remoteCheckedAt?: string | null;
remoteStatusError?: string | null;
}) {
if (!isRemoteExecutor) return null;
return (
<Tooltip>
<TooltipTrigger asChild>
<span className="inline-flex items-center">
{remoteStatusError ? (
<IconCloudOff className="h-4 w-4 text-destructive" />
) : (
<IconCloud className="h-4 w-4 text-muted-foreground" />
)}
</span>
</TooltipTrigger>
<TooltipContent side="bottom" className="space-y-0.5">
<div className="font-medium">{remoteExecutorName ?? remoteExecutorType ?? "Remote"}</div>
{remoteState && <div>State: {remoteState}</div>}
{remoteCreatedAt && <div>Created: {new Date(remoteCreatedAt).toLocaleString()}</div>}
{remoteCheckedAt && <div>Last check: {new Date(remoteCheckedAt).toLocaleString()}</div>}
{remoteStatusError && (
<div className="text-destructive">Status failed: {remoteStatusError}</div>
)}
</TooltipContent>
</Tooltip>
);
}
/** Left section: breadcrumbs, branch pill */
function TopBarLeft({
taskId,
activeSessionId,
taskTitle,
repositoryName,
displayBranch,
@@ -281,6 +242,8 @@ function TopBarLeft({
remoteCheckedAt,
remoteStatusError,
}: {
taskId?: string | null;
activeSessionId?: string | null;
taskTitle?: string;
repositoryName?: string | null;
displayBranch?: string;
@@ -343,15 +306,21 @@ function TopBarLeft({
/>
</div>
<RemoteExecutorIndicator
isRemoteExecutor={isRemoteExecutor}
remoteExecutorName={remoteExecutorName}
remoteExecutorType={remoteExecutorType}
remoteState={remoteState}
remoteCreatedAt={remoteCreatedAt}
remoteCheckedAt={remoteCheckedAt}
remoteStatusError={remoteStatusError}
/>
{isRemoteExecutor && (
<RemoteCloudTooltip
taskId={taskId ?? ""}
sessionId={activeSessionId}
fallbackName={remoteExecutorName ?? remoteExecutorType}
iconClassName="h-4 w-4"
status={{
remote_name: remoteExecutorName ?? undefined,
remote_state: remoteState ?? undefined,
remote_created_at: remoteCreatedAt ?? undefined,
remote_checked_at: remoteCheckedAt ?? undefined,
remote_status_error: remoteStatusError ?? undefined,
}}
/>
)}
</div>
);
}
+3 -3
View File
@@ -305,7 +305,7 @@ The session is NOT marked as failed -- the user can send a new message to start
| Agent | NativeSessionResume | HistoryContextInjection | Notes |
|-------|:---:|:---:|-------|
| Claude Code | - | - | Uses `--resume` CLI flag for its own session restore |
| Claude Code | - | - | Uses `--resume` CLI flag with stored resume token |
| Codex | yes | - | ACP `session/load` restores context |
| Copilot | yes | - | ACP `session/load` restores context |
| Auggie | - | yes | Fresh start + history injection on first prompt |
@@ -325,7 +325,7 @@ Persists runtime state for a session across backend restarts:
| Field | Purpose |
|-------|---------|
| `ResumeToken` | ACP session ID for native resume (`session/load`) |
| `ResumeToken` | ACP session ID (used for `session/load` or `--resume` CLI flag) |
| `LastMessageUUID` | For `--resume-session-at` flag |
| `ContainerID` | Docker container ID (for recovery) |
| `AgentctlURL` / `AgentctlPort` | agentctl connection info |
@@ -338,7 +338,7 @@ The resume token is stored/updated in `storeResumeToken()` (`internal/orchestrat
- `handleSessionStatusEvent` -- on session status updates
- `handleCompleteStreamEvent` -- on stream completion
If the agent's profile does NOT have `NativeSessionResume`, the token is cleared to prevent stale resume attempts.
The token is always stored regardless of `NativeSessionResume`. Agents with native resume use it for ACP `session/load`; others (e.g., Claude Code) use it for their `--resume` CLI flag.
---