bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
agent-orchestrator/backend/internal/lifecycle/manager_test.go

1053 lines
39 KiB
Go
Raw Blame History

package lifecycle
import (
"context"
"errors"
"strings"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
)
var ctx = context.Background()
type fakeStore struct {
sessions map[domain.SessionID]domain.SessionRecord
prs map[domain.SessionID][]domain.PullRequest
signatures map[string]string
signatureWriteErr error
signatureWrites int
}
func newFakeStore() *fakeStore {
return &fakeStore{sessions: map[domain.SessionID]domain.SessionRecord{}, prs: map[domain.SessionID][]domain.PullRequest{}, signatures: map[string]string{}}
}
func (f *fakeStore) GetSession(_ context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) {
r, ok := f.sessions[id]
return r, ok, nil
}
func (f *fakeStore) ListPRsBySession(_ context.Context, id domain.SessionID) ([]domain.PullRequest, error) {
return f.prs[id], nil
}
func (f *fakeStore) UpdateSession(_ context.Context, rec domain.SessionRecord) error {
f.sessions[rec.ID] = rec
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
}
type telemetrySink struct {
events []ports.TelemetryEvent
}
func (s *telemetrySink) Emit(_ context.Context, ev ports.TelemetryEvent) {
s.events = append(s.events, ev)
}
func (*telemetrySink) Close(context.Context) error { return nil }
func (f *fakeMessenger) Send(_ context.Context, _ domain.SessionID, msg string) error {
if f.err != nil {
return f.err
}
f.msgs = append(f.msgs, msg)
return nil
}
func newManager() (*Manager, *fakeStore, *fakeMessenger) {
st := newFakeStore()
msg := &fakeMessenger{}
return New(st, msg), st, msg
}
func working(id domain.SessionID) domain.SessionRecord {
return domain.SessionRecord{ID: id, ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: time.Now()}}
}
func TestRuntimeObservation_InferredDeathSetsTerminated(t *testing.T) {
m, st, _ := newManager()
rec := working("mer-1")
rec.Activity.LastActivityAt = time.Now().Add(-2 * time.Minute)
st.sessions["mer-1"] = rec
if err := m.ApplyRuntimeObservation(ctx, "mer-1", ports.RuntimeFacts{Probe: ports.ProbeDead}); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"]
if !got.IsTerminated || got.Activity.State != domain.ActivityExited {
t.Fatalf("want terminated/exited, got %+v", got)
}
}
func TestRuntimeObservation_FailedProbeDoesNotMutate(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
before := st.sessions["mer-1"]
if err := m.ApplyRuntimeObservation(ctx, "mer-1", ports.RuntimeFacts{Probe: ports.ProbeFailed}); err != nil {
t.Fatal(err)
}
if st.sessions["mer-1"] != before {
t.Fatalf("failed probe should not persist a state, got %+v", st.sessions["mer-1"])
}
}
func TestActivity_InvalidIsIgnored(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
before := st.sessions["mer-1"]
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: false, State: domain.ActivityIdle}); err != nil {
t.Fatal(err)
}
if st.sessions["mer-1"] != before {
t.Fatal("invalid signal must not mutate")
}
}
func TestActivity_MissingSessionReturnsNotFound(t *testing.T) {
m, _, _ := newManager()
err := m.ApplyActivitySignal(ctx, "missing-1", ports.ActivitySignal{Valid: true, State: domain.ActivityWaitingInput})
if !errors.Is(err, ports.ErrSessionNotFound) {
t.Fatalf("err = %v, want ErrSessionNotFound", err)
}
}
func TestMarkTerminated(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
if err := m.MarkTerminated(ctx, "mer-1"); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"]
if !got.IsTerminated || got.Activity.State != domain.ActivityExited {
t.Fatalf("want terminated/exited, got %+v", got)
}
}
func TestMarkSpawnedStoresRuntimeMetadata(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", IsTerminated: true}
metadata := domain.SessionMetadata{Branch: "b", WorkspacePath: "/ws", RuntimeHandleID: "h1", AgentSessionID: "agent", Prompt: "prompt"}
if err := m.MarkSpawned(ctx, "mer-1", metadata); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"]
if got.IsTerminated || got.Activity.State != domain.ActivityIdle || got.Metadata.RuntimeHandleID != "h1" {
t.Fatalf("spawn metadata wrong: %+v", got)
}
}
// TestMarkSpawned_StampsUTCActivity locks the lifecycle clock to UTC so
// activity-driven timestamps match the session manager's spawn timestamps. A
// local clock here left `ao session get` showing created in UTC but updated in
// local time.
func TestMarkSpawned_StampsUTCActivity(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", IsTerminated: true}
if err := m.MarkSpawned(ctx, "mer-1", domain.SessionMetadata{RuntimeHandleID: "h1"}); err != nil {
t.Fatal(err)
}
if loc := st.sessions["mer-1"].Activity.LastActivityAt.Location(); loc != time.UTC {
t.Fatalf("LastActivityAt location = %v, want UTC", loc)
}
}
func TestActivity_WaitingInputEntryAndExitEmitTelemetry(t *testing.T) {
st := newFakeStore()
sink := &telemetrySink{}
m := New(st, nil, WithTelemetry(sink))
now := time.Unix(100, 0).UTC()
m.clock = func() time.Time { return now }
st.sessions["mer-1"] = domain.SessionRecord{
ID: "mer-1",
ProjectID: "mer",
Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now.Add(-time.Minute)},
}
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityWaitingInput, Timestamp: now}); err != nil {
t.Fatal(err)
}
now = now.Add(3 * time.Second)
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: now}); err != nil {
t.Fatal(err)
}
if len(sink.events) != 2 {
t.Fatalf("events = %#v, want waiting_input entered/exited", sink.events)
}
if sink.events[0].Name != "ao.session.waiting_input_entered" || sink.events[1].Name != "ao.session.waiting_input_exited" {
t.Fatalf("event names = %#v", []string{sink.events[0].Name, sink.events[1].Name})
}
if got := sink.events[1].Payload["dwell_ms"]; got != int64(3000) {
t.Fatalf("dwell_ms = %#v, want 3000", got)
}
}
func TestPRObservation_CIFailingNudgesAgentWithLogs(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.PRObservation{Fetched: true, URL: "pr1", 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.Fatal(err)
}
if len(msg.msgs) != 1 || !strings.Contains(msg.msgs[0], "boom") {
t.Fatalf("want one CI nudge with log tail, got %v", msg.msgs)
}
}
func TestPRObservation_ReviewCommentsNudgeAgent(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.PRObservation{Fetched: true, URL: "pr1", Review: domain.ReviewChangesRequest, Comments: []ports.PRCommentObservation{{ID: "1", Author: "alice", Body: "fix this"}}}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 || !strings.Contains(msg.msgs[0], "fix this") {
t.Fatalf("want review nudge, got %v", msg.msgs)
}
}
func TestPRObservation_CINudgeSanitizesLogTailControlChars(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
// A CI log tail with an embedded ANSI escape sequence and a NUL byte; the
// agent's pane must receive the visible text without the control bytes.
o := ports.PRObservation{Fetched: true, URL: "pr1", CI: domain.CIFailing, Checks: []ports.PRCheckObservation{{Name: "build", CommitHash: "c1", Status: domain.PRCheckFailed, LogTail: "line1\x1b[2Jline2\x00\ttabbed"}}}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 {
t.Fatalf("want one CI nudge, got %v", msg.msgs)
}
got := msg.msgs[0]
if strings.ContainsRune(got, '\x1b') || strings.ContainsRune(got, '\x00') {
t.Fatalf("nudge still carries control bytes: %q", got)
}
if !strings.Contains(got, "line1") || !strings.Contains(got, "line2") || !strings.Contains(got, "\ttabbed") {
t.Fatalf("nudge dropped visible text or tab: %q", got)
}
}
func TestPRObservation_ReviewNudgeSanitizesCommentControlChars(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.PRObservation{Fetched: true, URL: "pr1", Review: domain.ReviewChangesRequest, Comments: []ports.PRCommentObservation{{ID: "1", Body: "please\x1b]0;pwned\afix this"}}}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 {
t.Fatalf("want one review nudge, got %v", msg.msgs)
}
got := msg.msgs[0]
if strings.ContainsRune(got, '\x1b') || strings.ContainsRune(got, '\a') {
t.Fatalf("review nudge still carries control bytes: %q", got)
}
if !strings.Contains(got, "please") || !strings.Contains(got, "fix this") {
t.Fatalf("review nudge dropped visible text: %q", got)
}
}
func TestSCMObservationProjectsToExistingPRReactions(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.SCMObservation{
Fetched: true,
PR: ports.SCMPRObservation{URL: "pr1", Number: 1},
CI: ports.SCMCIObservation{
Summary: string(domain.CIFailing),
HeadSHA: "c1",
FailedChecks: []ports.SCMCheckObservation{{
Name: "build", Status: string(domain.PRCheckFailed), LogTail: "boom",
}},
},
}
if err := m.ApplySCMObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 || !strings.Contains(msg.msgs[0], "boom") {
t.Fatalf("want SCM CI nudge with log tail, got %v", msg.msgs)
}
}
func TestSCMObservation_MissingSessionIsIgnored(t *testing.T) {
st := newFakeStore()
m := New(st, nil)
o := ports.SCMObservation{
Fetched: true,
PR: ports.SCMPRObservation{URL: "pr1", Number: 1},
}
if err := m.ApplySCMObservation(ctx, "missing-1", o); err != nil {
t.Fatalf("ApplySCMObservation missing session: %v", err)
}
}
func TestSCMObservationUsesPRHeadWhenCIHeadMissing(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.SCMObservation{
Fetched: true,
PR: ports.SCMPRObservation{URL: "pr1", HeadSHA: "c1"},
CI: ports.SCMCIObservation{
Summary: string(domain.CIFailing),
FailedChecks: []ports.SCMCheckObservation{{
Name: "build", Status: string(domain.PRCheckFailed),
}},
},
}
if err := m.ApplySCMObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
o.PR.HeadSHA = "c2"
if err := m.ApplySCMObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 2 {
t.Fatalf("want separate CI nudges for distinct PR heads when CI head is absent, got %d: %v", len(msg.msgs), msg.msgs)
}
}
func TestPRObservation_MergeConflictNudgesAgent(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.PRObservation{Fetched: true, URL: "pr1", Mergeability: domain.MergeConflicting}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 || !strings.Contains(msg.msgs[0], "merge conflicts") {
t.Fatalf("want merge-conflict nudge, got %v", msg.msgs)
}
}
func TestPRObservation_MergedTerminatesWithoutNudge(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{{URL: "pr1", Merged: true}}
if err := m.ApplyPRObservation(ctx, "mer-1", ports.PRObservation{Fetched: true, URL: "pr1", Merged: true}); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"]
if !got.IsTerminated || got.Activity.State != domain.ActivityExited {
t.Fatalf("merged PR should terminate session, got %+v", got)
}
if len(msg.msgs) != 0 {
t.Fatalf("merged PR should not send nudge, got %v", msg.msgs)
}
}
// A session with one merged PR and one still-open PR must NOT terminate: the
// completion bar is "no open PR remains AND at least one merged".
func TestPRObservation_MergedWithOpenSiblingDoesNotTerminate(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{
{URL: "pr1", Merged: true},
{URL: "pr2"},
}
if err := m.ApplyPRObservation(ctx, "mer-1", ports.PRObservation{Fetched: true, URL: "pr1", Merged: true}); err != nil {
t.Fatal(err)
}
if got := st.sessions["mer-1"]; got.IsTerminated {
t.Fatalf("session with an open sibling PR must stay alive, got %+v", got)
}
}
// Once the last open PR merges (all PRs now merged), the session terminates.
func TestPRObservation_LastMergeTerminatesSession(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{
{URL: "pr1", Merged: true},
{URL: "pr2", Merged: true},
}
if err := m.ApplyPRObservation(ctx, "mer-1", ports.PRObservation{Fetched: true, URL: "pr2", Merged: true}); err != nil {
t.Fatal(err)
}
if got := st.sessions["mer-1"]; !got.IsTerminated {
t.Fatalf("session should terminate once all PRs are merged, got %+v", got)
}
}
// A closed PR that leaves the session with an open sibling and no merge does not
// terminate; closing the last PR with no merge also does not terminate (nothing
// shipped).
func TestPRObservation_ClosedWithoutMergeDoesNotTerminate(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{{URL: "pr1", Closed: true}}
if err := m.ApplyPRObservation(ctx, "mer-1", ports.PRObservation{Fetched: true, URL: "pr1", Closed: true}); err != nil {
t.Fatal(err)
}
if got := st.sessions["mer-1"]; got.IsTerminated {
t.Fatalf("a closed-without-merge PR must not terminate the session, got %+v", got)
}
}
// A PR stacked on an open parent (its target branch is the parent's source
// branch) is exempt from the merge-conflict nudge: conflicts there are expected
// until the parent merges.
func TestPRObservation_StackedChildConflictSuppressed(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{
{URL: "parent", SourceBranch: "ao/x", TargetBranch: "main"},
{URL: "child", SourceBranch: "ao/x/auth", TargetBranch: "ao/x"},
}
o := ports.PRObservation{Fetched: true, URL: "child", Mergeability: domain.MergeConflicting}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 0 {
t.Fatalf("stacked child conflict should be suppressed, got %v", msg.msgs)
}
}
// The bottom-of-stack PR (not stacked on any open parent) still gets the
// merge-conflict nudge even when it has open stacked children.
func TestPRObservation_BottomOfStackConflictNudges(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
st.prs["mer-1"] = []domain.PullRequest{
{URL: "parent", SourceBranch: "ao/x", TargetBranch: "main"},
{URL: "child", SourceBranch: "ao/x/auth", TargetBranch: "ao/x"},
}
o := ports.PRObservation{Fetched: true, URL: "parent", Mergeability: domain.MergeConflicting}
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 || !strings.Contains(msg.msgs[0], "merge conflicts") {
t.Fatalf("bottom-of-stack conflict should nudge, got %v", msg.msgs)
}
}
// 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 TestApplyReviewResultSendsAndDedupsThroughPRSignature(t *testing.T) {
st := newFakeStore()
st.sessions["mer-1"] = working("mer-1")
msg := &fakeMessenger{}
m := New(st, msg)
result := ReviewResult{
RunID: "run-1",
WorkerID: "mer-1",
PRURL: "https://github.com/o/r/pull/1",
TargetSHA: "sha1",
Verdict: domain.VerdictChangesRequested,
Body: "fix the bug",
GithubReviewID: "98\x1b[2J765",
}
outcome, err := m.ApplyReviewResult(ctx, "mer-1", result)
if err != nil {
t.Fatalf("ApplyReviewResult: %v", err)
}
if outcome != ReviewDeliverySent || len(msg.msgs) != 1 {
t.Fatalf("outcome/messages = %q/%v, want sent once", outcome, msg.msgs)
}
got := msg.msgs[0]
for _, want := range []string{"[AO reviewer]", "PR: " + result.PRURL, "Verdict: changes_requested", "Review body:\nfix the bug", "GitHub review: 98[2J765"} {
if !strings.Contains(got, want) {
t.Fatalf("AO review nudge missing %q: %q", want, got)
}
}
if strings.Contains(got, "\x1b") {
t.Fatalf("AO review nudge should sanitize control bytes: %q", got)
}
if st.signatures[result.PRURL] == "" {
t.Fatal("AO review nudge did not persist sendOnce signature")
}
outcome, err = m.ApplyReviewResult(ctx, "mer-1", result)
if err != nil {
t.Fatalf("repeat ApplyReviewResult: %v", err)
}
if outcome != ReviewDeliverySent || len(msg.msgs) != 1 {
t.Fatalf("repeat should report delivered outcome and suppress duplicate send, outcome=%q msgs=%v", outcome, msg.msgs)
}
result.RunID = "run-2"
result.TargetSHA = "sha2"
outcome, err = m.ApplyReviewResult(ctx, "mer-1", result)
if err != nil {
t.Fatalf("new pass ApplyReviewResult: %v", err)
}
if outcome != ReviewDeliverySent || len(msg.msgs) != 2 {
t.Fatalf("new review pass should send again, outcome=%q msgs=%v", outcome, msg.msgs)
}
}
func TestApplyReviewBatchSendsCombinedAndDedups(t *testing.T) {
st := newFakeStore()
st.sessions["mer-1"] = working("mer-1")
msg := &fakeMessenger{}
m := New(st, msg)
results := []ReviewResult{
{RunID: "run-2", BatchID: "batch-1", WorkerID: "mer-1", PRURL: "https://github.com/o/r/pull/2", TargetSHA: "sha2", Verdict: domain.VerdictChangesRequested, Body: "fix tests", GithubReviewID: "102"},
{RunID: "run-1", BatchID: "batch-1", WorkerID: "mer-1", PRURL: "https://github.com/o/r/pull/1", TargetSHA: "sha1", Verdict: domain.VerdictChangesRequested, Body: "fix auth", GithubReviewID: "101"},
}
outcome, err := m.ApplyReviewBatch(ctx, "mer-1", "batch-1", results)
if err != nil {
t.Fatalf("ApplyReviewBatch: %v", err)
}
if outcome != ReviewDeliverySent || len(msg.msgs) != 1 {
t.Fatalf("outcome/messages = %q/%v, want sent once", outcome, msg.msgs)
}
got := msg.msgs[0]
for _, want := range []string{
"submitted 2 review(s) requesting changes",
"PR: https://github.com/o/r/pull/1",
"GitHub review: 101",
"Review body:\nfix auth",
"PR: https://github.com/o/r/pull/2",
"GitHub review: 102",
"Review body:\nfix tests",
} {
if !strings.Contains(got, want) {
t.Fatalf("batch nudge missing %q: %q", want, got)
}
}
if st.signatures["https://github.com/o/r/pull/1"] == "" {
t.Fatal("batch nudge did not persist signature on anchor PR")
}
outcome, err = m.ApplyReviewBatch(ctx, "mer-1", "batch-1", results)
if err != nil {
t.Fatalf("repeat ApplyReviewBatch: %v", err)
}
if outcome != ReviewDeliverySent || len(msg.msgs) != 1 {
t.Fatalf("repeat should suppress duplicate send, outcome=%q msgs=%v", outcome, msg.msgs)
}
}
func TestApplyReviewBatchNoopsWithoutDeliverableResults(t *testing.T) {
st := newFakeStore()
st.sessions["mer-1"] = working("mer-1")
msg := &fakeMessenger{}
m := New(st, msg)
outcome, err := m.ApplyReviewBatch(ctx, "mer-1", "batch-1", nil)
if err != nil {
t.Fatalf("ApplyReviewBatch: %v", err)
}
if outcome != ReviewDeliveryNoop || len(msg.msgs) != 0 || st.signatureWrites != 0 {
t.Fatalf("empty batch should no-op, outcome=%q msgs=%v signatureWrites=%d", outcome, msg.msgs, st.signatureWrites)
}
}
func TestApplyReviewResultNoopsWhenIrrelevant(t *testing.T) {
deliveredAt := time.Unix(100, 0).UTC()
tests := []struct {
name string
result ReviewResult
rec domain.SessionRecord
}{
{
name: "approved",
result: ReviewResult{RunID: "run-1", PRURL: "pr1", Verdict: domain.VerdictApproved},
rec: working("mer-1"),
},
{
name: "already delivered",
result: ReviewResult{RunID: "run-1", PRURL: "pr1", Verdict: domain.VerdictChangesRequested, DeliveredAt: &deliveredAt},
rec: working("mer-1"),
},
{
name: "terminated worker",
result: ReviewResult{RunID: "run-1", PRURL: "pr1", Verdict: domain.VerdictChangesRequested},
rec: func() domain.SessionRecord { r := working("mer-1"); r.IsTerminated = true; return r }(),
},
{
name: "worker waiting input",
result: ReviewResult{RunID: "run-1", PRURL: "pr1", Verdict: domain.VerdictChangesRequested},
rec: domain.SessionRecord{ID: "mer-1", Activity: domain.Activity{State: domain.ActivityWaitingInput}},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = tt.rec
outcome, err := m.ApplyReviewResult(ctx, "mer-1", tt.result)
if err != nil {
t.Fatalf("ApplyReviewResult: %v", err)
}
if outcome != ReviewDeliveryNoop || len(msg.msgs) != 0 || st.signatureWrites != 0 {
t.Fatalf("irrelevant result should no-op, outcome=%q msgs=%v signatureWrites=%d", outcome, msg.msgs, st.signatureWrites)
}
})
}
}
func TestApplyTrackerFacts_TerminalStateMarksTerminated(t *testing.T) {
for _, state := range []domain.NormalizedIssueState{domain.IssueDone, domain.IssueCancelled} {
t.Run(string(state), func(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: state},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
got := st.sessions["mer-1"]
if !got.IsTerminated || got.Activity.State != domain.ActivityExited {
t.Fatalf("want terminated/exited for state %q, got %+v", state, got)
}
if len(msg.msgs) != 0 {
t.Fatalf("terminal state should not nudge, got %v", msg.msgs)
}
})
}
}
func TestApplyTrackerFacts_AssigneeChangedIsLogOnly(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
before := st.sessions["mer-1"]
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: domain.IssueOpen, Assignee: "someone-else"},
Changed: ports.TrackerChanged{Assignee: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
if st.sessions["mer-1"] != before {
t.Fatalf("assignee-only change must not mutate the session row, got %+v", st.sessions["mer-1"])
}
if len(msg.msgs) != 0 {
t.Fatalf("assignee-only change must not nudge, got %v", msg.msgs)
}
}
func TestApplyTrackerFacts_NewBotCommentNudges(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: domain.IssueOpen},
Comments: []ports.TrackerCommentObservation{
{ID: "human-1", Author: "alice", Body: "human chime-in, must NOT nudge", IsBot: false},
{ID: "bot-1", Author: "ci-bot[bot]", Body: "please rerun the migration", IsBot: true},
},
Changed: ports.TrackerChanged{Comments: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 1 {
t.Fatalf("want one bot-mention nudge, got %d: %v", len(msg.msgs), msg.msgs)
}
if !strings.Contains(msg.msgs[0], "please rerun the migration") {
t.Fatalf("nudge should include the bot comment body, got %q", msg.msgs[0])
}
if strings.Contains(msg.msgs[0], "human chime-in") {
t.Fatalf("nudge must not include human comments, got %q", msg.msgs[0])
}
}
func TestApplyTrackerFacts_NudgeSuppressedOnRepeat(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: domain.IssueOpen},
Comments: []ports.TrackerCommentObservation{
{ID: "bot-1", Author: "ci-bot[bot]", Body: "please rerun the migration", IsBot: true},
},
Changed: ports.TrackerChanged{Comments: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("first ApplyTrackerFacts: %v", err)
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("second ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 1 {
t.Fatalf("repeat observation must dedup; got %d nudges: %v", len(msg.msgs), msg.msgs)
}
// A genuinely new bot comment still fires.
o.Comments = append(o.Comments, ports.TrackerCommentObservation{ID: "bot-2", Author: "ci-bot[bot]", Body: "now check the seed", IsBot: true})
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("third ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 2 {
t.Fatalf("new bot comment id should re-fire, got %d: %v", len(msg.msgs), msg.msgs)
}
}
func TestApplyTrackerFacts_BotCommentWithEmptyIDIsIgnored(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
// Bot comment lacks an ID — without one we cannot dedup, and the
// zero-value signature collides with m.react.seen's empty default and
// would silently suppress every future nudge for this issue. The
// reducer must skip it entirely.
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: domain.IssueOpen},
Comments: []ports.TrackerCommentObservation{
{ID: "", Author: "ci-bot[bot]", Body: "no id, must be skipped", IsBot: true},
},
Changed: ports.TrackerChanged{Comments: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 0 {
t.Fatalf("bot comment with empty ID must not nudge, got %v", msg.msgs)
}
// A subsequent, properly-formed bot comment must still nudge — the
// earlier empty-ID entry must not have polluted the dedup signature.
o.Comments = []ports.TrackerCommentObservation{
{ID: "bot-1", Author: "ci-bot[bot]", Body: "now with an id", IsBot: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("second ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 1 {
t.Fatalf("follow-up bot comment with real ID should nudge, got %d: %v", len(msg.msgs), msg.msgs)
}
}
func TestApplyTrackerFacts_NotFetchedIsNoop(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
before := st.sessions["mer-1"]
if err := m.ApplyTrackerFacts(ctx, "mer-1", ports.TrackerObservation{Fetched: false}); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
if st.sessions["mer-1"] != before {
t.Fatalf("not-fetched observation must not mutate state")
}
if len(msg.msgs) != 0 {
t.Fatalf("not-fetched observation must not nudge")
}
}
func TestApplyTrackerFacts_TerminatedSessionDoesNotRefireOrNudge(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", IsTerminated: true, Activity: domain.Activity{State: domain.ActivityExited}}
o := ports.TrackerObservation{
Fetched: true,
Issue: ports.TrackerIssueObservation{URL: "https://github.com/o/r/issues/1", State: domain.IssueOpen},
Comments: []ports.TrackerCommentObservation{
{ID: "bot-1", Body: "x", IsBot: true},
},
Changed: ports.TrackerChanged{Comments: true},
}
if err := m.ApplyTrackerFacts(ctx, "mer-1", o); err != nil {
t.Fatalf("ApplyTrackerFacts: %v", err)
}
if len(msg.msgs) != 0 {
t.Fatalf("terminated session must not receive nudges, got %v", msg.msgs)
}
}
func TestPRObservation_RetriesAfterMessengerFailure(t *testing.T) {
m, st, msg := newManager()
st.sessions["mer-1"] = working("mer-1")
o := ports.PRObservation{Fetched: true, URL: "pr1", Mergeability: domain.MergeConflicting}
msg.err = errors.New("temporary send failure")
if err := m.ApplyPRObservation(ctx, "mer-1", o); err == nil {
t.Fatal("want send error")
}
msg.err = nil
if err := m.ApplyPRObservation(ctx, "mer-1", o); err != nil {
t.Fatal(err)
}
if len(msg.msgs) != 1 {
t.Fatalf("want retry to send once, got %v", msg.msgs)
}
}
func TestActivity_FirstSignalStampsReceipt(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: time.Now()}}
// A same-state repeat (idle on an idle-seeded row) must still write: the
// receipt itself is the durable fact that clears no_signal.
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityIdle}); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"]
if got.FirstSignalAt.IsZero() {
t.Fatalf("first signal not stamped: %+v", got)
}
stamped := got.FirstSignalAt
// Later signals must not move the receipt.
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: time.Now().Add(time.Minute)}); err != nil {
t.Fatal(err)
}
if got := st.sessions["mer-1"]; !got.FirstSignalAt.Equal(stamped) {
t.Fatalf("first signal moved: %v -> %v", stamped, got.FirstSignalAt)
}
}
func TestActivity_SameStateRepeatAfterReceiptIsNoOp(t *testing.T) {
m, st, _ := newManager()
rec := working("mer-1")
rec.FirstSignalAt = time.Now()
st.sessions["mer-1"] = rec
before := st.sessions["mer-1"]
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive}); err != nil {
t.Fatal(err)
}
if st.sessions["mer-1"] != before {
t.Fatalf("same-state repeat after receipt must not rewrite: %+v", st.sessions["mer-1"])
}
}
func TestMarkSpawnedClearsFirstSignal(t *testing.T) {
m, st, _ := newManager()
rec := working("mer-1")
rec.FirstSignalAt = time.Now().Add(-time.Hour)
st.sessions["mer-1"] = rec
if err := m.MarkSpawned(ctx, "mer-1", domain.SessionMetadata{}); err != nil {
t.Fatal(err)
}
if got := st.sessions["mer-1"]; !got.FirstSignalAt.IsZero() {
t.Fatalf("spawn/restore must clear the receipt, got %+v", got)
}
}
type fakeNotificationSink struct {
intents []ports.NotificationIntent
err error
}
func (f *fakeNotificationSink) Notify(_ context.Context, intent ports.NotificationIntent) error {
f.intents = append(f.intents, intent)
return f.err
}
func TestActivity_WaitingInputTransitionEmitsNotification(t *testing.T) {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
now := time.Date(2026, 6, 11, 10, 0, 0, 0, time.UTC)
m.clock = func() time.Time { return now }
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", DisplayName: "checkout-flow", Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: now.Add(-time.Minute)}, FirstSignalAt: now.Add(-time.Minute)}
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityWaitingInput}); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 1 {
t.Fatalf("intents = %d, want 1", len(sink.intents))
}
intent := sink.intents[0]
if intent.Type != domain.NotificationNeedsInput || intent.SessionID != "mer-1" || intent.ProjectID != "mer" || intent.SessionDisplayName != "checkout-flow" {
t.Fatalf("intent = %+v", intent)
}
}
func TestActivity_WaitingInputSameStateDoesNotEmitNotification(t *testing.T) {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
now := time.Now()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityWaitingInput, LastActivityAt: now}, FirstSignalAt: now}
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityWaitingInput}); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 0 {
t.Fatalf("same-state waiting_input emitted %+v", sink.intents)
}
}
func TestActivity_TerminatedSessionDoesNotEmitNotification(t *testing.T) {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", IsTerminated: true, Activity: domain.Activity{State: domain.ActivityExited}}
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityWaitingInput}); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 0 {
t.Fatalf("terminated session emitted %+v", sink.intents)
}
}
func TestSCMObservation_Notifications(t *testing.T) {
for _, tc := range []struct {
name string
obs ports.SCMObservation
want domain.NotificationType
}{
{
name: "ready",
obs: ports.SCMObservation{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1, Title: "checkout"}, CI: ports.SCMCIObservation{Summary: string(domain.CIPassing)}, Review: ports.SCMReviewObservation{Decision: string(domain.ReviewApproved)}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
want: domain.NotificationReadyToMerge,
},
{
name: "merged",
obs: ports.SCMObservation{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/2", Number: 2, Merged: true}},
want: domain.NotificationPRMerged,
},
{
name: "closed",
obs: ports.SCMObservation{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/3", Number: 3, Closed: true}},
want: domain.NotificationPRClosedUnmerged,
},
} {
t.Run(tc.name, func(t *testing.T) {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
st.sessions["mer-1"] = working("mer-1")
if err := m.ApplySCMObservation(ctx, "mer-1", tc.obs); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 1 {
t.Fatalf("intents = %d, want 1", len(sink.intents))
}
if got := sink.intents[0]; got.Type != tc.want || got.PRURL != tc.obs.PR.URL || got.PRNumber != tc.obs.PR.Number {
t.Fatalf("intent = %+v, want type %s", got, tc.want)
}
})
}
}
func TestSCMObservation_NotReadyWhenCIOrReviewBlocks(t *testing.T) {
for _, obs := range []ports.SCMObservation{
{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, CI: ports.SCMCIObservation{Summary: string(domain.CIFailing)}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, CI: ports.SCMCIObservation{Summary: string(domain.CIPending)}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, CI: ports.SCMCIObservation{Summary: string(domain.CIUnknown)}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
{Fetched: true, PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1}, CI: ports.SCMCIObservation{Summary: string(domain.CIPassing)}, Review: ports.SCMReviewObservation{Decision: string(domain.ReviewChangesRequest)}, Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)}},
} {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
st.sessions["mer-1"] = working("mer-1")
if err := m.ApplySCMObservation(ctx, "mer-1", obs); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 0 {
t.Fatalf("blocked PR emitted %+v", sink.intents)
}
}
}
func TestSCMObservation_ReadyToMergeSuppressedWhileWaitingInput(t *testing.T) {
st := newFakeStore()
sink := &fakeNotificationSink{}
m := New(st, nil, WithNotificationSink(sink))
rec := working("mer-1")
rec.Activity.State = domain.ActivityWaitingInput
st.sessions["mer-1"] = rec
obs := ports.SCMObservation{
Fetched: true,
PR: ports.SCMPRObservation{URL: "https://github.com/o/r/pull/1", Number: 1},
CI: ports.SCMCIObservation{Summary: string(domain.CIPassing)},
Mergeability: ports.SCMMergeabilityObservation{State: string(domain.MergeMergeable)},
}
if err := m.ApplySCMObservation(ctx, "mer-1", obs); err != nil {
t.Fatal(err)
}
if len(sink.intents) != 0 {
t.Fatalf("waiting-input session emitted ready notification: %+v", sink.intents)
}
}
Reference in New Issue View Git Blame Copy Permalink