Persist merge intent so a restart mid-merge-turn doesn't lose it
This commit is contained in:
@@ -51,7 +51,6 @@ type Manager struct {
|
||||
live map[string]*live
|
||||
closing bool
|
||||
rateLimits map[api.AgentSessionVendor]map[string]any
|
||||
merges map[string]string
|
||||
leadAnswered map[string]bool
|
||||
paneFocused map[string]bool
|
||||
|
||||
@@ -99,7 +98,6 @@ func New(st *store.Store, limits Limits, worktreeRoot, leadRoot string) *Manager
|
||||
leadRoot: leadRoot,
|
||||
live: map[string]*live{},
|
||||
rateLimits: map[api.AgentSessionVendor]map[string]any{},
|
||||
merges: map[string]string{},
|
||||
leadAnswered: map[string]bool{},
|
||||
paneFocused: map[string]bool{},
|
||||
signals: make(chan api.Event, 1024),
|
||||
@@ -281,6 +279,9 @@ func (m *Manager) Recover(ctx context.Context) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if sess.Owner == "" {
|
||||
m.takeMerge(task.Id)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -699,6 +700,7 @@ func (m *Manager) pump(l *live) {
|
||||
slog.Error("record session exit", "session", l.id, "err", err)
|
||||
}
|
||||
if l.owner == "" {
|
||||
m.takeMerge(l.taskID)
|
||||
m.fileError(l, "The agent process exited mid-turn. Send a message to resume it.")
|
||||
}
|
||||
|
||||
@@ -823,8 +825,8 @@ func (m *Manager) handle(ctx context.Context, l *live, ev claude.Event) error {
|
||||
go m.continueFromHandoff(l.id, res.Result)
|
||||
return nil
|
||||
}
|
||||
mergeBy, merging := m.takeMerge(l.taskID)
|
||||
if merging && finished {
|
||||
mergeBy := m.takeMerge(l.taskID)
|
||||
if mergeBy != "" && finished {
|
||||
go m.finishMerge(l, mergeBy)
|
||||
return nil
|
||||
}
|
||||
|
||||
+11
-12
@@ -50,25 +50,24 @@ func (m *Manager) Merge(ctx context.Context, task api.Task, project api.Project,
|
||||
return api.AgentSession{}, err
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
m.merges[task.Id] = user.Login
|
||||
m.mu.Unlock()
|
||||
err = m.store.SetMergingBy(ctx, task.Id, user.Login)
|
||||
if err != nil {
|
||||
return api.AgentSession{}, err
|
||||
}
|
||||
|
||||
sess, err := m.MergeIn(ctx, task, user, project.DefaultBranch)
|
||||
if err != nil {
|
||||
m.mu.Lock()
|
||||
delete(m.merges, task.Id)
|
||||
m.mu.Unlock()
|
||||
m.takeMerge(task.Id)
|
||||
}
|
||||
return sess, err
|
||||
}
|
||||
|
||||
func (m *Manager) takeMerge(taskID string) (string, bool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
login, ok := m.merges[taskID]
|
||||
delete(m.merges, taskID)
|
||||
return login, ok
|
||||
func (m *Manager) takeMerge(taskID string) string {
|
||||
login, err := m.store.TakeMergingBy(context.Background(), taskID)
|
||||
if err != nil {
|
||||
slog.Error("take merge intent", "task", taskID, "err", err)
|
||||
}
|
||||
return login
|
||||
}
|
||||
|
||||
func (m *Manager) finishMerge(l *live, login string) {
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"g.fc.run/misc/byre/internal/api"
|
||||
"g.fc.run/misc/byre/internal/store"
|
||||
)
|
||||
|
||||
func TestMergingBySurvivesReopen(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
path := filepath.Join(t.TempDir(), "byre.db")
|
||||
st, err := store.Open(path, func(api.Event) {})
|
||||
if err != nil {
|
||||
t.Fatalf("open store: %v", err)
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
_, err = st.EnsureUser(ctx, "alice", "Alice")
|
||||
if err != nil {
|
||||
t.Fatalf("ensure user: %v", err)
|
||||
}
|
||||
err = st.CreateProject(ctx, api.Project{Id: "p1", Name: "demo", RepoPath: "/nonexistent", DefaultBranch: "main", Remote: "none", CreatedBy: "alice", CreatedAt: now})
|
||||
if err != nil {
|
||||
t.Fatalf("create project: %v", err)
|
||||
}
|
||||
err = st.CreateTask(ctx, api.Task{Id: "t1", ProjectId: "p1", Title: "t1", Branch: "t1", WorktreePath: "/nonexistent", BaseRef: "x", Status: api.TaskStatusActive, CreatedBy: "alice", CreatedAt: now}, "alice")
|
||||
if err != nil {
|
||||
t.Fatalf("create task: %v", err)
|
||||
}
|
||||
|
||||
err = st.SetMergingBy(ctx, "t1", "alice")
|
||||
if err != nil {
|
||||
t.Fatalf("set merging by: %v", err)
|
||||
}
|
||||
st.Close()
|
||||
|
||||
st, err = store.Open(path, func(api.Event) {})
|
||||
if err != nil {
|
||||
t.Fatalf("reopen store: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { st.Close() })
|
||||
|
||||
login, err := st.TakeMergingBy(ctx, "t1")
|
||||
if err != nil {
|
||||
t.Fatalf("take merging by: %v", err)
|
||||
}
|
||||
if login != "alice" {
|
||||
t.Fatalf("merging by = %q, want alice", login)
|
||||
}
|
||||
|
||||
login, err = st.TakeMergingBy(ctx, "t1")
|
||||
if err != nil {
|
||||
t.Fatalf("take merging by again: %v", err)
|
||||
}
|
||||
if login != "" {
|
||||
t.Fatalf("merging by after take = %q, want empty", login)
|
||||
}
|
||||
}
|
||||
@@ -248,6 +248,8 @@ CREATE INDEX lead_inputs_owner ON lead_inputs(owner, id);
|
||||
UPDATE users SET focus = json_set(focus, '$.project_id', (SELECT project_id FROM tasks WHERE id = json_extract(users.focus, '$.task_id')))
|
||||
WHERE json_extract(focus, '$.task_id') IS NOT NULL;
|
||||
DELETE FROM lead_inputs WHERE text LIKE 'The human switched the right pane %' OR text = 'The human cleared the right pane.';
|
||||
`}, {"add-task-merging-by", `
|
||||
ALTER TABLE tasks ADD COLUMN merging_by TEXT NOT NULL DEFAULT '';
|
||||
`}}
|
||||
|
||||
type Store struct {
|
||||
@@ -564,6 +566,29 @@ func (s *Store) SetMergeTitle(ctx context.Context, id, title, actor string) erro
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) SetMergingBy(ctx context.Context, id, login string) error {
|
||||
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
_, err := tx.ExecContext(ctx, `UPDATE tasks SET merging_by = ? WHERE id = ?`, login, id)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Store) TakeMergingBy(ctx context.Context, id string) (string, error) {
|
||||
login := ""
|
||||
err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
||||
err := tx.QueryRowContext(ctx, `SELECT merging_by FROM tasks WHERE id = ?`, id).Scan(&login)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(ctx, `UPDATE tasks SET merging_by = '' WHERE id = ?`, id)
|
||||
return err
|
||||
})
|
||||
return login, err
|
||||
}
|
||||
|
||||
func (s *Store) MergeTask(ctx context.Context, id, commit, pushed, actor string) (api.Task, error) {
|
||||
return s.closeTask(ctx, id, api.TaskStatusMerged, commit, actor, "task.merged", map[string]any{"id": id, "commit": commit, "pushed": pushed})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user