diff --git a/TODO.md b/TODO.md index 020ef76..7f19bc6 100644 --- a/TODO.md +++ b/TODO.md @@ -2,8 +2,6 @@ - Question answers given while the agent is working get lost Answering an inbox question while the task's agent is mid-turn sometimes never reaches the agent. Answers are meant to be held and delivered as one message once the session is idle with no blocking question still waiting (`Deliver` in `internal/agents/inbox.go`, called when a turn ends). Find the path where a held answer is dropped. Byre restarts often while agents deploy, and a restart cuts the turn off and resumes it with the restart notice (`Recover` in `internal/agents/manager.go`), so check that path first. Add debugging output before changing code. -- Merge is forgotten if the human switches tasks before the merge turn ends - Clicking Merge starts a merge-in turn and records, in memory only, that the task is to be squashed when that turn ends (`m.merges` in `internal/agents/mergein.go`, read by `takeMerge` at turn end). The human then switched to another task, and when the turn finished nothing was merged. Switching tasks in the UI shouldn't affect server state, so the likelier cause is that byre restarted during the turn, which loses the in-memory intent (as described under "Closing a task" in `docs/operations.md`). Confirm the cause, and make the merge intent survive until the turn ends. # M1 — One agent end to end diff --git a/docs/operations.md b/docs/operations.md index 989c390..b96c5c1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -28,9 +28,9 @@ A task's `merge_title` is the title of the squash commit Merge makes. It is the A task's participants are the users who put a message bubble with their name in its transcript, derived from its events (`Store.Participants`): the actors of its `user.message` events, joined to `users` so `byre` and `agent` drop out. That covers messages sent to the agent and inbox answers, whose delivery message carries the name of the first responder in it (`Deliver` in `internal/agents/inbox.go`); dismissing an item puts nothing in the transcript and doesn't count. They are ordered by their first such message. `GET /api/tasks/{id}/co-authors` returns the participants other than the caller, with their git name and email from `PUT /api/me` (empty for a user who hasn't set one). The Merge confirm fetches it and lists `Name ` for each co-author, and separately those with no git identity, who get no trailer. -`POST /api/tasks/{id}/merge` starts the same turn and records, in memory, that the task is to be merged by the requesting user, who must have a git identity (400 otherwise). When that turn ends normally (not interrupted, not an error), byre squashes the worktree, now containing the default branch, onto it with `git merge-tree --merge-base`, makes one commit titled with the task's `merge_title` with `git commit-tree`, authored and committed by the merging user with no agent attribution, its body one `Co-authored-by: Name ` trailer per co-author with a git identity (`squashMessage` in `internal/agents/mergein.go`; the co-authors are computed when the squash runs, relative to the user who clicked Merge), and pushes that commit to the default branch on the `origin` remote of the project's checkout (`git push --porcelain origin :refs/heads/`, run in the checkout). Only once the push is accepted does it run `git merge --ff-only` in the project's checkout (which must have the default branch checked out) and close the task; the `task.merged` event carries the commit and `pushed` (`origin/`), and the transcript shows both. When the merge changes nothing, no commit is made and nothing is pushed. +`POST /api/tasks/{id}/merge` starts the same turn and records in `tasks.merging_by` that the task is to be merged by the requesting user, who must have a git identity (400 otherwise). Every turn end of the task's session reads and clears it. It survives a byre restart during the turn: startup resumes the session with the restart notice, and the merge happens when that resumed turn ends. When that turn ends normally (not interrupted, not an error), byre squashes the worktree, now containing the default branch, onto it with `git merge-tree --merge-base`, makes one commit titled with the task's `merge_title` with `git commit-tree`, authored and committed by the merging user with no agent attribution, its body one `Co-authored-by: Name ` trailer per co-author with a git identity (`squashMessage` in `internal/agents/mergein.go`; the co-authors are computed when the squash runs, relative to the user who clicked Merge), and pushes that commit to the default branch on the `origin` remote of the project's checkout (`git push --porcelain origin :refs/heads/`, run in the checkout). Only once the push is accepted does it run `git merge --ff-only` in the project's checkout (which must have the default branch checked out) and close the task; the `task.merged` event carries the commit and `pushed` (`origin/`), and the transcript shows both. When the merge changes nothing, no commit is made and nothing is pushed. -Before starting the turn, Merge runs `git fetch origin ` and `git merge --ff-only origin/` in the checkout, so the agent merges in what the remote has; the request waits for the fetch, and a failure (network, or a checkout with commits the remote lacks) is returned as the request's error. If the push is still refused, for example because someone pushed during the turn, or fails for any other reason, byre files the error in the inbox and the task stays active; clicking Merge again fetches and goes round once more. A refused push leaves the checkout's default branch where it was, since the squash commit was never moved onto it. If the squash fails, for example because the default branch moved again during the turn, byre also files an error and the task stays active. An interrupted or failed turn, or a byre restart during it, leaves the task active without squashing. `POST /api/tasks/{id}/discard` removes everything without merging. +Before starting the turn, Merge runs `git fetch origin ` and `git merge --ff-only origin/` in the checkout, so the agent merges in what the remote has; the request waits for the fetch, and a failure (network, or a checkout with commits the remote lacks) is returned as the request's error. If the push is still refused, for example because someone pushed during the turn, or fails for any other reason, byre files the error in the inbox and the task stays active; clicking Merge again fetches and goes round once more. A refused push leaves the checkout's default branch where it was, since the squash commit was never moved onto it. If the squash fails, for example because the default branch moved again during the turn, byre also files an error and the task stays active. An interrupted or failed turn, an agent process that exits mid-turn, or a session that can't be resumed after a restart clears `merging_by` and leaves the task active without squashing. `POST /api/tasks/{id}/discard` removes everything without merging. ## Running as a service diff --git a/internal/agents/manager.go b/internal/agents/manager.go index cc7baf3..c0ecd9a 100644 --- a/internal/agents/manager.go +++ b/internal/agents/manager.go @@ -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 } diff --git a/internal/agents/mergein.go b/internal/agents/mergein.go index 0ce77bd..ef95c6c 100644 --- a/internal/agents/mergein.go +++ b/internal/agents/mergein.go @@ -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) { diff --git a/internal/store/merging_test.go b/internal/store/merging_test.go new file mode 100644 index 0000000..695b408 --- /dev/null +++ b/internal/store/merging_test.go @@ -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) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index ce0bf90..dae9133 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -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}) }