diff --git a/apps/backend/internal/agentctl/client/git.go b/apps/backend/internal/agentctl/client/git.go index da591b79c8..237de7be0d 100644 --- a/apps/backend/internal/agentctl/client/git.go +++ b/apps/backend/internal/agentctl/client/git.go @@ -403,18 +403,18 @@ func (c *Client) GetCumulativeDiff(ctx context.Context, baseCommit string) (*Cum // GitStatusResult represents the result of a git status query. type GitStatusResult struct { - Success bool `json:"success"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch"` - HeadCommit string `json:"head_commit"` - BaseCommit string `json:"base_commit"` // Merge-base with origin branch - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` + Success bool `json:"success"` + Branch string `json:"branch"` + RemoteBranch string `json:"remote_branch"` + HeadCommit string `json:"head_commit"` + BaseCommit string `json:"base_commit"` // Merge-base with origin branch + Ahead int `json:"ahead"` + Behind int `json:"behind"` + Modified []string `json:"modified"` + Added []string `json:"added"` + Deleted []string `json:"deleted"` + Untracked []string `json:"untracked"` + Renamed []string `json:"renamed"` Files map[string]interface{} `json:"files"` Timestamp string `json:"timestamp"` BranchAdditions int `json:"branch_additions,omitempty"` diff --git a/apps/backend/internal/agentctl/server/api/git.go b/apps/backend/internal/agentctl/server/api/git.go index a83286a3ed..3363c4f03d 100644 --- a/apps/backend/internal/agentctl/server/api/git.go +++ b/apps/backend/internal/agentctl/server/api/git.go @@ -581,18 +581,18 @@ func (s *Server) handleGitCumulativeDiff(c *gin.Context) { // GitStatusResult represents the result of a git status query. type GitStatusResult struct { - Success bool `json:"success"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch"` - HeadCommit string `json:"head_commit"` - BaseCommit string `json:"base_commit"` // Merge-base with origin branch - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` + Success bool `json:"success"` + Branch string `json:"branch"` + RemoteBranch string `json:"remote_branch"` + HeadCommit string `json:"head_commit"` + BaseCommit string `json:"base_commit"` // Merge-base with origin branch + Ahead int `json:"ahead"` + Behind int `json:"behind"` + Modified []string `json:"modified"` + Added []string `json:"added"` + Deleted []string `json:"deleted"` + Untracked []string `json:"untracked"` + Renamed []string `json:"renamed"` Files map[string]interface{} `json:"files"` Timestamp string `json:"timestamp"` BranchAdditions int `json:"branch_additions,omitempty"` @@ -628,18 +628,18 @@ func (s *Server) handleGitStatus(c *gin.Context) { } c.JSON(http.StatusOK, GitStatusResult{ - Success: true, - Branch: status.Branch, - RemoteBranch: status.RemoteBranch, - HeadCommit: status.HeadCommit, - BaseCommit: status.BaseCommit, - Ahead: status.Ahead, - Behind: status.Behind, - Modified: status.Modified, - Added: status.Added, - Deleted: status.Deleted, - Untracked: status.Untracked, - Renamed: status.Renamed, + Success: true, + Branch: status.Branch, + RemoteBranch: status.RemoteBranch, + HeadCommit: status.HeadCommit, + BaseCommit: status.BaseCommit, + Ahead: status.Ahead, + Behind: status.Behind, + Modified: status.Modified, + Added: status.Added, + Deleted: status.Deleted, + Untracked: status.Untracked, + Renamed: status.Renamed, Files: filesMap, Timestamp: status.Timestamp.Format("2006-01-02T15:04:05.000Z07:00"), BranchAdditions: status.BranchAdditions, diff --git a/apps/backend/internal/orchestrator/event_handlers_git.go b/apps/backend/internal/orchestrator/event_handlers_git.go index b9589daf43..008fd8e894 100644 --- a/apps/backend/internal/orchestrator/event_handlers_git.go +++ b/apps/backend/internal/orchestrator/event_handlers_git.go @@ -2,6 +2,10 @@ package orchestrator import ( "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "sync" "time" "go.uber.org/zap" @@ -10,8 +14,92 @@ import ( "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" ) +// gitSnapshotPersistInterval is the minimum time between persisted live git +// status snapshots for a single session when the underlying status hasn't +// changed. Writes still happen immediately when the status hash changes. +const gitSnapshotPersistInterval = 30 * time.Second + +// gitSnapshotCacheMaxEntries bounds the in-memory throttle map so a long-lived +// backend with many sessions can't grow it without limit. When the cache is +// full and a new session arrives, the oldest entry by lastWrite is evicted. +const gitSnapshotCacheMaxEntries = 4096 + +type gitSnapshotCacheEntry struct { + hash string + lastWrite time.Time +} + +// gitSnapshotCache throttles per-session writes to the live git snapshot cache +// table. It is process-local — first event after a restart will rewrite the +// row, which is fine because UpsertLatestLiveGitSnapshot is idempotent. +type gitSnapshotCache struct { + mu sync.Mutex + byID map[string]gitSnapshotCacheEntry + maxSize int +} + +func newGitSnapshotCache() *gitSnapshotCache { + return &gitSnapshotCache{ + byID: make(map[string]gitSnapshotCacheEntry), + maxSize: gitSnapshotCacheMaxEntries, + } +} + +// shouldWrite returns true if the new hash should be persisted now. Writes +// happen on hash change, or when the previous write is older than +// gitSnapshotPersistInterval (defensive: makes the cache eventually consistent +// even if hashing misses something). +func (c *gitSnapshotCache) shouldWrite(sessionID, hash string, now time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + prev, ok := c.byID[sessionID] + if ok && prev.hash == hash && now.Sub(prev.lastWrite) < gitSnapshotPersistInterval { + return false + } + if !ok && c.maxSize > 0 && len(c.byID) >= c.maxSize { + c.evictOldestLocked() + } + c.byID[sessionID] = gitSnapshotCacheEntry{hash: hash, lastWrite: now} + return true +} + +// evictOldestLocked drops the entry with the oldest lastWrite. Caller must +// hold c.mu. O(n) over the cache; only invoked when the cache is full, which +// is rare in practice. +func (c *gitSnapshotCache) evictOldestLocked() { + var oldestID string + var oldestAt time.Time + for id, entry := range c.byID { + if oldestID == "" || entry.lastWrite.Before(oldestAt) { + oldestID = id + oldestAt = entry.lastWrite + } + } + if oldestID != "" { + delete(c.byID, oldestID) + } +} + +// forget removes a session's cached entry. Called when a session is deleted +// so the cache doesn't retain stale state for sessions that will never +// receive another git event. +func (c *gitSnapshotCache) forget(sessionID string) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.byID, sessionID) +} + +func gitStatusHash(s *lifecycle.GitStatusData) string { + h := sha256.New() + _, _ = fmt.Fprintf(h, "%s|%s|%s|%s|%d|%d|%d|%d", + s.Branch, s.RemoteBranch, s.HeadCommit, s.BaseCommit, + s.Ahead, s.Behind, s.BranchAdditions, s.BranchDeletions) + return hex.EncodeToString(h.Sum(nil)) +} + // handleGitEvent handles unified git events and dispatches to appropriate handler func (s *Service) handleGitEvent(ctx context.Context, data watcher.GitEventData) { s.logger.Debug("handling git event", @@ -67,6 +155,54 @@ func (s *Service) handleGitStatusUpdate(ctx context.Context, data watcher.GitEve // Push detection: when ahead goes from >0 to 0, a push happened s.trackPushAndAssociatePR(ctx, data) + + // Persist a throttled cache of the live status so the sidebar diff badge + // works for tasks whose executor isn't currently running (and across + // backend restarts). Best-effort: errors are logged and swallowed. + s.persistGitStatusSnapshot(ctx, data) +} + +// persistGitStatusSnapshot writes a single cached "live monitor" snapshot per +// session, throttled by gitSnapshotCache. The cached row is read by +// appendDBSnapshotGitStatus when no live execution is available. +func (s *Service) persistGitStatusSnapshot(ctx context.Context, data watcher.GitEventData) { + if s.repo == nil || data.SessionID == "" || data.Status == nil { + return + } + if s.gitSnapshotCache == nil { + return + } + hash := gitStatusHash(data.Status) + if !s.gitSnapshotCache.shouldWrite(data.SessionID, hash, time.Now()) { + return + } + + st := data.Status + snapshot := &models.GitSnapshot{ + SessionID: data.SessionID, + Branch: st.Branch, + RemoteBranch: st.RemoteBranch, + HeadCommit: st.HeadCommit, + BaseCommit: st.BaseCommit, + Ahead: st.Ahead, + Behind: st.Behind, + Files: nil, // intentional: badge only needs totals + Metadata: map[string]interface{}{ + "branch_additions": st.BranchAdditions, + "branch_deletions": st.BranchDeletions, + "modified": st.Modified, + "added": st.Added, + "deleted": st.Deleted, + "untracked": st.Untracked, + "renamed": st.Renamed, + "timestamp": data.Timestamp, + }, + } + if err := s.repo.UpsertLatestLiveGitSnapshot(ctx, snapshot); err != nil { + s.logger.Debug("failed to persist live git snapshot", + zap.String("session_id", data.SessionID), + zap.Error(err)) + } } // trackPushAndAssociatePR detects git pushes by tracking the "ahead" count. diff --git a/apps/backend/internal/orchestrator/git_snapshot_cache_test.go b/apps/backend/internal/orchestrator/git_snapshot_cache_test.go new file mode 100644 index 0000000000..d3604790f8 --- /dev/null +++ b/apps/backend/internal/orchestrator/git_snapshot_cache_test.go @@ -0,0 +1,71 @@ +package orchestrator + +import ( + "strconv" + "testing" + "time" +) + +func TestGitSnapshotCacheShouldWrite(t *testing.T) { + c := newGitSnapshotCache() + now := time.Unix(1_000_000, 0) + + // First call always writes. + if !c.shouldWrite("s1", "h1", now) { + t.Fatal("first write should be allowed") + } + // Same hash within throttle window: skip. + if c.shouldWrite("s1", "h1", now.Add(5*time.Second)) { + t.Fatal("duplicate within throttle window should be skipped") + } + // Hash change: write immediately. + if !c.shouldWrite("s1", "h2", now.Add(5*time.Second)) { + t.Fatal("hash change should bypass throttle") + } + // Same hash after interval: write. + if !c.shouldWrite("s1", "h2", now.Add(5*time.Second+gitSnapshotPersistInterval)) { + t.Fatal("write after persist interval should be allowed") + } +} + +func TestGitSnapshotCacheEvictsOldestWhenFull(t *testing.T) { + c := newGitSnapshotCache() + c.maxSize = 3 + base := time.Unix(2_000_000, 0) + + // Fill the cache. Each entry has a strictly increasing lastWrite. + for i := 0; i < 3; i++ { + if !c.shouldWrite("session-"+strconv.Itoa(i), "h", base.Add(time.Duration(i)*time.Second)) { + t.Fatalf("fill #%d should write", i) + } + } + if got := len(c.byID); got != 3 { + t.Fatalf("expected cache size 3, got %d", got) + } + + // Adding a 4th distinct session should evict the oldest (session-0). + if !c.shouldWrite("session-3", "h", base.Add(10*time.Second)) { + t.Fatal("4th distinct session should write") + } + if got := len(c.byID); got != 3 { + t.Fatalf("expected cache size to remain at 3, got %d", got) + } + if _, ok := c.byID["session-0"]; ok { + t.Error("session-0 should have been evicted as the oldest") + } + if _, ok := c.byID["session-3"]; !ok { + t.Error("session-3 should be present after insert") + } +} + +func TestGitSnapshotCacheForget(t *testing.T) { + c := newGitSnapshotCache() + now := time.Unix(3_000_000, 0) + c.shouldWrite("s1", "h", now) + c.forget("s1") + if _, ok := c.byID["s1"]; ok { + t.Error("forget did not remove the entry") + } + // Forgetting an unknown session is a no-op. + c.forget("unknown") +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 3ec63b1c7c..c2637a9207 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -145,6 +145,7 @@ type sessionExecutorStore interface { // Git snapshots and commits GetLatestGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) CreateGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error + UpsertLatestLiveGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) DeleteSessionCommit(ctx context.Context, id string) error @@ -204,6 +205,10 @@ type Service struct { // Push tracker: sessionID -> last known ahead count pushTracker sync.Map + // gitSnapshotCache throttles per-session writes of the live git status + // snapshot (used by handleGitStatusUpdate -> persistGitStatusSnapshot). + gitSnapshotCache *gitSnapshotCache + // Active turns map: sessionID -> turnID activeTurns sync.Map @@ -283,6 +288,7 @@ func NewService( scheduler: sched, messageQueue: msgQueue, clarificationWatchdogTimeout: 15 * time.Second, + gitSnapshotCache: newGitSnapshotCache(), } // Wire executor state changes through the orchestrator so events are published diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 07559c08fa..2850ebb392 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -1089,6 +1089,12 @@ func (s *Service) DeleteSession(ctx context.Context, sessionID string) error { return fmt.Errorf("failed to delete session: %w", err) } + // Drop the in-memory git snapshot throttle entry — the session will + // never receive another git event, so its cache slot is dead weight. + if s.gitSnapshotCache != nil { + s.gitSnapshotCache.forget(sessionID) + } + // Auto-promote another session if we deleted the primary if wasPrimary { s.promoteNextPrimaryAfterRemoval(ctx, taskID, sessionID) diff --git a/apps/backend/internal/task/repository/git_snapshots_upsert_test.go b/apps/backend/internal/task/repository/git_snapshots_upsert_test.go new file mode 100644 index 0000000000..70f9ffa37e --- /dev/null +++ b/apps/backend/internal/task/repository/git_snapshots_upsert_test.go @@ -0,0 +1,175 @@ +package repository + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// seedTaskAndSession creates the parent rows the git_snapshots foreign key +// requires (task → task_session). Returns the session ID. +func seedTaskAndSession(t *testing.T, ctx context.Context, repo *sqlite.Repository, taskID, sessionID string) { + t.Helper() + if err := repo.CreateTask(ctx, &models.Task{ + ID: taskID, + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + Title: "Test Task", + }); err != nil { + t.Fatalf("create task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "profile-1", + State: models.TaskSessionStateStarting, + }); err != nil { + t.Fatalf("create task session: %v", err) + } +} + +// TestUpsertLatestLiveGitSnapshot covers the at-most-one-row invariant the +// sidebar diff badge cache depends on. See PR #556. +func TestUpsertLatestLiveGitSnapshot(t *testing.T) { + ctx := context.Background() + + t.Run("two consecutive upserts keep exactly one live row", func(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + + const taskID = "task-A" + const sessionID = "session-A" + seedTaskAndSession(t, ctx, repo, taskID, sessionID) + + first := &models.GitSnapshot{ + SessionID: sessionID, + Branch: "main", + HeadCommit: "abc123", + BaseCommit: "base000", + Ahead: 1, + Behind: 0, + Metadata: map[string]interface{}{"branch_additions": 5, "branch_deletions": 1}, + } + if err := repo.UpsertLatestLiveGitSnapshot(ctx, first); err != nil { + t.Fatalf("first upsert: %v", err) + } + + second := &models.GitSnapshot{ + SessionID: sessionID, + Branch: "main", + HeadCommit: "def456", + BaseCommit: "base000", + Ahead: 2, + Behind: 0, + Metadata: map[string]interface{}{"branch_additions": 9, "branch_deletions": 3}, + } + if err := repo.UpsertLatestLiveGitSnapshot(ctx, second); err != nil { + t.Fatalf("second upsert: %v", err) + } + + all, err := repo.GetGitSnapshotsBySession(ctx, sessionID, 0) + if err != nil { + t.Fatalf("list snapshots: %v", err) + } + if len(all) != 1 { + t.Fatalf("expected exactly 1 snapshot row, got %d", len(all)) + } + got := all[0] + if got.HeadCommit != "def456" { + t.Errorf("expected head_commit=def456, got %q", got.HeadCommit) + } + if got.TriggeredBy != sqlite.TriggeredByLiveMonitor { + t.Errorf("expected triggered_by=%q, got %q", sqlite.TriggeredByLiveMonitor, got.TriggeredBy) + } + if got.SnapshotType != models.SnapshotTypeStatusUpdate { + t.Errorf("expected snapshot_type=%q, got %q", models.SnapshotTypeStatusUpdate, got.SnapshotType) + } + if got.Metadata["branch_additions"] != float64(9) { + t.Errorf("expected branch_additions=9 in metadata, got %v", got.Metadata["branch_additions"]) + } + }) + + t.Run("non-live snapshots are untouched by upsert", func(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + + const taskID = "task-B" + const sessionID = "session-B" + seedTaskAndSession(t, ctx, repo, taskID, sessionID) + + // Pre-existing archive snapshot — must survive subsequent live upserts. + archive := &models.GitSnapshot{ + SessionID: sessionID, + SnapshotType: models.SnapshotTypeArchive, + Branch: "feature", + HeadCommit: "archive-head", + TriggeredBy: "archive", + } + if err := repo.CreateGitSnapshot(ctx, archive); err != nil { + t.Fatalf("create archive snapshot: %v", err) + } + + live := &models.GitSnapshot{ + SessionID: sessionID, + Branch: "feature", + HeadCommit: "live-head-1", + } + if err := repo.UpsertLatestLiveGitSnapshot(ctx, live); err != nil { + t.Fatalf("first live upsert: %v", err) + } + + live2 := &models.GitSnapshot{ + SessionID: sessionID, + Branch: "feature", + HeadCommit: "live-head-2", + } + if err := repo.UpsertLatestLiveGitSnapshot(ctx, live2); err != nil { + t.Fatalf("second live upsert: %v", err) + } + + all, err := repo.GetGitSnapshotsBySession(ctx, sessionID, 0) + if err != nil { + t.Fatalf("list snapshots: %v", err) + } + if len(all) != 2 { + t.Fatalf("expected 2 rows (1 archive + 1 live), got %d", len(all)) + } + + var sawArchive, sawLive bool + for _, snap := range all { + switch snap.SnapshotType { + case models.SnapshotTypeArchive: + sawArchive = true + if snap.HeadCommit != "archive-head" { + t.Errorf("archive head_commit mutated: got %q", snap.HeadCommit) + } + case models.SnapshotTypeStatusUpdate: + sawLive = true + if snap.HeadCommit != "live-head-2" { + t.Errorf("live head_commit not updated: got %q", snap.HeadCommit) + } + if snap.TriggeredBy != sqlite.TriggeredByLiveMonitor { + t.Errorf("live triggered_by wrong: got %q", snap.TriggeredBy) + } + } + } + if !sawArchive { + t.Error("archive snapshot disappeared after upsert") + } + if !sawLive { + t.Error("live snapshot missing after upsert") + } + }) + + t.Run("nil snapshot returns error", func(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + + if err := repo.UpsertLatestLiveGitSnapshot(ctx, nil); err == nil { + t.Fatal("expected error for nil snapshot, got nil") + } + }) +} diff --git a/apps/backend/internal/task/repository/sqlite/git_snapshots.go b/apps/backend/internal/task/repository/sqlite/git_snapshots.go index f83c9cc2fa..2f772800f6 100644 --- a/apps/backend/internal/task/repository/sqlite/git_snapshots.go +++ b/apps/backend/internal/task/repository/sqlite/git_snapshots.go @@ -12,6 +12,80 @@ import ( "github.com/kandev/kandev/internal/task/models" ) +// TriggeredByLiveMonitor identifies snapshots written by the orchestrator's live +// git status persistence path. Used to scope the upsert in +// UpsertLatestLiveGitSnapshot so we don't disturb archive/completion snapshots. +const TriggeredByLiveMonitor = "live_monitor" + +// 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. +func (r *Repository) UpsertLatestLiveGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error { + if snapshot == nil { + return fmt.Errorf("snapshot is nil") + } + snapshot.SnapshotType = models.SnapshotTypeStatusUpdate + snapshot.TriggeredBy = TriggeredByLiveMonitor + if snapshot.ID == "" { + snapshot.ID = uuid.New().String() + } + if snapshot.CreatedAt.IsZero() { + snapshot.CreatedAt = time.Now().UTC() + } + + tx, err := r.db.BeginTxx(ctx, nil) + if err != nil { + return fmt.Errorf("begin tx: %w", err) + } + defer func() { _ = tx.Rollback() }() + + 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 { + return fmt.Errorf("delete previous live snapshot: %w", err) + } + + filesJSON, metadataJSON, err := serializeSnapshotJSON(snapshot) + if err != nil { + return err + } + + if _, err := tx.ExecContext(ctx, r.db.Rebind(` + INSERT INTO task_session_git_snapshots ( + id, session_id, snapshot_type, branch, remote_branch, head_commit, base_commit, + ahead, behind, files, triggered_by, metadata, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `), snapshot.ID, snapshot.SessionID, string(snapshot.SnapshotType), snapshot.Branch, + snapshot.RemoteBranch, snapshot.HeadCommit, snapshot.BaseCommit, snapshot.Ahead, + snapshot.Behind, filesJSON, snapshot.TriggeredBy, metadataJSON, snapshot.CreatedAt); err != nil { + return fmt.Errorf("insert live snapshot: %w", err) + } + + return tx.Commit() +} + +func serializeSnapshotJSON(snapshot *models.GitSnapshot) (string, string, error) { + filesJSON := "{}" + if snapshot.Files != nil { + b, err := json.Marshal(snapshot.Files) + if err != nil { + return "", "", fmt.Errorf("failed to serialize git snapshot files: %w", err) + } + filesJSON = string(b) + } + metadataJSON := "{}" + if snapshot.Metadata != nil { + b, err := json.Marshal(snapshot.Metadata) + if err != nil { + return "", "", fmt.Errorf("failed to serialize git snapshot metadata: %w", err) + } + metadataJSON = string(b) + } + return filesJSON, metadataJSON, nil +} + // CreateGitSnapshot inserts a new git snapshot into the database. func (r *Repository) CreateGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error { if snapshot.ID == "" { @@ -21,25 +95,12 @@ func (r *Repository) CreateGitSnapshot(ctx context.Context, snapshot *models.Git snapshot.CreatedAt = time.Now().UTC() } - filesJSON := "{}" - if snapshot.Files != nil { - filesBytes, err := json.Marshal(snapshot.Files) - if err != nil { - return fmt.Errorf("failed to serialize git snapshot files: %w", err) - } - filesJSON = string(filesBytes) + filesJSON, metadataJSON, err := serializeSnapshotJSON(snapshot) + if err != nil { + return err } - metadataJSON := "{}" - if snapshot.Metadata != nil { - metadataBytes, err := json.Marshal(snapshot.Metadata) - if err != nil { - return fmt.Errorf("failed to serialize git snapshot metadata: %w", err) - } - metadataJSON = string(metadataBytes) - } - - _, err := r.db.ExecContext(ctx, r.db.Rebind(` + _, err = r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO task_session_git_snapshots ( id, session_id, snapshot_type, branch, remote_branch, head_commit, base_commit, ahead, behind, files, triggered_by, metadata, created_at diff --git a/apps/web/e2e/tests/task/sidebar-diff-stats.spec.ts b/apps/web/e2e/tests/task/sidebar-diff-stats.spec.ts new file mode 100644 index 0000000000..3e1b4cd02b --- /dev/null +++ b/apps/web/e2e/tests/task/sidebar-diff-stats.spec.ts @@ -0,0 +1,111 @@ +import { test, expect } from "../../fixtures/test-base"; +import { SessionPage } from "../../pages/session-page"; + +/** + * Regression test for the task sidebar diff badge bug. + * + * The sidebar bulk-subscribes to every task's primary session on connect and + * expects an initial git status snapshot per session, including the global + * branch_additions/branch_deletions diff against the merge-base. Before the + * fix, the backend's `tryGetLiveGitStatus` only returned data when an + * agentctl execution was actively running for that session. For any task + * whose execution had been torn down (e.g. after a backend restart), the + * fallback hit `appendDBSnapshotGitStatus` which only had data for archived + * tasks — so the badge silently disappeared for every non-active task. + * + * The fix persists the live monitor's last status to a single cached row per + * session in `task_session_git_snapshots` (triggered_by='live_monitor'), + * keeping the DB-snapshot fallback fresh across restarts and unavailability. + * + * This test creates two tasks that produce diffs, restarts the backend + * (which kills all running executors), then asserts the sidebar still shows + * +N/-N badges for BOTH tasks — not just an active one. + */ +test.describe("Task sidebar diff stats", () => { + test("badges survive backend restart for non-active tasks", async ({ + testPage, + apiClient, + seedData, + backend, + }) => { + test.setTimeout(180_000); + + // Create two tasks, each in its own worktree, each running the + // diff-update-setup scenario which leaves one modified, committed file + // and one unstaged modification → branch_additions / branch_deletions > 0. + const taskAlpha = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Diff Alpha", + seedData.agentProfileId, + { + description: "/e2e:diff-update-setup", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + executor_profile_id: seedData.worktreeExecutorProfileId, + }, + ); + const taskBeta = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Diff Beta", + seedData.agentProfileId, + { + description: "/e2e:diff-update-setup", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + executor_profile_id: seedData.worktreeExecutorProfileId, + }, + ); + + // Visit Alpha so we can wait for the agent's completion message and let + // the live git monitor fire at least once for both sessions. + await testPage.goto(`/t/${taskAlpha.id}`); + const alphaSession = new SessionPage(testPage); + await alphaSession.waitForLoad(); + await expect( + alphaSession.chat.getByText("diff-update-setup complete", { exact: false }), + ).toBeVisible({ timeout: 60_000 }); + + await testPage.goto(`/t/${taskBeta.id}`); + const betaSession = new SessionPage(testPage); + await betaSession.waitForLoad(); + await expect( + betaSession.chat.getByText("diff-update-setup complete", { exact: false }), + ).toBeVisible({ timeout: 60_000 }); + + // Both tasks now have diffs and the live monitor has run at least once + // — meaning the orchestrator should have persisted a live_monitor + // snapshot row for each session via UpsertLatestLiveGitSnapshot. + + // Restart the backend. This destroys all in-memory executions, so + // GetExecutionBySessionID will return nil for both sessions on the next + // session.subscribe — forcing the DB-snapshot fallback path to run. + await backend.restart(); + + // Reload and navigate back to one task to re-establish the WS connection. + // The sidebar bulk-subscribes to all primary sessions on connect. + await testPage.goto(`/t/${taskAlpha.id}`); + await alphaSession.waitForLoad(); + + // Both Alpha (active) and Beta (non-active) should display +N/-N badges. + // diff-update-setup leaves one modified file with 1 added line and 1 + // removed line, so we just assert the badge text pattern is present in + // each task row rather than a specific number (mock-agent details may + // change). The key assertion is that the badge appears at all for the + // non-active task — which is the bug. + const alphaRow = alphaSession.sidebar + .getByTestId("sidebar-task-item") + .filter({ hasText: "Diff Alpha" }); + const betaRow = alphaSession.sidebar + .getByTestId("sidebar-task-item") + .filter({ hasText: "Diff Beta" }); + + await expect(alphaRow).toBeVisible({ timeout: 15_000 }); + await expect(betaRow).toBeVisible({ timeout: 15_000 }); + + // Diff badge is rendered as "+N -N" inside a font-mono span. + await expect(alphaRow.getByText(/\+\d+\s+-\d+/)).toBeVisible({ timeout: 30_000 }); + await expect(betaRow.getByText(/\+\d+\s+-\d+/)).toBeVisible({ timeout: 30_000 }); + }); +});