kandev/apps/backend/internal/agent/runtime/lifecycle/manager_launch.go

package lifecycle

import (
	"context"
	"errors"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
	"go.uber.org/zap"

	"github.com/kandev/kandev/internal/agent/agents"
	"github.com/kandev/kandev/internal/agent/executor"
	"github.com/kandev/kandev/internal/agent/settings/cliflags"
	"github.com/kandev/kandev/internal/events"
	"github.com/kandev/kandev/internal/task/models"
)

// resolveAgentProfile resolves the agent profile and returns the agent type name and profile info.
func (m *Manager) resolveAgentProfile(ctx context.Context, req *LaunchRequest) (string, *AgentProfileInfo, error) {
	if m.profileResolver == nil {
		// Fallback: treat AgentProfileID as agent type directly (for backward compat)
		m.logger.Warn("no profile resolver configured, using profile ID as agent type",
			zap.String("agent_type", req.AgentProfileID))
		return req.AgentProfileID, nil, nil
	}
	profileInfo, err := m.profileResolver.ResolveProfile(ctx, req.AgentProfileID)
	if err != nil {
		return "", nil, fmt.Errorf("failed to resolve agent profile: %w", err)
	}
	applyRouteOverrideToProfile(profileInfo, req)
	m.logger.Debug("resolved agent profile",
		zap.String("profile_id", req.AgentProfileID),
		zap.String("agent_name", profileInfo.AgentName),
		zap.String("agent_type", profileInfo.AgentName))
	return profileInfo.AgentName, profileInfo, nil
}

// appendRouteOverrideFlags returns tokens with the route override's
// extra CLI flags appended (when set). Order: user-configured flags
// from the base profile first, route-override flags after.
func appendRouteOverrideFlags(tokens []string, req *LaunchRequest) []string {
	if req == nil || req.RouteOverride == nil || len(req.RouteOverride.Flags) == 0 {
		return tokens
	}
	out := make([]string, 0, len(tokens)+len(req.RouteOverride.Flags))
	out = append(out, tokens...)
	out = append(out, req.RouteOverride.Flags...)
	return out
}

// applyRouteOverrideToProfile mutates profileInfo in-place when the
// request carries a RouteOverride. Empty fields on the override are
// preserved on the profile — the override only replaces explicit values.
// Mode override is applied unconditionally because routing's mode
// belongs to the provider, not the base profile.
func applyRouteOverrideToProfile(profile *AgentProfileInfo, req *LaunchRequest) {
	if profile == nil || req == nil || req.RouteOverride == nil {
		return
	}
	ov := req.RouteOverride
	if ov.ProviderID != "" {
		profile.AgentName = ov.ProviderID
	}
	if ov.Model != "" {
		profile.Model = ov.Model
	}
	profile.Mode = ov.Mode
}

// buildLaunchMetadata builds runtime metadata for the Launch request.
func buildLaunchMetadata(req *LaunchRequest, mainRepoGitDir, worktreeID, worktreeBranch string) map[string]interface{} {
	metadata := make(map[string]interface{})
	for k, v := range req.Metadata {
		metadata[k] = v
	}
	if mainRepoGitDir != "" {
		metadata[MetadataKeyMainRepoGitDir] = mainRepoGitDir
	}
	if worktreeID != "" {
		metadata[MetadataKeyWorktreeID] = worktreeID
	}
	if worktreeBranch != "" {
		metadata[MetadataKeyWorktreeBranch] = worktreeBranch
	}
	// Pass repo info for remote executors (Sprites, remote docker, etc.)
	if req.RepositoryPath != "" {
		metadata[MetadataKeyRepositoryPath] = req.RepositoryPath
	}
	if req.SetupScript != "" {
		metadata[MetadataKeySetupScript] = req.SetupScript
	}
	if req.BaseBranch != "" {
		metadata[MetadataKeyBaseBranch] = req.BaseBranch
	}
	return metadata
}

// agentCommands holds the initial and continue command strings for an agent execution.
type agentCommands struct {
	initial   string
	continue_ string // continue command for one-shot agents (empty if not applicable)
}

// buildAgentCommand builds the agent command strings for the execution.
// Returns both the initial command and the continue command (for one-shot agents like Amp).
func (m *Manager) buildAgentCommand(req *LaunchRequest, profileInfo *AgentProfileInfo, agentConfig agents.Agent) agentCommands {
	model := ""
	autoApprove := false
	permissionValues := make(map[string]bool)
	var cliFlagTokens []string
	if profileInfo != nil {
		model = profileInfo.Model
		autoApprove = profileInfo.AutoApprove
		permissionValues["auto_approve"] = profileInfo.AutoApprove
		permissionValues["allow_indexing"] = profileInfo.AllowIndexing
		permissionValues["dangerously_skip_permissions"] = profileInfo.DangerouslySkipPermissions
		tokens, err := cliflags.Resolve(profileInfo.CLIFlags)
		if err != nil {
			m.logger.Warn("failed to resolve cli_flags for profile, launching without user-configured flags",
				zap.String("profile_id", profileInfo.ProfileID),
				zap.Error(err))
		} else {
			cliFlagTokens = tokens
		}
	}
	// Allow model override from request (for dynamic model switching)
	if req.ModelOverride != "" {
		model = req.ModelOverride
	}
	cliFlagTokens = appendRouteOverrideFlags(cliFlagTokens, req)
	// 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:        sessionID,
		AutoApprove:      autoApprove,
		PermissionValues: permissionValues,
		CLIFlagTokens:    cliFlagTokens,
		Runtime:          models.ExecutorType(req.ExecutorType).Runtime(),
	}
	return agentCommands{
		initial:   m.commandBuilder.BuildCommandString(agentConfig, cmdOpts),
		continue_: m.commandBuilder.BuildContinueCommandString(agentConfig, cmdOpts),
	}
}

// launchResolveWorkspacePath resolves the effective workspace path for non-worktree executors.
// For worktree executors, workspace resolution is handled by the WorktreePreparer.
// For tasks without repositories, creates a workspace directory in ~/.kandev/quick-chat/.
// Returns workspacePath, mainRepoGitDir, worktreeID, worktreeBranch.
// resolveResumeWorktreePath resolves workspace path for worktree resume using the provider.
func (m *Manager) resolveResumeWorktreePath(ctx context.Context, req *LaunchRequest) (string, string, string, string) {
	ws := m.resolveWorkspaceFromProvider(ctx, req)
	if ws == "" {
		m.logger.Warn("could not resolve workspace path for worktree resume",
			zap.String("session_id", req.SessionID))
		return "", "", "", ""
	}
	var mainRepoGitDir string
	if req.RepositoryPath != "" {
		mainRepoGitDir = filepath.Join(req.RepositoryPath, ".git")
	}
	return ws, mainRepoGitDir, req.WorktreeID, ""
}

// resolveWorkspaceFromProvider looks up the workspace path from the info provider.
func (m *Manager) resolveWorkspaceFromProvider(ctx context.Context, req *LaunchRequest) string {
	if m.workspaceInfoProvider == nil {
		return ""
	}
	info, err := m.workspaceInfoProvider.GetWorkspaceInfoForSession(ctx, req.TaskID, req.SessionID)
	if err != nil || info.WorkspacePath == "" {
		return ""
	}
	m.logger.Debug("resolved workspace from provider for resume",
		zap.String("session_id", req.SessionID),
		zap.String("path", info.WorkspacePath))
	return info.WorkspacePath
}

func (m *Manager) launchResolveWorkspacePath(ctx context.Context, req *LaunchRequest) (workspacePath, mainRepoGitDir, worktreeID, worktreeBranch string) {
	// Worktree mode requires a repository. Repo-less tasks fall through to the
	// scratch workspace path below — even if the executor type was worktree.
	useWorktree := req.UseWorktree && req.RepositoryPath != ""
	if useWorktree && req.ACPSessionID == "" {
		return "", "", "", ""
	}
	if useWorktree && req.ACPSessionID != "" {
		return m.resolveResumeWorktreePath(ctx, req)
	}
	workspacePath = req.WorkspacePath
	if req.RepositoryPath != "" && workspacePath == "" {
		workspacePath = req.RepositoryPath
	}
	if workspacePath == "" && req.ACPSessionID != "" {
		if resolved := m.resolveWorkspaceFromProvider(ctx, req); resolved != "" {
			return resolved, "", "", ""
		}
	}
	// For tasks without a repository, create a scratch workspace.
	// - Non-ephemeral repo-less tasks: <homeDir>/tasks/<workspaceID>/<taskID>/
	//   (task-scoped, persists across sessions, mirrors the worktree task layout).
	// - Ephemeral tasks (slack triage / quick chat): <dataDir>/quick-chat/<sessionID>/
	//   (session-scoped, cleaned up on task delete via performTaskCleanup).
	// Office tasks that have no repo (onboarding, planning) take the
	// non-ephemeral branch and land under <homeDir>/tasks/...
	if workspacePath == "" && req.SessionID != "" && m.dataDir != "" {
		workspacePath = m.resolveScratchWorkspace(ctx, req)
	}
	return
}

// resolveScratchWorkspace creates and returns the scratch workspace path for a
// repo-less task. Returns empty string when the path could not be created.
func (m *Manager) resolveScratchWorkspace(ctx context.Context, req *LaunchRequest) string {
	scratchPath := m.scratchWorkspacePath(req)
	if scratchPath == "" {
		return ""
	}
	if err := os.MkdirAll(scratchPath, 0755); err != nil {
		m.logger.Warn("failed to create scratch workspace, continuing without workspace",
			zap.String("session_id", req.SessionID),
			zap.String("workspace_path", scratchPath),
			zap.Error(err))
		return ""
	}
	if err := m.initGitRepo(ctx, scratchPath); err != nil {
		m.logger.Warn("failed to initialize git repository in scratch workspace",
			zap.String("session_id", req.SessionID),
			zap.String("workspace_path", scratchPath),
			zap.Error(err))
		// Continue anyway - git is optional for repo-less workspaces.
	}
	m.logger.Info("created scratch workspace",
		zap.String("session_id", req.SessionID),
		zap.String("task_id", req.TaskID),
		zap.String("workspace_path", scratchPath))
	return scratchPath
}

// scratchWorkspacePath computes the scratch workspace path for a launch request.
// Returns empty string if the inputs are invalid (path traversal guard, missing IDs).
func (m *Manager) scratchWorkspacePath(req *LaunchRequest) string {
	if req.IsEphemeral {
		// Legacy quick-chat path — session-scoped, kept for backward compat with
		// slack triage and other ephemeral one-shot flows.
		if strings.ContainsAny(req.SessionID, `/\`) {
			m.logger.Warn("session ID contains path separator, rejecting",
				zap.String("session_id", req.SessionID))
			return ""
		}
		return filepath.Join(m.dataDir, "quick-chat", req.SessionID)
	}
	// Non-ephemeral repo-less task: place under <homeDir>/tasks/<workspaceID>/<taskID>/
	// so it sits alongside repo-bound worktrees and persists across sessions.
	if req.TaskID == "" || req.WorkspaceID == "" {
		m.logger.Warn("scratch workspace requires task_id and workspace_id",
			zap.String("session_id", req.SessionID),
			zap.String("task_id", req.TaskID),
			zap.String("workspace_id", req.WorkspaceID))
		return ""
	}
	if strings.ContainsAny(req.TaskID, `/\`) || strings.ContainsAny(req.WorkspaceID, `/\`) {
		m.logger.Warn("task or workspace ID contains path separator, rejecting",
			zap.String("task_id", req.TaskID),
			zap.String("workspace_id", req.WorkspaceID))
		return ""
	}
	// m.dataDir is misnamed — cmd/kandev/agents.go passes cfg.ResolvedHomeDir()
	// (the kandev root, e.g. ~/.kandev), not ResolvedDataDir(). So scratch
	// workspaces live alongside the existing repo-bound worktree task dirs
	// at <kandevHome>/tasks/<workspaceID>/<taskID>/.
	return filepath.Join(m.dataDir, "tasks", req.WorkspaceID, req.TaskID)
}

// launchPrepareRequest copies the launch request, sets the resolved workspace path,
// populates metadata from the request fields, and injects profile environment variables.
func (m *Manager) launchPrepareRequest(req *LaunchRequest, profileInfo *AgentProfileInfo, workspacePath string) (LaunchRequest, string) {
	executionID := uuid.New().String()
	reqWithWorktree := *req
	reqWithWorktree.WorkspacePath = workspacePath

	if reqWithWorktree.Metadata == nil {
		reqWithWorktree.Metadata = make(map[string]interface{})
	}
	if req.TaskDescription != "" {
		reqWithWorktree.Metadata["task_description"] = req.TaskDescription
	}
	if len(req.Attachments) > 0 {
		reqWithWorktree.Metadata["attachments"] = req.Attachments
	}
	if req.SessionID != "" {
		reqWithWorktree.Metadata["session_id"] = req.SessionID
	}

	if profileInfo != nil {
		if reqWithWorktree.Env == nil {
			reqWithWorktree.Env = make(map[string]string)
		}
		if profileInfo.Model != "" {
			reqWithWorktree.Env["AGENT_MODEL"] = profileInfo.Model
		}
		if profileInfo.AutoApprove {
			reqWithWorktree.Env["AGENTCTL_AUTO_APPROVE_PERMISSIONS"] = "true"
		}
	}
	mergeRouteOverrideEnv(&reqWithWorktree)
	return reqWithWorktree, executionID
}

// mergeRouteOverrideEnv copies the routing override's Env map into the
// launch request's Env map (last-write-wins, route overrides base env).
func mergeRouteOverrideEnv(req *LaunchRequest) {
	if req == nil || req.RouteOverride == nil || len(req.RouteOverride.Env) == 0 {
		return
	}
	if req.Env == nil {
		req.Env = make(map[string]string, len(req.RouteOverride.Env))
	}
	for k, v := range req.RouteOverride.Env {
		req.Env[k] = v
	}
}

// newProgressCallback builds a PrepareProgressCallback that publishes progress events for a task/session.
func (m *Manager) newProgressCallback(taskID, sessionID string) PrepareProgressCallback {
	return func(step PrepareStep, stepIndex int, totalSteps int) {
		m.eventPublisher.PublishPrepareProgress(sessionID, &PrepareProgressEventPayload{
			TaskID:        taskID,
			SessionID:     sessionID,
			StepName:      step.Name,
			StepCommand:   step.Command,
			StepIndex:     stepIndex,
			TotalSteps:    totalSteps,
			Status:        string(step.Status),
			Output:        step.Output,
			Error:         step.Error,
			Warning:       step.Warning,
			WarningDetail: step.WarningDetail,
			StartedAt:     step.StartedAt,
			EndedAt:       step.EndedAt,
		})
	}
}

type prepareProgressRecorder struct {
	mu       sync.Mutex
	steps    []PrepareStep
	callback PrepareProgressCallback
}

func newPrepareProgressRecorder(callback PrepareProgressCallback) *prepareProgressRecorder {
	return &prepareProgressRecorder{callback: callback}
}

func (r *prepareProgressRecorder) Callback(offset int) PrepareProgressCallback {
	return func(step PrepareStep, stepIndex int, totalSteps int) {
		absoluteIndex := stepIndex + offset
		r.recordStep(step, absoluteIndex)
		if r.callback != nil {
			r.callback(step, absoluteIndex, totalSteps+offset)
		}
	}
}

func (r *prepareProgressRecorder) Merge(steps []PrepareStep) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for i, step := range steps {
		if i >= len(r.steps) {
			r.steps = append(r.steps, step)
			continue
		}
		if r.steps[i].Name == "" {
			r.steps[i] = step
		}
	}
}

func (r *prepareProgressRecorder) Len() int {
	r.mu.Lock()
	defer r.mu.Unlock()
	return len(r.steps)
}

func (r *prepareProgressRecorder) Steps() []PrepareStep {
	r.mu.Lock()
	defer r.mu.Unlock()
	out := make([]PrepareStep, len(r.steps))
	copy(out, r.steps)
	return out
}

func (r *prepareProgressRecorder) recordStep(step PrepareStep, index int) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for len(r.steps) <= index {
		r.steps = append(r.steps, PrepareStep{})
	}
	r.steps[index] = step
}

// launchBuildExecutorRequest resolves MCP servers, builds the ExecutorCreateRequest,
// and creates the runtime instance.
func (m *Manager) launchBuildExecutorRequest(ctx context.Context, executionID string, reqWithWorktree *LaunchRequest, agentConfig agents.Agent, profileInfo *AgentProfileInfo, mainRepoGitDir, worktreeID, worktreeBranch string, onProgress PrepareProgressCallback) (*ExecutorCreateRequest, *ExecutorInstance, ExecutorBackend, error) {
	rt, err := m.getExecutorBackend(reqWithWorktree.ExecutorType)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("no runtime configured: %w", err)
	}

	env := m.buildEnvForExecution(ctx, executionID, reqWithWorktree, agentConfig, profileInfo)

	acpMcpServers, err := m.resolveMcpServersWithParams(ctx, reqWithWorktree.AgentProfileID, reqWithWorktree.Metadata, agentConfig)
	if err != nil {
		m.logger.Warn("failed to resolve MCP servers for launch", zap.Error(err))
	}

	var mcpServers []McpServerConfig
	for _, srv := range acpMcpServers {
		mcpServers = append(mcpServers, McpServerConfig{
			Name:    srv.Name,
			URL:     srv.URL,
			Type:    srv.Type,
			Command: srv.Command,
			Args:    srv.Args,
			Env:     srv.Env,
			Headers: srv.Headers,
		})
	}

	metadata := buildLaunchMetadata(reqWithWorktree, mainRepoGitDir, worktreeID, worktreeBranch)

	execReq := &ExecutorCreateRequest{
		InstanceID:          executionID,
		TaskID:              reqWithWorktree.TaskID,
		TaskTitle:           reqWithWorktree.TaskTitle,
		SessionID:           reqWithWorktree.SessionID,
		TaskEnvironmentID:   reqWithWorktree.TaskEnvironmentID,
		AgentProfileID:      reqWithWorktree.AgentProfileID,
		WorkspacePath:       reqWithWorktree.WorkspacePath,
		Protocol:            string(agentConfig.Runtime().Protocol),
		Env:                 env,
		Metadata:            metadata,
		AgentConfig:         agentConfig,
		McpServers:          mcpServers,
		PreviousExecutionID: reqWithWorktree.PreviousExecutionID,
		McpMode:             reqWithWorktree.McpMode,
		AuthToken:           m.revealRuntimeSecret(ctx, metadata, MetadataKeyAuthTokenSecret),
		BootstrapNonce:      m.revealRuntimeSecret(ctx, metadata, MetadataKeyBootstrapNonceSecret),
		OnProgress:          onProgress,
	}

	if resumer, ok := rt.(RemoteSessionResumer); ok {
		if err := resumer.ResumeRemoteInstance(ctx, execReq); err != nil {
			return nil, nil, nil, fmt.Errorf("failed remote resume preflight: %w", err)
		}
	}

	execInstance, err := rt.CreateInstance(ctx, execReq)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to create execution: %w", err)
	}
	return execReq, execInstance, rt, nil
}

// runEnvironmentPreparer runs the environment preparer for the executor type, if one is registered.
// Returns the prepare result (nil if no preparer ran). Does NOT publish PrepareCompleted;
// the caller is responsible for publishing based on the returned result.
func (m *Manager) runEnvironmentPreparer(
	ctx context.Context,
	req *LaunchRequest,
	workspacePath string,
) *EnvPrepareResult {
	return m.runEnvironmentPreparerWithProgress(ctx, req, workspacePath, m.newProgressCallback(req.TaskID, req.SessionID))
}

func (m *Manager) runEnvironmentPreparerWithProgress(
	ctx context.Context,
	req *LaunchRequest,
	workspacePath string,
	onProgress PrepareProgressCallback,
) *EnvPrepareResult {
	if m.preparerRegistry == nil {
		return nil
	}
	// Preparer registry is keyed by ExecutorType (the "local"/"worktree"/
	// "local_docker"/... taxonomy), not Runtime — so executor types that
	// share a runtime backend (local + worktree both run on standalone)
	// can still get distinct preparation logic.
	execType := models.ExecutorType(req.ExecutorType)
	preparer := m.preparerRegistry.Get(execType)
	if preparer == nil && execType == "" {
		// Fall back to LocalPreparer only for a genuinely empty ExecutorType
		// — legacy task rows (e.g. PR-watcher-created tasks without an
		// explicit executor) rely on local environment prep, including
		// missing-branch detection. Typed-but-unregistered values like
		// "remote_docker" intentionally return nil so the caller skips prep
		// rather than running local git operations against a remote executor.
		preparer = m.preparerRegistry.Get(models.ExecutorTypeLocal)
	}
	if preparer == nil {
		return nil
	}
	// The EnvPrepareRequest carries the resolved Runtime (executor.Name),
	// which preparer_script.go uses for runtime-level decisions like
	// picking the default prepare template.
	execName := execType.Runtime()

	// Skip environment preparation for repo-less tasks (e.g. config chat).
	// Preparers assume a repository is available; without one the session
	// falls through to the quick-chat workspace path instead.
	if req.RepositoryPath == "" {
		m.logger.Debug("skipping environment preparer — no repository path",
			zap.String("task_id", req.TaskID),
			zap.String("session_id", req.SessionID),
			zap.String("preparer", preparer.Name()))
		return nil
	}

	prepReq := buildEnvPrepareRequest(req, workspacePath, execName)

	result, err := preparer.Prepare(ctx, prepReq, onProgress)
	if err != nil {
		m.logger.Warn("environment preparation failed",
			zap.String("task_id", req.TaskID),
			zap.String("preparer", preparer.Name()),
			zap.Error(err))
		return &EnvPrepareResult{
			Success:      false,
			ErrorMessage: err.Error(),
		}
	}

	return result
}

func buildEnvPrepareRequest(req *LaunchRequest, workspacePath string, execName executor.Name) *EnvPrepareRequest {
	repoSetupScript, _ := req.Metadata[MetadataKeyRepoSetupScript].(string)
	prepReq := &EnvPrepareRequest{
		TaskID:               req.TaskID,
		SessionID:            req.SessionID,
		TaskTitle:            req.TaskTitle,
		ExecutorType:         execName,
		WorkspacePath:        workspacePath,
		RepositoryPath:       req.RepositoryPath,
		RepositoryID:         req.RepositoryID,
		UseWorktree:          req.UseWorktree,
		WorktreeID:           req.WorktreeID,
		SetupScript:          req.SetupScript,
		RepoSetupScript:      repoSetupScript,
		BaseBranch:           req.BaseBranch,
		DefaultBranch:        req.DefaultBranch,
		CheckoutBranch:       req.CheckoutBranch,
		PRNumber:             req.PRNumber,
		WorktreeBranch:       getMetadataString(req.Metadata, MetadataKeyWorktreeBranch),
		WorktreeBranchPrefix: req.WorktreeBranchPrefix,
		PullBeforeWorktree:   req.PullBeforeWorktree,
		TaskDirName:          req.TaskDirName,
		RepoName:             req.RepoName,
		Env:                  req.Env,
	}
	// Multi-repo: forward the repo list when the launch request carries one.
	// Each per-repo entry inherits the request-level RepoSetupScript when its
	// own is empty so single-repo callers continue to work unchanged.
	if len(req.Repositories) > 0 {
		specs := make([]RepoPrepareSpec, 0, len(req.Repositories))
		for _, r := range req.Repositories {
			setup := r.RepoSetupScript
			if setup == "" {
				setup = repoSetupScript
			}
			specs = append(specs, RepoPrepareSpec{
				RepositoryID:         r.RepositoryID,
				RepositoryPath:       r.RepositoryPath,
				RepoName:             r.RepoName,
				BaseBranch:           r.BaseBranch,
				DefaultBranch:        r.DefaultBranch,
				CheckoutBranch:       r.CheckoutBranch,
				PRNumber:             r.PRNumber,
				WorktreeID:           r.WorktreeID,
				WorktreeBranchPrefix: r.WorktreeBranchPrefix,
				PullBeforeWorktree:   r.PullBeforeWorktree,
				RepoSetupScript:      setup,
			})
		}
		prepReq.Repositories = specs
	}
	return prepReq
}

// launchApplyPrepareResult applies workspace metadata from the preparer result.
// Returns an error if the preparer failed.
func (m *Manager) launchApplyPrepareResult(
	req *LaunchRequest,
	result *EnvPrepareResult,
	workspacePath, mainRepoGitDir, worktreeID, worktreeBranch *string,
) error {
	if !result.Success {
		m.eventPublisher.PublishPrepareCompleted(req.SessionID, &PrepareCompletedEventPayload{
			TaskID:       req.TaskID,
			SessionID:    req.SessionID,
			Success:      false,
			ErrorMessage: result.ErrorMessage,
			Steps:        result.Steps,
		})
		return fmt.Errorf("environment preparation failed: %s", result.ErrorMessage)
	}
	if result.WorkspacePath != "" {
		*workspacePath = result.WorkspacePath
	}
	if result.MainRepoGitDir != "" {
		*mainRepoGitDir = result.MainRepoGitDir
	}
	if result.WorktreeID != "" {
		*worktreeID = result.WorktreeID
	}
	if result.WorktreeBranch != "" {
		*worktreeBranch = result.WorktreeBranch
	}
	return nil
}

func (m *Manager) publishLaunchPrepareCompleted(req *LaunchRequest, result *EnvPrepareResult, recorder *prepareProgressRecorder, workspacePath string, success bool, err error) {
	if req.ACPSessionID != "" {
		return
	}

	steps := recorder.Steps()
	if len(steps) == 0 && result != nil {
		steps = result.Steps
	}

	payload := &PrepareCompletedEventPayload{
		TaskID:        req.TaskID,
		SessionID:     req.SessionID,
		Success:       success,
		WorkspacePath: workspacePath,
		Steps:         steps,
	}
	if result != nil {
		payload.DurationMs = result.Duration.Milliseconds()
		if payload.WorkspacePath == "" {
			payload.WorkspacePath = result.WorkspacePath
		}
	}
	if err != nil {
		payload.Success = false
		payload.ErrorMessage = err.Error()
	}
	m.eventPublisher.PublishPrepareCompleted(req.SessionID, payload)
}

// Launch launches a new agent for a task. Concurrent calls for the same
// req.SessionID are collapsed via the same singleflight bucket used by
// EnsureWorkspaceExecutionForSession and GetOrEnsureExecution — this closes
// the check-then-act race that previously could spawn a runtime instance,
// fail at executionStore.Add (race), and then have the orchestrator persist
// the orphan execution_id to disk before rollback completed (the original
// agent-execution-id divergence bug).
//
// If req.SessionID is empty (quick chat / pre-session contexts), no
// deduplication key exists and we fall through to direct execution.
func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecution, error) {
	if req.SessionID == "" {
		return m.launchInternal(ctx, req)
	}
	v, err, _ := m.ensureExecutionGroup.Do(req.SessionID, func() (interface{}, error) {
		return m.launchInternal(ctx, req)
	})
	if err != nil {
		return nil, err
	}
	execution := v.(*AgentExecution)
	// If this Launch call joined a workspace-only ensure peer's singleflight
	// slot (EnsureWorkspaceExecutionForSession / GetOrEnsureExecution), the
	// returned execution has no AgentCommand and the orchestrator's subsequent
	// StartAgentProcess() would fail with "no agent command configured".
	// Promote it in place so the agent subprocess can start against the
	// existing agentctl instance.
	if execution.AgentCommand == "" {
		if err := m.promoteWorkspaceExecution(ctx, execution, req); err != nil {
			return nil, err
		}
	}
	return execution, nil
}

// promoteWorkspaceExecution populates the agent command fields on a
// workspace-only execution so a subsequent StartAgentProcess() can configure
// and start the agent subprocess. Concurrent promoters serialize through a
// dedicated singleflight key so they don't race on the shared AgentExecution
// pointer.
func (m *Manager) promoteWorkspaceExecution(ctx context.Context, execution *AgentExecution, req *LaunchRequest) error {
	_, err, _ := m.ensureExecutionGroup.Do("promote:"+req.SessionID, func() (interface{}, error) {
		// Re-check after acquiring the slot — a peer Launch may have already
		// promoted while we were waiting.
		if execution.AgentCommand != "" {
			return nil, nil
		}
		agentTypeName, profileInfo, err := m.resolveAgentProfile(ctx, req)
		if err != nil {
			return nil, err
		}
		agentConfig, ok := m.registry.Get(agentTypeName)
		if !ok {
			return nil, fmt.Errorf("agent type %q not found in registry", agentTypeName)
		}
		if !agentConfig.Enabled() {
			return nil, fmt.Errorf("agent type %q is disabled", agentTypeName)
		}
		cmds := m.buildAgentCommand(req, profileInfo, agentConfig)
		execution.AgentCommand = cmds.initial
		execution.ContinueCommand = cmds.continue_
		if req.ACPSessionID != "" && execution.ACPSessionID == "" {
			execution.ACPSessionID = req.ACPSessionID
		}
		if req.PreviousExecutionID != "" {
			execution.isResumedSession = true
		}
		execution.IsPassthrough = req.IsPassthrough
		m.logger.Info("promoted workspace-only execution to agent execution",
			zap.String("execution_id", execution.ID),
			zap.String("session_id", req.SessionID),
			zap.String("agent_profile_id", req.AgentProfileID),
			zap.Bool("resume", req.ACPSessionID != ""))
		return nil, nil
	})
	return err
}

// launchInternal is the body of Launch run inside the per-session singleflight
// slot. Callers must not invoke this directly except via Launch.
func (m *Manager) launchInternal(ctx context.Context, req *LaunchRequest) (*AgentExecution, error) {
	m.logger.Debug("launching agent",
		zap.String("task_id", req.TaskID),
		zap.String("agent_profile_id", req.AgentProfileID),
		zap.Bool("use_worktree", req.UseWorktree))

	// 1. Resolve the agent profile to get agent type info
	agentTypeName, profileInfo, err := m.resolveAgentProfile(ctx, req)
	if err != nil {
		return nil, err
	}

	// 2. Get agent config from registry
	agentConfig, ok := m.registry.Get(agentTypeName)
	if !ok {
		return nil, fmt.Errorf("agent type %q not found in registry", agentTypeName)
	}
	if !agentConfig.Enabled() {
		return nil, fmt.Errorf("agent type %q is disabled", agentTypeName)
	}

	// 3. Check if session already has an agent running. A workspace-only
	// execution created by EnsureWorkspaceExecutionForSession /
	// GetOrEnsureExecution has no AgentCommand — return it so the outer Launch
	// can promote it instead of erroring as if a real agent were running.
	if req.SessionID != "" {
		if existingExecution, exists := m.executionStore.GetBySessionID(req.SessionID); exists {
			if existingExecution.AgentCommand == "" {
				return existingExecution, nil
			}
			return nil, fmt.Errorf("session %q already has an agent running (execution: %s)", req.SessionID, existingExecution.ID)
		}
	}

	// 4. Resolve workspace path (non-worktree executors use this directly)
	workspacePath, mainRepoGitDir, worktreeID, worktreeBranch := m.launchResolveWorkspacePath(ctx, req)
	progressRecorder := newPrepareProgressRecorder(m.newProgressCallback(req.TaskID, req.SessionID))

	// 4b. Run environment preparation (if preparer registered for this executor type).
	// Skip on resume (ACPSessionID set) — workspace was already prepared during initial launch.
	var prepResult *EnvPrepareResult
	if req.ACPSessionID == "" {
		prepResult = m.runEnvironmentPreparerWithProgress(ctx, req, workspacePath, progressRecorder.Callback(0))
	} else {
		m.logger.Debug("skipping environment preparation for resumed session",
			zap.String("task_id", req.TaskID),
			zap.String("session_id", req.SessionID))
	}
	if prepResult != nil {
		progressRecorder.Merge(prepResult.Steps)
		if err := m.launchApplyPrepareResult(req, prepResult, &workspacePath, &mainRepoGitDir, &worktreeID, &worktreeBranch); err != nil {
			return nil, err
		}
	}

	// 5 & 6. Prepare the request copy with metadata and profile env
	reqWithWorktree, executionID := m.launchPrepareRequest(req, profileInfo, workspacePath)

	// 6b. Deploy per-profile skills + custom prompt (ADR 0005 Wave A).
	// Best-effort: a deploy failure is logged but does not abort the launch
	// — the agent can still start with whatever skills were already on disk.
	m.runSkillDeploy(ctx, req, &reqWithWorktree)

	// 7. Build runtime request and create instance (agent not started yet)
	var runtimeProgress PrepareProgressCallback
	if req.ACPSessionID == "" {
		runtimeProgress = progressRecorder.Callback(progressRecorder.Len())
	}
	execReq, execInstance, rt, err := m.launchBuildExecutorRequest(ctx, executionID, &reqWithWorktree, agentConfig, profileInfo, mainRepoGitDir, worktreeID, worktreeBranch, runtimeProgress)
	if err != nil {
		m.publishLaunchPrepareCompleted(req, prepResult, progressRecorder, workspacePath, false, err)
		return nil, err
	}

	// Remote executors (Docker, Sprites) clone the workspace inside the
	// container, so the worktree path's host-side copy_files never ran.
	// Ship the bytes through agentctl now that the instance is up. The
	// worktree path is already gated by reqWithWorktree.UseWorktree, so
	// it's safe to skip when that's true. For multi-repo launches, loop
	// over every per-repo spec — each repo's CopyFiles ships into its
	// own RepoName subdir under the workspace.
	if !reqWithWorktree.UseWorktree && execInstance != nil && execInstance.Client != nil {
		shipRemoteCopyfilesForLaunch(ctx, m.logger, &reqWithWorktree, execInstance.Client, runtimeProgress, progressRecorder)
	}

	if prepResult != nil {
		prepResult.Steps = progressRecorder.Steps()
	}
	m.publishLaunchPrepareCompleted(req, prepResult, progressRecorder, workspacePath, true, nil)

	// Build the in-memory AgentExecution from the runtime instance. Extracted
	// to keep launchInternal under the cyclomatic-complexity budget.
	execution := m.buildExecutionFromInstance(req, execReq, execInstance, rt, profileInfo, agentConfig, prepResult)
	if profileInfo != nil && len(profileInfo.EnvVars) > 0 {
		m.cacheResolvedProfileEnv(execution, m.resolveAgentProfileEnvVars(ctx, profileInfo.EnvVars))
	}

	// Track + persist + publish. Returns the rollback error if Add lost a race.
	if err := m.registerAndPublishExecution(ctx, execution, rt, execInstance, req.SessionID); err != nil {
		return nil, err
	}

	m.logger.Debug("agentctl execution created (agent not started)",
		zap.String("execution_id", executionID),
		zap.String("task_id", req.TaskID),
		zap.Stringer("runtime", execution.RuntimeName))

	return execution, nil
}

// buildExecutionFromInstance turns the spawned ExecutorInstance + request shape
// into an in-memory *AgentExecution ready for Add. Pulled out of launchInternal
// to keep the orchestration loop's cyclomatic complexity within the linter budget.
func (m *Manager) buildExecutionFromInstance(
	req *LaunchRequest,
	execReq *ExecutorCreateRequest,
	execInstance *ExecutorInstance,
	rt ExecutorBackend,
	profileInfo *AgentProfileInfo,
	agentConfig agents.Agent,
	prepResult *EnvPrepareResult,
) *AgentExecution {
	execution := execInstance.ToAgentExecution(execReq)
	execution.RuntimeName = rt.Name()
	if req.ACPSessionID != "" {
		execution.ACPSessionID = req.ACPSessionID
	}
	execution.PrepareResult = prepResult
	if req.PreviousExecutionID != "" {
		execution.isResumedSession = true
	}
	execution.IsPassthrough = req.IsPassthrough
	cmds := m.buildAgentCommand(req, profileInfo, agentConfig)
	execution.AgentCommand = cmds.initial
	execution.ContinueCommand = cmds.continue_
	return execution
}

// registerAndPublishExecution does the post-spawn lockstep dance: track in the
// in-memory store, persist the executors_running row, publish events, kick off
// the readiness poll. On a session-conflict race during Add, rolls back the
// runtime instance so we never leak a subprocess.
func (m *Manager) registerAndPublishExecution(
	ctx context.Context,
	execution *AgentExecution,
	rt ExecutorBackend,
	execInstance *ExecutorInstance,
	sessionID string,
) error {
	if addErr := m.executionStore.Add(execution); addErr != nil {
		if errors.Is(addErr, ErrExecutionAlreadyExistsForSession) {
			m.rollbackRacedExecution(ctx, rt, execInstance, execution)
			return fmt.Errorf("session %q already has an agent running (race resolved during register)", sessionID)
		}
		return fmt.Errorf("failed to register execution: %w", addErr)
	}

	m.persistRuntimeSecrets(ctx, execInstance, execution)

	// Persist executors_running in lockstep with Add — see persistence.go for the
	// invariant. Carries forward resume_token / metadata from a prior row so the
	// lifecycle write doesn't clobber data the orchestrator's narrow CAS updates
	// wrote earlier (e.g., context_window from a previous run).
	m.persistExecutorRunning(ctx, execution)

	go m.pollOneRemoteStatus(context.Background(), execution)

	m.eventPublisher.PublishAgentEvent(ctx, events.AgentStarted, execution)
	m.eventPublisher.PublishAgentctlEvent(ctx, events.AgentctlStarting, execution, "")

	// Wait for agentctl to be ready (for shell/workspace access).
	// NOTE: This does NOT start the agent process — call StartAgentProcess() explicitly.
	go m.waitForAgentctlReady(execution)
	return nil
}

// SetExecutionDescription updates the task description stored in an execution's metadata.
// This is used when starting an agent on a workspace that was launched without a prompt.
func (m *Manager) SetExecutionDescription(_ context.Context, executionID string, description string) error {
	execution, exists := m.executionStore.Get(executionID)
	if !exists {
		return fmt.Errorf("execution %q not found", executionID)
	}
	if execution.Metadata == nil {
		execution.Metadata = make(map[string]interface{})
	}
	execution.Metadata["task_description"] = description
	return nil
}

// SetExecutionEnv stores per-run environment variables for the next agent subprocess start.
func (m *Manager) SetExecutionEnv(_ context.Context, executionID string, env map[string]string) error {
	execution, exists := m.executionStore.Get(executionID)
	if !exists {
		return fmt.Errorf("execution %q not found", executionID)
	}
	if execution.Metadata == nil {
		execution.Metadata = make(map[string]interface{})
	}
	execution.Metadata["runtime_env"] = cloneStringMap(env)
	return nil
}

func cloneStringMap(in map[string]string) map[string]string {
	if len(in) == 0 {
		return nil
	}
	out := make(map[string]string, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

// SetMcpMode changes the MCP tool mode on an existing execution's agentctl instance.
// This is used when a session transitions to plan/config mode after the workspace was
// already prepared with the default (task) mode.
func (m *Manager) SetMcpMode(ctx context.Context, executionID string, mode string) error {
	execution, exists := m.executionStore.Get(executionID)
	if !exists {
		return fmt.Errorf("execution %q not found", executionID)
	}
	if execution.agentctl == nil {
		return fmt.Errorf("execution %q has no agentctl client", executionID)
	}
	return execution.agentctl.SetMcpMode(ctx, mode)
}

// resolveApprovalPolicyAndDisplayName resolves the approval policy and agent display name
// from the execution's agent profile and registry.
func (m *Manager) resolveApprovalPolicyAndDisplayName(ctx context.Context, execution *AgentExecution) (string, string) {
	approvalPolicy := ""
	agentDisplayName := ""
	if execution.AgentProfileID == "" || m.profileResolver == nil {
		return approvalPolicy, agentDisplayName
	}
	profileInfo, err := m.profileResolver.ResolveProfile(ctx, execution.AgentProfileID)
	if err != nil {
		return approvalPolicy, agentDisplayName
	}
	if profileInfo.AutoApprove {
		approvalPolicy = "never"
	} else {
		approvalPolicy = "untrusted"
	}
	// Look up display name from registry (e.g. "Claude", "Auggie", "Codex")
	if agentCfg, ok := m.registry.Get(profileInfo.AgentName); ok && agentCfg.DisplayName() != "" {
		agentDisplayName = agentCfg.DisplayName()
	} else {
		agentDisplayName = profileInfo.AgentName
	}
	return approvalPolicy, agentDisplayName
}

// createBootMessage creates a boot message and starts the stderr polling goroutine.
// Returns the message and stop channel (both nil if bootMessageService is not configured).
func (m *Manager) createBootMessage(ctx context.Context, execution *AgentExecution, bootCommand, agentDisplayName string) (*models.Message, chan struct{}) {
	if m.bootMessageService == nil {
		return nil, nil
	}
	bootMsg, bootErr := m.bootMessageService.CreateMessage(ctx, &BootMessageRequest{
		TaskSessionID: execution.SessionID,
		TaskID:        execution.TaskID,
		Content:       "",
		AuthorType:    "agent",
		Type:          "script_execution",
		Metadata: map[string]interface{}{
			"script_type": "agent_boot",
			"agent_name":  agentDisplayName,
			"command":     bootCommand,
			"status":      "running",
			"is_resuming": execution.ACPSessionID != "",
			"started_at":  time.Now().UTC().Format(time.RFC3339Nano),
		},
	})
	if bootErr != nil {
		m.logger.Warn("failed to create boot message, continuing without boot output",
			zap.String("execution_id", execution.ID),
			zap.Error(bootErr))
		return nil, nil
	}
	bootStopCh := make(chan struct{})
	go m.pollAgentStderr(execution, execution.agentctl, bootMsg, bootStopCh)
	return bootMsg, bootStopCh
}

// getTaskDescriptionFromMetadata extracts the task description string from execution metadata.
func getTaskDescriptionFromMetadata(execution *AgentExecution) string {
	if execution.Metadata == nil {
		return ""
	}
	if desc, ok := execution.Metadata["task_description"].(string); ok {
		return desc
	}
	return ""
}

// getAttachmentsFromMetadata extracts attachments from execution metadata.
func getAttachmentsFromMetadata(execution *AgentExecution) []MessageAttachment {
	if execution.Metadata == nil {
		return nil
	}
	attachments, ok := execution.Metadata["attachments"].([]MessageAttachment)
	if ok {
		return attachments
	}
	return nil
}

// configureAndStartAgent configures the agent command and starts the agent subprocess.
// Returns the effective boot command (full command with adapter args, or base command).
func (m *Manager) configureAndStartAgent(ctx context.Context, execution *AgentExecution, taskDescription, approvalPolicy string) (string, error) {
	env := runtimeEnvFromMetadata(execution.Metadata)
	if taskDescription != "" {
		env["TASK_DESCRIPTION"] = taskDescription
	}
	m.mergeAgentProfileEnvForExecution(ctx, execution, env)

	if err := execution.agentctl.ConfigureAgent(ctx, execution.AgentCommand, env, approvalPolicy, execution.ContinueCommand); err != nil {
		return "", fmt.Errorf("failed to configure agent: %w", err)
	}

	fullCommand, err := execution.agentctl.Start(ctx)
	if err != nil {
		m.updateExecutionError(execution.ID, "failed to start agent: "+err.Error())
		return "", fmt.Errorf("failed to start agent: %w", err)
	}

	bootCommand := fullCommand
	if bootCommand == "" {
		bootCommand = execution.AgentCommand
	}
	return bootCommand, nil
}

func runtimeEnvFromMetadata(metadata map[string]interface{}) map[string]string {
	env := map[string]string{}
	if metadata == nil {
		return env
	}
	if typed, ok := metadata["runtime_env"].(map[string]string); ok {
		for k, v := range typed {
			env[k] = v
		}
	}
	if raw, ok := metadata["runtime_env"].(map[string]interface{}); ok {
		for k, v := range raw {
			if str, strOK := v.(string); strOK {
				env[k] = str
			}
		}
	}
	return env
}

// initializeAgentSession handles post-startup initialization: boot message, ACP session,
// MCP servers. It finalizes the boot message on success or failure.
func (m *Manager) initializeAgentSession(ctx context.Context, execution *AgentExecution, bootCommand, agentDisplayName, taskDescription string) error {
	bootMsg, bootStopCh := m.createBootMessage(ctx, execution, bootCommand, agentDisplayName)

	// Give the agent process a moment to initialize
	time.Sleep(500 * time.Millisecond)

	agentConfig, err := m.getAgentConfigForExecution(execution)
	if err != nil {
		m.finalizeBootMessage(execution, bootMsg, bootStopCh, execution.agentctl, "failed")
		return fmt.Errorf("failed to get agent config: %w", err)
	}

	mcpServers, err := m.resolveMcpServers(ctx, execution, agentConfig)
	if err != nil {
		m.finalizeBootMessage(execution, bootMsg, bootStopCh, execution.agentctl, "failed")
		m.updateExecutionError(execution.ID, "failed to resolve MCP config: "+err.Error())
		return fmt.Errorf("failed to resolve MCP config: %w", err)
	}

	attachments := getAttachmentsFromMetadata(execution)
	if err := m.initializeACPSession(ctx, execution, agentConfig, taskDescription, attachments, mcpServers); err != nil {
		m.finalizeBootMessage(execution, bootMsg, bootStopCh, execution.agentctl, "failed")
		m.updateExecutionError(execution.ID, "failed to initialize ACP: "+err.Error())
		return fmt.Errorf("failed to initialize ACP: %w", err)
	}

	m.finalizeBootMessage(execution, bootMsg, bootStopCh, execution.agentctl, containerStateExited)
	return nil
}

// initGitRepo initializes a git repository in the given directory.
// Creates an initial commit so the workspace has a clean git state.
// This function is idempotent - it skips initialization if .git already exists.
func (m *Manager) initGitRepo(ctx context.Context, workspacePath string) error {
	// Check if git repository already exists (idempotent)
	gitDir := filepath.Join(workspacePath, ".git")
	if info, err := os.Stat(gitDir); err == nil {
		if info.IsDir() {
			return nil // Already initialized
		}
	} else if !os.IsNotExist(err) {
		// Non-ENOENT error (permissions, I/O, etc.) - fail explicitly
		return fmt.Errorf("failed to check for .git directory: %w", err)
	}

	// Initialize git repository
	cmd := exec.CommandContext(ctx, "git", "init")
	cmd.Dir = workspacePath
	if output, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("git init failed: %w (output: %s)", err, string(output))
	}

	// Configure git user (required for initial commit)
	configName := exec.CommandContext(ctx, "git", "config", "user.name", "Kandev Quick Chat")
	configName.Dir = workspacePath
	_ = configName.Run() // Ignore error - might already be configured globally

	configEmail := exec.CommandContext(ctx, "git", "config", "user.email", "quickchat@kandev.local")
	configEmail.Dir = workspacePath
	_ = configEmail.Run() // Ignore error - might already be configured globally

	// Create initial commit with empty .gitkeep file
	gitkeepPath := filepath.Join(workspacePath, ".gitkeep")
	if err := os.WriteFile(gitkeepPath, []byte(""), 0644); err != nil {
		return fmt.Errorf("failed to create .gitkeep: %w", err)
	}

	addCmd := exec.CommandContext(ctx, "git", "add", ".gitkeep")
	addCmd.Dir = workspacePath
	if output, err := addCmd.CombinedOutput(); err != nil {
		return fmt.Errorf("git add failed: %w (output: %s)", err, string(output))
	}

	commitCmd := exec.CommandContext(ctx, "git", "commit", "-m", "Initial commit")
	commitCmd.Dir = workspacePath
	if output, err := commitCmd.CombinedOutput(); err != nil {
		return fmt.Errorf("git commit failed: %w (output: %s)", err, string(output))
	}

	return nil
}