From bfb6e9860b9947e067ab373bb844811c2e022ac4 Mon Sep 17 00:00:00 2001 From: Harshit Singh Bhandari Date: Fri, 5 Jun 2026 23:20:02 +0530 Subject: [PATCH] feat(scm): wire observer messenger + RepoOriginURL + persist dedup (#108) (#114) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(daemon): thread runtime messenger into Lifecycle Manager (#108) The daemon used to construct the LCM with a nil messenger, so every SCM-driven nudge dropped silently inside sendOnce. Move newSessionMessenger above startLifecycle and pass the real messenger through, so CI-failure, review-feedback, and merge-conflict nudges actually reach the agent. Co-Authored-By: Claude Opus 4.7 * feat(project): populate RepoOriginURL at add + lazy observer backfill (#108) project.Add now shells out to `git -C path remote get-url origin` and captures the result on the new project row, so the SCM observer can parse it on the first poll. A missing remote falls back to "" rather than failing project add — non-git roots and remoteless repos stay registerable. To cover projects added before this change, the observer's discoverSubjects lazily backfills RepoOriginURL via the same shell-out and persists it through UpsertProject, so subsequent polls skip the fork-exec. Co-Authored-By: Claude Opus 4.7 * feat(lifecycle): persist reaction-dedup signatures across restart (#108) Add migration 0005 with `pr.last_nudge_signature TEXT NOT NULL DEFAULT ''` and two scoped sqlc queries (Get/UpdatePRLastNudgeSignature). Lifecycle serialises the per-PR slice of its seen/attempts maps to that column as a small JSON document; sendOnce loads it lazily on first touch of each PR and persists after every successful send. This closes the post-restart re-nudge gap: the daemon used to lose the seen map on bounce, so a still-failing CI re-prompted the agent on the first post-restart observer poll even when it had already been told. Co-Authored-By: Claude Opus 4.7 * fix(lifecycle): silence nilerr on intentional corrupt-payload swallow golangci-lint's nilerr flagged the `if err := json.Unmarshal(...); err != nil { return nil }` path in loadPRSignaturesLocked. The swallow is deliberate (a corrupt persisted payload should not crash the lifecycle write path), so compare against nil directly so no `err` is bound and the lint goes quiet. Co-Authored-By: Claude Opus 4.7 * fix: silence nilerr, address reviewer notes, drop task-tagged comments - reactions.go: discard the json.Unmarshal error explicitly via `_ =` so golangci-lint's nilerr stops flagging the intentional corrupt-payload swallow; behavior unchanged. - reactions.go: document the Send → memory → persist order in sendOnce so the "one extra nudge on restart after a transient persist failure" trade-off is explicit (vs. the inverse risk of losing a real nudge). - service.go: stop reaching for slog.Default() in resolveGitOriginURL; align with the observer's identical helper that just returns "" on git failure rather than logging through the global logger. - tests: drop "issue #108" / "guards the regression from #X" framing in test docstrings — explain WHAT the test asserts, not the PR context. Co-Authored-By: Claude Opus 4.7 --- backend/internal/daemon/daemon.go | 12 +- backend/internal/daemon/lifecycle_wiring.go | 6 +- backend/internal/daemon/wiring_test.go | 65 ++++++++++ backend/internal/lifecycle/manager.go | 4 + backend/internal/lifecycle/manager_test.go | 101 +++++++++++++++- backend/internal/lifecycle/reactions.go | 112 ++++++++++++++++-- backend/internal/observe/scm/observer.go | 22 ++++ backend/internal/observe/scm/observer_test.go | 65 ++++++++++ backend/internal/service/project/service.go | 21 +++- .../internal/service/project/service_test.go | 44 +++++++ backend/internal/storage/sqlite/gen/pr.sql.go | 25 ++++ .../0005_pr_last_nudge_signature.sql | 18 +++ .../internal/storage/sqlite/queries/pr.sql | 6 + .../internal/storage/sqlite/store/pr_store.go | 25 ++++ 14 files changed, 504 insertions(+), 22 deletions(-) create mode 100644 backend/internal/storage/sqlite/migrations/0005_pr_last_nudge_signature.sql diff --git a/backend/internal/daemon/daemon.go b/backend/internal/daemon/daemon.go index 5cb7ae090..438de56a3 100644 --- a/backend/internal/daemon/daemon.go +++ b/backend/internal/daemon/daemon.go @@ -76,16 +76,18 @@ func Run() error { termMgr := terminal.NewManager(runtimeAdapter, cdcPipe.Broadcaster, log) defer termMgr.Close() + // The agent messenger sends validated user input to the session's live + // zellij pane. Keep this path small until durable inbox semantics are needed. + // Built before the Lifecycle Manager so the LCM can use it for SCM-driven + // agent nudges (CI failure, review feedback, merge conflict). + messenger := newSessionMessenger(store, runtimeAdapter, log) + // Bring up the Lifecycle Manager and the reaper first: it makes the session // lifecycle write path live (reducer write -> store -> DB trigger -> // change_log -> poller -> broadcaster) and gives startSession the shared LCM. - lcStack := startLifecycle(ctx, store, runtimeAdapter, log) + lcStack := startLifecycle(ctx, store, runtimeAdapter, messenger, log) lcStack.scmDone = startSCMObserver(ctx, store, lcStack.LCM, log) - // The agent messenger sends validated user input to the session's live - // zellij pane. Keep this path small until durable inbox semantics are needed. - messenger := newSessionMessenger(store, runtimeAdapter, log) - // Wire the controller-facing session service over the same store + LCM, the // zellij runtime, a gitworktree workspace, the per-session agent resolver // (AO_AGENT default, validated here), and the agent messenger, then mount it diff --git a/backend/internal/daemon/lifecycle_wiring.go b/backend/internal/daemon/lifecycle_wiring.go index 1356ffc84..a313a6db6 100644 --- a/backend/internal/daemon/lifecycle_wiring.go +++ b/backend/internal/daemon/lifecycle_wiring.go @@ -34,8 +34,10 @@ type lifecycleStack struct { // startLifecycle constructs the Lifecycle Manager over the store and starts the // reaper. The goroutine stops when ctx is cancelled; Stop waits for it to drain. -func startLifecycle(ctx context.Context, store *sqlite.Store, runtime ports.Runtime, logger *slog.Logger) *lifecycleStack { - lcm := lifecycle.New(store, nil) +// The messenger is the per-daemon agent messenger the LCM uses to nudge agents +// in response to SCM observations (CI failure, review feedback, merge conflict). +func startLifecycle(ctx context.Context, store *sqlite.Store, runtime ports.Runtime, messenger ports.AgentMessenger, logger *slog.Logger) *lifecycleStack { + lcm := lifecycle.New(store, messenger) rp := reaper.New(lcm, store, runtime, reaper.Config{Logger: logger}) return &lifecycleStack{LCM: lcm, reaperDone: rp.Start(ctx)} } diff --git a/backend/internal/daemon/wiring_test.go b/backend/internal/daemon/wiring_test.go index 93b023e73..963aa17a9 100644 --- a/backend/internal/daemon/wiring_test.go +++ b/backend/internal/daemon/wiring_test.go @@ -254,6 +254,71 @@ func TestWiring_SessionMessengerRejectsTerminatedSession(t *testing.T) { } } +type captureMessenger struct { + msgs []capturedMessage +} + +type capturedMessage struct { + id domain.SessionID + msg string +} + +func (c *captureMessenger) Send(_ context.Context, id domain.SessionID, msg string) error { + c.msgs = append(c.msgs, capturedMessage{id: id, msg: msg}) + return nil +} + +// TestWiring_StartLifecycleThreadsMessengerIntoLCM asserts startLifecycle +// constructs the LCM with a real messenger by driving an SCM observation +// through the wired stack and checking the messenger receives the CI-failure +// nudge — a nil messenger here would silently drop the send inside sendOnce. +func TestWiring_StartLifecycleThreadsMessengerIntoLCM(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + // Cancel must run BEFORE Stop so the reaper goroutine's ctx.Done() fires; + // Stop is a no-op otherwise. Cleanup is LIFO, so register Stop first. + store, err := sqlite.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + + if err := store.UpsertProject(ctx, domain.ProjectRecord{ID: "p", Path: "/repo/p", RegisteredAt: time.Now()}); err != nil { + t.Fatal(err) + } + rec, err := store.CreateSession(ctx, domain.SessionRecord{ + ProjectID: "p", Kind: domain.KindWorker, + Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: time.Now()}, + }) + if err != nil { + t.Fatal(err) + } + + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + messenger := &captureMessenger{} + stack := startLifecycle(ctx, store, zellij.New(zellij.Options{}), messenger, log) + t.Cleanup(stack.Stop) + t.Cleanup(cancel) + + obs := ports.SCMObservation{ + Fetched: true, + PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1, HeadSHA: "c1"}, + CI: ports.SCMCIObservation{ + Summary: string(domain.CIFailing), + HeadSHA: "c1", + FailedChecks: []ports.SCMCheckObservation{{Name: "build", Status: string(domain.PRCheckFailed), LogTail: "boom"}}, + }, + } + if err := stack.LCM.ApplySCMObservation(ctx, rec.ID, obs); err != nil { + t.Fatalf("ApplySCMObservation: %v", err) + } + if len(messenger.msgs) != 1 { + t.Fatalf("want one nudge to flow through the wired messenger, got %d", len(messenger.msgs)) + } + if messenger.msgs[0].id != rec.ID { + t.Fatalf("nudge sent to %q, want %q", messenger.msgs[0].id, rec.ID) + } +} + // TestProjectRepoResolver_ResolvesRegisteredProject asserts the DB-backed repo // resolver turns a registered project into its on-disk repo path (so spawns // materialise a worktree), and fails loudly for an unregistered project. diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index f76620fb3..8b7eb3f26 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -17,6 +17,10 @@ import ( type sessionStore interface { GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) UpdateSession(ctx context.Context, rec domain.SessionRecord) error + // GetPRLastNudgeSignature / UpdatePRLastNudgeSignature persist the + // reaction-dedup map so nudges survive a daemon restart. + GetPRLastNudgeSignature(ctx context.Context, prURL string) (string, error) + UpdatePRLastNudgeSignature(ctx context.Context, prURL, payload string) error } // Manager reduces runtime, activity, spawn, and termination observations into durable session facts. diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index 24fb3c6cb..6d86189e4 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/backend/internal/lifecycle/manager_test.go @@ -14,11 +14,15 @@ import ( var ctx = context.Background() type fakeStore struct { - sessions map[domain.SessionID]domain.SessionRecord + sessions map[domain.SessionID]domain.SessionRecord + signatures map[string]string + + signatureWriteErr error + signatureWrites int } func newFakeStore() *fakeStore { - return &fakeStore{sessions: map[domain.SessionID]domain.SessionRecord{}} + return &fakeStore{sessions: map[domain.SessionID]domain.SessionRecord{}, signatures: map[string]string{}} } func (f *fakeStore) GetSession(_ context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { @@ -31,6 +35,22 @@ func (f *fakeStore) UpdateSession(_ context.Context, rec domain.SessionRecord) e return nil } +func (f *fakeStore) GetPRLastNudgeSignature(_ context.Context, prURL string) (string, error) { + return f.signatures[prURL], nil +} + +func (f *fakeStore) UpdatePRLastNudgeSignature(_ context.Context, prURL, payload string) error { + if f.signatureWriteErr != nil { + return f.signatureWriteErr + } + if f.signatures == nil { + f.signatures = map[string]string{} + } + f.signatures[prURL] = payload + f.signatureWrites++ + return nil +} + type fakeMessenger struct { msgs []string err error @@ -216,6 +236,83 @@ func TestPRObservation_MergedTerminatesWithoutNudge(t *testing.T) { } } +// TestPRObservation_DedupSurvivesManagerRestart simulates a daemon restart by +// constructing a second Manager over the same store and asserts that an +// identical PR observation does not re-fire the nudge — the dedup signature +// must survive process restart, not just live in the Manager's maps. +func TestPRObservation_DedupSurvivesManagerRestart(t *testing.T) { + st := newFakeStore() + st.sessions["mer-1"] = working("mer-1") + + o := ports.PRObservation{ + Fetched: true, + URL: "https://github.com/o/r/pull/1", + CI: domain.CIFailing, + Checks: []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "boom"}}, + } + + first := &fakeMessenger{} + m1 := New(st, first) + if err := m1.ApplyPRObservation(ctx, "mer-1", o); err != nil { + t.Fatalf("first ApplyPRObservation: %v", err) + } + if len(first.msgs) != 1 { + t.Fatalf("first manager: want 1 nudge, got %d", len(first.msgs)) + } + if got := st.signatures[o.URL]; got == "" { + t.Fatalf("signature was not persisted; want a non-empty JSON payload for %q", o.URL) + } + + // Simulate daemon restart: the second Manager has no in-memory state but + // shares the same store, so it should hydrate seen/attempts from the + // persisted payload and suppress the re-send. + second := &fakeMessenger{} + m2 := New(st, second) + if err := m2.ApplyPRObservation(ctx, "mer-1", o); err != nil { + t.Fatalf("second ApplyPRObservation: %v", err) + } + if len(second.msgs) != 0 { + t.Fatalf("post-restart manager re-nudged on identical observation, got %d msgs: %v", len(second.msgs), second.msgs) + } + + // And a genuinely new signature (different log tail) still fires — proving + // the persisted state is per-signature, not a blanket "this PR was nudged". + o2 := o + o2.Checks = []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "different boom"}} + if err := m2.ApplyPRObservation(ctx, "mer-1", o2); err != nil { + t.Fatalf("third ApplyPRObservation: %v", err) + } + if len(second.msgs) != 1 { + t.Fatalf("new signature should send, got %d msgs", len(second.msgs)) + } +} + +func TestPRObservation_DedupPersistsAcrossPRs(t *testing.T) { + st := newFakeStore() + st.sessions["mer-1"] = working("mer-1") + msg := &fakeMessenger{} + m := New(st, msg) + + for _, url := range []string{"https://github.com/o/r/pull/1", "https://github.com/o/r/pull/2"} { + o := ports.PRObservation{ + Fetched: true, URL: url, CI: domain.CIFailing, + Checks: []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "boom"}}, + } + if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil { + t.Fatalf("ApplyPRObservation for %s: %v", url, err) + } + } + if len(msg.msgs) != 2 { + t.Fatalf("distinct PRs should each get one nudge, got %d", len(msg.msgs)) + } + if _, ok := st.signatures["https://github.com/o/r/pull/1"]; !ok { + t.Fatal("missing persisted signature for PR 1") + } + if _, ok := st.signatures["https://github.com/o/r/pull/2"]; !ok { + t.Fatal("missing persisted signature for PR 2") + } +} + func TestPRObservation_RetriesAfterMessengerFailure(t *testing.T) { m, st, msg := newManager() st.sessions["mer-1"] = working("mer-1") diff --git a/backend/internal/lifecycle/reactions.go b/backend/internal/lifecycle/reactions.go index abf65891c..c921a932c 100644 --- a/backend/internal/lifecycle/reactions.go +++ b/backend/internal/lifecycle/reactions.go @@ -2,6 +2,7 @@ package lifecycle import ( "context" + "encoding/json" "strings" "sync" @@ -15,10 +16,22 @@ type reactionState struct { mu sync.Mutex seen map[string]string attempts map[string]int + // loaded tracks PR URLs whose persisted dedup payload has been merged into + // seen/attempts during this process. Lazy: we only pay the DB read on the + // first reaction touching each PR after startup. + loaded map[string]bool } func newReactionState() reactionState { - return reactionState{seen: map[string]string{}, attempts: map[string]int{}} + return reactionState{seen: map[string]string{}, attempts: map[string]int{}, loaded: map[string]bool{}} +} + +// reactionPayload is the JSON document persisted in pr.last_nudge_signature. +// Keeping the schema explicit (and stable) lets the daemon restart and resume +// the existing dedup state without re-nudging an agent. +type reactionPayload struct { + Seen map[string]string `json:"seen,omitempty"` + Attempts map[string]int `json:"attempts,omitempty"` } // ApplyPRObservation reacts to a fetched PR observation after the PR service has @@ -49,7 +62,7 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o if ch.LogTail != "" { msg += "\n\nFailing output:\n" + ch.LogTail } - return m.sendOnce(ctx, id, "ci:"+o.URL+":"+ch.Name, ch.CommitHash+":"+ch.LogTail, msg, 0) + return m.sendOnce(ctx, id, o.URL, "ci:"+o.URL+":"+ch.Name, ch.CommitHash+":"+ch.LogTail, msg, 0) } } } @@ -62,10 +75,10 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o if sig == "" { sig = string(o.Review) } - return m.sendOnce(ctx, id, "review:"+o.URL, sig, msg, reviewMaxNudge) + return m.sendOnce(ctx, id, o.URL, "review:"+o.URL, sig, msg, reviewMaxNudge) } if o.Mergeability == domain.MergeConflicting { - return m.sendOnce(ctx, id, "merge-conflict:"+o.URL, string(o.Mergeability), "Your PR has merge conflicts. Rebase onto the base branch and resolve them.", 0) + return m.sendOnce(ctx, id, o.URL, "merge-conflict:"+o.URL, string(o.Mergeability), "Your PR has merge conflicts. Rebase onto the base branch and resolve them.", 0) } return nil } @@ -170,26 +183,107 @@ func reviewContent(comments []ports.PRCommentObservation) (string, string) { return strings.Join(bodies, "\n\n"), strings.Join(ids, ",") } -func (m *Manager) sendOnce(ctx context.Context, id domain.SessionID, key, sig, msg string, maxAttempts int) error { +func (m *Manager) sendOnce(ctx context.Context, id domain.SessionID, prURL, key, sig, msg string, maxAttempts int) error { if m.messenger == nil { return nil } m.react.mu.Lock() + defer m.react.mu.Unlock() + + if prURL != "" && !m.react.loaded[prURL] { + if err := m.loadPRSignaturesLocked(ctx, prURL); err != nil { + return err + } + m.react.loaded[prURL] = true + } + if m.react.seen[key] == sig { - m.react.mu.Unlock() return nil } attempts := m.react.attempts[key] if maxAttempts > 0 && attempts >= maxAttempts { - m.react.mu.Unlock() return nil } if err := m.messenger.Send(ctx, id, msg); err != nil { - m.react.mu.Unlock() return err } + // Order: Send → in-memory mutation → durable persist. Sending first means a + // transient persist failure does NOT swallow a real send (the agent saw the + // message; subsequent polls in this process suppress re-sends via the + // in-memory dedup). A persist failure that survives until a daemon restart + // degrades to one extra nudge — preferred over the inverse (persist before + // send, then crash mid-call) which would silently lose a real nudge. m.react.seen[key] = sig m.react.attempts[key] = attempts + 1 - m.react.mu.Unlock() + if prURL != "" { + if err := m.persistPRSignaturesLocked(ctx, prURL); err != nil { + return err + } + } return nil } + +// loadPRSignaturesLocked merges any previously persisted reaction-dedup state +// for prURL into the in-memory maps. Caller must hold m.react.mu. +func (m *Manager) loadPRSignaturesLocked(ctx context.Context, prURL string) error { + raw, err := m.store.GetPRLastNudgeSignature(ctx, prURL) + if err != nil { + return err + } + if raw == "" { + return nil + } + // A corrupt persisted payload must not crash the lifecycle write path; + // the worst case from a swallow is re-firing a nudge once. + var p reactionPayload + _ = json.Unmarshal([]byte(raw), &p) + for k, v := range p.Seen { + if _, ok := m.react.seen[k]; !ok { + m.react.seen[k] = v + } + } + for k, v := range p.Attempts { + if cur, ok := m.react.attempts[k]; !ok || v > cur { + m.react.attempts[k] = v + } + } + return nil +} + +// persistPRSignaturesLocked serialises every reaction-dedup entry whose key +// references prURL and writes the JSON payload back via the store. Caller must +// hold m.react.mu. A failed persist surfaces upward so the in-memory mutation +// (which the messenger already acted on) is not silently divergent from disk. +func (m *Manager) persistPRSignaturesLocked(ctx context.Context, prURL string) error { + payload := reactionPayload{Seen: map[string]string{}, Attempts: map[string]int{}} + for k, v := range m.react.seen { + if reactionKeyTargetsPR(k, prURL) { + payload.Seen[k] = v + } + } + for k, v := range m.react.attempts { + if reactionKeyTargetsPR(k, prURL) { + payload.Attempts[k] = v + } + } + raw, err := json.Marshal(payload) + if err != nil { + return err + } + return m.store.UpdatePRLastNudgeSignature(ctx, prURL, string(raw)) +} + +// reactionKeyTargetsPR matches the ":[:]" reaction keys used +// by ApplyPRObservation. Anchoring on the second colon-delimited segment keeps +// PR-specific keys grouped with the row that survives a restart. +func reactionKeyTargetsPR(key, prURL string) bool { + if prURL == "" { + return false + } + parts := strings.SplitN(key, ":", 2) + if len(parts) != 2 { + return false + } + rest := parts[1] + return rest == prURL || strings.HasPrefix(rest, prURL+":") +} diff --git a/backend/internal/observe/scm/observer.go b/backend/internal/observe/scm/observer.go index a1ebd8f79..2c1473b3c 100644 --- a/backend/internal/observe/scm/observer.go +++ b/backend/internal/observe/scm/observer.go @@ -12,6 +12,7 @@ import ( "errors" "fmt" "log/slog" + "os/exec" "strings" "time" @@ -47,6 +48,7 @@ type Provider interface { type Store interface { ListAllSessions(ctx context.Context) ([]domain.SessionRecord, error) GetProject(ctx context.Context, id string) (domain.ProjectRecord, bool, error) + UpsertProject(ctx context.Context, row domain.ProjectRecord) error ListPRsBySession(ctx context.Context, sessionID domain.SessionID) ([]domain.PullRequest, error) ListChecks(ctx context.Context, prURL string) ([]domain.PullRequestCheck, error) WriteSCMObservation(ctx context.Context, pr domain.PullRequest, checks []domain.PullRequestCheck, threads []domain.PullRequestReviewThread, comments []domain.PullRequestComment, reviewMode ports.ReviewWriteMode) error @@ -447,6 +449,14 @@ func (o *Observer) discoverSubjects(ctx context.Context) (map[string]*subject, e if !found || !p.ArchivedAt.IsZero() { continue } + if p.RepoOriginURL == "" && p.Path != "" { + if url := resolveGitOriginURL(p.Path); url != "" { + p.RepoOriginURL = url + if err := o.store.UpsertProject(ctx, p); err != nil { + o.logger.Warn("scm observer: backfill origin URL persist failed", "project", p.ID, "err", err) + } + } + } projects[sess.ProjectID] = p proj = p } @@ -1124,6 +1134,18 @@ func normalizePRState(draft, merged, closed bool) string { } } +// resolveGitOriginURL runs `git -C path remote get-url origin` and returns the +// trimmed URL, or "" if the command fails (missing repo, no origin remote, etc). +// The observer uses this to backfill projects that were registered before +// project.Add resolved origin URLs at add time. +func resolveGitOriginURL(path string) string { + out, err := exec.Command("git", "-C", path, "remote", "get-url", "origin").Output() + if err != nil { + return "" + } + return strings.TrimSpace(string(out)) +} + func scrubLine(s string) string { s = strings.ReplaceAll(s, "\n", " ") s = strings.ReplaceAll(s, "\r", " ") diff --git a/backend/internal/observe/scm/observer_test.go b/backend/internal/observe/scm/observer_test.go index 55c61823a..448f893b4 100644 --- a/backend/internal/observe/scm/observer_test.go +++ b/backend/internal/observe/scm/observer_test.go @@ -10,6 +10,7 @@ import ( "fmt" "io" "log/slog" + "os/exec" "sync" "testing" "time" @@ -69,6 +70,16 @@ func (s *fakeStore) GetProject(_ context.Context, id string) (domain.ProjectReco return p, ok, nil } +func (s *fakeStore) UpsertProject(_ context.Context, row domain.ProjectRecord) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.projects == nil { + s.projects = map[string]domain.ProjectRecord{} + } + s.projects[row.ID] = row + return nil +} + func (s *fakeStore) ListPRsBySession(_ context.Context, id domain.SessionID) ([]domain.PullRequest, error) { s.mu.Lock() defer s.mu.Unlock() @@ -831,3 +842,57 @@ func TestPoll_DuplicateTrackedPRKeepsFirstSession(t *testing.T) { t.Fatalf("duplicate owner overwrote first session: wrote session %s", store.writes[0].pr.SessionID) } } + +// TestDiscoverSubjects_BackfillsRepoOriginURL asserts that a project row with +// an empty RepoOriginURL but a real on-disk repo gets its origin filled in +// during discovery and persisted, so the same project becomes observable +// without re-running project add. +func TestDiscoverSubjects_BackfillsRepoOriginURL(t *testing.T) { + dir := t.TempDir() + if out, err := exec.Command("git", "init", dir).CombinedOutput(); err != nil { + t.Fatalf("git init: %v (%s)", err, out) + } + if out, err := exec.Command("git", "-C", dir, "remote", "add", "origin", "https://github.com/o/r.git").CombinedOutput(); err != nil { + t.Fatalf("git remote add: %v (%s)", err, out) + } + + store := &fakeStore{ + sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "feat"}}}, + projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir}}, // empty RepoOriginURL + prs: map[domain.SessionID][]domain.PullRequest{}, + checks: map[string][]domain.PullRequestCheck{}, + } + provider := &fakeProvider{} + obs := newTestObserver(store, provider, &fakeLifecycle{}, time.Unix(0, 0).UTC()) + + if _, err := obs.discoverSubjects(context.Background()); err != nil { + t.Fatalf("discoverSubjects: %v", err) + } + if got := store.projects["p"].RepoOriginURL; got != "https://github.com/o/r.git" { + t.Fatalf("RepoOriginURL after backfill = %q, want https://github.com/o/r.git", got) + } +} + +// TestDiscoverSubjects_NonGitPathDoesNotBackfill confirms the backfill is +// best-effort: a non-git project path leaves RepoOriginURL empty without +// erroring or persisting a stub, so the project is simply skipped. +func TestDiscoverSubjects_NonGitPathDoesNotBackfill(t *testing.T) { + dir := t.TempDir() + store := &fakeStore{ + sessions: []domain.SessionRecord{{ID: "p-1", ProjectID: "p", Metadata: domain.SessionMetadata{Branch: "feat"}}}, + projects: map[string]domain.ProjectRecord{"p": {ID: "p", Path: dir}}, + prs: map[domain.SessionID][]domain.PullRequest{}, + checks: map[string][]domain.PullRequestCheck{}, + } + obs := newTestObserver(store, &fakeProvider{}, &fakeLifecycle{}, time.Unix(0, 0).UTC()) + subjects, err := obs.discoverSubjects(context.Background()) + if err != nil { + t.Fatalf("discoverSubjects: %v", err) + } + if len(subjects) != 0 { + t.Fatalf("non-git project should be skipped, got %d subjects", len(subjects)) + } + if got := store.projects["p"].RepoOriginURL; got != "" { + t.Fatalf("RepoOriginURL = %q, want empty (no persist on failed backfill)", got) + } +} diff --git a/backend/internal/service/project/service.go b/backend/internal/service/project/service.go index f19d776b5..966f9bb4a 100644 --- a/backend/internal/service/project/service.go +++ b/backend/internal/service/project/service.go @@ -120,10 +120,11 @@ func (m *Service) Add(ctx context.Context, in AddInput) (Project, error) { } row := domain.ProjectRecord{ - ID: string(id), - Path: path, - DisplayName: name, - RegisteredAt: time.Now(), + ID: string(id), + Path: path, + RepoOriginURL: resolveGitOriginURL(path), + DisplayName: name, + RegisteredAt: time.Now(), } if err := m.store.UpsertProject(ctx, row); err != nil { return Project{}, apierr.Internal("PROJECT_ADD_FAILED", "Failed to register project") @@ -131,6 +132,18 @@ func (m *Service) Add(ctx context.Context, in AddInput) (Project, error) { return projectFromRow(row), nil } +// resolveGitOriginURL returns the project's `origin` remote URL via +// `git -C path remote get-url origin`. A missing remote, missing repo, or any +// other git error returns an empty string — `project add` must not fail just +// because no origin is configured (the SCM observer skips such projects). +func resolveGitOriginURL(path string) string { + out, err := exec.Command("git", "-C", path, "remote", "get-url", "origin").Output() + if err != nil { + return "" + } + return strings.TrimSpace(string(out)) +} + // Remove archives a project registration. func (m *Service) Remove(ctx context.Context, id domain.ProjectID) (RemoveResult, error) { if err := validateProjectID(id); err != nil { diff --git a/backend/internal/service/project/service_test.go b/backend/internal/service/project/service_test.go index 69e8fd807..e67f526f8 100644 --- a/backend/internal/service/project/service_test.go +++ b/backend/internal/service/project/service_test.go @@ -145,6 +145,50 @@ func TestManager_AddValidationAndConflicts(t *testing.T) { wantCode(t, err, "ID_ALREADY_REGISTERED") } +// gitRepoWithOrigin creates a real git repo with an `origin` remote pointing +// at `originURL`. Used to assert project.Add captures the origin at add time. +func gitRepoWithOrigin(t *testing.T, originURL string) string { + t.Helper() + dir := gitRepo(t) + if out, err := exec.Command("git", "-C", dir, "remote", "add", "origin", originURL).CombinedOutput(); err != nil { + t.Fatalf("git remote add: %v (%s)", err, out) + } + return dir +} + +func TestManager_AddPopulatesRepoOriginURL(t *testing.T) { + ctx := context.Background() + + for _, tc := range []struct { + name string + setup func(t *testing.T) string + wantURL string + }{ + { + name: "git repo with origin populates url", + setup: func(t *testing.T) string { return gitRepoWithOrigin(t, "https://github.com/o/r.git") }, + wantURL: "https://github.com/o/r.git", + }, + { + name: "git repo without origin leaves url empty", + setup: func(t *testing.T) string { return gitRepo(t) }, + wantURL: "", + }, + } { + t.Run(tc.name, func(t *testing.T) { + m := newManager(t) + path := tc.setup(t) + proj, err := m.Add(ctx, project.AddInput{Path: path, ProjectID: ptr("p")}) + if err != nil { + t.Fatalf("Add: %v", err) + } + if proj.Repo != tc.wantURL { + t.Fatalf("Repo = %q, want %q", proj.Repo, tc.wantURL) + } + }) + } +} + func TestManager_GetUpdateRemoveErrors(t *testing.T) { ctx := context.Background() m := newManager(t) diff --git a/backend/internal/storage/sqlite/gen/pr.sql.go b/backend/internal/storage/sqlite/gen/pr.sql.go index 4edc7cd53..6b2eb02e0 100644 --- a/backend/internal/storage/sqlite/gen/pr.sql.go +++ b/backend/internal/storage/sqlite/gen/pr.sql.go @@ -379,3 +379,28 @@ func (q *Queries) UpsertPR(ctx context.Context, arg UpsertPRParams) error { ) return err } + +const getPRLastNudgeSignature = `-- name: GetPRLastNudgeSignature :one +SELECT last_nudge_signature FROM pr WHERE url = ? +` + +func (q *Queries) GetPRLastNudgeSignature(ctx context.Context, url string) (string, error) { + row := q.db.QueryRowContext(ctx, getPRLastNudgeSignature, url) + var sig string + err := row.Scan(&sig) + return sig, err +} + +const updatePRLastNudgeSignature = `-- name: UpdatePRLastNudgeSignature :exec +UPDATE pr SET last_nudge_signature = ? WHERE url = ? +` + +type UpdatePRLastNudgeSignatureParams struct { + LastNudgeSignature string + URL string +} + +func (q *Queries) UpdatePRLastNudgeSignature(ctx context.Context, arg UpdatePRLastNudgeSignatureParams) error { + _, err := q.db.ExecContext(ctx, updatePRLastNudgeSignature, arg.LastNudgeSignature, arg.URL) + return err +} diff --git a/backend/internal/storage/sqlite/migrations/0005_pr_last_nudge_signature.sql b/backend/internal/storage/sqlite/migrations/0005_pr_last_nudge_signature.sql new file mode 100644 index 000000000..026e03962 --- /dev/null +++ b/backend/internal/storage/sqlite/migrations/0005_pr_last_nudge_signature.sql @@ -0,0 +1,18 @@ +-- Summary: persist per-PR reaction dedup signatures so agent nudges +-- (CI failure, review feedback, merge conflict) survive a daemon restart +-- instead of re-firing on the first post-restart observer poll. +-- +-- The column carries a small JSON document encoded by lifecycle.Manager: +-- {"seen":{:}, "attempts":{:}} +-- where reaction_key uniquely identifies a nudge target (e.g. "ci::", +-- "review:", "merge-conflict:") and signature is the content +-- fingerprint that gates whether a re-fire is warranted. +-- +goose Up +-- +goose StatementBegin +ALTER TABLE pr ADD COLUMN last_nudge_signature TEXT NOT NULL DEFAULT ''; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE pr DROP COLUMN last_nudge_signature; +-- +goose StatementEnd diff --git a/backend/internal/storage/sqlite/queries/pr.sql b/backend/internal/storage/sqlite/queries/pr.sql index 9eacbb3f4..7ef7009cf 100644 --- a/backend/internal/storage/sqlite/queries/pr.sql +++ b/backend/internal/storage/sqlite/queries/pr.sql @@ -89,6 +89,12 @@ FROM pr WHERE session_id = ? ORDER BY updated_at DESC; +-- name: GetPRLastNudgeSignature :one +SELECT last_nudge_signature FROM pr WHERE url = ?; + +-- name: UpdatePRLastNudgeSignature :exec +UPDATE pr SET last_nudge_signature = ? WHERE url = ?; + -- name: GetDisplayPRFactsBySession :one SELECT pr.url, diff --git a/backend/internal/storage/sqlite/store/pr_store.go b/backend/internal/storage/sqlite/store/pr_store.go index be6649126..17a91344f 100644 --- a/backend/internal/storage/sqlite/store/pr_store.go +++ b/backend/internal/storage/sqlite/store/pr_store.go @@ -137,6 +137,31 @@ func reviewThreadIDs(threads []domain.PullRequestReviewThread, comments []domain return out } +// GetPRLastNudgeSignature returns the persisted nudge-dedup JSON payload for a +// PR (empty string when the PR has no row or no signatures yet). The payload is +// opaque to storage; lifecycle.Manager owns its shape. +func (s *Store) GetPRLastNudgeSignature(ctx context.Context, url string) (string, error) { + sig, err := s.qr.GetPRLastNudgeSignature(ctx, url) + if errors.Is(err, sql.ErrNoRows) { + return "", nil + } + if err != nil { + return "", fmt.Errorf("get pr nudge signature %s: %w", url, err) + } + return sig, nil +} + +// UpdatePRLastNudgeSignature overwrites the persisted nudge-dedup JSON payload +// for a PR. A no-op when the URL has no pr row yet. +func (s *Store) UpdatePRLastNudgeSignature(ctx context.Context, url, payload string) error { + s.writeMu.Lock() + defer s.writeMu.Unlock() + if err := s.qw.UpdatePRLastNudgeSignature(ctx, gen.UpdatePRLastNudgeSignatureParams{LastNudgeSignature: payload, URL: url}); err != nil { + return fmt.Errorf("update pr nudge signature %s: %w", url, err) + } + return nil +} + // GetPR returns the PR facts for a URL, or ok=false if absent. func (s *Store) GetPR(ctx context.Context, url string) (domain.PullRequest, bool, error) { p, err := s.qr.GetPR(ctx, url)