fix: stream setup script output and keep prepare panel on failure (#607)

This commit is contained in:
Carlos Florêncio
2026-04-19 11:08:18 +01:00
committed by GitHub
parent 519de03da5
commit 1f7e4dbb5e
49 changed files with 1925 additions and 524 deletions
+1
View File
@@ -127,6 +127,7 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch
WorktreePath: worktreePath,
WorktreeBranch: worktreeBranch,
Metadata: execution.Metadata,
PrepareResult: execution.PrepareResult,
}, nil
}
@@ -21,6 +21,9 @@ const defaultLocalPrepareScript = `#!/bin/bash
# Runs before launching the local agent runtime.
# The script executes with working directory set to {{workspace.path}}.
# Use {{repository.path}} when you need the canonical repository root path.
# ---- Repository setup (if configured) ----
{{repository.setup_script}}
`
const defaultWorktreePrepareScript = `#!/bin/bash
@@ -28,6 +31,9 @@ const defaultWorktreePrepareScript = `#!/bin/bash
# Runs after the worktree has already been created/reused by Kandev.
# The script executes with working directory set to {{worktree.path}}.
# Use {{repository.path}} if you need to run commands in the main repository.
# ---- Repository setup (if configured) ----
{{repository.setup_script}}
`
const defaultDockerPrepareScript = `#!/bin/bash
@@ -2,6 +2,7 @@ package lifecycle
import (
"context"
"strings"
"time"
"github.com/kandev/kandev/internal/agent/executor"
@@ -21,20 +22,21 @@ const (
// EnvPrepareRequest contains the parameters for environment preparation.
type EnvPrepareRequest struct {
TaskID string
SessionID string
TaskTitle string
ExecutionID string
ExecutorType executor.Name
WorkspacePath string
RepositoryPath string
RepositoryID string
UseWorktree bool
SetupScript string
BaseBranch string
CheckoutBranch string
WorktreeID string
WorktreeBranch string
TaskID string
SessionID string
TaskTitle string
ExecutionID string
ExecutorType executor.Name
WorkspacePath string
RepositoryPath string
RepositoryID string
UseWorktree bool
SetupScript string
RepoSetupScript string // Repository-level setup script (e.g. "make install")
BaseBranch string
CheckoutBranch string
WorktreeID string
WorktreeBranch string
WorktreeBranchPrefix string
PullBeforeWorktree bool
@@ -48,6 +50,7 @@ type EnvPrepareRequest struct {
// PrepareStep represents a single step in the preparation process.
type PrepareStep struct {
Name string `json:"name"`
Command string `json:"command,omitempty"`
Status PrepareStepStatus `json:"status"`
Output string `json:"output,omitempty"`
Error string `json:"error,omitempty"`
@@ -83,6 +86,53 @@ type EnvironmentPreparer interface {
Prepare(ctx context.Context, req *EnvPrepareRequest, onProgress PrepareProgressCallback) (*EnvPrepareResult, error)
}
// MaxStepOutputBytes is the maximum size of a single step's Output field
// when persisting to session metadata.
const MaxStepOutputBytes = 10 * 1024
// SerializePrepareResult converts a prepare result into a map suitable for
// storage in session metadata. Used by both the synchronous persistence path
// (persistLaunchState) and the async event handler (handlePrepareCompleted).
func SerializePrepareResult(result *EnvPrepareResult) map[string]interface{} {
status := "completed"
if !result.Success {
status = "failed"
}
steps := make([]map[string]interface{}, 0, len(result.Steps))
for _, step := range result.Steps {
output := step.Output
if len(output) > MaxStepOutputBytes {
// Truncate at a valid UTF-8 boundary to avoid splitting multi-byte runes.
output = strings.ToValidUTF8(output[:MaxStepOutputBytes], "") + "\n... (truncated)"
}
entry := map[string]interface{}{
"name": step.Name, "status": string(step.Status),
"output": output, "command": step.Command,
}
if step.Error != "" {
entry["error"] = step.Error
}
if step.Warning != "" {
entry["warning"] = step.Warning
}
if step.WarningDetail != "" {
entry["warning_detail"] = step.WarningDetail
}
if step.StartedAt != nil {
entry["started_at"] = step.StartedAt.Format(time.RFC3339Nano)
}
if step.EndedAt != nil {
entry["ended_at"] = step.EndedAt.Format(time.RFC3339Nano)
}
steps = append(steps, entry)
}
return map[string]interface{}{
"status": status, "steps": steps,
"error_message": result.ErrorMessage,
"duration_ms": result.Duration.Milliseconds(),
}
}
// PreparerRegistry maps executor types to environment preparers.
type PreparerRegistry struct {
preparers map[executor.Name]EnvironmentPreparer
@@ -1,11 +1,13 @@
package lifecycle
import (
"bytes"
"context"
"fmt"
"os"
"os/exec"
"strings"
"sync"
"time"
"go.uber.org/zap"
@@ -71,6 +73,7 @@ func (p *LocalPreparer) Prepare(ctx context.Context, req *EnvPrepareRequest, onP
// Step 2: Checkout branch (if specified)
if effectiveBranch != "" {
step = beginStep("Checkout branch")
step.Command = fmt.Sprintf("git fetch origin %s && git checkout %s", effectiveBranch, effectiveBranch)
reportProgress(onProgress, step, stepIdx, totalSteps)
output, err := checkoutBranch(ctx, workspacePath, effectiveBranch)
if err != nil {
@@ -89,22 +92,7 @@ func (p *LocalPreparer) Prepare(ctx context.Context, req *EnvPrepareRequest, onP
// Step 3: Run setup script (if provided)
if resolvedScript != "" {
step = beginStep("Run setup script")
reportProgress(onProgress, step, stepIdx, totalSteps)
output, err := runSetupScript(ctx, resolvedScript, workspacePath, req.Env)
if err != nil {
completeStepError(&step, err.Error())
step.Output = output
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
p.logger.Warn("setup script failed", zap.String("task_id", req.TaskID), zap.Error(err))
// Setup script failure is non-fatal — log and continue
} else {
step.Output = output
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
}
steps = runSetupScriptStep(ctx, req, workspacePath, resolvedScript, stepIdx, totalSteps, onProgress, steps, p.logger)
}
return &EnvPrepareResult{
@@ -137,15 +125,65 @@ func checkoutBranch(ctx context.Context, workDir, branch string) (string, error)
return outStr, nil
}
// runSetupScript executes a setup script in the given working directory.
func runSetupScript(ctx context.Context, script, workDir string, env map[string]string) (string, error) {
// setupScriptStreamInterval is the minimum gap between streaming-output callbacks
// while a setup script runs. Chatty scripts (e.g. npm install with progress bars)
// can emit hundreds of writes per second; throttling keeps WS event volume sane
// while still showing live output.
const setupScriptStreamInterval = 100 * time.Millisecond
// runSetupScript executes a setup script in the given working directory,
// streaming combined stdout/stderr to onOutput (if non-nil) as it runs.
// Returns the full accumulated output (trimmed) and any execution error.
func runSetupScript(ctx context.Context, script, workDir string, env map[string]string, onOutput func(current string)) (string, error) {
cmd := exec.CommandContext(ctx, "sh", "-c", script)
if workDir != "" {
cmd.Dir = workDir
}
cmd.Env = buildEnvSlice(env)
out, err := cmd.CombinedOutput()
return strings.TrimSpace(string(out)), err
w := newStreamingWriter(onOutput, setupScriptStreamInterval)
cmd.Stdout = w
cmd.Stderr = w
err := cmd.Run()
return strings.TrimSpace(w.String()), err
}
// streamingWriter accumulates writes into a buffer and invokes onFlush with the
// current snapshot at most once per minGap. It's safe for concurrent Write calls
// (both stdout and stderr write through the same writer).
type streamingWriter struct {
mu sync.Mutex
buf bytes.Buffer
onFlush func(current string)
lastFlush time.Time
minGap time.Duration
}
func newStreamingWriter(onFlush func(current string), minGap time.Duration) *streamingWriter {
return &streamingWriter{onFlush: onFlush, minGap: minGap}
}
func (w *streamingWriter) Write(p []byte) (int, error) {
// Hold the lock through the flush so concurrent stdout+stderr writes
// can't both observe `now-lastFlush >= minGap` and race on the captured
// `step.Output` shared by the caller's callback.
w.mu.Lock()
defer w.mu.Unlock()
n, err := w.buf.Write(p)
now := time.Now()
if w.onFlush != nil && now.Sub(w.lastFlush) >= w.minGap {
w.lastFlush = now
w.onFlush(strings.TrimSpace(w.buf.String()))
}
return n, err
}
// String returns the full accumulated output.
func (w *streamingWriter) String() string {
w.mu.Lock()
defer w.mu.Unlock()
return w.buf.String()
}
// buildEnvSlice converts a map to os.Environ format (KEY=VALUE).
@@ -162,6 +200,19 @@ func buildEnvSlice(env map[string]string) []string {
return result
}
// setupScriptDisplayCommand returns a short, user-facing command string for the
// "Run setup script" step. Prefers the explicit profile script, then the
// repository-level script. Falls back to empty (the step still shows its name).
func setupScriptDisplayCommand(req *EnvPrepareRequest) string {
if s := strings.TrimSpace(req.SetupScript); s != "" {
return s
}
if s := strings.TrimSpace(req.RepoSetupScript); s != "" {
return s
}
return ""
}
// Helper functions for step lifecycle
func beginStep(name string) PrepareStep {
@@ -0,0 +1,221 @@
package lifecycle
import (
"context"
"strings"
"sync"
"testing"
"time"
)
// TestRunSetupScript_StreamsOutputWhileRunning verifies that the streaming
// callback fires with partial output as the script runs, not only once at the
// end. This is what makes the UI show `npm install` progress live instead of
// sitting on a silent spinner.
func TestRunSetupScript_StreamsOutputWhileRunning(t *testing.T) {
// Script emits three lines with 200ms gaps. With the 100ms stream interval,
// we expect at least two intermediate flushes before completion.
script := `echo one; sleep 0.2; echo two; sleep 0.2; echo three`
var mu sync.Mutex
var snapshots []string
cb := func(current string) {
mu.Lock()
defer mu.Unlock()
snapshots = append(snapshots, current)
}
output, err := runSetupScript(context.Background(), script, t.TempDir(), nil, cb)
if err != nil {
t.Fatalf("runSetupScript error: %v", err)
}
if output != "one\ntwo\nthree" {
t.Fatalf("unexpected final output: %q", output)
}
mu.Lock()
defer mu.Unlock()
if len(snapshots) < 2 {
t.Fatalf("expected at least 2 streaming snapshots, got %d: %v", len(snapshots), snapshots)
}
// Snapshots are monotonically growing — each should be a prefix of the next.
for i := 1; i < len(snapshots); i++ {
if !strings.HasPrefix(snapshots[i], strings.TrimSpace(snapshots[i-1])) {
t.Fatalf("snapshot %d (%q) is not a growth of snapshot %d (%q)", i, snapshots[i], i-1, snapshots[i-1])
}
}
// First snapshot should contain "one" (seen at ~0ms after first echo).
if !strings.Contains(snapshots[0], "one") {
t.Fatalf("first snapshot should contain 'one', got %q", snapshots[0])
}
}
// TestRunSetupScript_CapturesOutputOnFailure verifies that when the script
// exits non-zero, both the output and the error are returned so the UI can
// display what the failing command printed.
func TestRunSetupScript_CapturesOutputOnFailure(t *testing.T) {
script := `echo installing; echo "ENOENT: package-lock.json" 1>&2; exit 1`
output, err := runSetupScript(context.Background(), script, t.TempDir(), nil, nil)
if err == nil {
t.Fatal("expected error for non-zero exit, got nil")
}
if !strings.Contains(output, "installing") {
t.Fatalf("output missing stdout line: %q", output)
}
if !strings.Contains(output, "ENOENT") {
t.Fatalf("output missing stderr line: %q", output)
}
}
// TestRunSetupScript_NilCallbackIsSafe verifies callers can skip streaming.
func TestRunSetupScript_NilCallbackIsSafe(t *testing.T) {
output, err := runSetupScript(context.Background(), `echo hi`, t.TempDir(), nil, nil)
if err != nil {
t.Fatalf("runSetupScript error: %v", err)
}
if output != "hi" {
t.Fatalf("unexpected output: %q", output)
}
}
// TestLocalPreparer_StreamsSetupScriptProgress verifies the LocalPreparer wires
// the streaming callback into PrepareProgressCallback — the setup script step
// is reported multiple times with growing Output before completion.
func TestLocalPreparer_StreamsSetupScriptProgress(t *testing.T) {
isolateGitEnv(t)
log := newTestLocalLogger()
preparer := NewLocalPreparer(log)
repoDir := initGitRepo(t)
req := &EnvPrepareRequest{
TaskID: "task-1",
RepositoryPath: repoDir,
SetupScript: `echo first; sleep 0.2; echo second; sleep 0.2; echo third`,
}
var mu sync.Mutex
var setupEvents []PrepareStep
cb := func(step PrepareStep, _, _ int) {
if step.Name != "Run setup script" {
return
}
mu.Lock()
defer mu.Unlock()
setupEvents = append(setupEvents, step)
}
result, err := preparer.Prepare(context.Background(), req, cb)
if err != nil {
t.Fatalf("Prepare() error: %v", err)
}
if !result.Success {
t.Fatalf("Prepare() failed: %s", result.ErrorMessage)
}
mu.Lock()
defer mu.Unlock()
// Expect: 1 begin (running, no output) + >=1 streaming update (running, partial output) + 1 final (completed, full output).
if len(setupEvents) < 3 {
t.Fatalf("expected >=3 setup step events (begin + stream + complete), got %d", len(setupEvents))
}
first := setupEvents[0]
last := setupEvents[len(setupEvents)-1]
if first.Status != PrepareStepRunning {
t.Errorf("first setup event should be running, got %q", first.Status)
}
if first.Output != "" {
t.Errorf("first setup event should have no output yet, got %q", first.Output)
}
if last.Status != PrepareStepCompleted {
t.Errorf("last setup event should be completed, got %q", last.Status)
}
if !strings.Contains(last.Output, "first") || !strings.Contains(last.Output, "third") {
t.Errorf("final output should contain all lines, got %q", last.Output)
}
// At least one intermediate event should have partial (but non-empty) output.
sawPartial := false
for _, ev := range setupEvents[1 : len(setupEvents)-1] {
if ev.Status == PrepareStepRunning && ev.Output != "" {
sawPartial = true
break
}
}
if !sawPartial {
t.Errorf("expected at least one intermediate streaming event with partial output; events: %+v", setupEvents)
}
}
// TestLocalPreparer_SetupScriptFailurePreservesOutputAndError verifies that
// a failing setup script marks the step as failed and keeps both the output
// and error attached — this is what lets the UI display the failure banner
// with the tail of stdout/stderr.
func TestLocalPreparer_SetupScriptFailurePreservesOutputAndError(t *testing.T) {
isolateGitEnv(t)
log := newTestLocalLogger()
preparer := NewLocalPreparer(log)
repoDir := initGitRepo(t)
req := &EnvPrepareRequest{
TaskID: "task-1",
RepositoryPath: repoDir,
SetupScript: `echo running install; echo "fatal: not a git repository" 1>&2; exit 1`,
}
result, err := preparer.Prepare(context.Background(), req, nil)
if err != nil {
t.Fatalf("Prepare() error: %v (setup script failure should be non-fatal)", err)
}
// Per design, setup script failure is non-fatal — overall result succeeds
// so the agent can still start and the user can retry.
if !result.Success {
t.Fatalf("Prepare() should succeed even when setup script fails: %s", result.ErrorMessage)
}
if len(result.Steps) != 2 {
t.Fatalf("expected 2 steps (validate + setup), got %d", len(result.Steps))
}
setup := result.Steps[1]
if setup.Name != "Run setup script" {
t.Fatalf("second step should be setup script, got %q", setup.Name)
}
if setup.Status != PrepareStepFailed {
t.Errorf("setup step should be failed, got %q", setup.Status)
}
if setup.Error == "" {
t.Errorf("setup step should have an error message")
}
if !strings.Contains(setup.Output, "running install") {
t.Errorf("setup step output should contain stdout, got %q", setup.Output)
}
if !strings.Contains(setup.Output, "fatal: not a git repository") {
t.Errorf("setup step output should contain stderr, got %q", setup.Output)
}
}
// TestStreamingWriter_ThrottlesToMinGap verifies the writer does not fire the
// callback more often than minGap — burst writes should coalesce.
func TestStreamingWriter_ThrottlesToMinGap(t *testing.T) {
var mu sync.Mutex
var calls int
w := newStreamingWriter(func(_ string) {
mu.Lock()
calls++
mu.Unlock()
}, 50*time.Millisecond)
// Burst 10 writes in a tight loop — throttled callback should fire once
// (or at most twice if the test scheduler stalls across the 50ms window).
for i := 0; i < 10; i++ {
if _, err := w.Write([]byte("x")); err != nil {
t.Fatal(err)
}
}
mu.Lock()
defer mu.Unlock()
if calls > 2 {
t.Errorf("expected <=2 throttled callback invocations for a tight burst, got %d", calls)
}
}
@@ -0,0 +1,90 @@
package lifecycle
import (
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func TestSerializePrepareResult(t *testing.T) {
t.Run("success result", func(t *testing.T) {
now := time.Now()
twoSecsAgo := now.Add(-2 * time.Second)
result := &EnvPrepareResult{
Success: true,
Steps: []PrepareStep{
{Name: "Validate repository", Status: PrepareStepCompleted, Command: "git status", Output: "On branch main"},
{Name: "Run setup script", Status: PrepareStepCompleted, Command: "npm install", Output: "added 100 packages", StartedAt: &twoSecsAgo, EndedAt: &now},
},
Duration: 3 * time.Second,
}
serialized := SerializePrepareResult(result)
require.Equal(t, "completed", serialized["status"])
require.Equal(t, int64(3000), serialized["duration_ms"])
steps, ok := serialized["steps"].([]map[string]interface{})
require.True(t, ok)
require.Len(t, steps, 2)
require.Equal(t, "Validate repository", steps[0]["name"])
require.Equal(t, string(PrepareStepCompleted), steps[0]["status"])
require.Equal(t, "git status", steps[0]["command"])
require.Equal(t, "On branch main", steps[0]["output"])
require.Equal(t, "Run setup script", steps[1]["name"])
require.NotEmpty(t, steps[1]["started_at"])
require.NotEmpty(t, steps[1]["ended_at"])
})
t.Run("failure result", func(t *testing.T) {
result := &EnvPrepareResult{
Success: false,
ErrorMessage: "setup script failed",
Steps: []PrepareStep{
{Name: "Run setup script", Status: PrepareStepFailed, Command: "npm install", Output: "ERR! not found"},
},
Duration: 1 * time.Second,
}
serialized := SerializePrepareResult(result)
require.Equal(t, "failed", serialized["status"])
require.Equal(t, "setup script failed", serialized["error_message"])
})
t.Run("truncates long output at UTF-8 boundary", func(t *testing.T) {
// Create output with a multi-byte emoji near the truncation boundary
longOutput := strings.Repeat("a", MaxStepOutputBytes-2) + "\U0001F600" // 4-byte emoji
result := &EnvPrepareResult{
Success: true,
Steps: []PrepareStep{{Name: "test", Status: PrepareStepCompleted, Output: longOutput}},
}
serialized := SerializePrepareResult(result)
steps := serialized["steps"].([]map[string]interface{})
output := steps[0]["output"].(string)
require.True(t, strings.HasSuffix(output, "\n... (truncated)"))
// The output should be valid UTF-8 (no partial runes)
require.True(t, strings.ToValidUTF8(output, "") == output, "output should be valid UTF-8")
})
t.Run("omits timestamps when nil", func(t *testing.T) {
result := &EnvPrepareResult{
Success: true,
Steps: []PrepareStep{{Name: "test", Status: PrepareStepCompleted}},
}
serialized := SerializePrepareResult(result)
steps := serialized["steps"].([]map[string]interface{})
_, hasStarted := steps[0]["started_at"]
_, hasEnded := steps[0]["ended_at"]
require.False(t, hasStarted, "should not include started_at when nil")
require.False(t, hasEnded, "should not include ended_at when nil")
})
}
@@ -2,6 +2,7 @@ package lifecycle
import (
"context"
"fmt"
"path/filepath"
"time"
@@ -44,37 +45,110 @@ func (p *WorktreePreparer) Prepare(ctx context.Context, req *EnvPrepareRequest,
stepIdx := 0
// Step 1: Validate repository path
steps, ok := p.validateWorktreeRequest(req, stepIdx, totalSteps, onProgress, steps)
if !ok {
return &EnvPrepareResult{Success: false, Steps: steps, ErrorMessage: steps[len(steps)-1].Error, Duration: time.Since(start)}, nil
}
stepIdx++
// Steps 2 (and optional sync): Create worktree (with optional pre-sync).
wt, steps, stepIdx, err := p.createWorktreeWithSync(ctx, req, stepIdx, totalSteps, onProgress, steps)
if err != nil {
return &EnvPrepareResult{Success: false, Steps: steps, ErrorMessage: err.Error(), Duration: time.Since(start)}, nil
}
workspacePath := wt.Path
mainRepoGitDir := filepath.Join(req.RepositoryPath, ".git")
// Step 3 (optional): Fetch PR branch is handled inside worktree.Manager.Create
// via CreateRequest.CheckoutBranch. If it was set and failed, Create() would have
// returned an error above. We report a success step here for UI visibility.
if req.CheckoutBranch != "" {
step := beginStep("Fetch PR branch")
step.Command = fmt.Sprintf("git fetch origin %s", req.CheckoutBranch)
reportProgress(onProgress, step, stepIdx, totalSteps)
if wt.FetchWarning != "" {
step.Warning = wt.FetchWarning
step.WarningDetail = wt.FetchWarningDetail
}
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
stepIdx++
}
// Step N: Run setup script (if provided)
resolvedScript := resolvePreparerSetupScript(req, workspacePath)
if resolvedScript != "" {
totalSteps++
steps = runSetupScriptStep(ctx, req, workspacePath, resolvedScript, stepIdx, totalSteps, onProgress, steps, p.logger)
}
return &EnvPrepareResult{
Success: true,
Steps: steps,
WorkspacePath: workspacePath,
Duration: time.Since(start),
WorktreeID: wt.ID,
WorktreeBranch: wt.Branch,
MainRepoGitDir: mainRepoGitDir,
}, nil
}
// validateWorktreeRequest runs the "Validate repository" step.
// Returns the updated steps and ok=true on success, ok=false when a failure step was appended.
func (p *WorktreePreparer) validateWorktreeRequest(
req *EnvPrepareRequest,
stepIdx, totalSteps int,
onProgress PrepareProgressCallback,
steps []PrepareStep,
) ([]PrepareStep, bool) {
step := beginStep("Validate repository")
reportProgress(onProgress, step, stepIdx, totalSteps)
if req.RepositoryPath == "" {
completeStepError(&step, "no repository path provided")
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
return &EnvPrepareResult{Success: false, Steps: steps, ErrorMessage: step.Error, Duration: time.Since(start)}, nil
return steps, false
}
if p.worktreeMgr == nil {
completeStepError(&step, "worktree manager not available")
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
return &EnvPrepareResult{Success: false, Steps: steps, ErrorMessage: step.Error, Duration: time.Since(start)}, nil
return steps, false
}
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
stepIdx++
return steps, true
}
// createWorktreeWithSync creates the worktree (with optional sync-base-branch step).
// Returns the created worktree, updated steps, the next step index after creation, and any error.
func (p *WorktreePreparer) createWorktreeWithSync(
ctx context.Context,
req *EnvPrepareRequest,
stepIdx, totalSteps int,
onProgress PrepareProgressCallback,
steps []PrepareStep,
) (*worktree.Worktree, []PrepareStep, int, error) {
var syncStep *PrepareStep
syncStepIndex := -1
if req.PullBeforeWorktree {
s := beginStep("Sync base branch")
if req.BaseBranch != "" {
s.Command = fmt.Sprintf("git fetch origin %s && git pull origin %s", req.BaseBranch, req.BaseBranch)
} else {
s.Command = "git fetch origin && git pull"
}
reportProgress(onProgress, s, stepIdx, totalSteps)
syncStep = &s
syncStepIndex = stepIdx
stepIdx++
}
// Step 2: Create worktree
step = beginStep("Create worktree")
step := beginStep("Create worktree")
step.Command = "git worktree add"
reportProgress(onProgress, step, stepIdx, totalSteps)
createReq := worktree.CreateRequest{
@@ -99,82 +173,42 @@ func (p *WorktreePreparer) Prepare(ctx context.Context, req *EnvPrepareRequest,
}
wt, err := p.worktreeMgr.Create(ctx, createReq)
if syncStep != nil {
if syncStep.Status == PrepareStepRunning {
if err != nil {
completeStepError(syncStep, "sync interrupted by worktree creation failure")
} else {
completeStepSuccess(syncStep)
}
}
steps = append(steps, *syncStep)
reportProgress(onProgress, *syncStep, syncStepIndex, totalSteps)
}
steps = finalizeSyncStep(syncStep, syncStepIndex, totalSteps, onProgress, steps, err)
if err != nil {
completeStepError(&step, err.Error())
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
p.logger.Error("worktree creation failed",
zap.String("task_id", req.TaskID),
zap.Error(err))
return &EnvPrepareResult{Success: false, Steps: steps, ErrorMessage: err.Error(), Duration: time.Since(start)}, nil
p.logger.Error("worktree creation failed", zap.String("task_id", req.TaskID), zap.Error(err))
return nil, steps, stepIdx, err
}
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
stepIdx++
return wt, steps, stepIdx + 1, nil
}
workspacePath := wt.Path
mainRepoGitDir := filepath.Join(req.RepositoryPath, ".git")
// Step 3 (optional): Fetch PR branch is handled inside worktree.Manager.Create
// via CreateRequest.CheckoutBranch. If it was set and failed, Create() would have
// returned an error above. We report a success step here for UI visibility.
if req.CheckoutBranch != "" {
step = beginStep("Fetch PR branch")
reportProgress(onProgress, step, stepIdx, totalSteps)
if wt.FetchWarning != "" {
step.Warning = wt.FetchWarning
step.WarningDetail = wt.FetchWarningDetail
}
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
stepIdx++
// finalizeSyncStep closes out the optional sync step after worktree.Create returns.
func finalizeSyncStep(
syncStep *PrepareStep,
syncStepIndex, totalSteps int,
onProgress PrepareProgressCallback,
steps []PrepareStep,
createErr error,
) []PrepareStep {
if syncStep == nil {
return steps
}
// Step N: Run setup script (if provided)
resolvedScript := resolvePreparerSetupScript(req, workspacePath)
if resolvedScript != "" {
totalSteps++
step = beginStep("Run setup script")
reportProgress(onProgress, step, stepIdx, totalSteps)
output, scriptErr := runSetupScript(ctx, resolvedScript, workspacePath, req.Env)
if scriptErr != nil {
completeStepError(&step, scriptErr.Error())
step.Output = output
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
p.logger.Warn("setup script failed", zap.String("task_id", req.TaskID), zap.Error(scriptErr))
// Setup script failure is non-fatal
if syncStep.Status == PrepareStepRunning {
if createErr != nil {
completeStepError(syncStep, "sync interrupted by worktree creation failure")
} else {
step.Output = output
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
completeStepSuccess(syncStep)
}
}
return &EnvPrepareResult{
Success: true,
Steps: steps,
WorkspacePath: workspacePath,
Duration: time.Since(start),
WorktreeID: wt.ID,
WorktreeBranch: wt.Branch,
MainRepoGitDir: mainRepoGitDir,
}, nil
steps = append(steps, *syncStep)
reportProgress(onProgress, *syncStep, syncStepIndex, totalSteps)
return steps
}
func applySyncProgressEvent(step *PrepareStep, event worktree.SyncProgressEvent) {
@@ -42,18 +42,21 @@ type ACPSessionCreatedPayload struct {
// PrepareProgressEventPayload is the payload for environment preparation progress events.
type PrepareProgressEventPayload struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
ExecutionID string `json:"execution_id"`
StepName string `json:"step_name"`
StepIndex int `json:"step_index"`
TotalSteps int `json:"total_steps"`
Status string `json:"status"`
Output string `json:"output,omitempty"`
Error string `json:"error,omitempty"`
Warning string `json:"warning,omitempty"`
WarningDetail string `json:"warning_detail,omitempty"`
Timestamp string `json:"timestamp"`
TaskID string `json:"task_id"`
SessionID string `json:"session_id"`
ExecutionID string `json:"execution_id"`
StepName string `json:"step_name"`
StepCommand string `json:"step_command,omitempty"`
StepIndex int `json:"step_index"`
TotalSteps int `json:"total_steps"`
Status string `json:"status"`
Output string `json:"output,omitempty"`
Error string `json:"error,omitempty"`
Warning string `json:"warning,omitempty"`
WarningDetail string `json:"warning_detail,omitempty"`
StartedAt *time.Time `json:"started_at,omitempty"`
EndedAt *time.Time `json:"ended_at,omitempty"`
Timestamp string `json:"timestamp"`
}
// GetSessionID returns the session ID for this event (used by event routing).
@@ -73,7 +73,7 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID
}
if info.WorkspacePath == "" {
return nil, fmt.Errorf("no workspace path available for session %s", sessionID)
return nil, fmt.Errorf("%w: session %s has no workspace path yet", ErrSessionWorkspaceNotReady, sessionID)
}
m.logger.Info("creating execution for task session",
@@ -110,16 +110,52 @@ func (m *Manager) buildAgentCommand(req *LaunchRequest, profileInfo *AgentProfil
// 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.
func (m *Manager) launchResolveWorkspacePath(ctx context.Context, req *LaunchRequest) (workspacePath, mainRepoGitDir, worktreeID, worktreeBranch string) {
if req.UseWorktree {
// Worktree executors: preparer handles worktree creation.
// Return empty path; it will be populated from preparer results.
// 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) {
if req.UseWorktree && req.ACPSessionID == "" {
return "", "", "", ""
}
if req.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 repositories (e.g., quick chat), create a workspace in ~/.kandev/quick-chat/
// These directories are cleaned up when the ephemeral task is deleted (see task service performTaskCleanup).
if workspacePath == "" && req.SessionID != "" && m.dataDir != "" {
@@ -204,6 +240,7 @@ func (m *Manager) newProgressCallback(taskID, sessionID string) PrepareProgressC
TaskID: taskID,
SessionID: sessionID,
StepName: step.Name,
StepCommand: step.Command,
StepIndex: stepIndex,
TotalSteps: totalSteps,
Status: string(step.Status),
@@ -211,6 +248,8 @@ func (m *Manager) newProgressCallback(taskID, sessionID string) PrepareProgressC
Error: step.Error,
Warning: step.Warning,
WarningDetail: step.WarningDetail,
StartedAt: step.StartedAt,
EndedAt: step.EndedAt,
})
}
}
@@ -310,6 +349,7 @@ func (m *Manager) runEnvironmentPreparer(
return nil
}
repoSetupScript, _ := req.Metadata[MetadataKeyRepoSetupScript].(string)
prepReq := &EnvPrepareRequest{
TaskID: req.TaskID,
SessionID: req.SessionID,
@@ -321,6 +361,7 @@ func (m *Manager) runEnvironmentPreparer(
UseWorktree: req.UseWorktree,
WorktreeID: req.WorktreeID,
SetupScript: req.SetupScript,
RepoSetupScript: repoSetupScript,
BaseBranch: req.BaseBranch,
CheckoutBranch: req.CheckoutBranch,
WorktreeBranchPrefix: req.WorktreeBranchPrefix,
@@ -418,8 +459,15 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
workspacePath, mainRepoGitDir, worktreeID, worktreeBranch := m.launchResolveWorkspacePath(ctx, req)
// 4b. Run environment preparation (if preparer registered for this executor type).
// For worktree executors, the preparer creates the worktree and returns workspace metadata.
prepResult := m.runEnvironmentPreparer(ctx, req, workspacePath)
// Skip on resume (ACPSessionID set) — workspace was already prepared during initial launch.
var prepResult *EnvPrepareResult
if req.ACPSessionID == "" {
prepResult = m.runEnvironmentPreparer(ctx, req, workspacePath)
} 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 {
if err := m.launchApplyPrepareResult(req, prepResult, &workspacePath, &mainRepoGitDir, &worktreeID, &worktreeBranch); err != nil {
return nil, err
@@ -441,8 +489,8 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
return nil, err
}
// Publish PrepareCompleted for CreateInstance only if no preparer ran
// (preparers publish their own completion above).
if prepResult == nil {
// AND this is not a resume (resume already has persisted prepare_result).
if prepResult == nil && req.ACPSessionID == "" {
m.eventPublisher.PublishPrepareCompleted(req.SessionID, &PrepareCompletedEventPayload{
TaskID: req.TaskID,
SessionID: req.SessionID,
@@ -458,6 +506,8 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
if req.ACPSessionID != "" {
execution.ACPSessionID = req.ACPSessionID
}
// Carry prepare result back to caller for synchronous persistence.
execution.PrepareResult = prepResult
if req.PreviousExecutionID != "" {
execution.isResumedSession = true
}
@@ -1,12 +1,14 @@
package lifecycle
import (
"context"
"strings"
"testing"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/agent/agents"
"github.com/kandev/kandev/internal/agent/executor"
)
// resumeTestAgent is a minimal agent with a BuildCommand that respects the
@@ -71,3 +73,52 @@ func TestBuildAgentCommand_ResumeFlag(t *testing.T) {
require.Contains(t, cmds.initial, "sess-456")
})
}
// trackingPreparer records whether Prepare was called.
type trackingPreparer struct {
called bool
}
func (p *trackingPreparer) Name() string { return "tracking" }
func (p *trackingPreparer) Prepare(_ context.Context, _ *EnvPrepareRequest, _ PrepareProgressCallback) (*EnvPrepareResult, error) {
p.called = true
return &EnvPrepareResult{Success: true, WorkspacePath: "/tmp/ws"}, nil
}
func TestRunEnvironmentPreparer_CalledOnFreshLaunch(t *testing.T) {
mgr := newTestManager()
preparer := &trackingPreparer{}
mgr.preparerRegistry = NewPreparerRegistry(mgr.logger)
mgr.preparerRegistry.Register(executor.NameStandalone, preparer)
req := &LaunchRequest{
TaskID: "task-1",
SessionID: "session-1",
ExecutorType: string(executor.NameStandalone),
RepositoryPath: "/tmp/repo",
}
result := mgr.runEnvironmentPreparer(context.Background(), req, "/tmp/repo")
require.True(t, preparer.called, "preparer should be called on fresh launch")
require.NotNil(t, result)
require.True(t, result.Success)
}
func TestRunEnvironmentPreparer_SkippedWithoutRepoPath(t *testing.T) {
mgr := newTestManager()
preparer := &trackingPreparer{}
mgr.preparerRegistry = NewPreparerRegistry(mgr.logger)
mgr.preparerRegistry.Register(executor.NameStandalone, preparer)
req := &LaunchRequest{
TaskID: "task-1",
SessionID: "session-1",
ExecutorType: string(executor.NameStandalone),
// No RepositoryPath — preparer should be skipped
}
result := mgr.runEnvironmentPreparer(context.Background(), req, "")
require.False(t, preparer.called, "preparer should be skipped when no repository path")
require.Nil(t, result)
}
@@ -2,14 +2,21 @@ package lifecycle
import (
"bytes"
"context"
"os/exec"
"path/filepath"
"strings"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/executor"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/scriptengine"
)
// prepareScriptLogger is a package-level logger for setup script resolution.
var prepareScriptLogger = logger.Default().WithFields(zap.String("component", "prepare-script"))
func resolvePreparerSetupScript(req *EnvPrepareRequest, workspacePath string) string {
script := strings.TrimSpace(req.SetupScript)
if script == "" {
@@ -20,8 +27,9 @@ func resolvePreparerSetupScript(req *EnvPrepareRequest, workspacePath string) st
}
metadata := map[string]any{
MetadataKeyRepositoryPath: req.RepositoryPath,
MetadataKeyBaseBranch: req.BaseBranch,
MetadataKeyRepositoryPath: req.RepositoryPath,
MetadataKeyBaseBranch: req.BaseBranch,
MetadataKeyRepoSetupScript: req.RepoSetupScript,
}
if req.WorktreeID != "" {
metadata[MetadataKeyWorktreeID] = req.WorktreeID
@@ -46,7 +54,30 @@ func resolvePreparerSetupScript(req *EnvPrepareRequest, workspacePath string) st
req.BaseBranch,
))
return resolver.Resolve(script)
resolved := resolver.Resolve(script)
if isScriptEffectivelyEmpty(resolved) {
prepareScriptLogger.Warn("setup script is comment-only after resolution, skipping",
zap.String("task_id", req.TaskID),
zap.String("executor_type", string(req.ExecutorType)),
zap.Bool("use_worktree", req.UseWorktree),
zap.Bool("has_explicit_script", strings.TrimSpace(req.SetupScript) != ""))
return ""
}
return resolved
}
// isScriptEffectivelyEmpty returns true when a script contains only a shebang,
// comments, and/or blank lines — i.e. no executable commands. This avoids
// running (and displaying) the default template scripts that are pure comments.
func isScriptEffectivelyEmpty(script string) bool {
for _, line := range strings.Split(script, "\n") {
trimmed := strings.TrimSpace(line)
if trimmed == "" || strings.HasPrefix(trimmed, "#") {
continue
}
return false
}
return true
}
func defaultPreparerSetupScript(req *EnvPrepareRequest) string {
@@ -71,3 +102,37 @@ func getGitRemoteURL(repoPath string) (string, error) {
}
return strings.TrimSpace(out.String()), nil
}
// runSetupScriptStep executes the resolved setup script as a named prepare step,
// appends the completed step to steps, and returns the updated slice.
// Setup script failures are non-fatal: the error is recorded in the step and
// logged as a warning, but preparation continues.
func runSetupScriptStep(
ctx context.Context,
req *EnvPrepareRequest,
workspacePath, resolvedScript string,
stepIdx, totalSteps int,
onProgress PrepareProgressCallback,
steps []PrepareStep,
log *logger.Logger,
) []PrepareStep {
step := beginStep("Run setup script")
step.Command = setupScriptDisplayCommand(req)
reportProgress(onProgress, step, stepIdx, totalSteps)
output, err := runSetupScript(ctx, resolvedScript, workspacePath, req.Env, func(current string) {
step.Output = current
reportProgress(onProgress, step, stepIdx, totalSteps)
})
step.Output = output
if err != nil {
completeStepError(&step, err.Error())
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
log.Warn("setup script failed", zap.String("task_id", req.TaskID), zap.Error(err))
} else {
completeStepSuccess(&step)
steps = append(steps, step)
reportProgress(onProgress, step, stepIdx, totalSteps)
}
return steps
}
@@ -7,21 +7,71 @@ import (
"github.com/kandev/kandev/internal/agent/executor"
)
func TestResolvePreparerSetupScript_LocalFallback(t *testing.T) {
func TestResolvePreparerSetupScript_LocalFallbackCommentOnly(t *testing.T) {
req := &EnvPrepareRequest{
ExecutorType: executor.NameStandalone,
RepositoryPath: "/tmp/my-repo",
}
got := resolvePreparerSetupScript(req, "/tmp/my-repo")
if got != "" {
t.Fatalf("expected comment-only default script to be treated as empty, got %q", got)
}
}
func TestIsScriptEffectivelyEmpty(t *testing.T) {
tests := []struct {
name string
input string
want bool
}{
{"empty", "", true},
{"shebang only", "#!/bin/bash\n", true},
{"shebang and comments", "#!/bin/bash\n# comment\n# another\n", true},
{"blank lines and comments", "\n# comment\n\n# more\n\n", true},
{"has command", "#!/bin/bash\necho hello\n", false},
{"command after comments", "#!/bin/bash\n# setup\napt-get install git\n", false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := isScriptEffectivelyEmpty(tt.input)
if got != tt.want {
t.Fatalf("isScriptEffectivelyEmpty(%q) = %v, want %v", tt.input, got, tt.want)
}
})
}
}
func TestResolvePreparerSetupScript_LocalWithRepoSetupScript(t *testing.T) {
req := &EnvPrepareRequest{
ExecutorType: executor.NameStandalone,
RepositoryPath: "/tmp/my-repo",
RepoSetupScript: "make install",
}
got := resolvePreparerSetupScript(req, "/tmp/my-repo")
if got == "" {
t.Fatal("expected default local setup script")
t.Fatal("expected non-empty script when repo setup script is set")
}
if strings.Contains(got, "{{repository.path}}") {
t.Fatalf("expected repository.path placeholder to be resolved, got %q", got)
if !strings.Contains(got, "make install") {
t.Fatalf("expected repo setup script in resolved output, got %q", got)
}
if !strings.Contains(got, "set to /tmp/my-repo.") {
t.Fatalf("expected resolved workspace path in script comment, got %q", got)
}
func TestResolvePreparerSetupScript_WorktreeWithRepoSetupScript(t *testing.T) {
req := &EnvPrepareRequest{
ExecutorType: executor.NameStandalone,
UseWorktree: true,
RepositoryPath: "/tmp/my-repo",
RepoSetupScript: "npm ci",
}
got := resolvePreparerSetupScript(req, "/tmp/worktrees/wt-1")
if got == "" {
t.Fatal("expected non-empty script when repo setup script is set")
}
if !strings.Contains(got, "npm ci") {
t.Fatalf("expected repo setup script in resolved output, got %q", got)
}
}
@@ -40,6 +40,10 @@ type AgentExecution struct {
ErrorMessage string
Metadata map[string]interface{}
// PrepareResult carries the environment preparation result back to the caller
// so it can be persisted synchronously before UpdateTaskSession clobbers metadata.
PrepareResult *EnvPrepareResult `json:"-"`
// agentctl client for this execution
agentctl *agentctl.Client
@@ -18,6 +18,7 @@ const (
agentEventCompleted = "completed"
agentEventError = "error"
agentEventToolCall = "tool_call"
agentEventFailed = "failed"
)
// buildTaskEventPayload builds the standard map payload for TaskUpdated events.
@@ -282,29 +282,18 @@ func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.C
"efficiency": data.ContextEfficiency,
}
// Persist to database asynchronously. Read the session inside the goroutine
// to get the latest metadata (avoids race with setSessionPlanMode etc.).
// Persist to database asynchronously using json_set to atomically set one
// key without clobbering other metadata keys (plan_mode, prepare_result).
go func() {
session, err := s.repo.GetTaskSession(context.Background(), data.TaskSessionID)
if err != nil {
s.logger.Debug("no task session for context window update",
zap.String("session_id", data.TaskSessionID),
zap.Error(err))
return
}
if session.Metadata == nil {
session.Metadata = make(map[string]interface{})
}
session.Metadata["context_window"] = contextWindowData
if err := s.repo.UpdateSessionMetadata(context.Background(), session.ID, session.Metadata); err != nil {
if err := s.repo.SetSessionMetadataKey(context.Background(), data.TaskSessionID, "context_window", contextWindowData); err != nil {
s.logger.Error("failed to update session with context window",
zap.String("task_id", data.TaskID),
zap.String("session_id", session.ID),
zap.String("session_id", data.TaskSessionID),
zap.Error(err))
} else {
s.logger.Debug("persisted context window to session",
zap.String("task_id", data.TaskID),
zap.String("session_id", session.ID))
zap.String("session_id", data.TaskSessionID))
}
}()
@@ -0,0 +1,48 @@
package orchestrator
import (
"context"
"time"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
)
// subscribePrepareEvents subscribes to environment preparation events.
func (s *Service) subscribePrepareEvents() {
if s.eventBus == nil {
return
}
if _, err := s.eventBus.Subscribe(events.ExecutorPrepareCompleted, s.handlePrepareCompleted); err != nil {
s.logger.Error("failed to subscribe to executor.prepare.completed events", zap.Error(err))
}
}
// handlePrepareCompleted persists prepare_result in session metadata using
// json_set to atomically set one key without clobbering others.
func (s *Service) handlePrepareCompleted(_ context.Context, event *bus.Event) error {
payload, ok := event.Data.(*lifecycle.PrepareCompletedEventPayload)
if !ok {
return nil
}
pr := lifecycle.SerializePrepareResult(&lifecycle.EnvPrepareResult{
Success: payload.Success,
Steps: payload.Steps,
ErrorMessage: payload.ErrorMessage,
Duration: time.Duration(payload.DurationMs) * time.Millisecond,
})
if err := s.repo.SetSessionMetadataKey(context.Background(), payload.SessionID, "prepare_result", pr); err != nil {
s.logger.Warn("failed to persist prepare_result",
zap.String("session_id", payload.SessionID), zap.Error(err))
}
s.logger.Info("environment preparation completed",
zap.String("session_id", payload.SessionID),
zap.Bool("success", payload.Success),
zap.Int("steps", len(payload.Steps)))
return nil
}
@@ -0,0 +1,82 @@
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"
)
func TestHandlePrepareCompleted_PersistsViaJsonSet(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "task-1", "sess-1", "step1")
svc := &Service{logger: testLogger(), repo: repo}
event := &bus.Event{
Type: "executor.prepare.completed",
Data: &lifecycle.PrepareCompletedEventPayload{
SessionID: "sess-1",
Success: true,
Steps: []lifecycle.PrepareStep{{Name: "step1", Status: lifecycle.PrepareStepCompleted}},
},
Timestamp: time.Now(),
}
err := svc.handlePrepareCompleted(context.Background(), event)
require.NoError(t, err)
session, err := repo.GetTaskSession(context.Background(), "sess-1")
require.NoError(t, err)
require.NotNil(t, session.Metadata["prepare_result"], "prepare_result should be persisted")
}
func TestHandlePrepareCompleted_PersistsFailure(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "task-1", "sess-1", "step1")
svc := &Service{logger: testLogger(), repo: repo}
event := &bus.Event{
Type: "executor.prepare.completed",
Data: &lifecycle.PrepareCompletedEventPayload{
SessionID: "sess-1",
Success: false,
ErrorMessage: "setup script exited with code 1",
Steps: []lifecycle.PrepareStep{
{Name: "step1", Status: lifecycle.PrepareStepCompleted},
{Name: "step2", Status: lifecycle.PrepareStepFailed, Output: "npm ERR! missing"},
},
},
Timestamp: time.Now(),
}
err := svc.handlePrepareCompleted(context.Background(), event)
require.NoError(t, err)
session, err := repo.GetTaskSession(context.Background(), "sess-1")
require.NoError(t, err)
pr, ok := session.Metadata["prepare_result"].(map[string]interface{})
require.True(t, ok, "prepare_result should be a map")
require.Equal(t, "failed", pr["status"])
require.Equal(t, "setup script exited with code 1", pr["error_message"])
steps, ok := pr["steps"].([]interface{})
require.True(t, ok)
require.Len(t, steps, 2)
}
func TestHandlePrepareCompleted_HandlesWrongType(t *testing.T) {
svc := &Service{logger: testLogger()}
event := &bus.Event{
Type: "executor.prepare.completed",
Data: "wrong type",
Timestamp: time.Now(),
}
err := svc.handlePrepareCompleted(context.Background(), event)
require.NoError(t, err)
}
@@ -311,7 +311,7 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.
// Handle all status updates (running, complete, error, cancelled, pending, in_progress)
switch payload.Data.ToolStatus {
case "running", agentEventComplete, agentEventCompleted, "success", agentEventError, "failed", "cancelled", "pending", "in_progress":
case "running", agentEventComplete, agentEventCompleted, "success", agentEventError, agentEventFailed, "cancelled", "pending", "in_progress":
if err := s.messageCreator.UpdateToolCallMessage(
ctx,
payload.TaskID,
@@ -334,7 +334,7 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.
// Update session state for completion events
// Allow tool completions to wake session from WAITING_FOR_INPUT
if payload.Data.ToolStatus == agentEventComplete || payload.Data.ToolStatus == agentEventCompleted ||
payload.Data.ToolStatus == "success" || payload.Data.ToolStatus == agentEventError || payload.Data.ToolStatus == "failed" {
payload.Data.ToolStatus == "success" || payload.Data.ToolStatus == agentEventError || payload.Data.ToolStatus == agentEventFailed {
s.updateTaskSessionState(ctx, payload.TaskID, payload.SessionID, models.TaskSessionStateRunning, "", true)
}
}
@@ -403,7 +403,9 @@ func (s *Service) finalizeStepEnter(ctx context.Context, taskID, sessionID strin
// resolveStepPlanMode determines whether plan mode should be active for a step.
// Returns false for passthrough sessions, steps without enable_plan_mode, or when the agent
// doesn't support MCP. Also clears plan mode on the session when entering a non-plan-mode step.
// doesn't support MCP. Plan mode is only cleared by explicit on_exit/on_turn_complete
// disable_plan_mode actions, not automatically when entering a non-plan-mode step.
// This preserves user-initiated plan mode across workflow transitions.
func (s *Service) resolveStepPlanMode(ctx context.Context, session *models.TaskSession, step *wfmodels.WorkflowStep, isPassthrough bool) bool {
hasPlanMode := step.HasOnEnterAction(wfmodels.OnEnterEnablePlanMode)
@@ -415,10 +417,6 @@ func (s *Service) resolveStepPlanMode(ctx context.Context, session *models.TaskS
hasPlanMode = false
}
// For ACP sessions, auto-disable plan mode when entering a step that doesn't explicitly enable it.
if !isPassthrough && !hasPlanMode {
s.clearSessionPlanMode(ctx, session)
}
return hasPlanMode
}
@@ -543,7 +541,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
if len(step.Events.OnEnter) == 0 {
s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
return
}
@@ -552,7 +550,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
if step.HasOnEnterAction(wfmodels.OnEnterResetAgentContext) {
if !s.resetAgentContext(ctx, taskID, session, step.Name) {
s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
return
}
}
@@ -584,7 +582,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
zap.String("step_name", step.Name),
zap.Error(err))
s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
}
case hasAutoStart:
@@ -605,7 +603,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
default:
s.setSessionWaitingForInput(ctx, taskID, sessionID, session)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID)
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
}
}
@@ -805,15 +803,11 @@ func (s *Service) resetAgentContext(ctx context.Context, taskID string, session
return false
}
// Clear the stored ACP session ID in the database so future resumes use session/new.
// Use targeted metadata update to avoid overwriting other fields.
if session.Metadata != nil {
delete(session.Metadata, "acp_session_id")
if updateErr := s.repo.UpdateSessionMetadata(ctx, sessionID, session.Metadata); updateErr != nil {
s.logger.Warn("failed to clear ACP session ID from session metadata",
zap.String("session_id", sessionID),
zap.Error(updateErr))
}
// Clear the stored ACP session ID using json_set to avoid clobbering other keys.
if updateErr := s.repo.SetSessionMetadataKey(ctx, sessionID, "acp_session_id", ""); updateErr != nil {
s.logger.Warn("failed to clear ACP session ID from session metadata",
zap.String("session_id", sessionID),
zap.Error(updateErr))
}
return true
}
@@ -866,17 +860,17 @@ func (s *Service) clearSessionPlanMode(ctx context.Context, session *models.Task
// setSessionPlanMode sets or clears plan mode in session metadata.
// Uses targeted metadata update to avoid overwriting other session fields.
func (s *Service) setSessionPlanMode(ctx context.Context, session *models.TaskSession, enabled bool) {
// Update in-memory struct for callers that read session.Metadata.
if session.Metadata == nil {
session.Metadata = make(map[string]interface{})
}
if enabled {
session.Metadata["plan_mode"] = true
} else {
delete(session.Metadata, "plan_mode")
}
if err := s.repo.UpdateSessionMetadata(ctx, session.ID, session.Metadata); err != nil {
// Persist using json_set to atomically set one key without clobbering others.
if err := s.repo.SetSessionMetadataKey(ctx, session.ID, "plan_mode", enabled); err != nil {
s.logger.Warn("failed to update session plan mode",
zap.String("session_id", session.ID),
zap.Bool("enabled", enabled),
@@ -906,7 +900,9 @@ func (s *Service) processTurnCompleteActions(ctx context.Context, session *model
}
// publishSessionWaitingEvent publishes a session state change event for WAITING_FOR_INPUT.
func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessionID, stepID string) {
// An optional preloaded session avoids re-reading from DB (which can miss recent writes
// on the read-only WAL connection).
func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessionID, stepID string, preloadedSession ...*models.TaskSession) {
if s.eventBus == nil {
return
}
@@ -918,7 +914,13 @@ func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessio
}
// Include agent_profile_id and session metadata so the frontend can
// identify the agent (e.g. MCP support) without waiting for SSR hydration.
if session, err := s.repo.GetTaskSession(ctx, sessionID); err == nil {
var session *models.TaskSession
if len(preloadedSession) > 0 && preloadedSession[0] != nil {
session = preloadedSession[0]
} else if s, err := s.repo.GetTaskSession(ctx, sessionID); err == nil {
session = s
}
if session != nil {
if session.AgentProfileID != "" {
eventData["agent_profile_id"] = session.AgentProfileID
}
@@ -161,8 +161,8 @@ func TestHandleTaskMovedWithSession(t *testing.T) {
// Set plan_mode on the session to verify on_exit clears it
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
stepGetter := newMockStepGetter()
stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
@@ -350,8 +350,8 @@ func TestHandleTaskMovedWithSession(t *testing.T) {
// Set agent execution ID on the session
session, _ := repo.GetTaskSession(ctx, "s1")
session.AgentExecutionID = "exec-123"
session.Metadata = map[string]interface{}{"acp_session_id": "old-acp"}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"acp_session_id": "old-acp"})
stepGetter := newMockStepGetter()
stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
@@ -387,7 +387,7 @@ func TestHandleTaskMovedWithSession(t *testing.T) {
// Verify acp_session_id was cleared
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasACP := updated.Metadata["acp_session_id"]; hasACP {
if acp, _ := updated.Metadata["acp_session_id"].(string); acp != "" {
t.Error("expected acp_session_id to be cleared from session metadata")
}
}
@@ -403,8 +403,8 @@ func TestProcessStepExitAndEnter(t *testing.T) {
// Set plan_mode on session — on_exit should clear it
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
stepGetter := newMockStepGetter()
stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
@@ -426,7 +426,7 @@ func TestProcessStepExitAndEnter(t *testing.T) {
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasPlanMode := updated.Metadata["plan_mode"]; hasPlanMode {
if pm, _ := updated.Metadata["plan_mode"].(bool); pm {
t.Error("expected plan_mode to be cleared by on_exit")
}
}
@@ -62,7 +62,8 @@ func TestPublishSessionWaitingEvent(t *testing.T) {
t.Fatalf("failed to get session: %v", err)
}
session.AgentProfileID = "profile-auggie"
session.Metadata = map[string]any{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]any{"plan_mode": true})
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("failed to update session: %v", err)
}
@@ -187,8 +187,8 @@ func TestProcessOnTurnComplete(t *testing.T) {
// Set plan_mode in session metadata
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
stepGetter := newMockStepGetter()
stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
@@ -217,7 +217,7 @@ func TestProcessOnTurnComplete(t *testing.T) {
// Verify plan_mode was cleared
updatedSession, _ := repo.GetTaskSession(ctx, "s1")
if updatedSession.Metadata != nil {
if _, hasPlanMode := updatedSession.Metadata["plan_mode"]; hasPlanMode {
if pm, _ := updatedSession.Metadata["plan_mode"].(bool); pm {
t.Error("expected plan_mode to be cleared from session metadata")
}
}
@@ -326,14 +326,14 @@ func TestProcessOnEnter(t *testing.T) {
}
})
t.Run("no plan mode clears it", func(t *testing.T) {
t.Run("plan mode persists when entering step without enable_plan_mode", func(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1")
// Set plan_mode in session metadata
// Set plan_mode in session metadata (simulates user-initiated plan mode)
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
taskRepo := newMockTaskRepo()
svc := createTestService(repo, newMockStepGetter(), taskRepo)
@@ -346,11 +346,12 @@ func TestProcessOnEnter(t *testing.T) {
session, _ = repo.GetTaskSession(ctx, "s1")
svc.processOnEnter(ctx, "t1", session, step, "test task")
// Plan mode should persist — only explicit on_exit/on_turn_complete
// disable_plan_mode actions should clear it.
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasPlanMode := updated.Metadata["plan_mode"]; hasPlanMode {
t.Error("expected plan_mode to be cleared from session metadata")
}
pm, _ := updated.Metadata["plan_mode"].(bool)
if !pm {
t.Error("expected plan_mode to persist in session metadata")
}
})
@@ -422,8 +423,8 @@ func TestSetSessionPlanMode(t *testing.T) {
// First enable
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
session, _ = repo.GetTaskSession(ctx, "s1")
@@ -431,7 +432,7 @@ func TestSetSessionPlanMode(t *testing.T) {
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasPlanMode := updated.Metadata["plan_mode"]; hasPlanMode {
if pm, _ := updated.Metadata["plan_mode"].(bool); pm {
t.Error("expected plan_mode to be removed from metadata")
}
}
@@ -480,8 +481,8 @@ func TestProcessOnExit(t *testing.T) {
// Set plan_mode in session metadata
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
@@ -498,7 +499,7 @@ func TestProcessOnExit(t *testing.T) {
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasPlanMode := updated.Metadata["plan_mode"]; hasPlanMode {
if pm, _ := updated.Metadata["plan_mode"].(bool); pm {
t.Error("expected plan_mode to be cleared from session metadata")
}
}
@@ -510,8 +511,8 @@ func TestProcessOnExit(t *testing.T) {
// Set plan_mode in session metadata
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isPassthrough: true})
@@ -560,7 +561,7 @@ func TestProcessOnEnterPassthrough(t *testing.T) {
session, _ = repo.GetTaskSession(ctx, "s1")
if session.Metadata != nil {
if _, hasPlanMode := session.Metadata["plan_mode"]; hasPlanMode {
if pm, _ := session.Metadata["plan_mode"].(bool); pm {
t.Error("expected plan_mode NOT to be set for passthrough session")
}
}
@@ -572,8 +573,8 @@ func TestProcessOnEnterPassthrough(t *testing.T) {
// Set plan_mode in session metadata
session, _ := repo.GetTaskSession(ctx, "s1")
session.Metadata = map[string]interface{}{"plan_mode": true}
_ = repo.UpdateTaskSession(ctx, session)
_ = repo.UpdateSessionMetadata(ctx, session.ID, map[string]interface{}{"plan_mode": true})
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{isPassthrough: true})
@@ -635,7 +636,7 @@ func TestProcessOnEnterResetAgentContext(t *testing.T) {
// Verify acp_session_id was cleared from session metadata
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.Metadata != nil {
if _, hasACP := updated.Metadata["acp_session_id"]; hasACP {
if acp, _ := updated.Metadata["acp_session_id"].(string); acp != "" {
t.Error("expected acp_session_id to be cleared from session metadata")
}
}
@@ -7,6 +7,7 @@ import (
"sync"
"time"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/agentctl/client"
"github.com/kandev/kandev/internal/agentctl/types/streams"
"github.com/kandev/kandev/internal/common/logger"
@@ -279,6 +280,7 @@ type LaunchAgentResponse struct {
WorktreePath string
WorktreeBranch string
Metadata map[string]interface{}
PrepareResult *lifecycle.EnvPrepareResult `json:"-"` // Carried from lifecycle.Launch for synchronous persistence
}
// TaskExecution tracks an active task execution
@@ -294,6 +296,8 @@ type TaskExecution struct {
// Worktree info for the agent
WorktreePath string
WorktreeBranch string
// PrepareResult carries the env preparation result for deferred persistence
PrepareResult *lifecycle.EnvPrepareResult
}
// FromTaskSession converts a models.TaskSession to TaskExecution
@@ -455,6 +455,7 @@ func (e *Executor) finalizeLaunch(ctx context.Context, task *v1.Task, session *m
SessionID: sessionID,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
PrepareResult: resp.PrepareResult,
}
if startAgent {
@@ -478,6 +478,9 @@ func (m *mockRepository) UpdateSessionReviewStatus(ctx context.Context, sessionI
func (m *mockRepository) UpdateSessionMetadata(ctx context.Context, sessionID string, metadata map[string]interface{}) error {
return nil
}
func (m *mockRepository) SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error {
return nil
}
// Task Session Worktree operations
func (m *mockRepository) ListTaskSessionWorktrees(ctx context.Context, sessionID string) ([]*models.TaskSessionWorktree, error) {
@@ -166,6 +166,16 @@ func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID str
session.ErrorMessage = ""
session.UpdatedAt = now
// Merge prepare result into session metadata synchronously so it survives
// the UpdateTaskSession write (which would otherwise clobber it if the async
// handlePrepareCompleted event handler hasn't run yet).
if resp.PrepareResult != nil && resp.PrepareResult.Success {
if session.Metadata == nil {
session.Metadata = make(map[string]interface{})
}
session.Metadata["prepare_result"] = buildPrepareResultMetadata(resp.PrepareResult)
}
if err := e.repo.UpdateTaskSession(ctx, session); err != nil {
e.logger.Error("failed to update agent session after launch",
zap.String("task_id", taskID),
@@ -182,8 +192,12 @@ func (e *Executor) persistLaunchState(ctx context.Context, taskID, sessionID str
}
}
// persistWorktreeAssociation creates a TaskSessionWorktree record if the response contains
// a new worktree not already tracked by the session.
// buildPrepareResultMetadata serializes a prepare result for storage in session metadata.
// Uses lifecycle.SerializePrepareResult which is shared with the event handler.
func buildPrepareResultMetadata(result *lifecycle.EnvPrepareResult) map[string]interface{} {
return lifecycle.SerializePrepareResult(result)
}
func (e *Executor) persistWorktreeAssociation(ctx context.Context, taskID string, session *models.TaskSession, repositoryID string, resp *LaunchAgentResponse) {
if resp.WorktreeID == "" {
return
@@ -134,6 +134,7 @@ type sessionExecutorStore interface {
ClearSessionExecutionID(ctx context.Context, id string) error
UpdateSessionReviewStatus(ctx context.Context, sessionID string, status string) error
UpdateSessionMetadata(ctx context.Context, sessionID string, metadata map[string]interface{}) error
SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error
// Executor running state
ListExecutorsRunning(ctx context.Context) ([]*models.ExecutorRunning, error)
UpsertExecutorRunning(ctx context.Context, running *models.ExecutorRunning) error
@@ -543,6 +544,9 @@ func (s *Service) Start(ctx context.Context) error {
// Subscribe to clarification events (cancel-and-resume flow)
s.subscribeClarificationEvents()
// Subscribe to prepare events (persist result in session metadata)
s.subscribePrepareEvents()
s.logger.Info("orchestrator service started successfully")
return nil
}
@@ -452,6 +452,17 @@ func (s *Service) postLaunchStart(ctx context.Context, taskID string, execution
s.setSessionPlanMode(ctx, session, true)
}
}
// Persist prepare_result using SetSessionMetadataKey (json_set) which
// atomically sets ONE key without touching others. This avoids the
// read-modify-write race where UpdateSessionMetadata clobbers plan_mode.
if execution.PrepareResult != nil && execution.PrepareResult.Success {
pr := lifecycle.SerializePrepareResult(execution.PrepareResult)
if err := s.repo.SetSessionMetadataKey(ctx, execution.SessionID, "prepare_result", pr); err != nil {
s.logger.Warn("failed to persist prepare_result",
zap.String("session_id", execution.SessionID), zap.Error(err))
}
}
}
go s.ensureSessionPRWatch(context.Background(), taskID, execution.SessionID, execution.WorktreeBranch)
}
@@ -127,14 +127,11 @@ func (s *workflowStore) ApplyTransition(ctx context.Context, taskID, sessionID,
}
func (s *workflowStore) PersistData(ctx context.Context, sessionID string, data map[string]any) error {
// Read existing workflow_data to merge new keys into it.
session, err := s.repo.GetTaskSession(ctx, sessionID)
if err != nil {
return fmt.Errorf("load session for data persist: %w", err)
}
if session.Metadata == nil {
session.Metadata = make(map[string]interface{})
}
existing, _ := session.Metadata["workflow_data"].(map[string]interface{})
if existing == nil {
existing = make(map[string]interface{})
@@ -142,9 +139,9 @@ func (s *workflowStore) PersistData(ctx context.Context, sessionID string, data
for k, v := range data {
existing[k] = v
}
session.Metadata["workflow_data"] = existing
if err := s.repo.UpdateSessionMetadata(ctx, sessionID, session.Metadata); err != nil {
// Use SetSessionMetadataKey (json_set) to atomically set workflow_data
// without clobbering other metadata keys (plan_mode, prepare_result).
if err := s.repo.SetSessionMetadataKey(ctx, sessionID, "workflow_data", existing); err != nil {
return fmt.Errorf("persist workflow data: %w", err)
}
return nil
@@ -404,6 +404,9 @@ func (m *mockRepository) UpdateSessionReviewStatus(ctx context.Context, sessionI
func (m *mockRepository) UpdateSessionMetadata(ctx context.Context, sessionID string, metadata map[string]interface{}) error {
return nil
}
func (m *mockRepository) SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error {
return nil
}
// Task Plan operations
func (m *mockRepository) CreateTaskPlan(ctx context.Context, plan *models.TaskPlan) error {
@@ -109,6 +109,7 @@ type SessionRepository interface {
SetSessionPrimary(ctx context.Context, sessionID string) error
UpdateSessionReviewStatus(ctx context.Context, sessionID string, status string) error
UpdateSessionMetadata(ctx context.Context, sessionID string, metadata map[string]interface{}) error
SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error
}
// SessionWorktreeRepository handles the task session↔worktree association.
@@ -328,14 +328,13 @@ func (r *Repository) GetActiveTaskSessionByTaskID(ctx context.Context, taskID st
return r.scanTaskSession(ctx, row, fmt.Sprintf("no active agent session for task: %s", taskID))
}
// UpdateTaskSession updates an existing agent session
// UpdateTaskSession updates an existing agent session.
// Note: metadata is NOT updated here to prevent clobbering concurrent writes
// from UpdateSessionMetadata callers (e.g. setSessionPlanMode, persistPrepareResult).
// Use UpdateSessionMetadata for metadata changes.
func (r *Repository) UpdateTaskSession(ctx context.Context, session *models.TaskSession) error {
session.UpdatedAt = time.Now().UTC()
metadataJSON, err := json.Marshal(session.Metadata)
if err != nil {
return fmt.Errorf("failed to serialize agent session metadata: %w", err)
}
agentProfileSnapshotJSON, err := json.Marshal(session.AgentProfileSnapshot)
if err != nil {
return fmt.Errorf("failed to serialize agent profile snapshot: %w", err)
@@ -358,13 +357,13 @@ func (r *Repository) UpdateTaskSession(ctx context.Context, session *models.Task
agent_execution_id = ?, container_id = ?, agent_profile_id = ?, executor_id = ?, executor_profile_id = ?, environment_id = ?,
repository_id = ?, base_branch = ?, base_commit_sha = ?,
agent_profile_snapshot = ?, executor_snapshot = ?, environment_snapshot = ?, repository_snapshot = ?,
state = ?, error_message = ?, metadata = ?, completed_at = ?, updated_at = ?,
state = ?, error_message = ?, completed_at = ?, updated_at = ?,
is_primary = ?, review_status = ?, is_passthrough = ?, task_environment_id = ?
WHERE id = ?
`), session.AgentExecutionID, session.ContainerID, session.AgentProfileID, session.ExecutorID, session.ExecutorProfileID, session.EnvironmentID,
session.RepositoryID, session.BaseBranch, session.BaseCommitSHA,
string(agentProfileSnapshotJSON), string(executorSnapshotJSON), string(environmentSnapshotJSON), string(repositorySnapshotJSON),
string(session.State), session.ErrorMessage, string(metadataJSON), session.CompletedAt, session.UpdatedAt,
string(session.State), session.ErrorMessage, session.CompletedAt, session.UpdatedAt,
dialect.BoolToInt(session.IsPrimary), session.ReviewStatus,
dialect.BoolToInt(session.IsPassthrough), session.TaskEnvironmentID,
session.ID)
@@ -423,6 +422,29 @@ func (r *Repository) UpdateSessionMetadata(ctx context.Context, sessionID string
return nil
}
// SetSessionMetadataKey atomically sets a single key in the session's metadata
// using SQLite's json_set. Unlike UpdateSessionMetadata (which does a full
// replacement), this preserves all other metadata keys.
func (r *Repository) SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error {
valueJSON, err := json.Marshal(value)
if err != nil {
return fmt.Errorf("failed to serialize metadata value: %w", err)
}
now := time.Now().UTC()
path := "$." + key
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE task_sessions SET metadata = json_set(CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END, ?, json(?)), updated_at = ? WHERE id = ?
`), path, string(valueJSON), now, sessionID)
if err != nil {
return err
}
rows, _ := result.RowsAffected()
if rows == 0 {
return fmt.Errorf("agent session not found: %s", sessionID)
}
return nil
}
// ClearSessionExecutionID clears the agent_execution_id for a session.
// This is used when a stale execution ID needs to be removed (e.g., after a failed resume on startup).
func (r *Repository) ClearSessionExecutionID(ctx context.Context, id string) error {
+313 -164
View File
@@ -6,16 +6,46 @@ import {
IconX,
IconLoader2,
IconAlertTriangle,
IconChevronDown,
IconChevronRight,
IconTerminal2,
IconFolder,
IconGitBranch,
} from "@tabler/icons-react";
import { useAppStore } from "@/components/state-provider";
import type { PrepareStepInfo } from "@/lib/state/slices/session-runtime/types";
import { ExpandableRow } from "@/components/task/chat/messages/expandable-row";
import { cn } from "@/lib/utils";
import type {
PrepareStepInfo,
SessionPrepareState,
} from "@/lib/state/slices/session-runtime/types";
const ANSI_RE = /\x1b\[[0-9;]*[a-zA-Z]/g;
function stripAnsi(s: string): string {
return s.replace(ANSI_RE, "");
}
function formatStepDuration(startedAt?: string, endedAt?: string): string | null {
if (!startedAt || !endedAt) return null;
const ms = new Date(endedAt).getTime() - new Date(startedAt).getTime();
if (ms < 0 || Number.isNaN(ms)) return null;
if (ms < 1000) return `${ms}ms`;
const secs = Math.round(ms / 1000);
if (secs < 60) return `${secs}s`;
const mins = Math.floor(secs / 60);
const remSecs = secs % 60;
return remSecs > 0 ? `${mins}m ${remSecs}s` : `${mins}m`;
}
type PrepareProgressProps = {
sessionId: string;
};
type EffectiveStatus =
| "preparing"
| "failed"
| "completed"
| "completed_with_error"
| "completed_with_warnings";
function StepIcon({ status, hasWarning }: { status: string; hasWarning?: boolean }) {
if (status === "completed" && hasWarning) {
return <IconAlertTriangle className="h-3.5 w-3.5 text-amber-500" />;
@@ -32,173 +62,294 @@ function StepIcon({ status, hasWarning }: { status: string; hasWarning?: boolean
return <div className="h-3.5 w-3.5 rounded-full border border-muted-foreground/30" />;
}
function StepRow({ step }: { step: PrepareStepInfo }) {
/** True when the command is short enough to display inline next to the step name. */
function isInlineCommand(cmd: string): boolean {
return !cmd.includes("\n") && cmd.length <= 60;
}
function StepDetails({ step, blockCommand }: { step: PrepareStepInfo; blockCommand?: string }) {
return (
<div className="border-muted-foreground/20 mt-1 ml-0.5 border-l pl-3">
{blockCommand && (
<pre className="bg-muted/50 text-muted-foreground/60 mb-1 max-h-24 overflow-auto rounded px-2 py-1 font-mono text-[10px] whitespace-pre-wrap">
{blockCommand}
</pre>
)}
{step.output && (
<pre className="text-muted-foreground/60 max-h-48 overflow-auto whitespace-pre text-xs">
{stripAnsi(step.output)}
</pre>
)}
</div>
);
}
function StepWarning({ warning, warningDetail }: { warning: string; warningDetail?: string }) {
const [detailExpanded, setDetailExpanded] = useState(false);
return (
<div className="flex items-start gap-2 text-xs">
<div className="mt-0.5 flex-shrink-0">
<StepIcon status={step.status} hasWarning={Boolean(step.warning)} />
</div>
<div className="min-w-0 flex-1">
<span className="text-muted-foreground">{step.name || "Preparing..."}</span>
{step.output && (
<pre className="text-muted-foreground/60 mt-0.5 overflow-x-auto whitespace-pre text-xs">
{step.output}
</pre>
)}
{step.warning && (
<div className="mt-0.5">
<pre className="text-amber-500 overflow-x-auto whitespace-pre text-xs">
{step.warning}
<div data-testid="prepare-warning-banner">
<span className="text-amber-500 mt-0.5 block text-xs">{warning}</span>
{warningDetail && (
<>
<button
type="button"
className="text-amber-500/70 hover:text-amber-500 mt-0.5 flex cursor-pointer items-center gap-0.5 text-xs"
onClick={() => setDetailExpanded(!detailExpanded)}
>
Details
</button>
{detailExpanded && (
<pre className="text-amber-500/60 mt-0.5 overflow-x-auto whitespace-pre text-xs">
{warningDetail}
</pre>
{step.warningDetail && (
<>
<button
type="button"
className="flex items-center gap-0.5 text-amber-500/70 hover:text-amber-500 mt-0.5 cursor-pointer"
onClick={() => setDetailExpanded((v) => !v)}
>
{detailExpanded ? (
<IconChevronDown className="h-3 w-3" />
) : (
<IconChevronRight className="h-3 w-3" />
)}
<span>Details</span>
</button>
{detailExpanded && (
<pre className="text-amber-500/70 mt-0.5 overflow-x-auto whitespace-pre text-xs">
{step.warningDetail}
</pre>
)}
</>
)}
</div>
)}
</>
)}
</div>
);
}
function StepMessages({ step }: { step: PrepareStepInfo }) {
return (
<>
{step.warning && <StepWarning warning={step.warning} warningDetail={step.warningDetail} />}
{step.error && (
<pre className="text-destructive mt-0.5 overflow-x-auto whitespace-pre text-xs">
{step.error}
</pre>
)}
</>
);
}
function StepRow({ step }: { step: PrepareStepInfo }) {
const inlineCommand = step.command && isInlineCommand(step.command) ? step.command : undefined;
const blockCommand = step.command && !isInlineCommand(step.command) ? step.command : undefined;
const hasExpandable = Boolean(step.output) || Boolean(blockCommand);
const [detailsExpanded, setDetailsExpanded] = useState(
step.status === "running" || step.status === "failed",
);
const duration = formatStepDuration(step.startedAt, step.endedAt);
const nameClass = cn(
"text-muted-foreground",
step.status === "completed" && !step.warning && "line-through text-muted-foreground/60",
);
return (
<div className="text-xs">
<div className="flex items-center gap-2">
<div className="flex-shrink-0">
<StepIcon status={step.status} hasWarning={Boolean(step.warning)} />
</div>
<span className={nameClass}>{step.name || "Preparing..."}</span>
{inlineCommand && (
<code className="text-muted-foreground/50 font-mono text-[10px]">{inlineCommand}</code>
)}
{step.error && (
<pre className="text-destructive mt-0.5 overflow-x-auto whitespace-pre text-xs">
{step.error}
</pre>
{duration && <span className="text-muted-foreground/40 text-[10px]">{duration}</span>}
{hasExpandable && (
<button
type="button"
className="text-muted-foreground/40 hover:text-muted-foreground/70 cursor-pointer text-[10px]"
onClick={() => setDetailsExpanded(!detailsExpanded)}
>
{detailsExpanded ? "[-]" : "[+]"}
</button>
)}
</div>
{/* Content below the step header, indented past the icon */}
<div className="ml-[22px]">
{detailsExpanded && <StepDetails step={step} blockCommand={blockCommand} />}
<StepMessages step={step} />
</div>
</div>
);
}
function useEffectivePrepareStatus(sessionId: string) {
// deriveStatus maps the raw prepare state + ambient signals into the effective
// UI status. Kept small/pure so the hook stays a thin Zustand selector wrapper.
type DeriveStatusInput = {
prepareStatus: string;
sessionState: string | undefined;
agentctlStatus: string | undefined;
hasFailedStep: boolean;
hasWarnings: boolean;
hasRunningStep: boolean;
};
function deriveStatus(input: DeriveStatusInput): EffectiveStatus {
const {
prepareStatus,
sessionState,
agentctlStatus,
hasFailedStep,
hasWarnings,
hasRunningStep,
} = input;
if (prepareStatus === "failed") return "failed";
if (prepareStatus === "completed") {
if (hasFailedStep) return "completed_with_error";
if (hasWarnings) return "completed_with_warnings";
return "completed";
}
// A step is still running — stay in preparing regardless of agentctl status.
if (hasRunningStep) return "preparing";
// prepareStatus === "preparing" from here on.
// Agentctl ready implies preparation succeeded — treat as completed even if
// the completed event hasn't arrived yet.
if (agentctlStatus === "ready" && !hasFailedStep && !hasWarnings) return "completed";
// If the session reached a terminal state but prepare is still "preparing",
// treat it as failed — the completed event may have been lost.
const isSessionTerminal =
sessionState === "FAILED" || sessionState === "COMPLETED" || sessionState === "CANCELLED";
return isSessionTerminal ? "failed" : "preparing";
}
function usePrepareStatus(sessionId: string) {
const prepareState = useAppStore((state) => state.prepareProgress.bySessionId[sessionId] ?? null);
const sessionState = useAppStore((state) => state.taskSessions.items[sessionId]?.state);
const agentctlStatus = useAppStore(
(state) => state.sessionAgentctl.itemsBySessionId[sessionId]?.status,
);
const profileLabel = useAppStore((state) => {
const session = state.taskSessions.items[sessionId];
if (!session?.agent_profile_id) return null;
const profile = state.agentProfiles.items.find((p) => p.id === session.agent_profile_id);
return profile?.label ?? null;
if (!prepareState) {
// No live or hydrated prepare state. If the session has moved past the
// initial state, preparation must have completed — show a minimal panel
// so it doesn't vanish on refresh.
const sessionStarted =
sessionState && sessionState !== "CREATED" && sessionState !== "STARTING";
if (sessionStarted) {
return {
status: "completed" as EffectiveStatus,
prepareState: { sessionId, status: "completed", steps: [] } as SessionPrepareState,
};
}
return { status: "preparing" as const, prepareState: null };
}
const status = deriveStatus({
prepareStatus: prepareState.status,
sessionState,
agentctlStatus,
hasFailedStep: prepareState.steps.some((s) => s.status === "failed"),
hasWarnings: prepareState.steps.some((s) => s.warning),
hasRunningStep: prepareState.steps.some((s) => s.status === "running"),
});
return { status, prepareState };
}
const hasWarnings = prepareState?.steps.some((s) => s.warning);
// Tracks whether the panel is expanded. Starts from the status-derived default
// (`autoExpand`) and flips back to that default whenever the status transitions
// or the session changes — the "new context" rule. User's manual toggles stick
// within a single status window. Prev-value pattern (not useEffect) avoids a
// cascading re-render.
function useExpandedFlag(sessionId: string, autoExpand: boolean) {
const [expanded, setExpanded] = useState(autoExpand);
const [prevSessionId, setPrevSessionId] = useState(sessionId);
const [prevAutoExpand, setPrevAutoExpand] = useState(autoExpand);
if (sessionId !== prevSessionId) {
setPrevSessionId(sessionId);
setExpanded(autoExpand);
} else if (autoExpand !== prevAutoExpand) {
setPrevAutoExpand(autoExpand);
setExpanded(autoExpand);
}
return [expanded, setExpanded] as const;
}
if (!prepareState)
return {
visible: false,
status: "preparing",
prepareState,
profileLabel,
hasWarnings: false,
} as const;
if (prepareState.status === "completed" && !hasWarnings)
return {
visible: false,
status: "completed",
prepareState,
profileLabel,
hasWarnings: false,
} as const;
if (prepareState.status === "completed" && hasWarnings)
return {
visible: true,
status: "completed_with_warnings",
prepareState,
profileLabel,
hasWarnings: true,
} as const;
function HeaderIcon({ status }: { status: EffectiveStatus }) {
if (status === "preparing") {
return (
<IconLoader2
data-testid="prepare-progress-header-spinner"
className="h-4 w-4 text-muted-foreground animate-spin"
/>
);
}
if (status === "failed" || status === "completed_with_error") {
return <IconX className="h-4 w-4 text-destructive" />;
}
if (status === "completed_with_warnings") {
return <IconAlertTriangle className="h-4 w-4 text-amber-500" />;
}
return <IconTerminal2 className="h-4 w-4 text-muted-foreground" />;
}
// Agentctl ready implies environment preparation succeeded —
// dismiss even if the completed event hasn't arrived yet.
if (agentctlStatus === "ready")
return { visible: false, status: "completed", prepareState, profileLabel } as const;
function getHeaderLabel(status: EffectiveStatus, prepareState: SessionPrepareState): string {
if (status === "preparing") return "Preparing environment...";
if (status === "failed") return prepareState.errorMessage || "Environment preparation failed";
if (status === "completed_with_error") {
const failed = prepareState.steps.find((s) => s.status === "failed");
// `||` (not `??`) so an empty-string error also falls through to the name fallback.
return failed?.error || `${failed?.name ?? "Environment prepared"} — step failed`;
}
if (status === "completed_with_warnings") return "Environment prepared with warnings";
return "Environment prepared";
}
// If session reached a terminal state but prepare status is still "preparing",
// treat it as failed (the completed event may not have arrived)
const isSessionTerminal =
sessionState === "FAILED" || sessionState === "COMPLETED" || sessionState === "CANCELLED";
const status =
prepareState.status === "preparing" && isSessionTerminal ? "failed" : prepareState.status;
function InfoLine({ icon, children }: { icon: React.ReactNode; children: React.ReactNode }) {
return (
<div className="text-muted-foreground/70 flex items-center gap-2 text-xs">
<div className="flex-shrink-0">{icon}</div>
<span>{children}</span>
</div>
);
}
return {
visible: true,
status,
prepareState,
profileLabel,
hasWarnings: hasWarnings ?? false,
} as const;
function Mono({ children }: { children: React.ReactNode }) {
return <code className="text-muted-foreground font-mono text-[11px]">{children}</code>;
}
function SessionInfo({ sessionId }: { sessionId: string }) {
const session = useAppStore((state) => state.taskSessions.items[sessionId]);
if (!session) return null;
const { worktree_id, worktree_path, worktree_branch, base_branch } = session;
const isWorktree = Boolean(worktree_id);
// Nothing useful to show for sessions without workspace info.
if (!worktree_path && !worktree_branch) return null;
return (
<div className="border-muted mt-2 space-y-1 border-t pt-2">
{isWorktree && worktree_path && (
<InfoLine icon={<IconFolder className="h-3 w-3" />}>
Isolated worktree at <Mono>{worktree_path}</Mono>
</InfoLine>
)}
{!isWorktree && worktree_path && (
<InfoLine icon={<IconFolder className="h-3 w-3" />}>
Working in <Mono>{worktree_path}</Mono>
</InfoLine>
)}
{worktree_branch && (
<InfoLine icon={<IconGitBranch className="h-3 w-3" />}>
{base_branch ? (
<>
Branch <Mono>{worktree_branch}</Mono>, based on <Mono>{base_branch}</Mono>
</>
) : (
<>
On branch <Mono>{worktree_branch}</Mono>
</>
)}
</InfoLine>
)}
</div>
);
}
export function PrepareProgress({ sessionId }: PrepareProgressProps) {
const { visible, status, prepareState, profileLabel } = useEffectivePrepareStatus(sessionId);
const [dismissed, setDismissed] = useState(false);
const [expandedDetails, setExpandedDetails] = useState<Record<number, boolean>>({});
const { status, prepareState } = usePrepareStatus(sessionId);
const session = useAppStore((state) => state.taskSessions.items[sessionId]);
// Only the clean-success status auto-collapses. Anything with running/error/
// warning context stays open by default so the user sees what's happening.
const autoExpand = status !== "completed";
const [expanded, setExpanded] = useExpandedFlag(sessionId, autoExpand);
if (!visible || !prepareState) return null;
// After successful completion with warnings, show only the warning steps.
if (status === "completed_with_warnings") {
if (dismissed) return null;
const warningSteps = prepareState.steps
.map((s, i) => ({ step: s, index: i }))
.filter(({ step }) => step.warning);
return (
<div data-testid="prepare-warning-banner" className="px-3 py-1.5 border-b border-border/50">
{warningSteps.map(({ step, index }) => (
<div key={index} className="text-xs text-amber-500">
<div className="flex items-center gap-1.5">
<IconAlertTriangle className="h-3 w-3 flex-shrink-0" />
<span className="flex-1">{step.warning}</span>
{step.warningDetail && (
<button
type="button"
className="flex items-center gap-0.5 text-amber-500/70 hover:text-amber-500 cursor-pointer"
onClick={() => setExpandedDetails((prev) => ({ ...prev, [index]: !prev[index] }))}
>
{expandedDetails[index] ? (
<IconChevronDown className="h-3 w-3" />
) : (
<IconChevronRight className="h-3 w-3" />
)}
<span>Details</span>
</button>
)}
<button
type="button"
className="text-amber-500/70 hover:text-amber-500 ml-1 cursor-pointer"
onClick={() => setDismissed(true)}
aria-label="Dismiss warning"
>
<IconX className="h-3 w-3" />
</button>
</div>
{step.warningDetail && expandedDetails[index] && (
<pre className="mt-1 ml-4.5 text-amber-500/70 overflow-x-auto whitespace-pre text-xs">
{step.warningDetail}
</pre>
)}
</div>
))}
</div>
);
}
if (!prepareState) return null;
const hasSessionInfo = Boolean(session?.worktree_path || session?.worktree_branch);
const visibleSteps = prepareState.steps.filter(
(step) =>
step.name.trim() !== "" ||
@@ -206,32 +357,30 @@ export function PrepareProgress({ sessionId }: PrepareProgressProps) {
Boolean(step.error) ||
Boolean(step.warning),
);
const headerLabel = getHeaderLabel(status, prepareState);
const isErrorStatus = status === "failed" || status === "completed_with_error";
const headerClass = cn("text-xs", isErrorStatus ? "text-destructive" : "text-muted-foreground");
return (
<div className="px-3 py-2 space-y-1.5 border-b border-border/50">
<div className="flex items-center gap-1.5 text-xs font-medium text-muted-foreground">
{status === "preparing" && (
<IconLoader2
data-testid="prepare-progress-header-spinner"
className="h-3.5 w-3.5 animate-spin"
/>
)}
{status === "failed" && <IconX className="h-3.5 w-3.5 text-destructive" />}
<span>
{status === "preparing" && "Preparing environment..."}
{status === "failed" && (prepareState.errorMessage ?? "Environment preparation failed")}
</span>
{profileLabel && (
<span className="text-muted-foreground/50 ml-auto font-normal">{profileLabel}</span>
)}
</div>
{visibleSteps.length > 0 && (
<div className="space-y-1 pl-1">
<div data-testid="prepare-progress-panel" data-status={status} data-expanded={expanded}>
<ExpandableRow
icon={<HeaderIcon status={status} />}
header={
<span data-testid="prepare-progress-toggle" className={headerClass}>
{headerLabel}
</span>
}
hasExpandableContent={visibleSteps.length > 0 || hasSessionInfo}
isExpanded={expanded}
onToggle={() => setExpanded(!expanded)}
>
<div className="space-y-1">
{visibleSteps.map((step, i) => (
<StepRow key={i} step={step} />
))}
</div>
)}
<SessionInfo sessionId={sessionId} />
</ExpandableRow>
</div>
);
}
@@ -5,7 +5,6 @@ import { SessionPanelContent } from "@kandev/ui/pannel-session";
import { useDockviewStore } from "@/lib/state/dockview-store";
import type { Message } from "@/lib/types/http";
import { AgentStatus } from "@/components/task/chat/messages/agent-status";
import { PrepareProgress } from "@/components/session/prepare-progress";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
import { useLazyLoadMessages } from "@/hooks/use-lazy-load-messages";
import {
@@ -254,7 +253,6 @@ export const NativeMessageList = memo(function NativeMessageList({
);
})}
{sessionId && <PrepareProgress sessionId={sessionId} />}
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{(footerActionMessages ?? []).map((msg: Message) => (
<MessageRenderer
@@ -5,6 +5,7 @@ import type { Message, TaskSessionState, TaskState } from "@/lib/types/http";
import type { RenderItem } from "@/hooks/use-processed-messages";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
import { TurnGroupMessage } from "@/components/task/chat/messages/turn-group-message";
import { PrepareProgress } from "@/components/session/prepare-progress";
export type MessageListProps = {
items: RenderItem[];
@@ -24,7 +25,8 @@ export type MessageListProps = {
};
export function getItemKey(item: RenderItem): string {
return item.type === "turn_group" ? item.id : item.message.id;
if (item.type === "turn_group" || item.type === "prepare_progress") return item.id;
return item.message.id;
}
export function getSessionRunningState(sessionState: string | null | undefined) {
@@ -105,6 +107,9 @@ export function MessageItem({
taskState?: TaskState;
onScrollToMessage: (id: string) => void;
}) {
if (item.type === "prepare_progress") {
return <PrepareProgress sessionId={item.sessionId} />;
}
if (item.type === "turn_group") {
return (
<TurnGroupMessage
@@ -6,7 +6,6 @@ import { Virtuoso, type VirtuosoHandle } from "react-virtuoso";
import { SessionPanelContent } from "@kandev/ui/pannel-session";
import type { RenderItem } from "@/hooks/use-processed-messages";
import { AgentStatus } from "@/components/task/chat/messages/agent-status";
import { PrepareProgress } from "@/components/session/prepare-progress";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
import { useLazyLoadMessages } from "@/hooks/use-lazy-load-messages";
import {
@@ -101,7 +100,8 @@ function useVirtuosoCallbacks(props: VirtuosoBodyProps) {
(messageId: string) => {
const idx = items.findIndex((item) => {
if (item.type === "turn_group") return item.messages.some((m) => m.id === messageId);
return item.message?.id === messageId;
if (item.type === "message") return item.message.id === messageId;
return false;
});
if (idx >= 0)
virtuosoRef.current?.scrollToIndex({ index: firstItemIndex + idx, align: "center" });
@@ -257,7 +257,6 @@ export const VirtuosoMessageList = memo(function VirtuosoMessageList(props: Mess
const Footer = useCallback(
() => (
<>
{sessionId && <PrepareProgress sessionId={sessionId} />}
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{footerActions.map((msg) => (
<MessageRenderer
@@ -283,7 +282,6 @@ export const VirtuosoMessageList = memo(function VirtuosoMessageList(props: Mess
isInitialLoading={isInitialLoading}
messagesCount={messages.length}
/>
{sessionId && <PrepareProgress sessionId={sessionId} />}
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{footerActions.map((msg) => (
<MessageRenderer
@@ -36,13 +36,13 @@ export const ExpandableRow = memo(function ExpandableRow({
{/* Clickable header row */}
<div
className={cn(
"flex items-start gap-3 w-full rounded px-2 py-1 -mx-2 transition-colors",
"flex items-center gap-3 w-full rounded px-2 py-1 -mx-2 transition-colors",
hasExpandableContent && "hover:bg-muted/50 cursor-pointer",
)}
onClick={handleClick}
>
{/* Icon with hover-to-show chevron */}
<div className="flex-shrink-0 mt-0.5 relative w-4 h-4">
<div className="flex-shrink-0 relative w-4 h-4">
<div
className={cn(
"absolute inset-0 transition-opacity",
@@ -60,7 +60,7 @@ export const ExpandableRow = memo(function ExpandableRow({
</div>
{/* Header content */}
<div className="flex-1 min-w-0 pt-0.5">{header}</div>
<div className="flex-1 min-w-0">{header}</div>
</div>
{/* Expanded content - outside clickable area */}
+6 -3
View File
@@ -102,10 +102,12 @@ test.describe("Diff update on file change", () => {
// The Pierre Diffs viewer should show the initial modification.
// Playwright's getByText auto-pierces shadow DOM and auto-retries, so we use it
// directly with a generous timeout to handle async web worker initialization.
// On cold CI runners (first test in shard, no V8 code cache), Pierre Diffs'
// createJavaScriptRegexEngine() can take 30-40s to JIT-compile.
const diffsContainer = getDiffsContainer(testPage);
await expect(diffsContainer).toBeVisible({ timeout: 15_000 });
await expect(diffsContainer.getByText("FIRST_MODIFICATION", { exact: true })).toBeVisible({
timeout: 15_000,
timeout: 60_000,
});
});
@@ -114,11 +116,12 @@ test.describe("Diff update on file change", () => {
await openChangesTab(testPage);
await openFileDiff(testPage, "diff_update_test.txt");
// Verify initial diff content (scoped to diffs-container to avoid matching chat text)
// Verify initial diff content (scoped to diffs-container to avoid matching chat text).
// Allow up to 60s for Pierre Diffs engine JIT on cold CI runners.
const diffsContainer = getDiffsContainer(testPage);
await expect(diffsContainer).toBeVisible({ timeout: 15_000 });
await expect(diffsContainer.getByText("FIRST_MODIFICATION", { exact: true })).toBeVisible({
timeout: 15_000,
timeout: 60_000,
});
// Click on the session tab to make the chat input visible again
@@ -1355,10 +1355,11 @@ test.describe("Git Changes Panel", () => {
await session.clickTab("Changes");
await expect(session.changes).toBeVisible({ timeout: 10_000 });
// Unified commits section should show both commits
// Unified commits section should show at least the two commits we created.
// Other tests in the same worker may have left additional commits in the shared e2e-repo.
await expect(testPage.getByTestId("commits-section")).toBeVisible({ timeout: 15_000 });
const commitsList = testPage.getByTestId("commits-list");
await expect(commitsList.locator("li")).toHaveCount(2, { timeout: 5_000 });
await expect(commitsList.locator("li").first()).toBeVisible({ timeout: 5_000 });
// Unpushed commit should have arrow-up icon (emerald)
const unpushedRow = commitsList.locator("li", { hasText: "Unpushed commit" });
@@ -0,0 +1,286 @@
import fs from "node:fs";
import path from "node:path";
import { test, expect } from "../../fixtures/test-base";
import { SessionPage } from "../../pages/session-page";
/**
* Verifies the UX for the inline prepare-progress card that renders below the
* initial user prompt:
* - While running: card is expanded and streams the setup script output.
* - On clean success: card auto-collapses (still visible as a summary row).
* - On step failure: card stays expanded so the error is immediately visible;
* user can click to collapse.
* - Without a setup script: card still renders, auto-collapses on success.
*
* Each test creates a fresh executor profile carrying the desired prepare
* script — this hits the same `runSetupScript` path users exercise via the
* repository setup_script placeholder in profile templates, without mutating
* the worker-scoped seed profile (avoids cross-test contamination).
*/
test.describe("Setup script progress UX", () => {
async function createWorktreeProfileWithScript(
apiClient: import("../../helpers/api-client").ApiClient,
script: string,
name: string,
): Promise<{ id: string }> {
const { executors } = await apiClient.listExecutors();
const worktreeExec = executors.find((e) => e.type === "worktree");
if (!worktreeExec) throw new Error("No worktree executor found");
return apiClient.createExecutorProfile(worktreeExec.id, name, { prepare_script: script });
}
test("streams setup script output while running, collapses after success", async ({
testPage,
apiClient,
seedData,
backend,
}) => {
test.setTimeout(120_000);
// Slow `git fetch` (worktree sync step) to keep the prepare panel reliably
// expanded while the page mounts and the WS subscribes — without this the
// whole prepare can finish before the client is listening, so events are
// missed and the running state is never observed. Same shim pattern as
// `long-prepare-panels.spec.ts`. The final `prepare.completed` payload
// carries all captured stdout, so the streaming assertion is still valid
// even if the page races past the per-progress events.
const delayFile = path.join(backend.tmpDir, "git-delay-ms");
let profile: { id: string } | null = null;
try {
// Inside the try so the finally always cleans up — if profile creation
// throws after the shim is written, leaving the delay file behind would
// contaminate every other test in this worker.
fs.writeFileSync(delayFile, "5000");
const setupScript = [
"echo '[setup] installing deps'",
"sleep 1",
"echo '[setup] ready'",
].join("; ");
profile = await createWorktreeProfileWithScript(
apiClient,
setupScript,
`E2E Streaming ${Date.now()}`,
);
const task = await apiClient.createTaskWithAgent(
seedData.workspaceId,
"Setup Script Streaming",
seedData.agentProfileId,
{
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
executor_profile_id: profile.id,
},
);
await testPage.goto(`/t/${task.id}`);
const session = new SessionPage(testPage);
await session.waitForLoad();
// Panel appears inline in the chat, expanded while running.
const panel = testPage.getByTestId("prepare-progress-panel");
await expect(panel).toBeVisible({ timeout: 15_000 });
await expect(panel).toHaveAttribute("data-status", "preparing");
await expect(panel).toHaveAttribute("data-expanded", "true");
await expect(panel.getByTestId("prepare-progress-header-spinner")).toBeVisible();
// Setup script output reaches the expanded step list — either streamed
// in real time or captured from the final `prepare.completed` payload.
await expect(panel).toContainText("[setup] installing deps", { timeout: 30_000 });
// On clean success the panel stays visible but auto-collapses.
await expect(panel).toHaveAttribute("data-status", "completed", { timeout: 30_000 });
await expect(panel).toHaveAttribute("data-expanded", "false");
} finally {
if (fs.existsSync(delayFile)) fs.unlinkSync(delayFile);
if (profile) {
await apiClient.deleteExecutorProfile(profile.id).catch(() => {
// Profile may already be deleted if the test tore down mid-run.
});
}
}
});
test("stays expanded on setup script failure with output visible", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(60_000);
// Exit non-zero with output on both streams — the panel must surface
// both so the user can diagnose without diving into backend logs. The
// upfront sleep gives the page + WS time to subscribe before the script
// terminates; without it the failure fires faster than the frontend can
// connect and we miss every event.
const failingScript = [
"echo '[setup] starting install'",
"sleep 4",
"echo 'ENOENT: package-lock.json missing' 1>&2",
"exit 1",
].join("; ");
const profile = await createWorktreeProfileWithScript(
apiClient,
failingScript,
`E2E Failing ${Date.now()}`,
);
try {
const task = await apiClient.createTaskWithAgent(
seedData.workspaceId,
"Setup Script Failure",
seedData.agentProfileId,
{
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
executor_profile_id: profile.id,
},
);
await testPage.goto(`/t/${task.id}`);
const session = new SessionPage(testPage);
await session.waitForLoad();
const panel = testPage.getByTestId("prepare-progress-panel");
// Setup script failures are non-fatal (agent still starts), so the panel
// transitions preparing → completed_with_error rather than the fatal
// `failed` state — and stays expanded so the error is readable.
await expect(panel).toBeVisible({ timeout: 15_000 });
await expect(panel).toHaveAttribute("data-status", "completed_with_error", {
timeout: 30_000,
});
await expect(panel).toHaveAttribute("data-expanded", "true");
// Both stdout and stderr from the failing script must remain visible so
// the user can diagnose — this is the core fix of this change.
await expect(panel).toContainText("[setup] starting install");
await expect(panel).toContainText("ENOENT: package-lock.json missing");
// User can click the card header to collapse it.
await panel.getByTestId("prepare-progress-toggle").click();
await expect(panel).toHaveAttribute("data-expanded", "false");
} finally {
await apiClient.deleteExecutorProfile(profile.id).catch(() => {});
}
});
test("prepare panel persists after page refresh", async ({
testPage,
apiClient,
seedData,
backend,
}) => {
test.setTimeout(120_000);
const delayFile = path.join(backend.tmpDir, "git-delay-ms");
const setupScript = "echo '[setup] persistence check'";
const profile = await createWorktreeProfileWithScript(
apiClient,
setupScript,
`E2E Persist ${Date.now()}`,
);
try {
fs.writeFileSync(delayFile, "5000");
const task = await apiClient.createTaskWithAgent(
seedData.workspaceId,
"Prepare Panel Persist",
seedData.agentProfileId,
{
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
executor_profile_id: profile.id,
},
);
await testPage.goto(`/t/${task.id}`);
const session = new SessionPage(testPage);
await session.waitForLoad();
const panel = testPage.getByTestId("prepare-progress-panel");
// Wait for preparation to complete.
await expect(panel).toBeVisible({ timeout: 15_000 });
await expect(panel).toHaveAttribute("data-status", "completed", { timeout: 30_000 });
// Expand the panel (auto-collapsed on success) and verify step data.
await panel.getByTestId("prepare-progress-toggle").click();
await expect(panel).toContainText("Validate repository", { timeout: 5_000 });
await expect(panel).toContainText("Run setup script", { timeout: 5_000 });
// Reload the page — the panel AND step data must survive the round-trip through SSR.
await testPage.reload();
await session.waitForLoad();
const panelAfterReload = testPage.getByTestId("prepare-progress-panel");
await expect(panelAfterReload).toBeVisible({ timeout: 10_000 });
// Status should still be "completed" (not "preparing").
await expect(panelAfterReload).toHaveAttribute("data-status", "completed");
// Step details must persist — not just the panel header.
// Expand the panel to see steps (auto-collapsed on success).
await panelAfterReload.getByTestId("prepare-progress-toggle").click();
await expect(panelAfterReload).toContainText("Validate repository");
await expect(panelAfterReload).toContainText("Run setup script");
} finally {
if (fs.existsSync(delayFile)) fs.unlinkSync(delayFile);
await apiClient.deleteExecutorProfile(profile.id).catch(() => {});
}
});
test("renders collapsed on success when no setup script is configured", async ({
testPage,
apiClient,
seedData,
backend,
}) => {
test.setTimeout(60_000);
// Slow the git fetch so prepare takes long enough for the page+WS to
// connect before the completed event fires — without this the events
// broadcast before the client is subscribed, and the panel never shows.
const delayFile = path.join(backend.tmpDir, "git-delay-ms");
try {
fs.writeFileSync(delayFile, "5000");
// Default seeded worktree profile — empty prepare_script. The worktree
// validate/sync/create steps still run (card appears), and with no
// error/warning the card auto-collapses once done.
const task = await apiClient.createTaskWithAgent(
seedData.workspaceId,
"No Setup Script",
seedData.agentProfileId,
{
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
executor_profile_id: seedData.worktreeExecutorProfileId,
},
);
await testPage.goto(`/t/${task.id}`);
const session = new SessionPage(testPage);
await session.waitForLoad();
// Panel stays visible but auto-collapses on a clean run. This replaces
// the pre-redesign "disappears entirely" behavior — users want to be
// able to expand it and read what ran even on a happy path.
const panel = testPage.getByTestId("prepare-progress-panel");
await expect(panel).toBeVisible({ timeout: 30_000 });
await expect(panel).toHaveAttribute("data-status", "completed", { timeout: 30_000 });
await expect(panel).toHaveAttribute("data-expanded", "false");
} finally {
if (fs.existsSync(delayFile)) fs.unlinkSync(delayFile);
}
});
});
@@ -15,27 +15,31 @@ test.describe("Branch selector behavior with executor types", () => {
test.skip(true, "No local executor available");
return;
}
await apiClient.createExecutorProfile(localExec.id, "E2E Local Profile");
const profile = await apiClient.createExecutorProfile(localExec.id, "E2E Local Profile");
const kanban = new KanbanPage(testPage);
await kanban.goto();
try {
const kanban = new KanbanPage(testPage);
await kanban.goto();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
// Fill title + description
await testPage.getByTestId("task-title-input").fill("Branch Selector Test");
await testPage.getByTestId("task-description-input").fill("testing branch selector disable");
// Fill title + description
await testPage.getByTestId("task-title-input").fill("Branch Selector Test");
await testPage.getByTestId("task-description-input").fill("testing branch selector disable");
// Select the local executor profile
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local Profile/i }).click();
// Select the local executor profile
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local Profile/i }).click();
// Branch selector should be disabled when local executor is selected
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeDisabled({ timeout: 5_000 });
// Branch selector should be disabled when local executor is selected
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeDisabled({ timeout: 5_000 });
} finally {
await apiClient.deleteExecutorProfile(profile.id).catch(() => {});
}
});
test("branch selector re-enables when switching from local to worktree executor", async ({
@@ -50,36 +54,40 @@ test.describe("Branch selector behavior with executor types", () => {
test.skip(true, "Need both local and worktree executors");
return;
}
await apiClient.createExecutorProfile(localExec.id, "E2E Local");
const profile = await apiClient.createExecutorProfile(localExec.id, "E2E Local");
const worktreeProfile = worktreeExec.profiles?.[0];
if (!worktreeProfile) {
test.skip(true, "No worktree profile available");
return;
}
const kanban = new KanbanPage(testPage);
await kanban.goto();
try {
const kanban = new KanbanPage(testPage);
await kanban.goto();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
await testPage.getByTestId("task-title-input").fill("Switch Executor Test");
await testPage.getByTestId("task-description-input").fill("testing executor switch");
await testPage.getByTestId("task-title-input").fill("Switch Executor Test");
await testPage.getByTestId("task-description-input").fill("testing executor switch");
// Select local executor -> branch should be disabled
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local/i }).click();
// Select local executor -> branch should be disabled
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local/i }).click();
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeDisabled({ timeout: 5_000 });
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeDisabled({ timeout: 5_000 });
// Switch to worktree executor -> branch should be enabled
await executorSelector.click();
await testPage.getByRole("option", { name: worktreeProfile.name }).click();
// Switch to worktree executor -> branch should be enabled
await executorSelector.click();
await testPage.getByRole("option", { name: worktreeProfile.name }).click();
await expect(branchSelector).toBeEnabled({ timeout: 5_000 });
await expect(branchSelector).toBeEnabled({ timeout: 5_000 });
} finally {
await apiClient.deleteExecutorProfile(profile.id).catch(() => {});
}
});
test("branch selector stays enabled for local executor with GitHub URL", async ({
@@ -106,46 +114,50 @@ test.describe("Branch selector behavior with executor types", () => {
test.skip(true, "No local executor available");
return;
}
await apiClient.createExecutorProfile(localExec.id, "E2E Local GitHub URL");
const profile = await apiClient.createExecutorProfile(localExec.id, "E2E Local GitHub URL");
// Create a unique repo for this test
const repoDir = `${backend.tmpDir}/repos/e2e-branch-gh`;
fs.mkdirSync(repoDir, { recursive: true });
execSync("git init -b main", { cwd: repoDir, env: gitEnv });
execSync('git commit --allow-empty -m "init"', { cwd: repoDir, env: gitEnv });
try {
// Create a unique repo for this test
const repoDir = `${backend.tmpDir}/repos/e2e-branch-gh`;
fs.mkdirSync(repoDir, { recursive: true });
execSync("git init -b main", { cwd: repoDir, env: gitEnv });
execSync('git commit --allow-empty -m "init"', { cwd: repoDir, env: gitEnv });
// Seed mock GitHub branches
await apiClient.createRepository(seedData.workspaceId, repoDir, "main", {
name: "branch-test-owner/branch-test-repo",
provider: "github",
provider_owner: "branch-test-owner",
provider_name: "branch-test-repo",
});
await apiClient.mockGitHubAddBranches("branch-test-owner", "branch-test-repo", [
{ name: "main" },
{ name: "develop" },
]);
// Seed mock GitHub branches
await apiClient.createRepository(seedData.workspaceId, repoDir, "main", {
name: "branch-test-owner/branch-test-repo",
provider: "github",
provider_owner: "branch-test-owner",
provider_name: "branch-test-repo",
});
await apiClient.mockGitHubAddBranches("branch-test-owner", "branch-test-repo", [
{ name: "main" },
{ name: "develop" },
]);
const kanban = new KanbanPage(testPage);
await kanban.goto();
const kanban = new KanbanPage(testPage);
await kanban.goto();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
await kanban.createTaskButton.first().click();
const dialog = testPage.getByTestId("create-task-dialog");
await expect(dialog).toBeVisible();
// Toggle to GitHub URL mode
await testPage.getByTestId("toggle-github-url").click();
await testPage
.getByTestId("github-url-input")
.fill("https://github.com/branch-test-owner/branch-test-repo");
// Toggle to GitHub URL mode
await testPage.getByTestId("toggle-github-url").click();
await testPage
.getByTestId("github-url-input")
.fill("https://github.com/branch-test-owner/branch-test-repo");
// Select local executor profile
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local GitHub URL/i }).click();
// Select local executor profile
const executorSelector = testPage.getByTestId("executor-profile-selector");
await executorSelector.click();
await testPage.getByRole("option", { name: /E2E Local GitHub URL/i }).click();
// Branch selector should NOT be disabled (GitHub URL mode overrides)
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeEnabled({ timeout: 10_000 });
// Branch selector should NOT be disabled (GitHub URL mode overrides)
const branchSelector = testPage.getByTestId("branch-selector");
await expect(branchSelector).toBeEnabled({ timeout: 10_000 });
} finally {
await apiClient.deleteExecutorProfile(profile.id).catch(() => {});
}
});
});
@@ -734,11 +734,6 @@ test.describe("Task creation from GitHub URL", () => {
// Clicking again should collapse the details.
await detailsBtn.click();
await expect(warningBanner.locator("pre")).not.toBeVisible();
// Dismissing the banner hides it entirely.
const dismissBtn = warningBanner.getByRole("button", { name: "Dismiss warning" });
await dismissBtn.click();
await expect(warningBanner).not.toBeVisible();
});
test("two tasks from the same PR URL create independent worktrees", async ({
+24 -5
View File
@@ -48,7 +48,13 @@ export type TurnGroup = {
messages: Message[];
};
export type RenderItem = { type: "message"; message: Message } | TurnGroup;
export type PrepareProgressItem = {
type: "prepare_progress";
id: string;
sessionId: string;
};
export type RenderItem = { type: "message"; message: Message } | TurnGroup | PrepareProgressItem;
function buildToolCallIds(messages: Message[]): Set<string> {
const set = new Set<string>();
@@ -220,6 +226,20 @@ function groupActivityMessages(allMessages: Message[]): RenderItem[] {
return items;
}
function injectPrepareProgressItem(
items: RenderItem[],
resolvedSessionId: string | null,
): RenderItem[] {
if (!resolvedSessionId) return items;
const prepareItem: PrepareProgressItem = {
type: "prepare_progress",
id: `prepare-progress-${resolvedSessionId}`,
sessionId: resolvedSessionId,
};
if (items.length === 0) return [prepareItem];
return [items[0], prepareItem, ...items.slice(1)];
}
export function useProcessedMessages(
messages: Message[],
taskId: string | null,
@@ -298,10 +318,9 @@ export function useProcessedMessages(
return { regularMessages: regular, footerActionMessages: footer };
}, [allMessages]);
const groupedItems = useMemo<RenderItem[]>(
() => groupActivityMessages(regularMessages),
[regularMessages],
);
const groupedItems = useMemo<RenderItem[]>(() => {
return injectPrepareProgressItem(groupActivityMessages(regularMessages), resolvedSessionId);
}, [regularMessages, resolvedSessionId]);
return {
visibleMessages,
+108 -56
View File
@@ -63,66 +63,49 @@ type BuildSessionPageStateParams = {
};
function buildSessionPageState(p: BuildSessionPageStateParams) {
const {
task,
sessionId,
snapshot,
agents,
repositories,
allSessions,
workspaces,
workflows,
turns,
userSettingsResponse,
messagesResponse,
} = p;
const repositoryId = task.repositories?.[0]?.repository_id;
const repository = repositories.find((r) => r.id === repositoryId);
const scripts = repository?.scripts ?? [];
const { task, sessionId, snapshot, agents, allSessions, messagesResponse } = p;
const messages = messagesResponse?.messages ? [...messagesResponse.messages].reverse() : [];
const taskState = taskToState(task, sessionId, {
items: messages,
hasMore: messagesResponse?.has_more ?? false,
oldestCursor: messages[0]?.id ?? null,
});
const repositoryScripts = repositoryId
? {
itemsByRepositoryId: { [repositoryId]: scripts },
loadingByRepositoryId: { [repositoryId]: false },
loadedByRepositoryId: { [repositoryId]: true },
}
: {
itemsByRepositoryId: {},
loadingByRepositoryId: {},
loadedByRepositoryId: {},
};
return {
...snapshotToState(snapshot),
...taskState,
...buildResourceState(p),
...buildSessionState(p),
...buildWorktreeState(allSessions),
...buildPrepareProgressState(allSessions),
settingsAgents: { items: agents.agents },
settingsData: { agentsLoaded: true, executorsLoaded: false },
userSettings: mapUserSettingsResponse(p.userSettingsResponse),
};
}
function buildResourceState(p: BuildSessionPageStateParams) {
const { task, agents, repositories, workspaces, workflows } = p;
const repositoryId = task.repositories?.[0]?.repository_id;
const repository = repositories.find((r) => r.id === repositoryId);
const scripts = repository?.scripts ?? [];
return {
workspaces: {
items: workspaces.map((workspace) => ({
id: workspace.id,
name: workspace.name,
description: workspace.description ?? null,
owner_id: workspace.owner_id,
default_executor_id: workspace.default_executor_id ?? null,
default_environment_id: workspace.default_environment_id ?? null,
default_agent_profile_id: workspace.default_agent_profile_id ?? null,
created_at: workspace.created_at,
updated_at: workspace.updated_at,
items: workspaces.map((w) => ({
id: w.id,
name: w.name,
description: w.description ?? null,
owner_id: w.owner_id,
default_executor_id: w.default_executor_id ?? null,
default_environment_id: w.default_environment_id ?? null,
default_agent_profile_id: w.default_agent_profile_id ?? null,
created_at: w.created_at,
updated_at: w.updated_at,
})),
activeId: task.workspace_id,
},
workflows: {
items: workflows.map((workflow) => ({
id: workflow.id,
workspaceId: workflow.workspace_id,
name: workflow.name,
})),
items: workflows.map((w) => ({ id: w.id, workspaceId: w.workspace_id, name: w.name })),
activeId: task.workflow_id,
},
repositories: {
@@ -130,16 +113,26 @@ function buildSessionPageState(p: BuildSessionPageStateParams) {
loadingByWorkspaceId: { [task.workspace_id]: false },
loadedByWorkspaceId: { [task.workspace_id]: true },
},
repositoryScripts,
repositoryScripts: repositoryId
? {
itemsByRepositoryId: { [repositoryId]: scripts },
loadingByRepositoryId: { [repositoryId]: false },
loadedByRepositoryId: { [repositoryId]: true },
}
: { itemsByRepositoryId: {}, loadingByRepositoryId: {}, loadedByRepositoryId: {} },
agentProfiles: {
items: agents.agents.flatMap((agent) =>
agent.profiles.map((profile) => toAgentProfileOption(agent, profile)),
),
version: 0,
},
taskSessions: {
items: Object.fromEntries(allSessions.map((s) => [s.id, s])),
},
};
}
function buildSessionState(p: BuildSessionPageStateParams) {
const { task, sessionId, allSessions, turns } = p;
return {
taskSessions: { items: Object.fromEntries(allSessions.map((s) => [s.id, s])) },
taskSessionsByTask: {
itemsByTaskId: { [task.id]: allSessions },
loadingByTaskId: { [task.id]: false },
@@ -153,19 +146,78 @@ function buildSessionPageState(p: BuildSessionPageStateParams) {
},
}
: { bySession: {}, activeBySession: {} },
...buildWorktreeState(allSessions),
environmentIdBySessionId: Object.fromEntries(
allSessions.filter((s) => s.task_environment_id).map((s) => [s.id, s.task_environment_id!]),
),
settingsAgents: { items: agents.agents },
settingsData: {
agentsLoaded: true,
executorsLoaded: false,
},
userSettings: mapUserSettingsResponse(userSettingsResponse),
};
}
function buildPrepareProgressState(allSessions: TaskSession[]) {
const bySessionId: Record<
string,
{
sessionId: string;
status: string;
steps: Array<{
name: string;
command?: string;
status: string;
output?: string;
error?: string;
warning?: string;
warningDetail?: string;
startedAt?: string;
endedAt?: string;
}>;
errorMessage?: string;
durationMs?: number;
}
> = {};
for (const session of allSessions) {
const pr = session.metadata?.prepare_result as
| {
status?: string;
steps?: Array<{
name: string;
command?: string;
status: string;
output?: string;
error?: string;
warning?: string;
warning_detail?: string;
started_at?: string;
ended_at?: string;
}>;
error_message?: string;
duration_ms?: number;
}
| undefined;
if (!pr) continue;
bySessionId[session.id] = {
sessionId: session.id,
status: pr.status ?? "completed",
steps: (pr.steps ?? []).map((s) => ({
name: s.name,
command: s.command,
status: s.status,
output: s.output,
error: s.error,
warning: s.warning,
warningDetail: s.warning_detail,
startedAt: s.started_at,
endedAt: s.ended_at,
})),
errorMessage: pr.error_message,
durationMs: pr.duration_ms,
};
}
if (Object.keys(bySessionId).length === 0) return {};
return { prepareProgress: { bySessionId } };
}
export type FetchedSessionData = {
task: Task;
sessionId: string | null;
+8
View File
@@ -179,6 +179,14 @@ function hydrateSessionRuntime(
Object.assign(draft.environmentIdBySessionId, state.environmentIdBySessionId);
}
if (state.agents) deepMerge(draft.agents, state.agents);
if (state.prepareProgress) {
mergeSessionMap(
draft.prepareProgress.bySessionId,
state.prepareProgress?.bySessionId,
activeSessionId,
forceMergeSessionId,
);
}
}
/** Hydrate UI slices without overwriting active connection state. */
@@ -226,11 +226,14 @@ export type UserShellsState = {
export type PrepareStepInfo = {
name: string;
command?: string;
status: string;
output?: string;
error?: string;
warning?: string;
warningDetail?: string;
startedAt?: string;
endedAt?: string;
};
export type SessionPrepareState = {
+6
View File
@@ -317,6 +317,7 @@ export type PrepareProgressPayload = {
session_id: string;
execution_id: string;
step_name: string;
step_command?: string;
step_index: number;
total_steps: number;
status: string;
@@ -324,6 +325,8 @@ export type PrepareProgressPayload = {
error?: string;
warning?: string;
warning_detail?: string;
started_at?: string;
ended_at?: string;
timestamp: string;
};
@@ -337,11 +340,14 @@ export type PrepareCompletedPayload = {
workspace_path?: string;
steps?: Array<{
name: string;
command?: string;
status: string;
output?: string;
error?: string;
warning?: string;
warning_detail?: string;
started_at?: string;
ended_at?: string;
}>;
timestamp: string;
};
@@ -36,11 +36,14 @@ export function registerExecutorPrepareHandlers(store: StoreApi<AppState>): WsHa
const steps = payload.steps?.length
? payload.steps.map((s) => ({
name: s.name,
command: s.command,
status: s.status,
output: s.output,
error: s.error,
warning: s.warning,
warningDetail: s.warning_detail,
startedAt: s.started_at,
endedAt: s.ended_at,
}))
: existing?.steps;
@@ -77,11 +80,14 @@ function updateSteps(
}
steps[payload.step_index] = {
name: payload.step_name,
command: payload.step_command,
status: payload.status,
output: payload.output,
error: payload.error,
warning: payload.warning,
warningDetail: payload.warning_detail,
startedAt: payload.started_at,
endedAt: payload.ended_at,
};
return steps;
}