* 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com>
This commit is contained in:
parent
6da5d93241
commit
bfb6e9860b
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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 "<type>:<url>[:<extra>]" 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+":")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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", " ")
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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":{<reaction_key>:<signature>}, "attempts":{<reaction_key>:<count>}}
|
||||
-- where reaction_key uniquely identifies a nudge target (e.g. "ci:<url>:<check>",
|
||||
-- "review:<url>", "merge-conflict:<url>") 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
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue