fix(orchestrator): persist live git status snapshot for sidebar diff badges (#556)
This commit is contained in:
@@ -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"`
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user