diff --git a/apps/backend/internal/backendapp/git_status_sources.go b/apps/backend/internal/backendapp/git_status_sources.go new file mode 100644 index 0000000000..45ed7fb8a7 --- /dev/null +++ b/apps/backend/internal/backendapp/git_status_sources.go @@ -0,0 +1,238 @@ +package backendapp + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +type gitStatusSources struct { + environmentID string + workspacePath string + sessionIDs []string +} + +func resolveGitStatusSources( + ctx context.Context, + taskRepo *sqliterepo.Repository, + requested *models.TaskSession, + log *logger.Logger, +) (*gitStatusSources, bool) { + if requested == nil || requested.ID == "" { + return nil, false + } + if requested.TaskEnvironmentID == "" { + return &gitStatusSources{sessionIDs: []string{requested.ID}}, true + } + if taskRepo == nil { + log.Debug("cannot resolve task environment for git status", + zap.String("session_id", requested.ID), + zap.String("reason", "task_repository_unavailable")) + return nil, false + } + + env, ok := loadGitStatusEnvironment(ctx, taskRepo, requested, log) + if !ok { + return nil, false + } + sessions, recordedPaths, ok := loadGitStatusCandidates(ctx, taskRepo, requested, env.ID, log) + if !ok { + return nil, false + } + sourceIDs := collectGitStatusSourceIDs(sessions, env, recordedPaths, log) + if len(sourceIDs) == 0 { + log.Debug("no eligible workspace source for git status", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", env.ID), + zap.String("reason", "no_matching_session")) + return nil, false + } + + // Prefer the requested session when it is canonical, then inspect sibling + // sessions in the repository's stable order. + ordered := orderGitStatusSourceIDs(sourceIDs, requested.ID) + log.Debug("resolved task-environment git status sources", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", env.ID), + zap.Int("sources", len(ordered))) + return &gitStatusSources{ + environmentID: env.ID, + workspacePath: env.WorkspacePath, + sessionIDs: ordered, + }, true +} + +func loadGitStatusEnvironment( + ctx context.Context, + taskRepo *sqliterepo.Repository, + requested *models.TaskSession, + log *logger.Logger, +) (*models.TaskEnvironment, bool) { + env, err := taskRepo.GetTaskEnvironment(ctx, requested.TaskEnvironmentID) + if err != nil { + log.Debug("cannot resolve task environment for git status", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", requested.TaskEnvironmentID), + zap.String("reason", "environment_lookup_failed"), + zap.Error(err)) + return nil, false + } + if env == nil || env.WorkspacePath == "" { + log.Debug("cannot resolve canonical workspace for git status", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", requested.TaskEnvironmentID), + zap.String("reason", "canonical_workspace_missing")) + return nil, false + } + return env, true +} + +func loadGitStatusCandidates( + ctx context.Context, + taskRepo *sqliterepo.Repository, + requested *models.TaskSession, + environmentID string, + log *logger.Logger, +) ([]*models.TaskSession, map[string]string, bool) { + sessions, err := taskRepo.ListTaskSessionsByTaskEnvironment(ctx, environmentID) + if err != nil { + log.Debug("cannot load sibling sessions for git status", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", environmentID), + zap.String("reason", "session_lookup_failed"), + zap.Error(err)) + return nil, nil, false + } + recordedPaths, err := taskRepo.GetTaskSessionWorkspacePathsByTaskEnvironment(ctx, environmentID) + if err != nil { + log.Debug("cannot load recorded session workspaces for git status", + zap.String("session_id", requested.ID), + zap.String("task_environment_id", environmentID), + zap.String("reason", "workspace_lookup_failed"), + zap.Error(err)) + return nil, nil, false + } + return sessions, recordedPaths, true +} + +func collectGitStatusSourceIDs( + sessions []*models.TaskSession, + env *models.TaskEnvironment, + recordedPaths map[string]string, + log *logger.Logger, +) []string { + sourceIDs := make([]string, 0, len(sessions)) + seen := make(map[string]struct{}, len(sessions)) + for _, candidate := range sessions { + if !eligibleGitStatusSession(candidate, env.ID, env.WorkspacePath, recordedPaths, log) { + continue + } + if _, exists := seen[candidate.ID]; exists { + continue + } + seen[candidate.ID] = struct{}{} + sourceIDs = append(sourceIDs, candidate.ID) + } + return sourceIDs +} + +func orderGitStatusSourceIDs(sourceIDs []string, requestedID string) []string { + ordered := make([]string, 0, len(sourceIDs)) + for _, sourceID := range sourceIDs { + if sourceID == requestedID { + ordered = append([]string{requestedID}, ordered...) + continue + } + ordered = append(ordered, sourceID) + } + return ordered +} + +func eligibleGitStatusSession( + candidate *models.TaskSession, + environmentID, canonicalWorkspacePath string, + recordedPaths map[string]string, + log *logger.Logger, +) bool { + if candidate == nil { + return false + } + if candidate.TaskEnvironmentID != environmentID { + log.Debug("rejecting git status source", + zap.String("source_session_id", candidate.ID), + zap.String("task_environment_id", environmentID), + zap.String("reason", "environment_mismatch")) + return false + } + workspacePath, recorded := recordedPaths[candidate.ID] + if !recorded || workspacePath == "" { + log.Debug("rejecting git status source", + zap.String("source_session_id", candidate.ID), + zap.String("task_environment_id", environmentID), + zap.String("reason", "workspace_unverified")) + return false + } + if workspacePath != canonicalWorkspacePath { + log.Debug("rejecting git status source", + zap.String("source_session_id", candidate.ID), + zap.String("task_environment_id", environmentID), + zap.String("reason", "workspace_mismatch")) + return false + } + return true +} + +func newestGitStatusSnapshotsByRepository(snapshots []*models.GitSnapshot) map[string]*models.GitSnapshot { + newest := make(map[string]*models.GitSnapshot) + for _, snapshot := range snapshots { + if snapshot == nil { + continue + } + repositoryName := gitStatusRepositoryName(snapshot) + current := newest[repositoryName] + if current == nil || newerGitStatusSnapshot(snapshot, current) { + newest[repositoryName] = snapshot + } + } + return newest +} + +func gitStatusRepositoryName(snapshot *models.GitSnapshot) string { + if snapshot != nil && snapshot.Metadata != nil { + if repositoryName, ok := snapshot.Metadata["repository_name"].(string); ok { + return repositoryName + } + } + return "" +} + +func newerGitStatusSnapshot(candidate, current *models.GitSnapshot) bool { + candidateAt := gitStatusObservationTime(candidate) + currentAt := gitStatusObservationTime(current) + if !candidateAt.Equal(currentAt) { + return candidateAt.After(currentAt) + } + if !candidate.CreatedAt.Equal(current.CreatedAt) { + return candidate.CreatedAt.After(current.CreatedAt) + } + return candidate.ID > current.ID +} + +func gitStatusObservationTime(snapshot *models.GitSnapshot) time.Time { + if snapshot != nil && snapshot.Metadata != nil { + if timestamp, ok := snapshot.Metadata["timestamp"].(string); ok { + if parsed, err := time.Parse(time.RFC3339Nano, timestamp); err == nil { + return parsed + } + } + } + if snapshot == nil { + return time.Time{} + } + return snapshot.CreatedAt +} diff --git a/apps/backend/internal/backendapp/helpers.go b/apps/backend/internal/backendapp/helpers.go index 2742136d12..bcbe108c0e 100644 --- a/apps/backend/internal/backendapp/helpers.go +++ b/apps/backend/internal/backendapp/helpers.go @@ -2,13 +2,13 @@ package backendapp import ( "context" - "database/sql" "errors" "fmt" "io/fs" "net/http" "net/url" "os" + "sort" "strings" "time" @@ -266,48 +266,89 @@ func appendSessionStateMessageWithCancellation( return result } -// appendLiveGitStatusMessage adds git status notification(s) by querying -// agentctl for live status. Multi-repo workspaces emit one notification per +// appendLiveGitStatusMessage adds git status notification(s) from the canonical +// task-environment workspace. Multi-repo workspaces emit one notification per // repo (stamped with repository_name); single-repo emits a single untagged -// notification. Falls back to DB snapshot if no execution exists (archived -// sessions only — the snapshot is workspace-wide, not per-repo). +// notification. Sessions without a task environment retain the legacy +// session-scoped behavior. func appendLiveGitStatusMessage(ctx context.Context, taskRepo *sqliterepo.Repository, lifecycleMgr *lifecycle.Manager, sessionID string, session *models.TaskSession, result []*ws.Message, log *logger.Logger) []*ws.Message { - if msgs := tryGetLiveGitStatus(ctx, lifecycleMgr, sessionID, log); len(msgs) > 0 { + sources, ok := resolveGitStatusSources(ctx, taskRepo, session, log) + if !ok { + return result + } + if msgs := tryGetLiveGitStatus(ctx, lifecycleMgr, sessionID, sources, log); len(msgs) > 0 { return append(result, msgs...) } - return appendDBSnapshotGitStatus(ctx, taskRepo, sessionID, result, log) + return appendDBSnapshotGitStatus(ctx, taskRepo, sessionID, sources, result, log) } // tryGetLiveGitStatus attempts to get live git status from agentctl. // Returns one notification per repo (one entry for single-repo workspaces). -// Returns nil when the session has no live execution or agentctl is stuck. -func tryGetLiveGitStatus(ctx context.Context, lifecycleMgr *lifecycle.Manager, sessionID string, log *logger.Logger) []*ws.Message { +// Returns nil when no eligible source has a live execution or agentctl is +// stuck. +func tryGetLiveGitStatus(ctx context.Context, lifecycleMgr *lifecycle.Manager, requestedSessionID string, sources *gitStatusSources, log *logger.Logger) []*ws.Message { if lifecycleMgr == nil { return nil } - execution, ok := lifecycleMgr.GetExecutionBySessionID(sessionID) - if !ok { - log.Debug("no execution found for session, will fall back to DB snapshot", - zap.String("session_id", sessionID)) - return nil - } + // Bound the complete source probe, not each source independently. Shared + // environments can have several eligible executions, and a stuck agentctl + // must not add another two seconds for every sibling. + rpcCtx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + for _, sourceSessionID := range sources.sessionIDs { + if err := rpcCtx.Err(); err != nil { + return nil + } + execution, ok := lifecycleMgr.GetExecutionBySessionID(sourceSessionID) + if !ok { + continue + } + if !executionMatchesGitStatusSource(sources, execution, sourceSessionID, log) { + continue + } + if msgs := tryGetLiveGitStatusFromExecution(rpcCtx, execution, requestedSessionID, sourceSessionID, log); len(msgs) > 0 { + return msgs + } + } + return nil +} + +func executionMatchesGitStatusSource(sources *gitStatusSources, execution *lifecycle.AgentExecution, sessionID string, log *logger.Logger) bool { + if execution == nil || sources.environmentID == "" { + return execution != nil + } + if execution.TaskEnvironmentID != "" && execution.TaskEnvironmentID != sources.environmentID { + log.Debug("rejecting live git status source", + zap.String("source_session_id", sessionID), + zap.String("task_environment_id", sources.environmentID), + zap.String("reason", "environment_mismatch")) + return false + } + if execution.WorkspacePath != sources.workspacePath { + log.Debug("rejecting live git status source", + zap.String("source_session_id", sessionID), + zap.String("task_environment_id", sources.environmentID), + zap.String("reason", "workspace_mismatch")) + return false + } + return true +} + +func tryGetLiveGitStatusFromExecution(ctx context.Context, execution *lifecycle.AgentExecution, requestedSessionID, sourceSessionID string, log *logger.Logger) []*ws.Message { agentClient := execution.GetAgentCtlClient() if agentClient == nil { log.Debug("no agentctl client available for session, will fall back to DB snapshot", - zap.String("session_id", sessionID)) + zap.String("source_session_id", sourceSessionID)) return nil } - // Use bounded timeout to prevent blocking session hydration if agentctl is stuck. - rpcCtx, cancel := context.WithTimeout(ctx, 2*time.Second) - defer cancel() // Force fresh git query: cache can wedge when the poll loop misses a HEAD change. - multi, err := agentClient.GetGitStatusMultiFresh(rpcCtx) + multi, err := agentClient.GetGitStatusMultiFresh(ctx) if err != nil { log.Debug("failed to get live git status, will fall back to DB snapshot", - zap.String("session_id", sessionID), + zap.String("source_session_id", sourceSessionID), zap.Error(err)) return nil } @@ -320,7 +361,7 @@ func tryGetLiveGitStatus(ctx context.Context, lifecycleMgr *lifecycle.Manager, s if !repo.Status.Success { continue } - notification := buildGitStatusNotification(sessionID, repo.RepositoryName, repo.Status) + notification := buildGitStatusNotification(requestedSessionID, repo.RepositoryName, repo.Status) if notification != nil { out = append(out, notification) } @@ -329,7 +370,8 @@ func tryGetLiveGitStatus(ctx context.Context, lifecycleMgr *lifecycle.Manager, s return nil } log.Debug("got live git status from agentctl", - zap.String("session_id", sessionID), + zap.String("requested_session_id", requestedSessionID), + zap.String("source_session_id", sourceSessionID), zap.Int("repos", len(out))) return out } @@ -378,61 +420,95 @@ func buildGitStatusNotification(sessionID, repositoryName string, status client. return notification } -// appendDBSnapshotGitStatus appends a git status notification from DB snapshot. -func appendDBSnapshotGitStatus(ctx context.Context, taskRepo *sqliterepo.Repository, sessionID string, result []*ws.Message, log *logger.Logger) []*ws.Message { - log.Debug("falling back to DB snapshot for git status", - zap.String("session_id", sessionID)) +// appendDBSnapshotGitStatus appends a git status notification from the newest +// eligible DB snapshot. The selected source is always routed as the requested +// subscription session. +func appendDBSnapshotGitStatus(ctx context.Context, taskRepo *sqliterepo.Repository, sessionID string, sources *gitStatusSources, result []*ws.Message, log *logger.Logger) []*ws.Message { + logFields := []zap.Field{zap.String("requested_session_id", sessionID)} + if sources.environmentID != "" { + logFields = append(logFields, zap.String("task_environment_id", sources.environmentID)) + } + log.Debug("falling back to DB snapshot for git status", logFields...) - latestSnapshot, err := taskRepo.GetLatestGitSnapshot(ctx, sessionID) + snapshots, err := taskRepo.GetLatestGitStatusSnapshotsBySessionIDs(ctx, sources.sessionIDs) if err != nil { - if errors.Is(err, sql.ErrNoRows) { - // Expected for sessions that have not produced a snapshot yet. - return result - } - log.Warn("failed to load DB snapshot for session", - zap.String("session_id", sessionID), + log.Warn("failed to load DB snapshots for git status", + zap.String("requested_session_id", sessionID), zap.Error(err)) return result } - if latestSnapshot == nil { - log.Debug("no DB snapshot found for session", - zap.String("session_id", sessionID)) + latestByRepository := newestGitStatusSnapshotsByRepository(snapshots) + if len(latestByRepository) == 0 { + log.Debug("no eligible DB snapshot found for git status", + zap.String("requested_session_id", sessionID)) return result } + repositoryNames := make([]string, 0, len(latestByRepository)) + for repositoryName := range latestByRepository { + repositoryNames = append(repositoryNames, repositoryName) + } + sort.Strings(repositoryNames) + for _, repositoryName := range repositoryNames { + snapshot := latestByRepository[repositoryName] + logFields := []zap.Field{ + zap.String("requested_session_id", sessionID), + zap.String("source_session_id", snapshot.SessionID), + } + if sources.environmentID != "" { + logFields = append(logFields, zap.String("task_environment_id", sources.environmentID)) + } + if repositoryName != "" { + logFields = append(logFields, zap.String("repository_name", repositoryName)) + } + log.Debug("selected DB snapshot for git status", logFields...) + if notification := buildGitSnapshotNotification(sessionID, repositoryName, snapshot); notification != nil { + result = append(result, notification) + } + } + return result +} - metadata := latestSnapshot.Metadata +func buildGitSnapshotNotification(sessionID, repositoryName string, snapshot *models.GitSnapshot) *ws.Message { + if snapshot == nil { + return nil + } + metadata := snapshot.Metadata + statusPayload := map[string]interface{}{ + branchFieldKey: snapshot.Branch, + "remote_branch": snapshot.RemoteBranch, + "head_commit": snapshot.HeadCommit, + "base_commit": snapshot.BaseCommit, + "ahead": snapshot.Ahead, + "behind": snapshot.Behind, + "remote_ahead": metadata["remote_ahead"], + "remote_behind": metadata["remote_behind"], + "remote_head_commit": metadata["remote_head_commit"], + "files": snapshot.Files, + "modified": metadata["modified"], + addedFieldKey: metadata[addedFieldKey], + deletedFieldKey: metadata[deletedFieldKey], + "untracked": metadata["untracked"], + "renamed": metadata["renamed"], + branchAdditionsFieldKey: metadata[branchAdditionsFieldKey], + branchDeletionsFieldKey: metadata[branchDeletionsFieldKey], + "comparison_target": metadata["comparison_target"], + "comparison_status": metadata["comparison_status"], + "comparison_error_code": metadata["comparison_error_code"], + } + if repositoryName != "" { + statusPayload["repository_name"] = repositoryName + } gitEventData := map[string]interface{}{ "type": "status_update", "session_id": sessionID, "timestamp": metadata["timestamp"], - "status": map[string]interface{}{ - branchFieldKey: latestSnapshot.Branch, - "remote_branch": latestSnapshot.RemoteBranch, - "head_commit": latestSnapshot.HeadCommit, - "base_commit": latestSnapshot.BaseCommit, - "ahead": latestSnapshot.Ahead, - "behind": latestSnapshot.Behind, - "remote_ahead": metadata["remote_ahead"], - "remote_behind": metadata["remote_behind"], - "remote_head_commit": metadata["remote_head_commit"], - "files": latestSnapshot.Files, - "modified": metadata["modified"], - addedFieldKey: metadata[addedFieldKey], - deletedFieldKey: metadata[deletedFieldKey], - "untracked": metadata["untracked"], - "renamed": metadata["renamed"], - branchAdditionsFieldKey: metadata[branchAdditionsFieldKey], - branchDeletionsFieldKey: metadata[branchDeletionsFieldKey], - "comparison_target": metadata["comparison_target"], - "comparison_status": metadata["comparison_status"], - "comparison_error_code": metadata["comparison_error_code"], - }, + "status": statusPayload, } notification, err := ws.NewNotification(ws.ActionSessionGitEvent, gitEventData) - if err == nil { - result = append(result, notification) + if err != nil { + return nil } - return result + return notification } // appendContextWindowMessage adds a context window notification to result if available. diff --git a/apps/backend/internal/backendapp/helpers_git_status_materialized_test.go b/apps/backend/internal/backendapp/helpers_git_status_materialized_test.go new file mode 100644 index 0000000000..b330117073 --- /dev/null +++ b/apps/backend/internal/backendapp/helpers_git_status_materialized_test.go @@ -0,0 +1,85 @@ +package backendapp + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +// TestAppendLiveGitStatusMessageReplaysTerminalMaterializedSessionSnapshot +// covers the post-materialization shape written by the executor: the session +// owns the canonical environment path, the agent is terminal so no live source +// is available, and the status notification must still replay its snapshot. +func TestAppendLiveGitStatusMessageReplaysTerminalMaterializedSessionSnapshot(t *testing.T) { + harness := newBootStateTestHarness(t) + ctx := context.Background() + const ( + taskID = "git-status-materialized-task" + environmentID = "git-status-materialized-environment" + sessionID = "git-status-materialized-session" + canonicalPath = "/tasks/git-status/materialized" + ) + now := time.Date(2026, 8, 31, 12, 0, 0, 0, time.UTC) + + if err := harness.taskRepo.CreateTask(ctx, &models.Task{ID: taskID, Title: taskID}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := harness.taskRepo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{ + ID: environmentID, + TaskID: taskID, + ExecutorType: string(models.ExecutorTypeWorktree), + Status: models.TaskEnvironmentStatusReady, + WorkspacePath: canonicalPath, + }); err != nil { + t.Fatalf("CreateTaskEnvironment: %v", err) + } + if err := harness.taskRepo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + TaskEnvironmentID: environmentID, + WorkspacePath: canonicalPath, + State: models.TaskSessionStateCompleted, + StartedAt: now, + UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + if err := harness.taskRepo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-materialized-snapshot", + SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{ + "materialized.go": map[string]interface{}{"status": "modified"}, + }, + Metadata: map[string]interface{}{ + "timestamp": "2026-08-31T12:05:00Z", + "modified": []string{"materialized.go"}, + }, + CreatedAt: now.Add(5 * time.Minute), + }); err != nil { + t.Fatalf("CreateGitSnapshot: %v", err) + } + + requested, err := harness.taskRepo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + msgs := appendLiveGitStatusMessage(ctx, harness.taskRepo, nil, sessionID, requested, nil, newTestLogger()) + if len(msgs) != 1 { + t.Fatalf("expected one replayed status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["materialized.go"]; !ok { + t.Fatalf("terminal materialized snapshot was not replayed: files = %#v", files) + } +} diff --git a/apps/backend/internal/backendapp/helpers_git_status_test.go b/apps/backend/internal/backendapp/helpers_git_status_test.go new file mode 100644 index 0000000000..8465790f86 --- /dev/null +++ b/apps/backend/internal/backendapp/helpers_git_status_test.go @@ -0,0 +1,574 @@ +package backendapp + +import ( + "context" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "testing" + "time" + + client "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/models" + sqlitetaskrepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +type gitStatusEnvironmentFixture struct { + repo *sqlitetaskrepo.Repository + env *models.TaskEnvironment + requested *models.TaskSession +} + +func newGitStatusEnvironmentFixture(t *testing.T) gitStatusEnvironmentFixture { + t.Helper() + harness := newBootStateTestHarness(t) + ctx := context.Background() + const ( + taskID = "git-status-task" + environmentID = "git-status-environment" + requestedID = "git-status-requested" + canonicalID = "git-status-canonical" + canonicalPath = "/tasks/git-status/canonical" + mismatchedPath = "/tasks/git-status/mismatched" + ) + now := time.Date(2026, 8, 30, 12, 0, 0, 0, time.UTC) + if err := harness.taskRepo.CreateTask(ctx, &models.Task{ID: taskID, Title: taskID}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + env := &models.TaskEnvironment{ + ID: environmentID, + TaskID: taskID, + ExecutorType: string(models.ExecutorTypeLocal), + Status: models.TaskEnvironmentStatusReady, + WorkspacePath: canonicalPath, + } + if err := harness.taskRepo.CreateTaskEnvironment(ctx, env); err != nil { + t.Fatalf("CreateTaskEnvironment: %v", err) + } + for _, session := range []*models.TaskSession{ + { + ID: requestedID, + TaskID: taskID, + TaskEnvironmentID: environmentID, + WorkspacePath: mismatchedPath, + State: models.TaskSessionStateWaitingForInput, + StartedAt: now, + UpdatedAt: now, + }, + { + ID: canonicalID, + TaskID: taskID, + TaskEnvironmentID: environmentID, + WorkspacePath: canonicalPath, + State: models.TaskSessionStateWaitingForInput, + StartedAt: now.Add(time.Minute), + UpdatedAt: now.Add(time.Minute), + }, + } { + if err := harness.taskRepo.CreateTaskSession(ctx, session); err != nil { + t.Fatalf("CreateTaskSession(%s): %v", session.ID, err) + } + } + if err := harness.taskRepo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-mismatched-snapshot", + SessionID: requestedID, + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{ + "mismatched.go": map[string]interface{}{"status": "modified"}, + }, + Metadata: map[string]interface{}{ + "timestamp": "2026-08-30T12:05:00Z", + "modified": []string{"mismatched.go"}, + }, + CreatedAt: now.Add(5 * time.Minute), + }); err != nil { + t.Fatalf("CreateGitSnapshot(mismatched): %v", err) + } + if err := harness.taskRepo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-canonical-snapshot", + SessionID: canonicalID, + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{ + "canonical.go": map[string]interface{}{"status": "modified"}, + }, + Metadata: map[string]interface{}{ + "timestamp": "2026-08-30T12:03:00Z", + "modified": []string{"canonical.go"}, + }, + CreatedAt: now.Add(3 * time.Minute), + }); err != nil { + t.Fatalf("CreateGitSnapshot(canonical): %v", err) + } + + requested, err := harness.taskRepo.GetTaskSession(ctx, requestedID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + return gitStatusEnvironmentFixture{repo: harness.taskRepo, env: env, requested: requested} +} + +func TestAppendLiveGitStatusMessageSelectsCanonicalEnvironmentSnapshot(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + msgs := appendLiveGitStatusMessage( + context.Background(), fixture.repo, nil, fixture.requested.ID, fixture.requested, nil, newTestLogger(), + ) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + if got := payload["session_id"]; got != fixture.requested.ID { + t.Fatalf("event session_id = %v, want requested session %q", got, fixture.requested.ID) + } + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["canonical.go"]; !ok { + t.Fatalf("canonical snapshot was not selected: files = %#v", files) + } + if _, ok := files["mismatched.go"]; ok { + t.Fatalf("mismatched snapshot was selected: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageSelectsCanonicalSiblingLiveStatus(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + log := newTestLogger() + agentClient, closeServer := newGitStatusClient(t, log, client.GitStatusResult{ + Success: true, + Modified: []string{"canonical-live.go"}, + Timestamp: "2026-08-30T12:04:00Z", + Files: map[string]interface{}{"canonical-live.go": map[string]interface{}{"status": "modified"}}, + }) + defer closeServer() + + mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), log) + if err := mgr.ExecutionStoreForTesting().Add(&lifecycle.AgentExecution{ + ID: "git-status-canonical-execution", + TaskID: fixture.requested.TaskID, + SessionID: "git-status-canonical", + // Recovered executions can predate the environment identity on the + // in-memory lifecycle record. The canonical workspace path remains a + // sufficient binding after the session source has been verified. + TaskEnvironmentID: "", + WorkspacePath: fixture.env.WorkspacePath, + }); err != nil { + t.Fatalf("add execution: %v", err) + } + execution, ok := mgr.GetExecutionBySessionID("git-status-canonical") + if !ok { + t.Fatal("canonical execution was not registered") + } + execution.SetAgentCtlClientForTesting(agentClient) + + msgs := appendLiveGitStatusMessage( + context.Background(), fixture.repo, mgr, fixture.requested.ID, fixture.requested, nil, log, + ) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + if got := payload["session_id"]; got != fixture.requested.ID { + t.Fatalf("event session_id = %v, want requested session %q", got, fixture.requested.ID) + } + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["canonical-live.go"]; !ok { + t.Fatalf("canonical sibling live status was not selected: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageFallsBackWhenRequestedSessionIsNotCanonical(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + log := newTestLogger() + agentClient, closeServer := newGitStatusClient(t, log, client.GitStatusResult{ + Success: true, + Modified: []string{"mismatched-live.go"}, + Timestamp: "2026-08-30T12:04:00Z", + Files: map[string]interface{}{"mismatched-live.go": map[string]interface{}{"status": "modified"}}, + }) + defer closeServer() + + mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), log) + if err := mgr.ExecutionStoreForTesting().Add(&lifecycle.AgentExecution{ + ID: "git-status-mismatched-execution", + TaskID: fixture.requested.TaskID, + SessionID: fixture.requested.ID, + TaskEnvironmentID: fixture.env.ID, + WorkspacePath: "/tasks/git-status/mismatched-runtime", + }); err != nil { + t.Fatalf("add execution: %v", err) + } + execution, ok := mgr.GetExecutionBySessionID(fixture.requested.ID) + if !ok { + t.Fatal("mismatched execution was not registered") + } + execution.SetAgentCtlClientForTesting(agentClient) + + msgs := appendLiveGitStatusMessage( + context.Background(), fixture.repo, mgr, fixture.requested.ID, fixture.requested, nil, log, + ) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["canonical.go"]; !ok { + t.Fatalf("mismatched live execution bypassed canonical snapshot: files = %#v", files) + } + if _, ok := files["mismatched-live.go"]; ok { + t.Fatalf("mismatched live status was selected: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageRejectsMismatchedLiveExecutionWorkspace(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + ctx := context.Background() + if _, err := fixture.repo.DB().ExecContext(ctx, ` + UPDATE task_sessions SET workspace_path = ? WHERE id = ? + `, fixture.env.WorkspacePath, fixture.requested.ID); err != nil { + t.Fatalf("update requested workspace path: %v", err) + } + requested, err := fixture.repo.GetTaskSession(ctx, fixture.requested.ID) + if err != nil { + t.Fatalf("reload requested session: %v", err) + } + if err := fixture.repo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-requested-fallback-snapshot", + SessionID: requested.ID, + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{"fallback.go": map[string]interface{}{"status": "modified"}}, + Metadata: map[string]interface{}{ + "timestamp": "2026-08-30T12:06:00Z", + "modified": []string{"fallback.go"}, + }, + CreatedAt: time.Date(2026, 8, 30, 12, 6, 0, 0, time.UTC), + }); err != nil { + t.Fatalf("create fallback snapshot: %v", err) + } + + log := newTestLogger() + agentClient, closeServer := newGitStatusClient(t, log, client.GitStatusResult{ + Success: true, + Modified: []string{"mismatched-live.go"}, + Timestamp: "2026-08-30T12:07:00Z", + Files: map[string]interface{}{"mismatched-live.go": map[string]interface{}{"status": "modified"}}, + }) + defer closeServer() + + mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), log) + if err := mgr.ExecutionStoreForTesting().Add(&lifecycle.AgentExecution{ + ID: "git-status-requested-mismatched-runtime", + TaskID: requested.TaskID, + SessionID: requested.ID, + TaskEnvironmentID: fixture.env.ID, + WorkspacePath: "/tasks/git-status/mismatched-runtime", + }); err != nil { + t.Fatalf("add execution: %v", err) + } + execution, ok := mgr.GetExecutionBySessionID(requested.ID) + if !ok { + t.Fatal("mismatched execution was not registered") + } + execution.SetAgentCtlClientForTesting(agentClient) + + msgs := appendLiveGitStatusMessage(ctx, fixture.repo, mgr, requested.ID, requested, nil, log) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["fallback.go"]; !ok { + t.Fatalf("mismatched runtime did not fall back to the eligible snapshot: files = %#v", files) + } + if _, ok := files["mismatched-live.go"]; ok { + t.Fatalf("mismatched runtime status was selected: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageRejectsUnrecordedWorkspace(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + ctx := context.Background() + if _, err := fixture.repo.DB().ExecContext(ctx, ` + UPDATE task_sessions SET workspace_path = '' WHERE id = ? + `, fixture.requested.ID); err != nil { + t.Fatalf("clear requested workspace path: %v", err) + } + requested, err := fixture.repo.GetTaskSession(ctx, fixture.requested.ID) + if err != nil { + t.Fatalf("reload requested session: %v", err) + } + + msgs := appendLiveGitStatusMessage(ctx, fixture.repo, nil, requested.ID, requested, nil, newTestLogger()) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["canonical.go"]; !ok { + t.Fatalf("canonical sibling snapshot was not selected: files = %#v", files) + } + if _, ok := files["mismatched.go"]; ok { + t.Fatalf("unrecorded session snapshot was selected: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageSelectsSharedEnvironmentAcrossTasks(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + ctx := context.Background() + const ownerTaskID = "git-status-owner-task" + if err := fixture.repo.CreateTask(ctx, &models.Task{ID: ownerTaskID, Title: ownerTaskID}); err != nil { + t.Fatalf("create owner task: %v", err) + } + if _, err := fixture.repo.DB().ExecContext(ctx, ` + UPDATE task_environments SET task_id = ? WHERE id = ? + `, ownerTaskID, fixture.env.ID); err != nil { + t.Fatalf("move environment owner: %v", err) + } + ownerSessionID := "git-status-owner-session" + ownerSessionTime := time.Date(2026, 8, 30, 12, 7, 0, 0, time.UTC) + if err := fixture.repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: ownerSessionID, + TaskID: ownerTaskID, + TaskEnvironmentID: fixture.env.ID, + WorkspacePath: fixture.env.WorkspacePath, + State: models.TaskSessionStateWaitingForInput, + StartedAt: ownerSessionTime, + UpdatedAt: ownerSessionTime, + }); err != nil { + t.Fatalf("create owner session: %v", err) + } + if err := fixture.repo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-owner-snapshot", + SessionID: ownerSessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{"owner.go": map[string]interface{}{"status": "modified"}}, + Metadata: map[string]interface{}{ + "timestamp": "2026-08-30T12:10:00Z", + "modified": []string{"owner.go"}, + }, + CreatedAt: time.Date(2026, 8, 30, 12, 10, 0, 0, time.UTC), + }); err != nil { + t.Fatalf("create owner snapshot: %v", err) + } + + msgs := appendLiveGitStatusMessage(ctx, fixture.repo, nil, fixture.requested.ID, fixture.requested, nil, newTestLogger()) + if len(msgs) != 1 { + t.Fatalf("expected one git status message, got %d", len(msgs)) + } + payload := decodePayload(t, msgs[0].Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + if _, ok := files["owner.go"]; !ok { + t.Fatalf("shared environment owner snapshot was not selected: files = %#v", files) + } + if _, ok := files["canonical.go"]; ok { + t.Fatalf("same-task snapshot won over newer shared-environment snapshot: files = %#v", files) + } +} + +func TestAppendLiveGitStatusMessageSelectsNewestSnapshotPerRepository(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + ctx := context.Background() + if err := fixture.repo.CreateGitSnapshot(ctx, &models.GitSnapshot{ + ID: "git-status-frontend-snapshot", + SessionID: "git-status-canonical", + SnapshotType: models.SnapshotTypeStatusUpdate, + Files: map[string]interface{}{"frontend.go": map[string]interface{}{"status": "modified"}}, + Metadata: map[string]interface{}{ + "repository_name": "frontend", + "timestamp": "2026-08-30T12:04:00Z", + "modified": []string{"frontend.go"}, + }, + CreatedAt: time.Date(2026, 8, 30, 12, 4, 0, 0, time.UTC), + }); err != nil { + t.Fatalf("create frontend snapshot: %v", err) + } + + msgs := appendLiveGitStatusMessage(ctx, fixture.repo, nil, fixture.requested.ID, fixture.requested, nil, newTestLogger()) + if len(msgs) != 2 { + t.Fatalf("expected one message per repository, got %d", len(msgs)) + } + seen := make(map[string]map[string]interface{}, len(msgs)) + for _, msg := range msgs { + payload := decodePayload(t, msg.Payload) + status, ok := payload["status"].(map[string]interface{}) + if !ok { + t.Fatalf("status payload = %#v, want object", payload["status"]) + } + repositoryName, _ := status["repository_name"].(string) + files, ok := status["files"].(map[string]interface{}) + if !ok { + t.Fatalf("status files = %#v, want object", status["files"]) + } + seen[repositoryName] = files + } + if _, ok := seen[""]["canonical.go"]; !ok { + t.Fatalf("root repository snapshot missing: %#v", seen) + } + if _, ok := seen["frontend"]["frontend.go"]; !ok { + t.Fatalf("frontend repository snapshot missing: %#v", seen) + } + if _, ok := seen[""]["mismatched.go"]; ok { + t.Fatalf("mismatched repository snapshot selected: %#v", seen) + } +} + +func TestTryGetLiveGitStatusUsesSingleTimeoutAcrossSources(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + log := newTestLogger() + firstClient, closeFirst := newGitStatusClientWithHandler(t, log, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + })) + defer closeFirst() + secondCalled := make(chan struct{}, 1) + secondClient, closeSecond := newGitStatusClientWithHandler(t, log, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case secondCalled <- struct{}{}: + default: + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(client.MultiRepoGitStatusResult{ + Success: true, + Repos: []client.PerRepoGitStatus{{Status: client.GitStatusResult{ + Success: true, + Timestamp: "2026-08-30T12:04:00Z", + Files: map[string]interface{}{"should-not-be-used.go": map[string]interface{}{"status": "modified"}}, + }}}, + }) + })) + defer closeSecond() + + mgr := lifecycle.NewManager(nil, nil, nil, nil, nil, nil, lifecycle.ExecutorFallbackDeny, t.TempDir(), log) + for _, item := range []struct { + id string + client *client.Client + }{ + {id: fixture.requested.ID, client: firstClient}, + {id: "git-status-canonical", client: secondClient}, + } { + if err := mgr.ExecutionStoreForTesting().Add(&lifecycle.AgentExecution{ + ID: item.id + "-execution", + TaskID: fixture.requested.TaskID, + SessionID: item.id, + TaskEnvironmentID: fixture.env.ID, + WorkspacePath: fixture.env.WorkspacePath, + }); err != nil { + t.Fatalf("add execution %s: %v", item.id, err) + } + execution, ok := mgr.GetExecutionBySessionID(item.id) + if !ok { + t.Fatalf("execution %s was not registered", item.id) + } + execution.SetAgentCtlClientForTesting(item.client) + } + + sources := &gitStatusSources{ + environmentID: fixture.env.ID, + workspacePath: fixture.env.WorkspacePath, + sessionIDs: []string{fixture.requested.ID, "git-status-canonical"}, + } + msgs := tryGetLiveGitStatus(context.Background(), mgr, fixture.requested.ID, sources, log) + if len(msgs) != 0 { + t.Fatalf("expected the shared timeout to stop live probing after the first source, got %d messages", len(msgs)) + } + select { + case <-secondCalled: + t.Fatal("second source was probed after the shared live-status deadline expired") + default: + } +} + +func TestAppendLiveGitStatusMessageDoesNotFallbackForUnverifiedEnvironment(t *testing.T) { + fixture := newGitStatusEnvironmentFixture(t) + fixture.env.WorkspacePath = "/tasks/git-status/unverified" + if err := fixture.repo.UpdateTaskEnvironment(context.Background(), fixture.env); err != nil { + t.Fatalf("UpdateTaskEnvironment: %v", err) + } + + msgs := appendLiveGitStatusMessage( + context.Background(), fixture.repo, nil, fixture.requested.ID, fixture.requested, nil, newTestLogger(), + ) + if len(msgs) != 0 { + t.Fatalf("expected no status for an unverified environment, got %d messages", len(msgs)) + } +} + +func newGitStatusClient(t *testing.T, log *logger.Logger, status client.GitStatusResult) (*client.Client, func()) { + t.Helper() + return newGitStatusClientWithHandler(t, log, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/git/status/multi" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(client.MultiRepoGitStatusResult{ + Success: true, + Repos: []client.PerRepoGitStatus{{Status: status}}, + }) + })) +} + +func newGitStatusClientWithHandler(t *testing.T, log *logger.Logger, handler http.Handler) (*client.Client, func()) { + t.Helper() + server := httptest.NewServer(handler) + + parsed, err := url.Parse(server.URL) + if err != nil { + server.Close() + t.Fatalf("parse test agentctl URL: %v", err) + } + host, portText, err := net.SplitHostPort(parsed.Host) + if err != nil { + server.Close() + t.Fatalf("split test agentctl address: %v", err) + } + port, err := strconv.Atoi(portText) + if err != nil { + server.Close() + t.Fatalf("parse test agentctl port: %v", err) + } + return client.NewClient(host, port, log), server.Close +} diff --git a/apps/backend/internal/orchestrator/executor/executor_environment_workspace_test.go b/apps/backend/internal/orchestrator/executor/executor_environment_workspace_test.go new file mode 100644 index 0000000000..14f72e0db5 --- /dev/null +++ b/apps/backend/internal/orchestrator/executor/executor_environment_workspace_test.go @@ -0,0 +1,152 @@ +package executor + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +// TestPersistTaskEnvironmentBindsCanonicalWorkspaceOnEverySuccessPath pins +// the session-row invariant consumed by shared-environment Git status source +// selection. The launch request may still carry the source checkout (or no +// path at all), but a successful environment bind must persist the effective +// environment workspace on the session before persistLaunchState writes it. +func TestPersistTaskEnvironmentBindsCanonicalWorkspaceOnEverySuccessPath(t *testing.T) { + const canonicalPath = "/tasks/task-1/materialized" + const sourcePath = "/source/checkout" + + tests := []struct { + name string + run func(t *testing.T, repo *mockRepository, exec *Executor, session *models.TaskSession) + }{ + { + name: "new environment", + run: func(t *testing.T, repo *mockRepository, exec *Executor, session *models.TaskSession) { + err := exec.persistTaskEnvironment( + context.Background(), session.TaskID, session, nil, + &LaunchAgentRequest{ + TaskID: session.TaskID, + ExecutorType: string(models.ExecutorTypeWorktree), + RepositoryPath: sourcePath, + }, + &LaunchAgentResponse{WorkspacePath: canonicalPath}, + executorConfig{ExecutorID: models.ExecutorIDWorktree}, + ) + if err != nil { + t.Fatalf("persistTaskEnvironment: %v", err) + } + if got := repo.taskEnvironments[session.TaskEnvironmentID].WorkspacePath; got != canonicalPath { + t.Fatalf("new environment workspace = %q, want %q", got, canonicalPath) + } + }, + }, + { + name: "initial materializer", + run: func(t *testing.T, repo *mockRepository, exec *Executor, session *models.TaskSession) { + env := &models.TaskEnvironment{ + ID: "env-materializer", + TaskID: session.TaskID, + ExecutorType: string(models.ExecutorTypeWorktree), + Status: models.TaskEnvironmentStatusCreating, + MaterializationSessionID: session.ID, + } + repo.taskEnvironments[env.ID] = env + session.TaskEnvironmentID = env.ID + err := exec.persistTaskEnvironment( + context.Background(), session.TaskID, session, env, + &LaunchAgentRequest{ + TaskID: session.TaskID, + ExecutorType: string(models.ExecutorTypeWorktree), + RepositoryPath: sourcePath, + }, + &LaunchAgentResponse{ + WorktreePath: canonicalPath, + Worktrees: []RepoWorktreeResult{{ + RepositoryID: "repo-1", WorktreePath: canonicalPath, + }}, + }, + executorConfig{ExecutorID: models.ExecutorIDWorktree}, + ) + if err != nil { + t.Fatalf("persistTaskEnvironment: %v", err) + } + }, + }, + { + name: "ready sibling", + run: func(t *testing.T, repo *mockRepository, exec *Executor, session *models.TaskSession) { + env := &models.TaskEnvironment{ + ID: "env-ready", + TaskID: session.TaskID, + ExecutorType: string(models.ExecutorTypeWorktree), + Status: models.TaskEnvironmentStatusReady, + WorkspacePath: canonicalPath, + } + repo.taskEnvironments[env.ID] = env + session.TaskEnvironmentID = env.ID + err := exec.persistTaskEnvironment( + context.Background(), session.TaskID, session, env, + &LaunchAgentRequest{TaskID: session.TaskID, ExecutorType: string(models.ExecutorTypeWorktree)}, + &LaunchAgentResponse{WorkspacePath: canonicalPath}, + executorConfig{ExecutorID: models.ExecutorIDWorktree}, + ) + if err != nil { + t.Fatalf("persistTaskEnvironment: %v", err) + } + }, + }, + { + name: "cross-task guest", + run: func(t *testing.T, repo *mockRepository, exec *Executor, session *models.TaskSession) { + env := &models.TaskEnvironment{ + ID: "env-owner", + TaskID: "task-owner", + ExecutorType: string(models.ExecutorTypeWorktree), + Status: models.TaskEnvironmentStatusReady, + WorkspacePath: canonicalPath, + } + repo.taskEnvironments[env.ID] = env + session.TaskEnvironmentID = env.ID + err := exec.persistTaskEnvironment( + context.Background(), session.TaskID, session, env, + &LaunchAgentRequest{TaskID: session.TaskID, ExecutorType: string(models.ExecutorTypeWorktree)}, + &LaunchAgentResponse{}, executorConfig{ExecutorID: models.ExecutorIDWorktree}, + ) + if err != nil { + t.Fatalf("persistTaskEnvironment: %v", err) + } + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + repo := newMockRepository() + exec := newTestExecutor(t, &mockAgentManager{}, repo) + session := &models.TaskSession{ + ID: "session-" + test.name, + TaskID: "task-1", + State: models.TaskSessionStateCompleted, + WorkspacePath: sourcePath, + } + repo.sessions[session.ID] = &models.TaskSession{ + ID: session.ID, TaskID: session.TaskID, State: session.State, + } + test.run(t, repo, exec, session) + if session.WorkspacePath != canonicalPath { + t.Fatalf("session workspace = %q, want canonical %q", session.WorkspacePath, canonicalPath) + } + if err := exec.persistLaunchState( + context.Background(), session.TaskID, session.ID, session, + &LaunchAgentResponse{}, false, time.Now().UTC(), + ); err != nil { + t.Fatalf("persistLaunchState: %v", err) + } + if got := repo.sessions[session.ID].WorkspacePath; got != canonicalPath { + t.Fatalf("persisted session workspace = %q, want canonical %q", got, canonicalPath) + } + }) + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index f2cacff73b..40996dba33 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -1601,6 +1601,14 @@ func (e *Executor) finalizeLaunch(ctx context.Context, task *v1.Task, session *m return execution, nil } +func bindSessionToTaskEnvironment(session *models.TaskSession, env *models.TaskEnvironment) { + if session == nil || env == nil { + return + } + session.TaskEnvironmentID = env.ID + session.WorkspacePath = env.WorkspacePath +} + func assignLaunchTaskEnvironmentID(session *models.TaskSession, existingEnv *models.TaskEnvironment) { if existingEnv != nil && existingEnv.ID != "" { session.TaskEnvironmentID = existingEnv.ID @@ -2247,7 +2255,7 @@ func (e *Executor) persistTaskEnvironment( // session elected to materialize a still-CREATING canonical environment // (shared_group), which must run the normal finalize path below. if existingEnv.TaskID != "" && existingEnv.TaskID != taskID && !isInitialMaterializer { - session.TaskEnvironmentID = existingEnv.ID + bindSessionToTaskEnvironment(session, existingEnv) return nil } previousStatus := existingEnv.Status @@ -2303,7 +2311,7 @@ func (e *Executor) persistTaskEnvironment( zap.String("task_id", taskID), zap.String("env_id", existingEnv.ID), zap.Error(err)) return fmt.Errorf("finalize task environment materialization: %w", err) } - session.TaskEnvironmentID = existingEnv.ID + bindSessionToTaskEnvironment(session, existingEnv) e.selfHealTaskRepositoryBaseBranches(ctx, taskID, req, resp) return nil } @@ -2355,7 +2363,7 @@ func (e *Executor) persistTaskEnvironment( zap.Error(err)) return fmt.Errorf("update task environment: %w", err) } - session.TaskEnvironmentID = existingEnv.ID + bindSessionToTaskEnvironment(session, existingEnv) e.selfHealTaskRepositoryBaseBranches(ctx, taskID, req, resp) return nil } @@ -2385,7 +2393,7 @@ func (e *Executor) persistTaskEnvironment( zap.Error(err)) return fmt.Errorf("create task environment: %w", err) } - session.TaskEnvironmentID = env.ID + bindSessionToTaskEnvironment(session, env) e.selfHealTaskRepositoryBaseBranches(ctx, taskID, req, resp) return nil } diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots.go b/apps/backend/internal/task/repository/sqlite/git_snapshots.go index 5c36fee591..125c77316a 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" + "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/task/models" ) @@ -63,10 +64,59 @@ const snapshotRankExpr = ` created_at DESC, id DESC` +// snapshotRepositoryExpr returns the normalized repository partition key used +// by the multi-repository status selector. Missing and explicit-empty names +// both represent the root repository and therefore share the empty key. +func snapshotRepositoryExpr(driver, tableName string) string { + return "COALESCE(" + dialect.JSONExtract(driver, tableName+".metadata", "repository_name") + ", '')" +} + +// snapshotRankExprForRepository mirrors snapshotRankExpr for the per-repository +// status selector. The legacy session-wide selectors intentionally keep using +// snapshotRankExpr. Every correlated generation check below compares the +// candidate row with the same normalized repository key used by the window +// partition, so another repository cannot change this row's rank. +func snapshotRankExprForRepository(driver string) string { + outerRepository := snapshotRepositoryExpr(driver, "task_session_git_snapshots") + newerRepository := snapshotRepositoryExpr(driver, "newer") + archiveRepository := snapshotRepositoryExpr(driver, "archive") + return fmt.Sprintf(` + CASE + WHEN snapshot_type = 'archive' AND ( + EXISTS ( + SELECT 1 FROM task_sessions ts + JOIN tasks t ON t.id = ts.task_id + WHERE ts.id = task_session_git_snapshots.session_id + AND t.archived_at IS NOT NULL + ) + OR NOT EXISTS ( + SELECT 1 FROM task_session_git_snapshots newer + WHERE newer.session_id = task_session_git_snapshots.session_id + AND %s = %s + AND newer.snapshot_type <> 'archive' + AND newer.created_at > task_session_git_snapshots.created_at + ) + ) THEN 0 + WHEN triggered_by = 'agent_completed' AND EXISTS ( + SELECT 1 FROM task_session_git_snapshots archive + WHERE archive.session_id = task_session_git_snapshots.session_id + AND %s = %s + AND archive.snapshot_type = 'archive' + AND archive.created_at > task_session_git_snapshots.created_at + ) THEN 3 + WHEN triggered_by = 'agent_completed' THEN 1 + WHEN snapshot_type = 'archive' THEN 3 + ELSE 2 + END, + created_at DESC, + id DESC`, newerRepository, outerRepository, archiveRepository, outerRepository) +} + // UpsertLatestLiveGitSnapshot keeps at most one cached "live monitor" snapshot -// per session by deleting any previous live row and inserting the new one in a -// single transaction. This is the cache that backs the sidebar diff badge for -// tasks whose executor isn't currently running. +// per session and repository by deleting the previous row for that repository +// and inserting the new one in a single transaction. This is the cache that +// backs the sidebar diff badge for tasks whose executor isn't currently +// running. func (r *Repository) UpsertLatestLiveGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error { if snapshot == nil { return fmt.Errorf("snapshot is nil") @@ -86,10 +136,12 @@ func (r *Repository) UpsertLatestLiveGitSnapshot(ctx context.Context, snapshot * } defer func() { _ = tx.Rollback() }() + repositoryName := gitSnapshotRepositoryName(snapshot) + repositoryExpr := "COALESCE(" + dialect.JSONExtract(r.db.DriverName(), "metadata", "repository_name") + ", '')" if _, err := tx.ExecContext(ctx, r.db.Rebind(` DELETE FROM task_session_git_snapshots - WHERE session_id = ? AND snapshot_type = ? AND triggered_by = ? - `), snapshot.SessionID, string(models.SnapshotTypeStatusUpdate), TriggeredByLiveMonitor); err != nil { + WHERE session_id = ? AND snapshot_type = ? AND triggered_by = ? AND `+repositoryExpr+` = ? + `), snapshot.SessionID, string(models.SnapshotTypeStatusUpdate), TriggeredByLiveMonitor, repositoryName); err != nil { return fmt.Errorf("delete previous live snapshot: %w", err) } @@ -283,6 +335,68 @@ func (r *Repository) GetLatestGitSnapshotsBySessionIDs( return result, nil } +// GetLatestGitStatusSnapshotsBySessionIDs loads one authoritative snapshot per +// session and repository. Unlike GetLatestGitSnapshotsBySessionIDs, this +// preserves every repository needed to rebuild a multi-repository status +// notification while retaining the same archive/completion/live ranking. +func (r *Repository) GetLatestGitStatusSnapshotsBySessionIDs( + ctx context.Context, + sessionIDs []string, +) ([]*models.GitSnapshot, error) { + result := make([]*models.GitSnapshot, 0) + if len(sessionIDs) == 0 { + return result, nil + } + repositoryExpr := snapshotRepositoryExpr(r.db.DriverName(), "task_session_git_snapshots") + rankExpr := snapshotRankExprForRepository(r.db.DriverName()) + for _, chunk := range chunkIDs(sessionIDs, sqliteMaxHostParams) { + placeholders, args := buildInPlaceholders(chunk) + query := ` + SELECT id, session_id, snapshot_type, branch, remote_branch, head_commit, base_commit, + ahead, behind, files, triggered_by, metadata, created_at + FROM ( + SELECT id, session_id, snapshot_type, branch, remote_branch, head_commit, base_commit, + ahead, behind, files, triggered_by, metadata, created_at, + ROW_NUMBER() OVER ( + PARTITION BY session_id, ` + repositoryExpr + ` + ORDER BY` + rankExpr + ` + ) AS row_number + FROM task_session_git_snapshots + WHERE session_id IN (` + placeholders + `) + ) ranked + WHERE row_number = 1 + ORDER BY session_id, id + ` + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, err + } + for rows.Next() { + snapshot, scanErr := scanGitSnapshot(rows) + if scanErr != nil { + _ = rows.Close() + return nil, scanErr + } + result = append(result, snapshot) + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + _ = rows.Close() + } + return result, nil +} + +func gitSnapshotRepositoryName(snapshot *models.GitSnapshot) string { + if snapshot != nil && snapshot.Metadata != nil { + if repositoryName, ok := snapshot.Metadata["repository_name"].(string); ok { + return repositoryName + } + } + return "" +} + type gitSnapshotScanner interface { Scan(dest ...interface{}) error } diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_postgres_test.go b/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_postgres_test.go new file mode 100644 index 0000000000..9b9a97d676 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_postgres_test.go @@ -0,0 +1,81 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/testutil" +) + +// TestPostgresGitStatusSnapshotsRankEachRepositoryIndependently executes the +// same interleaved ranking scenario as the SQLite regression against the +// Postgres JSON expression and ROW_NUMBER() window. It skips unless the +// Postgres test database is configured. +func TestPostgresGitStatusSnapshotsRankEachRepositoryIndependently(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + const ( + taskID = "task-snapshot-repository-ranking-pg" + sessionID = "session-snapshot-repository-ranking-pg" + ) + seedPostgresTask(t, repo, taskID) + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ID: sessionID, TaskID: taskID}); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + + base := time.Date(2026, 8, 31, 10, 0, 0, 0, time.UTC) + snapshots := []*models.GitSnapshot{ + { + ID: "root-live-old-pg", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: TriggeredByLiveMonitor, + HeadCommit: "root-live-old", CreatedAt: base.Add(-time.Hour), + Metadata: map[string]interface{}{"repository_name": ""}, + }, + { + ID: "named-completed-pg", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: "agent_completed", + HeadCommit: "named-completed", CreatedAt: base, + Metadata: map[string]interface{}{"repository_name": "named"}, + }, + { + ID: "root-archive-pg", SessionID: sessionID, + SnapshotType: models.SnapshotTypeArchive, + HeadCommit: "root-archive", CreatedAt: base.Add(time.Hour), + }, + { + ID: "named-live-new-pg", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: TriggeredByLiveMonitor, + HeadCommit: "named-live-new", CreatedAt: base.Add(2 * time.Hour), + Metadata: map[string]interface{}{"repository_name": "named"}, + }, + } + for _, snapshot := range snapshots { + if err := repo.CreateGitSnapshot(ctx, snapshot); err != nil { + t.Fatalf("CreateGitSnapshot(%s): %v", snapshot.ID, err) + } + } + + got, err := repo.GetLatestGitStatusSnapshotsBySessionIDs(ctx, []string{sessionID}) + if err != nil { + t.Fatalf("GetLatestGitStatusSnapshotsBySessionIDs: %v", err) + } + if len(got) != 2 { + t.Fatalf("selected %d snapshots (%v), want one per repository", len(got), got) + } + byRepository := make(map[string]string, len(got)) + for _, snapshot := range got { + byRepository[gitSnapshotRepositoryName(snapshot)] = snapshot.ID + } + if got := byRepository[""]; got != "root-archive-pg" { + t.Errorf("root repository selected %q, want root-archive-pg", got) + } + if got := byRepository["named"]; got != "named-completed-pg" { + t.Errorf("named repository selected %q, want named-completed-pg", got) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_test.go b/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_test.go new file mode 100644 index 0000000000..de4f39ead5 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots_repository_test.go @@ -0,0 +1,78 @@ +package sqlite + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +// TestGetLatestGitStatusSnapshotsBySessionIDsRanksEachRepositoryIndependently +// covers the multi-repository selector's ranking boundary. A root archive must +// not be demoted by a newer live row from a named repository, and a named +// agent-completed row must not be demoted by a newer root archive. The root +// rows deliberately use absent and explicit-empty repository names to verify +// both normalize to the same partition. +func TestGetLatestGitStatusSnapshotsBySessionIDsRanksEachRepositoryIndependently(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + const ( + taskID = "task-snapshot-repository-ranking" + sessionID = "session-snapshot-repository-ranking" + ) + seedSessionForGit(t, repo, taskID, sessionID) + + base := time.Date(2026, 8, 31, 10, 0, 0, 0, time.UTC) + snapshots := []*models.GitSnapshot{ + { + ID: "root-live-old", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: TriggeredByLiveMonitor, + HeadCommit: "root-live-old", CreatedAt: base.Add(-time.Hour), + Metadata: map[string]interface{}{"repository_name": ""}, + }, + { + ID: "named-completed", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: "agent_completed", + HeadCommit: "named-completed", CreatedAt: base, + Metadata: map[string]interface{}{"repository_name": "named"}, + }, + { + ID: "root-archive", SessionID: sessionID, + SnapshotType: models.SnapshotTypeArchive, + HeadCommit: "root-archive", CreatedAt: base.Add(time.Hour), + // A nil metadata map serializes without a repository_name key. It + // must share the root partition with root-live-old's explicit "". + }, + { + ID: "named-live-new", SessionID: sessionID, + SnapshotType: models.SnapshotTypeStatusUpdate, TriggeredBy: TriggeredByLiveMonitor, + HeadCommit: "named-live-new", CreatedAt: base.Add(2 * time.Hour), + Metadata: map[string]interface{}{"repository_name": "named"}, + }, + } + for _, snapshot := range snapshots { + if err := repo.CreateGitSnapshot(ctx, snapshot); err != nil { + t.Fatalf("CreateGitSnapshot(%s): %v", snapshot.ID, err) + } + } + + got, err := repo.GetLatestGitStatusSnapshotsBySessionIDs(ctx, []string{sessionID}) + if err != nil { + t.Fatalf("GetLatestGitStatusSnapshotsBySessionIDs: %v", err) + } + if len(got) != 2 { + t.Fatalf("selected %d snapshots (%v), want one per repository", len(got), got) + } + + byRepository := make(map[string]string, len(got)) + for _, snapshot := range got { + byRepository[gitSnapshotRepositoryName(snapshot)] = snapshot.ID + } + if got := byRepository[""]; got != "root-archive" { + t.Errorf("root repository selected %q, want root-archive", got) + } + if got := byRepository["named"]; got != "named-completed" { + t.Errorf("named repository selected %q, want named-completed", got) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go b/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go index 7541b52a3c..c18f5ca6f2 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots_test.go @@ -508,6 +508,55 @@ func TestUpsertLatestLiveGitSnapshotKeepsOneLiveRowPerSession(t *testing.T) { assertJSONMapEqual(t, "Files", all[0].Files, map[string]interface{}{"a.go": "modified"}) } +func TestUpsertLatestLiveGitSnapshotKeepsOneLiveRowPerRepository(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + seedSessionForGit(t, repo, "task-snap-multi-repo", "session-snap-multi-repo") + base := time.Date(2026, 6, 1, 9, 0, 0, 0, time.UTC) + + for _, snapshot := range []*models.GitSnapshot{ + { + ID: "live-root-1", SessionID: "session-snap-multi-repo", HeadCommit: "root-1", + Metadata: map[string]interface{}{"repository_name": "", "timestamp": "2026-06-01T09:01:00Z"}, CreatedAt: base.Add(time.Minute), + }, + { + ID: "live-frontend-1", SessionID: "session-snap-multi-repo", HeadCommit: "frontend-1", + Metadata: map[string]interface{}{"repository_name": "frontend", "timestamp": "2026-06-01T09:01:00Z"}, CreatedAt: base.Add(time.Minute), + }, + { + ID: "live-root-2", SessionID: "session-snap-multi-repo", HeadCommit: "root-2", + Metadata: map[string]interface{}{"repository_name": "", "timestamp": "2026-06-01T09:02:00Z"}, CreatedAt: base.Add(2 * time.Minute), + }, + } { + if err := repo.UpsertLatestLiveGitSnapshot(ctx, snapshot); err != nil { + t.Fatalf("UpsertLatestLiveGitSnapshot(%s): %v", snapshot.ID, err) + } + } + + liveRows := countRows(t, repo, + `SELECT COUNT(1) FROM task_session_git_snapshots WHERE session_id = ? AND triggered_by = ?`, + "session-snap-multi-repo", TriggeredByLiveMonitor) + if liveRows != 2 { + t.Fatalf("live_monitor rows = %d, want one row per repository", liveRows) + } + + all, err := repo.GetGitSnapshotsBySession(ctx, "session-snap-multi-repo", 0) + if err != nil { + t.Fatalf("GetGitSnapshotsBySession: %v", err) + } + byRepository := make(map[string]string, len(all)) + for _, snapshot := range all { + repositoryName, _ := snapshot.Metadata["repository_name"].(string) + byRepository[repositoryName] = snapshot.HeadCommit + } + if got := byRepository[""]; got != "root-2" { + t.Errorf("root live snapshot = %q, want root-2", got) + } + if got := byRepository["frontend"]; got != "frontend-1" { + t.Errorf("frontend live snapshot = %q, want frontend-1", got) + } +} + func TestUpsertLatestLiveGitSnapshotGeneratesIDAndRejectsNil(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 03628747f6..31c9e4ad24 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -2386,6 +2386,64 @@ func (r *Repository) ListTaskSessions(ctx context.Context, taskID string) ([]*mo return r.loadWorktreesBatch(ctx, sessions) } +// ListTaskSessionsByTaskEnvironment returns every session bound to an +// environment, including inherited sessions whose task differs from the +// environment owner's task. The environment binding, rather than task row +// ownership, is the source-selection boundary for shared workspaces. +func (r *Repository) ListTaskSessionsByTaskEnvironment(ctx context.Context, environmentID string) ([]*models.TaskSession, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTaskSessionsByTaskEnvironment") + defer span.End() + if environmentID == "" { + return []*models.TaskSession{}, nil + } + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind( + `SELECT `+taskSessionSelectCols+` `+taskSessionFromClause+` WHERE ts.task_environment_id = ? ORDER BY ts.started_at DESC`, + ), environmentID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + sessions, err := r.scanTaskSessions(ctx, rows) + if err != nil { + return nil, err + } + return r.loadWorktreesBatch(ctx, sessions) +} + +// GetTaskSessionWorkspacePathsByTaskEnvironment returns only the raw workspace +// paths recorded on session rows bound to an environment. It deliberately does +// not use the effective environment projection because an empty or historical +// session path must remain distinguishable from the canonical path. +func (r *Repository) GetTaskSessionWorkspacePathsByTaskEnvironment(ctx context.Context, environmentID string) (map[string]string, error) { + paths := make(map[string]string) + if environmentID == "" { + return paths, nil + } + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` + SELECT id, workspace_path + FROM task_sessions + WHERE task_environment_id = ? + `), environmentID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var sessionID string + var workspacePath sql.NullString + if err := rows.Scan(&sessionID, &workspacePath); err != nil { + return nil, err + } + paths[sessionID] = workspacePath.String + } + if err := rows.Err(); err != nil { + return nil, err + } + return paths, nil +} + // ListActiveTaskSessions returns all active agent sessions across all tasks func (r *Repository) ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) { ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListActiveTaskSessions") diff --git a/apps/web/e2e/tests/session/session-tab-management.spec.ts b/apps/web/e2e/tests/session/session-tab-management.spec.ts index 48baafd7bc..1b5e35086a 100644 --- a/apps/web/e2e/tests/session/session-tab-management.spec.ts +++ b/apps/web/e2e/tests/session/session-tab-management.spec.ts @@ -340,6 +340,65 @@ test.describe("Session tab management — close behavior", () => { } }); + test("keeps dirty Changes status after sibling hydration", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(150_000); + // Attach before setup so the capture also observes the reload-driven + // subscriptions used by this regression. + const traffic = attachGatewayTrafficCapture(testPage); + const { session, session1Id, session2Id } = await createTaskWithTwoSessions( + testPage, + apiClient, + seedData, + "Sibling Hydration Keeps Changes", + ); + const localChange = "sibling-hydration-change.txt"; + const localChangePath = path.join(seedData.repositoryPath, localChange); + fs.writeFileSync(localChangePath, "keep this change visible\n"); + + const receivedGitEvent = (sessionId: string, fromIndex: number) => + traffic.frames + .slice(fromIndex) + .some( + (frame) => + frame.direction === "received" && + frame.action === "session.git.event" && + frame.sessionId === sessionId, + ); + + try { + traffic.frames.length = 0; + const beforeReload = traffic.frames.length; + await testPage.reload(); + await session.waitForLoad(); + + await session.sessionTabBySessionId(session1Id).click(); + await expect + .poll(() => receivedGitEvent(session1Id, beforeReload), { + timeout: 20_000, + message: "waiting for the first sibling git-status hydration", + }) + .toBe(true); + await session.clickTab("Changes"); + await expect(session.changesFileRow(localChange)).toBeVisible({ timeout: 20_000 }); + + await session.sessionTabBySessionId(session2Id).click(); + await expect + .poll(() => receivedGitEvent(session2Id, beforeReload), { + timeout: 20_000, + message: "waiting for the second sibling git-status hydration", + }) + .toBe(true); + await session.clickTab("Changes"); + await expect(session.changesFileRow(localChange)).toBeVisible({ timeout: 20_000 }); + } finally { + fs.rmSync(localChangePath, { force: true }); + } + }); + test("deleting then immediately switching tasks does not resurrect the tab", async ({ testPage, apiClient, diff --git a/apps/web/hooks/domains/session/use-session-messages.test.ts b/apps/web/hooks/domains/session/use-session-messages.test.ts index 02ef523293..2b17d20899 100644 --- a/apps/web/hooks/domains/session/use-session-messages.test.ts +++ b/apps/web/hooks/domains/session/use-session-messages.test.ts @@ -356,11 +356,13 @@ describe("session subscription hydration ordering", () => { it("does not request messages before subscription acknowledgement", async () => { const readiness = deferred(); + const response = deferred<{ messages: Message[]; has_more: boolean }>(); mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(readiness.promise); mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({ ready: readiness.promise, unsubscribe: vi.fn(), }); + mockWebSocketClient.request.mockReturnValue(response.promise); const { unmount } = renderHook(() => useSessionMessages("sess-1")); @@ -379,6 +381,11 @@ describe("session subscription hydration ordering", () => { 10000, ); expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1); + + await act(async () => { + response.resolve({ messages: [], has_more: false }); + await response.promise; + }); unmount(); }); diff --git a/apps/web/lib/state/slices/session-runtime/git-status-state.ts b/apps/web/lib/state/slices/session-runtime/git-status-state.ts new file mode 100644 index 0000000000..0efe4c1148 --- /dev/null +++ b/apps/web/lib/state/slices/session-runtime/git-status-state.ts @@ -0,0 +1,314 @@ +import type { FileChangeFacet, FileInfo, GitStatusEntry, SessionRuntimeSliceState } from "./types"; +import { createDebugLogger, isDebug } from "@/lib/debug/log"; + +const debugGit = createDebugLogger("git-status:store"); + +/** Compute total additions/deletions across all files. */ +function computeFileStats(files: Record | undefined): { + additions: number; + deletions: number; +} { + if (!files) return { additions: 0, deletions: 0 }; + let additions = 0; + let deletions = 0; + for (const f of Object.values(files)) { + additions += f.additions || 0; + deletions += f.deletions || 0; + } + return { additions, deletions }; +} + +function sameStringList(existing: string[] | undefined, incoming: string[] | undefined): boolean { + const a = existing ?? []; + const b = incoming ?? []; + if (a.length !== b.length) return false; + const sortedA = [...a].sort(); + const sortedB = [...b].sort(); + return sortedA.every((value, index) => value === sortedB[index]); +} + +const COMPARABLE_FILE_FIELDS = [ + "path", + "status", + "staged", + "additions", + "deletions", + "old_path", + "diff", + "diff_skip_reason", + "staged_change", + "unstaged_change", + "repository_name", + "is_submodule", +] as const; + +function comparableChangeFacet(facet: FileChangeFacet | undefined): string { + if (!facet) return ""; + return [ + facet.status, + facet.additions ?? 0, + facet.deletions ?? 0, + facet.old_path ?? "", + facet.diff ?? "", + facet.diff_skip_reason ?? "", + ].join("\0"); +} + +function comparableFileInfo(file: FileInfo) { + return { + path: file.path, + status: file.status, + staged: file.staged, + additions: file.additions ?? 0, + deletions: file.deletions ?? 0, + old_path: file.old_path ?? "", + diff: file.diff ?? "", + diff_skip_reason: file.diff_skip_reason ?? "", + staged_change: comparableChangeFacet(file.staged_change), + unstaged_change: comparableChangeFacet(file.unstaged_change), + repository_name: file.repository_name ?? "", + is_submodule: file.is_submodule ?? false, + }; +} + +function sameFileInfo(existing: FileInfo | undefined, incoming: FileInfo | undefined): boolean { + if (!existing || !incoming) return existing === incoming; + const a = comparableFileInfo(existing); + const b = comparableFileInfo(incoming); + return COMPARABLE_FILE_FIELDS.every((field) => a[field] === b[field]); +} + +function sameFiles( + existingFiles: Record | undefined, + newFiles: Record | undefined, +): boolean { + if (!existingFiles || !newFiles) return existingFiles === newFiles; + const existingFileKeys = Object.keys(existingFiles).sort(); + const newFileKeys = Object.keys(newFiles).sort(); + if (existingFileKeys.length !== newFileKeys.length) return false; + for (let i = 0; i < existingFileKeys.length; i += 1) { + const key = existingFileKeys[i]; + if (key !== newFileKeys[i]) return false; + if (!sameFileInfo(existingFiles[key], newFiles[key])) return false; + } + return true; +} + +function hasBranchSummaryChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + return ( + hasBranchReferencesChanged(existing, incoming) || + hasComparisonSummaryChanged(existing, incoming) || + existing.ahead !== incoming.ahead || + existing.behind !== incoming.behind || + existing.remote_ahead !== incoming.remote_ahead || + existing.remote_behind !== incoming.remote_behind || + existing.remote_head_commit !== incoming.remote_head_commit || + (existing.repository_name ?? "") !== (incoming.repository_name ?? "") || + existing.is_submodule !== incoming.is_submodule || + existing.branch_additions !== incoming.branch_additions || + existing.branch_deletions !== incoming.branch_deletions + ); +} + +function hasBranchReferencesChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + return ( + existing.branch !== incoming.branch || + existing.remote_branch !== incoming.remote_branch || + existing.head_commit !== incoming.head_commit || + existing.base_commit !== incoming.base_commit + ); +} + +function hasComparisonSummaryChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + return ( + existing.comparison_target !== incoming.comparison_target || + existing.comparison_status !== incoming.comparison_status || + existing.comparison_error_code !== incoming.comparison_error_code + ); +} + +function hasFileListsChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + return ( + !sameStringList(existing.modified, incoming.modified) || + !sameStringList(existing.added, incoming.added) || + !sameStringList(existing.deleted, incoming.deleted) || + !sameStringList(existing.untracked, incoming.untracked) || + !sameStringList(existing.renamed, incoming.renamed) + ); +} + +function hasFileStatsChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + // Fast early-exit: aggregate totals differ → sameFiles would also return false, + // but this avoids the per-file deep comparison when the gross numbers differ. + const existingTotal = computeFileStats(existing.files); + const newTotal = computeFileStats(incoming.files); + return ( + existingTotal.additions !== newTotal.additions || existingTotal.deletions !== newTotal.deletions + ); +} + +/** Compare two git status entries to determine if a meaningful change occurred. */ +export function hasGitStatusChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + // The backend also emits fresh snapshots for focus/startup/poll events. Those + // can carry a new timestamp for identical git data, so timestamp alone must + // not force a store update or diff-cache invalidation. + return ( + hasBranchSummaryChanged(existing, incoming) || + hasFileListsChanged(existing, incoming) || + hasFileStatsChanged(existing, incoming) || + !sameFiles(existing.files, incoming.files) + ); +} + +function gitStatusTimestamp(timestamp: string | null | undefined): number | null { + if (!timestamp) return null; + const parsed = Date.parse(timestamp); + return Number.isFinite(parsed) ? parsed : null; +} + +function acceptsGitStatusTimestamp( + existing: GitStatusEntry | undefined, + incoming: GitStatusEntry, +): boolean { + const existingTimestamp = gitStatusTimestamp(existing?.timestamp); + const incomingTimestamp = gitStatusTimestamp(incoming.timestamp); + if (existingTimestamp !== null && incomingTimestamp === null) return false; + if ( + existingTimestamp !== null && + incomingTimestamp !== null && + incomingTimestamp < existingTimestamp + ) { + return false; + } + return true; +} + +function advancesGitStatusTimestamp(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { + const existingTimestamp = gitStatusTimestamp(existing.timestamp); + const incomingTimestamp = gitStatusTimestamp(incoming.timestamp); + return ( + incomingTimestamp !== null && + (existingTimestamp === null || incomingTimestamp > existingTimestamp) + ); +} + +type GitStatusUpdateContext = { + state: SessionRuntimeSliceState; + sessionId: string; + envKey: string; + repoName: string; + repoMap: Record; + existing: GitStatusEntry | undefined; + incoming: GitStatusEntry; + changed: boolean; + timestampAdvanced: boolean; +}; + +function logRejectedGitStatus( + sessionId: string, + envKey: string, + repoName: string, + existing: GitStatusEntry | undefined, + incoming: GitStatusEntry, +) { + if (!isDebug()) return; + debugGit("setGitStatus", { + sessionId, + envKey, + usingFallbackKey: envKey === sessionId, + repoName, + prevTimestamp: existing?.timestamp, + nextTimestamp: incoming.timestamp, + orderingRejected: true, + }); +} + +function logAcceptedGitStatus(context: GitStatusUpdateContext) { + if (!isDebug()) return; + debugGit("setGitStatus", { + sessionId: context.sessionId, + envKey: context.envKey, + usingFallbackKey: context.envKey === context.sessionId, + repoName: context.repoName, + prevFileCount: Object.keys(context.existing?.files ?? {}).length, + nextFileCount: Object.keys(context.incoming.files ?? {}).length, + prevRepoKeys: Object.keys(context.repoMap), + willMutate: context.changed, + prevTimestamp: context.existing?.timestamp, + nextTimestamp: context.incoming.timestamp, + orderingRejected: false, + timestampAdvanced: context.timestampAdvanced, + }); +} + +function updateRepoStatus(context: GitStatusUpdateContext) { + if (context.changed) { + context.repoMap[context.repoName] = context.incoming; + } else if (context.existing && context.timestampAdvanced) { + // Preserve the deep-content identity and cumulative diff cache while + // advancing the ordering watermark for a duplicate status snapshot. + context.existing.timestamp = context.incoming.timestamp; + } +} + +function mirrorLegacyGitStatus(context: GitStatusUpdateContext): boolean { + if (context.repoName !== "") { + // Multi-repo: only mirror into the legacy map when this repo's entry changed. + if (context.changed) context.state.gitStatus.byEnvironmentId[context.envKey] = context.incoming; + else if ( + context.timestampAdvanced && + context.state.gitStatus.byEnvironmentId[context.envKey]?.repository_name === context.repoName + ) { + context.state.gitStatus.byEnvironmentId[context.envKey].timestamp = + context.incoming.timestamp; + } + return context.changed; + } + + // The empty-repo entry and byEnvironmentId track together (written and cleared + // as a pair), so existingRepo presence/equality matches the env entry — reuse + // changed instead of comparing diffs again. + const legacyChanged = !context.state.gitStatus.byEnvironmentId[context.envKey] || context.changed; + if (legacyChanged) context.state.gitStatus.byEnvironmentId[context.envKey] = context.incoming; + else if (context.timestampAdvanced && context.state.gitStatus.byEnvironmentId[context.envKey]) { + context.state.gitStatus.byEnvironmentId[context.envKey].timestamp = context.incoming.timestamp; + } + return legacyChanged; +} + +/** Write a git-status update into the env/per-repo maps, skipping writes when + * nothing meaningfully changed. Returns whether git state changed so the WS + * handler can invalidate derived caches without repeating the deep comparison. */ +export function applyGitStatus( + state: SessionRuntimeSliceState, + sessionId: string, + gitStatus: GitStatusEntry, +): boolean { + const envKey = state.environmentIdBySessionId[sessionId] ?? sessionId; + const repoName = gitStatus.repository_name ?? ""; + const repoMap = (state.gitStatus.byEnvironmentRepo[envKey] ??= {}); + const existingRepo = repoMap[repoName]; + if (!acceptsGitStatusTimestamp(existingRepo, gitStatus)) { + logRejectedGitStatus(sessionId, envKey, repoName, existingRepo, gitStatus); + return false; + } + + const timestampAdvanced = existingRepo + ? advancesGitStatusTimestamp(existingRepo, gitStatus) + : true; + const changed = !existingRepo || hasGitStatusChanged(existingRepo, gitStatus); + const context: GitStatusUpdateContext = { + state, + sessionId, + envKey, + repoName, + repoMap, + existing: existingRepo, + incoming: gitStatus, + changed, + timestampAdvanced, + }; + logAcceptedGitStatus(context); + updateRepoStatus(context); + return mirrorLegacyGitStatus(context); +} diff --git a/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts b/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts index 195664a869..d2fe3c738a 100644 --- a/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts +++ b/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts @@ -1,16 +1,7 @@ import type { StateCreator } from "zustand"; -import type { - SessionRuntimeSlice, - SessionRuntimeSliceState, - SessionPollMode, - GitStatusEntry, - FileInfo, - FileChangeFacet, -} from "./types"; -import { createDebugLogger, isDebug } from "@/lib/debug/log"; +import type { SessionRuntimeSlice, SessionRuntimeSliceState, SessionPollMode } from "./types"; import { normalizeGitStatusEntry } from "./git-status-normalizer"; - -const debugGit = createDebugLogger("git-status:store"); +import { applyGitStatus } from "./git-status-state"; const maxProcessOutputBytes = 2 * 1024 * 1024; // Shell + terminal streams are unbounded over a session's lifetime; cap them at @@ -18,209 +9,6 @@ const maxProcessOutputBytes = 2 * 1024 * 1024; // store without limit (the xterm view only renders the tail anyway). const maxShellOutputBytes = 2 * 1024 * 1024; -/** Compute total additions/deletions across all files. */ -function computeFileStats(files: Record | undefined): { - additions: number; - deletions: number; -} { - if (!files) return { additions: 0, deletions: 0 }; - let additions = 0; - let deletions = 0; - for (const f of Object.values(files)) { - additions += f.additions || 0; - deletions += f.deletions || 0; - } - return { additions, deletions }; -} - -function sameStringList(existing: string[] | undefined, incoming: string[] | undefined): boolean { - const a = existing ?? []; - const b = incoming ?? []; - if (a.length !== b.length) return false; - const sortedA = [...a].sort(); - const sortedB = [...b].sort(); - return sortedA.every((value, index) => value === sortedB[index]); -} - -const COMPARABLE_FILE_FIELDS = [ - "path", - "status", - "staged", - "additions", - "deletions", - "old_path", - "diff", - "diff_skip_reason", - "staged_change", - "unstaged_change", - "repository_name", - "is_submodule", -] as const; - -function comparableChangeFacet(facet: FileChangeFacet | undefined): string { - if (!facet) return ""; - return [ - facet.status, - facet.additions ?? 0, - facet.deletions ?? 0, - facet.old_path ?? "", - facet.diff ?? "", - facet.diff_skip_reason ?? "", - ].join("\0"); -} - -function comparableFileInfo(file: FileInfo) { - return { - path: file.path, - status: file.status, - staged: file.staged, - additions: file.additions ?? 0, - deletions: file.deletions ?? 0, - old_path: file.old_path ?? "", - diff: file.diff ?? "", - diff_skip_reason: file.diff_skip_reason ?? "", - staged_change: comparableChangeFacet(file.staged_change), - unstaged_change: comparableChangeFacet(file.unstaged_change), - repository_name: file.repository_name ?? "", - is_submodule: file.is_submodule ?? false, - }; -} - -function sameFileInfo(existing: FileInfo | undefined, incoming: FileInfo | undefined): boolean { - if (!existing || !incoming) return existing === incoming; - const a = comparableFileInfo(existing); - const b = comparableFileInfo(incoming); - return COMPARABLE_FILE_FIELDS.every((field) => a[field] === b[field]); -} - -function sameFiles( - existingFiles: Record | undefined, - newFiles: Record | undefined, -): boolean { - if (!existingFiles || !newFiles) return existingFiles === newFiles; - const existingFileKeys = Object.keys(existingFiles).sort(); - const newFileKeys = Object.keys(newFiles).sort(); - if (existingFileKeys.length !== newFileKeys.length) return false; - for (let i = 0; i < existingFileKeys.length; i += 1) { - const key = existingFileKeys[i]; - if (key !== newFileKeys[i]) return false; - if (!sameFileInfo(existingFiles[key], newFiles[key])) return false; - } - return true; -} - -function hasBranchSummaryChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - return ( - hasBranchReferencesChanged(existing, incoming) || - hasComparisonSummaryChanged(existing, incoming) || - existing.ahead !== incoming.ahead || - existing.behind !== incoming.behind || - existing.remote_ahead !== incoming.remote_ahead || - existing.remote_behind !== incoming.remote_behind || - existing.remote_head_commit !== incoming.remote_head_commit || - (existing.repository_name ?? "") !== (incoming.repository_name ?? "") || - existing.is_submodule !== incoming.is_submodule || - existing.branch_additions !== incoming.branch_additions || - existing.branch_deletions !== incoming.branch_deletions - ); -} - -function hasBranchReferencesChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - return ( - existing.branch !== incoming.branch || - existing.remote_branch !== incoming.remote_branch || - existing.head_commit !== incoming.head_commit || - existing.base_commit !== incoming.base_commit - ); -} - -function hasComparisonSummaryChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - return ( - existing.comparison_target !== incoming.comparison_target || - existing.comparison_status !== incoming.comparison_status || - existing.comparison_error_code !== incoming.comparison_error_code - ); -} - -function hasFileListsChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - return ( - !sameStringList(existing.modified, incoming.modified) || - !sameStringList(existing.added, incoming.added) || - !sameStringList(existing.deleted, incoming.deleted) || - !sameStringList(existing.untracked, incoming.untracked) || - !sameStringList(existing.renamed, incoming.renamed) - ); -} - -function hasFileStatsChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - // Fast early-exit: aggregate totals differ → sameFiles would also return false, - // but this avoids the per-file deep comparison when the gross numbers differ. - const existingTotal = computeFileStats(existing.files); - const newTotal = computeFileStats(incoming.files); - return ( - existingTotal.additions !== newTotal.additions || existingTotal.deletions !== newTotal.deletions - ); -} - -/** Compare two git status entries to determine if a meaningful change occurred. */ -export function hasGitStatusChanged(existing: GitStatusEntry, incoming: GitStatusEntry): boolean { - // The backend also emits fresh snapshots for focus/startup/poll events. Those - // can carry a new timestamp for identical git data, so timestamp alone must - // not force a store update or diff-cache invalidation. - return ( - hasBranchSummaryChanged(existing, incoming) || - hasFileListsChanged(existing, incoming) || - hasFileStatsChanged(existing, incoming) || - !sameFiles(existing.files, incoming.files) - ); -} - -/** Write a git-status update into the env/per-repo maps, skipping writes when - * nothing meaningfully changed. Returns whether git state changed so the WS - * handler can invalidate derived caches without repeating the deep (per-file, - * full-diff-string) comparison — the dominant cost under a massive rebase. */ -function applyGitStatus( - state: SessionRuntimeSliceState, - sessionId: string, - gitStatus: GitStatusEntry, -): boolean { - const envKey = state.environmentIdBySessionId[sessionId] ?? sessionId; - // Multi-repo: when the update is tagged with repository_name, route it into - // the per-repo map. Single-repo updates (no name) keep the legacy single- - // status path; the per-repo map mirrors the same entry under an empty key so - // consumers using only byEnvironmentRepo still see it. - const repoName = gitStatus.repository_name ?? ""; - const repoMap = (state.gitStatus.byEnvironmentRepo[envKey] ??= {}); - const existingRepo = repoMap[repoName]; - const repoChanged = !existingRepo || hasGitStatusChanged(existingRepo, gitStatus); - if (isDebug()) { - debugGit("setGitStatus", { - sessionId, - envKey, - usingFallbackKey: envKey === sessionId, - repoName, - prevFileCount: Object.keys(existingRepo?.files ?? {}).length, - nextFileCount: Object.keys(gitStatus.files ?? {}).length, - prevRepoKeys: Object.keys(repoMap), - willMutate: repoChanged, - }); - } - if (repoChanged) { - repoMap[repoName] = gitStatus; - } - if (repoName !== "") { - // Multi-repo: only mirror into the legacy map when this repo's entry changed. - if (repoChanged) state.gitStatus.byEnvironmentId[envKey] = gitStatus; - return repoChanged; - } - // The empty-repo entry and byEnvironmentId track together (written and cleared - // as a pair), so existingRepo presence/equality matches the env entry — reuse - // repoChanged instead of comparing diffs again. - const changed = !state.gitStatus.byEnvironmentId[envKey] || repoChanged; - if (changed) state.gitStatus.byEnvironmentId[envKey] = gitStatus; - return changed; -} - function trimTailBytes(value: string, maxBytes: number) { if (value.length <= maxBytes) { return value; diff --git a/apps/web/lib/state/slices/session-runtime/set-git-status-return.test.ts b/apps/web/lib/state/slices/session-runtime/set-git-status-return.test.ts index c0b6deaf2b..87a3aff716 100644 --- a/apps/web/lib/state/slices/session-runtime/set-git-status-return.test.ts +++ b/apps/web/lib/state/slices/session-runtime/set-git-status-return.test.ts @@ -46,11 +46,17 @@ describe("setGitStatus change reporting (single deep compare)", () => { ); }); - it("does not rewrite byEnvironmentId for a duplicate snapshot", () => { + it("advances the timestamp watermark for a duplicate snapshot", () => { store.getState().setGitStatus(SESSION, status()); - const ref = store.getState().gitStatus.byEnvironmentId[SESSION]; store.getState().setGitStatus(SESSION, status({ timestamp: NEWER_TIMESTAMP })); - expect(store.getState().gitStatus.byEnvironmentId[SESSION]).toBe(ref); + expect(store.getState().gitStatus.byEnvironmentId[SESSION]).toMatchObject({ + modified: ["a.ts"], + timestamp: NEWER_TIMESTAMP, + }); + expect(store.getState().gitStatus.byEnvironmentRepo[SESSION][""]).toMatchObject({ + modified: ["a.ts"], + timestamp: NEWER_TIMESTAMP, + }); }); it("returns true when a file's diff content changes", () => { diff --git a/apps/web/lib/ws/handlers/git-status.test.ts b/apps/web/lib/ws/handlers/git-status.test.ts index 073762456e..bc45977b39 100644 --- a/apps/web/lib/ws/handlers/git-status.test.ts +++ b/apps/web/lib/ws/handlers/git-status.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, vi, beforeEach } from "vitest"; +import { describe, it, expect, vi } from "vitest"; import { create, type StoreApi } from "zustand"; import { immer } from "zustand/middleware/immer"; import { createSessionRuntimeSlice } from "@/lib/state/slices/session-runtime/session-runtime-slice"; @@ -50,10 +50,21 @@ function gitStatusHandler(store: StoreApi) { return handler; } -function statusUpdateEvent(timestamp: string, diff = "-old\n+new"): GitStatusUpdateEvent { +function freshStore() { + invalidateCumulativeDiffCacheMock.mockClear(); + const store = makeStore(); + seedSessionCommits(store); + return store; +} + +function statusUpdateEvent( + timestamp: string, + diff = "-old\n+new", + sessionId = SESSION, +): GitStatusUpdateEvent { return { type: "status_update", - session_id: SESSION, + session_id: sessionId, timestamp, status: { branch: "main", @@ -135,16 +146,9 @@ function seedSessionCommits(store: StoreApi) { ]); } -describe("git-status WS handler — stale-while-revalidate", () => { - let store: StoreApi; - - beforeEach(() => { - invalidateCumulativeDiffCacheMock.mockClear(); - store = makeStore(); - seedSessionCommits(store); - }); - +describe("git-status WS handler — commit events", () => { it("commits_reset bumps refetchTrigger and keeps existing commits visible", () => { + const store = freshStore(); const handler = gitStatusHandler(store); handler( @@ -167,6 +171,7 @@ describe("git-status WS handler — stale-while-revalidate", () => { }); it("branch_switched bumps refetchTrigger and keeps existing commits visible", () => { + const store = freshStore(); const handler = gitStatusHandler(store); handler( @@ -188,16 +193,8 @@ describe("git-status WS handler — stale-while-revalidate", () => { expect(state.sessionCommits.byEnvironmentId[SESSION]).toHaveLength(1); }); - it("does not invalidate cumulative diff for duplicate status snapshots", () => { - const handler = gitStatusHandler(store); - - handler(gitEvent(statusUpdateEvent(STATUS_TIME_1))); - handler(gitEvent(statusUpdateEvent(STATUS_TIME_2))); - - expect(invalidateCumulativeDiffCacheMock).toHaveBeenCalledTimes(1); - }); - it("stores the status-level submodule marker", () => { + const store = freshStore(); const handler = gitStatusHandler(store); const event = statusUpdateEvent(STATUS_TIME_1); event.status.is_submodule = true; @@ -208,6 +205,7 @@ describe("git-status WS handler — stale-while-revalidate", () => { }); it("retains the commit and upstream evidence from a status event", () => { + const store = freshStore(); const handler = gitStatusHandler(store); const event = statusUpdateEvent(STATUS_TIME_1); event.status = { @@ -229,8 +227,22 @@ describe("git-status WS handler — stale-while-revalidate", () => { remote_behind: 3, }); }); +}); + +describe("git-status WS handler — status updates", () => { + it("does not invalidate cumulative diff for duplicate status snapshots", () => { + const store = freshStore(); + const handler = gitStatusHandler(store); + + handler(gitEvent(statusUpdateEvent(STATUS_TIME_1))); + handler(gitEvent(statusUpdateEvent(STATUS_TIME_2))); + + expect(invalidateCumulativeDiffCacheMock).toHaveBeenCalledTimes(1); + expect(store.getState().gitStatus.byEnvironmentId[SESSION].timestamp).toBe(STATUS_TIME_2); + }); it("invalidates cumulative diff when status diff content changes", () => { + const store = freshStore(); const handler = gitStatusHandler(store); handler(gitEvent(statusUpdateEvent(STATUS_TIME_1, "-old\n+new"))); @@ -240,16 +252,73 @@ describe("git-status WS handler — stale-while-revalidate", () => { }); it("does not invalidate or overwrite env status for duplicate sibling-repo snapshots", () => { + const store = freshStore(); const handler = gitStatusHandler(store); handler(gitEvent(repoStatusUpdateEvent(STATUS_TIME_1, "frontend", "frontend.tsx"))); handler(gitEvent(repoStatusUpdateEvent(STATUS_TIME_2, "backend", "backend.go"))); - const envAfterBackend = store.getState().gitStatus.byEnvironmentId[SESSION]; invalidateCumulativeDiffCacheMock.mockClear(); handler(gitEvent(repoStatusUpdateEvent("2026-05-28T00:00:03Z", "backend", "backend.go"))); - expect(store.getState().gitStatus.byEnvironmentId[SESSION]).toBe(envAfterBackend); + expect(store.getState().gitStatus.byEnvironmentId[SESSION]).toMatchObject({ + modified: ["backend.go"], + timestamp: "2026-05-28T00:00:03Z", + }); expect(invalidateCumulativeDiffCacheMock).not.toHaveBeenCalled(); }); }); + +describe("git-status WS handler — shared environment ordering", () => { + it("does not overwrite newer environment status with an older sibling event", () => { + const store = freshStore(); + const handler = gitStatusHandler(store); + store.getState().registerSessionEnvironment(SESSION, "env-1"); + store.getState().registerSessionEnvironment("sess-2", "env-1"); + + handler(gitEvent(statusUpdateEvent(STATUS_TIME_2))); + const olderCleanEvent = statusUpdateEvent(STATUS_TIME_1, "", "sess-2"); + olderCleanEvent.status = { + ...olderCleanEvent.status, + modified: [], + files: {}, + }; + handler(gitEvent(olderCleanEvent)); + + expect(store.getState().gitStatus.byEnvironmentId["env-1"].modified).toEqual(["a.ts"]); + expect(invalidateCumulativeDiffCacheMock).toHaveBeenCalledTimes(1); + }); + + it("does not let an undated sibling event replace dated environment state", () => { + const store = freshStore(); + const handler = gitStatusHandler(store); + store.getState().registerSessionEnvironment(SESSION, "env-1"); + store.getState().registerSessionEnvironment("sess-2", "env-1"); + + handler(gitEvent(statusUpdateEvent(STATUS_TIME_1))); + invalidateCumulativeDiffCacheMock.mockClear(); + const undatedCleanEvent = statusUpdateEvent("not-a-timestamp", "", "sess-2"); + undatedCleanEvent.status = { + ...undatedCleanEvent.status, + modified: [], + files: {}, + }; + handler(gitEvent(undatedCleanEvent)); + + expect(store.getState().gitStatus.byEnvironmentId["env-1"].modified).toEqual(["a.ts"]); + expect(invalidateCumulativeDiffCacheMock).not.toHaveBeenCalled(); + }); + + it("accepts equal-time content changes", () => { + const store = freshStore(); + const handler = gitStatusHandler(store); + + handler(gitEvent(statusUpdateEvent(STATUS_TIME_1, "-old\n+new"))); + handler(gitEvent(statusUpdateEvent(STATUS_TIME_1, "-old\n+newer"))); + + expect(store.getState().gitStatus.byEnvironmentId[SESSION].files["a.ts"].diff).toBe( + "-old\n+newer", + ); + expect(invalidateCumulativeDiffCacheMock).toHaveBeenCalledTimes(2); + }); +}); diff --git a/docs/decisions/2026-08-30-environment-owned-git-status.md b/docs/decisions/2026-08-30-environment-owned-git-status.md new file mode 100644 index 0000000000..ac26776348 --- /dev/null +++ b/docs/decisions/2026-08-30-environment-owned-git-status.md @@ -0,0 +1,73 @@ +# ADR-2026-08-30-environment-owned-git-status: Project Current Git Status From the Task Environment + +**Status:** proposed +**Date:** 2026-08-30 +**Area:** backend, frontend, protocol + +## Context + +Multiple task sessions share one `TaskEnvironment` and one physical workspace. +Some sessions can be inherited or shared across task boundaries while still +binding to that environment. +Git-status transport is still session-scoped. Each mounted consumer can +subscribe with a different session ID, and each status event is then stored +under the shared environment key in the frontend. + +This creates competing writers. A delayed sibling session can publish a status +from an obsolete workspace or an older database snapshot. The frontend then +replaces the current environment status with that result. In the observed +incident, 91 dirty files appeared and then disappeared when an older sibling's +clean snapshot arrived later. + +## Decision + +Treat the task environment as the authority for current Git status. Treat a +session ID only as a request and delivery handle. + +Before the backend publishes current status, it must prove that the source +session is bound to the task environment and has a raw recorded workspace path +equal to the environment's canonical workspace. A live execution must match +that workspace; a recovered execution may have an empty environment ID only +after the session binding has been verified. A persisted snapshot is eligible +only when its source session has the same canonical binding. Among eligible +snapshots, the newest observation wins independently for each repository. The +outgoing event keeps the requested session ID for protocol compatibility. + +The frontend keeps Git status under the environment key and applies a monotonic +timestamp rule per repository. An older observation cannot replace a newer +one. This is a defense against response reordering after the backend has +validated workspace authority. + +Keep snapshots attributed to sessions. Resolve current environment authority +at projection time. Do not add a second persisted Git-status owner. + +## Consequences + +- Sibling subscriptions can no longer publish conflicting current status for + one task environment. +- A clean canonical status can replace dirty state. The rule does not prefer + non-empty file lists. +- Historical sessions with an obsolete workspace path remain in the audit + record but cannot supply current environment status. +- Multi-repository status remains session-attributed and is retained one row per + repository for fallback hydration. +- A lookup failure can temporarily produce no hydration event. It cannot clear + correct frontend state with unverified data. +- PR #3167 remains necessary because it repairs the workspace path used when a + session resumes. This decision protects the projection boundary before and + after that repair. +- No database migration or WebSocket action change is required. + +## Alternatives Considered + +1. **Accept the last event received.** Rejected because network and hydration + order do not define workspace authority. +2. **Reject only older timestamps in the frontend.** Rejected as the complete + fix because a newer event can still come from the wrong workspace. +3. **Prefer non-empty status.** Rejected because a real clean workspace must be + able to clear old dirty state. +4. **Unsubscribe hidden session components.** Rejected because another current + or future consumer can recreate the same competing-writer bug. +5. **Move all snapshots to the task-environment row.** Rejected because it + removes useful session attribution and adds a second persistence model when + projection-time validation is sufficient. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index eec80cdfd5..88b0f8d0a0 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -220,3 +220,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 2026-08-30-empty-remote-bootstrap-publication | [Keep Empty Remote Bootstrap Local Until Publication](2026-08-30-empty-remote-bootstrap-publication.md) | accepted | backend, frontend, protocol, security | 2026-08-30 | | 2026-08-30-bound-untracked-dependency-enumeration | [Bound Untracked Dependency Enumeration](2026-08-30-bound-untracked-dependency-enumeration.md) | accepted | backend | 2026-08-30 | | 2026-08-30-e2e-browser-cache | [Cache host-runner E2E browser provisioning](2026-08-30-e2e-browser-cache.md) | accepted | infra, workflow | 2026-08-30 | +| 2026-08-30-environment-owned-git-status | [Project Current Git Status From the Task Environment](2026-08-30-environment-owned-git-status.md) | proposed | backend, frontend, protocol | 2026-08-30 | diff --git a/docs/plans/environment-owned-git-status/plan.md b/docs/plans/environment-owned-git-status/plan.md new file mode 100644 index 0000000000..bd011d8124 --- /dev/null +++ b/docs/plans/environment-owned-git-status/plan.md @@ -0,0 +1,108 @@ +--- +spec: docs/specs/tasks/requirements/additional-session-workspace-reuse.md +design: docs/specs/tasks/system-design/environment-owned-git-status.md +created: 2026-08-30 +status: complete +--- + +# Fix Plan: Keep Shared-Environment Git Status Authoritative + +## Overview + +The Changes panel caches Git status by task environment, but backend hydration +currently reads status by requested session. Two sibling subscriptions can +therefore write different workspace results into one frontend cache entry. + +The fix makes the backend select status only from a session bound to the task +environment's canonical workspace. It also makes the frontend reject older +environment observations. A focused end-to-end test proves that dirty files +remain visible through reload and sibling hydration. + +## Confirmed root cause + +For task `c086d141-1966-4502-923a-c5ede67876d3`, two sessions shared task +environment `9d364cad-dde8-4e02-b29b-a07c83f66165` but recorded different +workspace paths. The canonical session had 91 dirty files. The older sibling +pointed at a clean checkout. + +Both live Git requests timed out. `appendDBSnapshotGitStatus` then loaded a +snapshot by each requested session ID. The frontend mapped both events to the +same environment key. It stored the 91-file snapshot first and replaced it +147 milliseconds later with the sibling's clean snapshot. + +PR #3167 fixes the recovery path that can preserve an incorrect session +workspace. It does not make current Git-status projection environment-owned or +order frontend status observations. + +## Approach + +1. Resolve eligible Git-status source sessions from the canonical task + environment before live or snapshot hydration. +2. Route the canonical result through the requested subscription session ID. +3. Keep a monotonic timestamp watermark per environment and repository in the + frontend store. +4. Extend the two-session E2E coverage to prove that Changes remains populated + after reload and sibling hydration. + +Do not add a database table, change the WebSocket action, prefer dirty state, +or hide the problem by removing sibling subscriptions. + +## Task waves + +### Wave 1: backend authority + +- [x] [Task 01: Select canonical Git-status sources](task-01-canonical-status-source.md) + +### Wave 2: ordering and user regression + +- [x] [Task 02: Reject older environment status](task-02-status-ordering-and-e2e.md) + +Task 02 depends on Task 01. The end-to-end result requires the backend authority +rule and the frontend ordering safeguard together. + +## Verification strategy + +- Backend unit tests reproduce same-environment sessions with different + workspace paths and prove live and snapshot source selection. +- Frontend unit tests reproduce a newer dirty event followed by an older clean + sibling event. +- A Chromium E2E test creates a dirty file in a shared environment and proves + that it remains visible after reload and sibling-session hydration. +- Specification lint and `git diff --check` validate the delivery records. + +## Risks and mitigations + +- **Remote workspace paths:** compare the same effective workspace field that + the task environment and session persist. Test host execution now and avoid + host-only path resolution such as symlink evaluation. +- **Missing historical paths:** fail closed for environment-owned hydration. + A later valid live update can recover the view. +- **Equal timestamps:** reject only strictly older valid timestamps. Preserve + meaningful equal-time updates for compatibility. +- **Duplicate updates:** advance the timestamp watermark without invalidating + the cumulative diff cache when Git content is unchanged. +- **PR #3167 overlap:** rebase before delivery and keep its recovery-path tests. + This package owns projection authority and ordering, not workspace recovery. + +## Mobile parity + +Desktop and mobile Changes surfaces share the same environment-keyed store. +This fix changes no layout, navigation, control, gesture, or copy. The store +unit test and shared-state E2E evidence cover the mobile data contract. A new +mobile E2E scenario duplicates the same state assertion. It does not test a +mobile-specific interaction. + +## Implementation results + +- Task 01 selects live and persisted Git-status sources from the canonical task + environment and keeps the requested session ID on the outgoing event. +- Task 02 applies per-environment and per-repository timestamp ordering, keeps + duplicate-content watermarks without invalidating cumulative diffs, and adds + the shared-environment Chromium regression. +- PR fixup coverage now verifies raw workspace provenance, inherited + cross-task environment bindings, recovered executions, one deadline across + live source probes, and one persisted fallback status per repository. +- The shared desktop/mobile store contract is covered without a duplicate + mobile E2E case because presentation and interaction are unchanged. +- The exact RED/GREEN evidence and verification results are recorded in the + two work orders. diff --git a/docs/plans/environment-owned-git-status/task-01-canonical-status-source.md b/docs/plans/environment-owned-git-status/task-01-canonical-status-source.md new file mode 100644 index 0000000000..9919286e8f --- /dev/null +++ b/docs/plans/environment-owned-git-status/task-01-canonical-status-source.md @@ -0,0 +1,109 @@ +--- +id: "01-canonical-status-source" +title: "Select canonical Git-status sources" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/tasks/requirements/additional-session-workspace-reuse.md" +design: "../../specs/tasks/system-design/environment-owned-git-status.md" +--- + +# Task 01: Select Canonical Git-Status Sources + +## Goal + +Make session subscription and explicit Git refresh project current status only +from a session and execution bound to the task environment's canonical +workspace. + +## Scope + +- Add focused source-selection helpers in + `apps/backend/internal/backendapp/helpers.go` and + `apps/backend/internal/backendapp/git_status_sources.go`. +- Resolve the requested session's task environment and every session bound to + that environment, including inherited sessions from another task. +- Validate session and live-execution workspace identity before a live query. +- Batch-load eligible snapshots and select the newest observation per + repository after live status is unavailable. +- Route the selected result with the requested subscription session ID. +- Add bounded debug logs for source choice and rejection reason. + +## Exclusions + +- Do not change workspace recovery or resume path selection from PR #3167. +- Do not move snapshots to a task-environment table. +- Do not expose workspace paths in logs or WebSocket payloads. +- Do not prefer dirty or non-empty status. + +## Requirements + +- `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001` +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001.5` + +## RED regression + +Add backend tests that seed two sessions with one task-environment ID and two +different workspace paths. The requested sibling points at the non-canonical +path. The canonical sibling has a dirty snapshot. + +Before production changes, prove that hydration returns the requested sibling's +clean snapshot. Also cover a mismatched live execution so it cannot bypass the +snapshot authority rule. + +Suggested test names: + +- `TestAppendLiveGitStatusMessageSelectsCanonicalEnvironmentSnapshot` +- `TestAppendLiveGitStatusMessageRejectsMismatchedLiveExecution` + +## Acceptance + +- A mismatched sibling cannot supply live or persisted current status for the + shared environment. +- A canonical sibling can supply status while the outgoing event remains routed + to the requested subscription session. +- A lookup error or an environment with no eligible source emits no suspect + status and does not fall back to the mismatched session. + +## Verification + +```bash +cd apps/backend && go test -tags fts5 ./internal/backendapp -run 'TestAppendLiveGitStatusMessage.*(Canonical|Mismatched)' -count=1 +``` + +## Files likely touched + +- `apps/backend/internal/backendapp/helpers.go` +- `apps/backend/internal/backendapp/git_status_sources.go` +- `apps/backend/internal/backendapp/helpers_git_status_test.go` +- `apps/backend/internal/task/repository/sqlite/session.go` + +## Dependencies + +None. Rebase on PR #3167 before delivery if it merges first. + +## Output contract + +Report RED and GREEN evidence, the source-selection order, changed files, test +results, and residual compatibility behavior for sessions without an +environment. Update this work order and `plan.md` in the implementation turn. + +## Results + +- RED evidence: the new canonical snapshot test stored no status, and the + mismatched-live test selected the wrong live status before the production + changes. +- GREEN evidence: the targeted canonical and mismatched-source test command + passed 3 tests. +- Source order: a matching live execution is preferred; when live status is + unavailable, the newest timestamped snapshot from canonical-bound sessions + is selected. The requested session ID remains on the published event. +- Environment-owned hydration fails closed when the task environment is + missing, mismatched, or has no eligible source. Sessions without an + environment retain the existing session-scoped behavior. +- The fixup adds raw workspace provenance checks, environment-wide source + enumeration across task boundaries, recovered-execution handling, a single + live-probe deadline, and per-repository snapshot fallback. +- `go test -tags fts5 ./internal/backendapp -count=1` passed 754 tests, and + `make -C apps/backend build` passed. diff --git a/docs/plans/environment-owned-git-status/task-02-status-ordering-and-e2e.md b/docs/plans/environment-owned-git-status/task-02-status-ordering-and-e2e.md new file mode 100644 index 0000000000..13d6e25f15 --- /dev/null +++ b/docs/plans/environment-owned-git-status/task-02-status-ordering-and-e2e.md @@ -0,0 +1,121 @@ +--- +id: "02-status-ordering-and-e2e" +title: "Reject older environment status" +status: done +wave: 2 +depends_on: + - "01-canonical-status-source" +plan: "plan.md" +spec: "../../specs/tasks/requirements/additional-session-workspace-reuse.md" +design: "../../specs/tasks/system-design/environment-owned-git-status.md" +--- + +# Task 02: Reject Older Environment Status + +## Goal + +Prevent a late, older Git-status event from replacing newer state in the shared +environment cache, and prove the user-visible Changes result across sibling +session hydration. + +## Scope + +- Add timestamp ordering to `applyGitStatus` for each environment and repository + key. +- Keep the newest valid timestamp as a watermark, including when Git content is + unchanged. +- Reject older dated entries. Also reject undated entries when the current + state has a valid timestamp. +- Keep equal-time meaningful updates compatible. +- Extend focused WebSocket/store tests. +- Add or extend a two-session Chromium E2E test. Create an uncommitted file, + reload the task, and allow both sessions to hydrate. Switch between the + sibling tabs, and prove that the file remains in Changes. + +## Exclusions + +- Do not add layout, touch, navigation, or user-copy changes. +- Do not add a special rule that preserves non-empty file lists. +- Do not remove or conditionally mount existing session subscriptions. +- Do not add a separate mobile implementation or duplicate mobile E2E case. + +## Requirements + +- `REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001` +- `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001.5` + +## RED regressions + +First add a unit test that maps two sessions to one environment. Deliver a +newer dirty status through one session and an older clean status through the +other. Prove that the current store incorrectly becomes clean before the fix. + +Then add the user regression in +`apps/web/e2e/tests/session/session-tab-management.spec.ts`. Confirm that the +test reaches both session hydrations and observes the dirty file through the +Changes UI. + +Suggested unit test name: + +- `does not overwrite newer environment status with an older sibling event` + +## Acceptance + +- Older or unknown-age events cannot replace a newer dated environment status. +- Duplicate content can advance the timestamp watermark without invalidating + the cumulative diff cache. +- The Changes panel keeps the uncommitted file visible through reload and a + sibling-session switch. + +## Verification + +```bash +cd apps/web && pnpm test -- lib/ws/handlers/git-status.test.ts +make -C apps/backend build +cd apps/web && pnpm run build:e2e +cd apps/web && pnpm e2e:run tests/session/session-tab-management.spec.ts --project=chromium -g "keeps dirty Changes status after sibling hydration" +``` + +## Files likely touched + +- `apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts` +- `apps/web/lib/state/slices/session-runtime/git-status-state.ts` +- `apps/web/lib/ws/handlers/git-status.test.ts` +- `apps/web/e2e/tests/session/session-tab-management.spec.ts` + +## Dependencies + +Task 01. Rebase on PR #3167 before delivery if it merges first. + +## Mobile parity + +Desktop and mobile Changes surfaces read the same status store. This task does +not change presentation or interaction. The unit regression proves the shared +data rule, and the Chromium E2E test proves the user-visible flow. No +mobile-specific behavior needs a second browser case. + +## Output contract + +Report RED and GREEN evidence, timestamp handling for invalid and equal values, +changed files, unit and E2E results, and any observed flake risk. Update this +work order and `plan.md` in the implementation turn. + +## Results + +- RED evidence: the shared-environment handler test allowed an older clean + sibling event to replace the newer dirty status before timestamp ordering + was implemented. +- GREEN evidence: the focused handler, session-runtime return-value, and + multi-repository tests passed 25 tests. The exact work-order command + `pnpm test -- lib/ws/handlers/git-status.test.ts` also passes. +- Valid timestamps are monotonic per environment and repository. Older valid + timestamps and undated entries after dated state are rejected. Equal-time + content changes remain accepted, and newer duplicate content advances the + watermark without invalidating the cumulative diff cache. +- The managed Chromium regression passed: it reloads the shared task, waits + for both sibling `session.git.event` hydrations, and keeps the dirty file + visible in Changes after switching sessions. No flake or retry was needed. +- `pnpm run typecheck`, `pnpm run lint`, `pnpm run build:e2e`, and the focused + Chromium E2E command passed. Mobile uses the same store and has no + presentation or interaction change, so no duplicate mobile scenario was + added. diff --git a/docs/specs/tasks/README.md b/docs/specs/tasks/README.md index 22b82c8ac3..2a2bdb2675 100644 --- a/docs/specs/tasks/README.md +++ b/docs/specs/tasks/README.md @@ -110,6 +110,7 @@ signals, and task-scoped scheduling contracts. ### System design - [Additional Session Workspace Reuse](system-design/additional-session-workspace-reuse.md) +- [Environment-Owned Git Status](system-design/environment-owned-git-status.md) - [Attach Workspace Sources](system-design/attach-workspace-sources.md) - [Quick Chat Agent Titles](system-design/quick-chat-agent-titles.md) - [Quick Chat Session Resumption](system-design/quick-chat-session-resumption.md) diff --git a/docs/specs/tasks/requirements/additional-session-workspace-reuse.md b/docs/specs/tasks/requirements/additional-session-workspace-reuse.md index f4b019ce57..955f0ade04 100644 --- a/docs/specs/tasks/requirements/additional-session-workspace-reuse.md +++ b/docs/specs/tasks/requirements/additional-session-workspace-reuse.md @@ -34,6 +34,10 @@ preserving independent session runtime state. - **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001.4:** Unsafe or unsupported reuse shall fail with a typed, recoverable API error without creating a session or replacement workspace. +- **AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001.5:** Desktop and mobile + workspace views shall show one current Git status for sessions that share a + task environment. A response from a non-canonical workspace or an older + observation shall not replace that status. ### REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-002: Canonical Workspace Identity Continuity @@ -115,6 +119,8 @@ an optional session name remains best effort after a successful launch. - A named or unnamed additional session sees the same uncommitted tracked and untracked files as the task's first session. +- A late status response from a sibling session does not clear or replace the + current task-environment status in desktop or mobile workspace views. - Git worktree inventory, HEAD, index, branch and status remain unchanged by an additional launch. - A terminal primary or zero-session task can reuse its retained ready diff --git a/docs/specs/tasks/system-design/environment-owned-git-status.md b/docs/specs/tasks/system-design/environment-owned-git-status.md new file mode 100644 index 0000000000..a66df18cd8 --- /dev/null +++ b/docs/specs/tasks/system-design/environment-owned-git-status.md @@ -0,0 +1,158 @@ +--- +status: draft +system: tasks +requirements: + - REQ-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001 +--- + +# Environment-Owned Git Status System Design + +## Purpose and boundaries + +The task environment owns the current workspace. It also owns the current Git +status that desktop and mobile workspace views show. A task session is a +transport handle for status requests and events. It is not a separate owner of +workspace state. + +This design covers status hydration, WebSocket delivery, frontend cache +ordering, and the Changes surfaces. It does not change Git polling inside +agentctl, Git operations, commit history, or workspace materialization. + +The workspace-path repair in PR #3167 prevents a resumed session from keeping +an incorrect workspace path. This design adds a separate boundary. It prevents +historical or delayed sibling-session data from becoming authoritative after a +task environment already has a canonical workspace. + +## Requirement mapping + +| Requirement | Design section | +| --------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- | +| `AC-TASKS-ADDITIONAL-SESSION-WORKSPACE-REUSE-001.5` | [Backend status authority](#backend-status-authority), [Frontend ordering](#frontend-ordering), [Workspace views](#workspace-views) | + +## Backend status authority + +`buildSessionDataProvider` and `buildSessionGitDataProvider` in +`apps/backend/internal/backendapp/helpers.go` accept a session ID because the +WebSocket protocol subscribes to sessions. Before either provider reads live +or persisted Git status, it resolves the session's `TaskEnvironment` and the +all sessions bound to that environment. This includes inherited or shared +sessions whose task differs from the environment owner's task. + +The provider selects a source session with these rules: + +1. If the requested session has no task environment, keep the legacy + session-scoped behavior. +2. Read the canonical workspace path from + `task_environments.workspace_path`. +3. Keep only sessions that reference the same task environment and whose raw + recorded workspace path is present and exactly matches the canonical path. + The projected environment path cannot prove that a historical session was + recorded against the canonical workspace. +4. Prefer a matching live execution. Its workspace path must match the + canonical environment. A recovered execution with an empty task-environment + ID is accepted only after the session binding and workspace checks pass. A + non-empty mismatched ID is rejected before the provider asks agentctl for + status. +5. If no live query succeeds, load the latest ranked snapshot for each + matching source session and repository, then select the newest observation + for each repository. +6. Publish the selected status with the requested subscription session ID so + existing WebSocket routing and environment mapping remain compatible. A + multi-repository workspace publishes one event per repository. + +The provider does not use a clean status, a non-empty file list, or a primary +session flag as authority. A clean canonical workspace is valid. Session +selection uses workspace identity, and snapshot selection uses observation +time only after the workspace identity matches. + +The implementation uses environment-bound session queries, +`GetTaskEnvironment`, and a per-session/per-repository snapshot query. It does +not add a database table or copy Git status onto the task-environment row. + +## Frontend ordering + +`session.git.event` continues to carry a session ID. The session-runtime store +continues to map that ID through `environmentIdBySessionId` and to cache status +under `gitStatus.byEnvironmentId` and `gitStatus.byEnvironmentRepo`. + +For each environment and repository key, `setGitStatus` keeps the newest valid +status timestamp it has observed. It rejects an incoming entry with an older +valid timestamp. If an entry has the same Git content and a newer timestamp, +the store advances the timestamp watermark without invalidating the cumulative +diff cache. This prevents an older hydration response from winning only because +it arrived later. + +An entry with no valid timestamp cannot replace an entry that already has a +valid timestamp. This rule keeps a legacy snapshot with unknown age from +replacing a dated live observation. Existing undated state can still accept a +new entry, which preserves compatibility during initial hydration. + +Timestamp ordering is a safeguard, not the source-authority rule. The backend +must first exclude data from a different workspace. + +## Workspace views + +Desktop and mobile Changes surfaces use the same environment-keyed store and +Git-status hooks. This change does not add controls, change layout, or alter +touch behavior. Both surfaces receive the same corrected state without separate +presentation logic. + +The desktop end-to-end regression uses two sessions in one environment. It +creates an uncommitted file, reloads the task, and proves that the file stays +visible after both session hydrations and a sibling-session switch. A focused +store test proves that an older sibling event cannot replace newer environment +state. No separate mobile end-to-end test is required because this design does +not change mobile composition or interaction, and the mobile surface reads the +same store contract. + +## Failure and recovery + +If environment or sibling-session lookup fails, the provider logs the failure +and does not publish an unverified session snapshot. The next Git poll, focus +refresh, reconnect, or explicit refresh can retry. The provider keeps legacy +session behavior only when the session has no task-environment identity. + +If a live status request times out, the provider can use only a snapshot whose +source session matches the canonical environment workspace. If none exists, it +returns no status rather than clearing the current frontend state with suspect +data. The live probe has one deadline for all eligible sources, so a stuck +sibling cannot extend hydration once per source. + +A corrected session path from PR #3167 makes that session eligible on its next +hydration. No database migration or historical snapshot rewrite is required. + +## Persistence + +Git snapshots remain session-attributed history in +`task_session_git_snapshots`. The source session preserves the audit trail. +Environment authority is resolved when current status is projected. This keeps +historical records intact and avoids a second status ownership model. Live +monitor snapshots are retained one per session and repository so fallback can +rebuild a multi-repository status. + +## Observability + +Backend logs for live-query and snapshot fallback include the requested session +ID, selected source session ID, task-environment ID, and a bounded reason when a +candidate is rejected. Logs must not include workspace paths. + +Frontend debug logs include the environment key, repository name, incoming and +stored timestamps, and whether ordering rejected the event. Existing file-count +and cache-invalidation fields remain. + +## Verification + +- Backend tests seed same-environment sessions, including an inherited session + from another task, with different workspace paths. They prove that live and + snapshot hydration select only canonical sources, preserve one event per + repository, and route results to the requested subscription. +- Frontend tests deliver a newer dirty status and then an older clean status + from a sibling session. They prove that the environment cache stays dirty and + the cumulative diff cache is not invalidated by the rejected event. +- The desktop end-to-end test proves that uncommitted files remain visible + through reload and sibling-session hydration. + +## Related decisions + +- [Keep Worktree Ownership at the Task Lifecycle](../../../decisions/2026-08-08-task-owned-worktree-lifetime.md) +- [Project Current Git Status From the Task Environment](../../../decisions/2026-08-30-environment-owned-git-status.md)