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

Refactoring Reviews feature code and idempotency check (#377) 

* fix: dedupe AO review change nudges

* fix: satisfy review delivery lint

* fix: renumber review delivery migration

* fix: gate review trigger idempotency on verdict

* fix: preserve running review trigger idempotency

* fix: fail stale running review runs

* Update review.go

* fix: avoid review context import ambiguity

* docs: clarify defensive review idempotency branch

* fix: keep review submit persistence single-sourced

* fix: remove unused review nudge send result

* fix: preserve existing pr review nudge copy

* fix: simplify merge conflict nudge return
This commit is contained in:
neversettle 2026-06-22 15:40:00 +05:30 committed by GitHub
parent cbd2a1baba
commit f85b0d2ffe
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 712 additions and 292 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/internal/daemon/lifecycle_wiring.go
View File

@ -121,12 +121,8 @@ func startSession(cfg config.Config, runtime *zellij.Runtime, store *sqlite.Stor
PRs: store,
Projects: store,
Launcher: reviewcore.NewLauncher(reviewers, runtime),
// On a changes_requested verdict the engine nudges the worker's live pane
// directly, using the same per-daemon messenger lifecycle uses for SCM
// nudges (issue #337).
Messenger: messenger,
})
reviewSvc := reviewsvc.New(reviewEngine)
reviewSvc := reviewsvc.New(reviewEngine, store, reviewsvc.WithLifecycleReducer(lcm))
return sessionSvc, reviewSvc, nil
}

2
backend/internal/domain/review.go
View File

@ -48,6 +48,7 @@ type ReviewRun struct {
// worker so the worker knows exactly which review to address and reply to.
GithubReviewID string `json:"githubReviewId"`
CreatedAt time.Time `json:"createdAt"`
DeliveredAt *time.Time `json:"deliveredAt,omitempty"`
}
// ReviewRunStatus is the lifecycle state of a single review pass.
@ -57,6 +58,7 @@ type ReviewRunStatus string
const (
ReviewRunRunning ReviewRunStatus = "running"
ReviewRunComplete ReviewRunStatus = "complete"
ReviewRunDelivered ReviewRunStatus = "delivered"
ReviewRunFailed ReviewRunStatus = "failed"
)

5
backend/internal/httpd/apispec/openapi.yaml
View File

@ -1897,6 +1897,11 @@ components:
createdAt:
format: date-time
type: string
deliveredAt:
format: date-time
type:
- "null"
- string
githubReviewId:
type: string
harness:

97
backend/internal/lifecycle/manager_test.go
View File

@ -222,7 +222,7 @@ func TestPRObservation_CIFailingNudgesAgentWithLogs(t *testing.T) {
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", Body: "fix this"}}}
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)
}
@ -520,6 +520,101 @@ func TestPRObservation_DedupPersistsAcrossPRs(t *testing.T) {
}
}
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]
if !strings.Contains(got, "[AO reviewer]") || !strings.Contains(got, "fix the bug") || !strings.Contains(got, "98[2J765") {
t.Fatalf("AO review nudge missing label/body/review id: %q", 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 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) {

63
backend/internal/lifecycle/reactions.go
View File

@ -3,9 +3,11 @@ package lifecycle
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"strings"
"sync"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
@ -13,6 +15,33 @@ import (
const reviewMaxNudge = 3
// ReviewDeliveryOutcome reports what ApplyReviewResult did with a completed
// AO-internal review pass.
type ReviewDeliveryOutcome string
const (
// ReviewDeliveryNoop means lifecycle did not send or confirm a review nudge
// because the result was not relevant for delivery.
ReviewDeliveryNoop ReviewDeliveryOutcome = "no_op"
// ReviewDeliverySent means the worker nudge was sent or was already covered
// by sendOnce dedup state and may be stamped delivered.
ReviewDeliverySent ReviewDeliveryOutcome = "sent"
)
// ReviewResult is the already-persisted result of an AO-internal review pass.
// Lifecycle treats it as input to the reaction reducer; it does not write the
// review_run row.
type ReviewResult struct {
RunID string
WorkerID domain.SessionID
PRURL string
TargetSHA string
Verdict domain.ReviewVerdict
Body string
GithubReviewID string
DeliveredAt *time.Time
}
type reactionState struct {
mu sync.Mutex
seen map[string]string
@ -108,6 +137,40 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o
return nil
}
// ApplyReviewResult reacts to a completed AO-internal review pass after the
// review service has persisted the run result. It mirrors ApplyPRObservation:
// no change_log reads, no review_run writes, only lifecycle side effects.
func (m *Manager) ApplyReviewResult(ctx context.Context, workerID domain.SessionID, r ReviewResult) (ReviewDeliveryOutcome, error) {
if r.Verdict != domain.VerdictChangesRequested || r.DeliveredAt != nil {
return ReviewDeliveryNoop, nil
}
rec, ok, err := m.store.GetSession(ctx, workerID)
if err != nil || !ok {
return ReviewDeliveryNoop, err
}
if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput {
return ReviewDeliveryNoop, nil
}
if m.messenger == nil {
return ReviewDeliveryNoop, nil
}
msg := "[AO reviewer] AO's internal code reviewer requested changes on your PR. Review the feedback below and address it."
if r.GithubReviewID != "" {
safeReviewID := domain.SanitizeControlChars(r.GithubReviewID)
msg += fmt.Sprintf(" This feedback is GitHub review %s. Once you have addressed it, reply on that review referencing id %s with how you addressed it, then resolve the review comment threads you addressed.", safeReviewID, safeReviewID)
}
if r.Body != "" {
msg += "\n\n" + domain.SanitizeControlChars(r.Body)
}
key := "review:" + r.PRURL + ":ao:" + r.RunID
sig := strings.Join([]string{r.TargetSHA, r.RunID, r.GithubReviewID, r.Body}, "\x00")
err = m.sendOnce(ctx, workerID, r.PRURL, key, sig, msg, reviewMaxNudge)
if err != nil {
return ReviewDeliveryNoop, err
}
return ReviewDeliverySent, nil
}
// sessionComplete reports whether the session has reached the multi-PR
// completion bar: at least one PR merged and no PR still open. A session with no
// PRs, or with any open PR, is not complete.

150
backend/internal/review/review.go
View File

@ -9,7 +9,7 @@
package review
import (
"context"
stdctx "context"
"errors"
"fmt"
"sync"
@ -18,7 +18,6 @@ import (
"github.com/google/uuid"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
)
// ErrInvalid and ErrNotFound let the transport layer map failures to 422/404.
@ -30,28 +29,30 @@ var (
// Store is the persistence surface the engine needs. *sqlite.Store satisfies it
// in production; tests use a fake.
type Store interface {
UpsertReview(ctx context.Context, r domain.Review) error
GetReviewBySession(ctx context.Context, id domain.SessionID) (domain.Review, bool, error)
InsertReviewRun(ctx context.Context, r domain.ReviewRun) error
UpdateReviewRunResult(ctx context.Context, id string, status domain.ReviewRunStatus, verdict domain.ReviewVerdict, body, githubReviewID string) (bool, error)
GetReviewRun(ctx context.Context, id string) (domain.ReviewRun, bool, error)
GetReviewRunBySessionAndSHA(ctx context.Context, id domain.SessionID, targetSHA string) (domain.ReviewRun, bool, error)
ListReviewRunsBySession(ctx context.Context, id domain.SessionID) ([]domain.ReviewRun, error)
UpsertReview(ctx stdctx.Context, r domain.Review) error
GetReviewBySession(ctx stdctx.Context, id domain.SessionID) (domain.Review, bool, error)
InsertReviewRun(ctx stdctx.Context, r domain.ReviewRun) error
UpdateReviewRunResult(ctx stdctx.Context, id string, status domain.ReviewRunStatus, verdict domain.ReviewVerdict, body, githubReviewID string) (bool, error)
SupersedeReviewRun(ctx stdctx.Context, id, body string) (bool, error)
SupersedeStaleRunningReviewRuns(ctx stdctx.Context, sessionID domain.SessionID, targetSHA, body string) (int64, error)
GetReviewRun(ctx stdctx.Context, id string) (domain.ReviewRun, bool, error)
GetReviewRunBySessionAndSHA(ctx stdctx.Context, id domain.SessionID, targetSHA string) (domain.ReviewRun, bool, error)
ListReviewRunsBySession(ctx stdctx.Context, id domain.SessionID) ([]domain.ReviewRun, error)
}
// Sessions resolves the worker session under review.
type Sessions interface {
GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error)
GetSession(ctx stdctx.Context, id domain.SessionID) (domain.SessionRecord, bool, error)
}
// PRs resolves the PR a worker owns.
type PRs interface {
ListPRsBySession(ctx context.Context, id domain.SessionID) ([]domain.PullRequest, error)
ListPRsBySession(ctx stdctx.Context, id domain.SessionID) ([]domain.PullRequest, error)
}
// Projects resolves the per-project reviewer config.
type Projects interface {
GetProject(ctx context.Context, id string) (domain.ProjectRecord, bool, error)
GetProject(ctx stdctx.Context, id string) (domain.ProjectRecord, bool, error)
}
// Deps wires the engine.
@ -61,7 +62,6 @@ type Deps struct {
PRs PRs
Projects Projects
Launcher Launcher
Messenger ports.AgentMessenger
// Clock and NewID are injectable for deterministic tests.
Clock func() time.Time
@ -75,7 +75,6 @@ type Engine struct {
prs PRs
projects Projects
launcher Launcher
messenger ports.AgentMessenger
clock func() time.Time
newID func() string
@ -102,7 +101,6 @@ func New(d Deps) *Engine {
prs: d.PRs,
projects: d.Projects,
launcher: d.Launcher,
messenger: d.Messenger,
clock: clock,
newID: newID,
triggerLocks: make(map[domain.SessionID]*sync.Mutex),
@ -151,7 +149,7 @@ type SessionReviews struct {
// new commit; if not, a fresh reviewer is spawned;
// - the run is recorded before launch so startup failures leave a visible
// failed pass instead of an empty gap.
func (e *Engine) Trigger(ctx context.Context, workerID domain.SessionID) (TriggerResult, error) {
func (e *Engine) Trigger(ctx stdctx.Context, workerID domain.SessionID) (TriggerResult, error) {
if workerID == "" {
return TriggerResult{}, fmt.Errorf("%w: worker session id is required", ErrInvalid)
}
@ -188,12 +186,29 @@ func (e *Engine) Trigger(ctx context.Context, workerID domain.SessionID) (Trigge
return TriggerResult{}, err
}
// Idempotency: return a non-failed pass as-is. Failed passes stay visible
// but can be retried after the user fixes the underlying issue.
// Idempotency: a pass for this commit is reusable while it is still running
// or once it carries a verdict. The fallback branch below is defensive for
// any non-running, non-failed row that somehow lacks a verdict; normal
// Submit paths complete a run only with a valid verdict (#342).
if existing, ok, err := e.store.GetReviewRunBySessionAndSHA(ctx, workerID, targetSHA); err != nil {
return TriggerResult{}, err
} else if ok && existing.Status != domain.ReviewRunFailed {
} else if ok && (existing.Status == domain.ReviewRunRunning || existing.Verdict != domain.VerdictNone) {
return TriggerResult{Run: existing, ReviewerHandleID: review.ReviewerHandleID, Created: false}, nil
} else if ok && existing.Status != domain.ReviewRunFailed {
superseded, err := e.store.SupersedeReviewRun(ctx, existing.ID, "superseded by a new review trigger")
if err != nil {
return TriggerResult{}, err
}
if !superseded {
if latest, ok, err := e.store.GetReviewRun(ctx, existing.ID); err != nil {
return TriggerResult{}, err
} else if ok {
return TriggerResult{Run: latest, ReviewerHandleID: review.ReviewerHandleID, Created: false}, nil
}
}
}
if _, err := e.store.SupersedeStaleRunningReviewRuns(ctx, workerID, targetSHA, "superseded by a review trigger for a newer commit"); err != nil {
return TriggerResult{}, err
}
harness, err := e.reviewerHarness(ctx, worker)
@ -276,97 +291,8 @@ func (e *Engine) Trigger(ctx context.Context, workerID domain.SessionID) (Trigge
return TriggerResult{Run: run, ReviewerHandleID: handleID, Created: true}, nil
}
// Submit records the reviewer's result for a specific worker review pass: it
// marks the run complete and stores the verdict, body, and the GitHub review id
// the reviewer posted. AO does not post the review — the reviewer agent posts it
// to the PR itself.
//
// On a changes_requested verdict, Submit also messages the worker session with
// the review feedback directly, so the worker learns about it event-driven
// rather than via the SCM poll loop (which never observes CHANGES_REQUESTED for
// self-reviews or COMMENT-state reviews; issue #337). When a GitHub review id is
// known, it is included so the worker knows exactly which review to address and
// reply to.
func (e *Engine) Submit(ctx context.Context, workerID domain.SessionID, runID string, verdict domain.ReviewVerdict, body, githubReviewID string) (domain.ReviewRun, error) {
if workerID == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: worker session id is required", ErrInvalid)
}
if runID == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: review run id is required", ErrInvalid)
}
if !verdict.Valid() {
return domain.ReviewRun{}, fmt.Errorf("%w: verdict must be %q or %q", ErrInvalid, domain.VerdictApproved, domain.VerdictChangesRequested)
}
if verdict == domain.VerdictChangesRequested && body == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: a changes_requested review requires a body", ErrInvalid)
}
run, ok, err := e.store.GetReviewRun(ctx, runID)
if err != nil {
return domain.ReviewRun{}, err
}
if !ok {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q", ErrNotFound, runID)
}
if run.SessionID != workerID {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q does not belong to worker %q", ErrInvalid, runID, workerID)
}
if run.Status != domain.ReviewRunRunning {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q is not running", ErrInvalid, runID)
}
// Notify the worker before marking the run complete. If the message fails,
// the run stays 'running' so a retried `ao review submit` runs again instead
// of tripping the status='running' guard above on an already-completed run. A
// message that lands but a DB write that then fails degrades to one extra
// nudge on retry — the same trade lifecycle's sendOnce makes.
if verdict == domain.VerdictChangesRequested {
if err := e.notifyWorkerChangesRequested(ctx, workerID, body, githubReviewID); err != nil {
return domain.ReviewRun{}, err
}
}
updated, err := e.store.UpdateReviewRunResult(ctx, run.ID, domain.ReviewRunComplete, verdict, body, githubReviewID)
if err != nil {
return domain.ReviewRun{}, err
}
if !updated {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q is not running", ErrInvalid, runID)
}
run.Status = domain.ReviewRunComplete
run.Verdict = verdict
run.Body = body
run.GithubReviewID = githubReviewID
return run, nil
}
// notifyWorkerChangesRequested injects the AO reviewer's feedback into the
// worker's live agent pane via the same messenger lifecycle uses for SCM nudges.
//
// When the GitHub review id is known, the worker is asked to reply on that
// review referencing its id with how it addressed the feedback and to resolve
// the review comment threads it addressed. The reviewer posts inline comments
// (per its prompt), so the per-finding threads are resolvable via `gh api`
// GraphQL resolveReviewThread; the top-level review object itself is not
// resolvable, hence the reply. The body is reviewer-authored text pasted into a
// PTY, so it is sanitized first (matching the lifecycle reaction path).
func (e *Engine) notifyWorkerChangesRequested(ctx context.Context, workerID domain.SessionID, body, githubReviewID string) error {
if e.messenger == nil {
return nil
}
msg := "An AO code reviewer requested changes on your PR. Review the feedback below and address it."
if githubReviewID != "" {
safeReviewID := domain.SanitizeControlChars(githubReviewID)
msg += fmt.Sprintf(" This feedback is GitHub review %s. Once you have addressed it, reply on that review referencing id %s with how you addressed it, then resolve the review comment threads you addressed.", safeReviewID, safeReviewID)
}
if body != "" {
msg += "\n\n" + domain.SanitizeControlChars(body)
}
return e.messenger.Send(ctx, workerID, msg)
}
// List returns a worker's review state: the live reviewer handle and its passes.
func (e *Engine) List(ctx context.Context, workerID domain.SessionID) (SessionReviews, error) {
func (e *Engine) List(ctx stdctx.Context, workerID domain.SessionID) (SessionReviews, error) {
if workerID == "" {
return SessionReviews{}, fmt.Errorf("%w: worker session id is required", ErrInvalid)
}
@ -383,7 +309,7 @@ func (e *Engine) List(ctx context.Context, workerID domain.SessionID) (SessionRe
return SessionReviews{ReviewerHandleID: handle, Runs: runs}, nil
}
func (e *Engine) workerPR(ctx context.Context, workerID domain.SessionID) (domain.PullRequest, error) {
func (e *Engine) workerPR(ctx stdctx.Context, workerID domain.SessionID) (domain.PullRequest, error) {
prs, err := e.prs.ListPRsBySession(ctx, workerID)
if err != nil {
return domain.PullRequest{}, err
@ -397,7 +323,7 @@ func (e *Engine) workerPR(ctx context.Context, workerID domain.SessionID) (domai
// reviewerHarness resolves which harness reviews the worker's PR: a configured
// reviewer wins, otherwise the worker's own harness is reused (falling back to
// claude-code), per domain.ResolveReviewerHarness.
func (e *Engine) reviewerHarness(ctx context.Context, worker domain.SessionRecord) (domain.ReviewerHarness, error) {
func (e *Engine) reviewerHarness(ctx stdctx.Context, worker domain.SessionRecord) (domain.ReviewerHarness, error) {
var cfg domain.ProjectConfig
if e.projects != nil {
if proj, ok, err := e.projects.GetProject(ctx, string(worker.ProjectID)); err != nil {
@ -409,7 +335,7 @@ func (e *Engine) reviewerHarness(ctx context.Context, worker domain.SessionRecor
return cfg.ResolveReviewerHarness(worker.Harness), nil
}
func (e *Engine) upsertReview(ctx context.Context, worker domain.SessionRecord, harness domain.ReviewerHarness, prURL, handleID string, now time.Time) (domain.Review, error) {
func (e *Engine) upsertReview(ctx stdctx.Context, worker domain.SessionRecord, harness domain.ReviewerHarness, prURL, handleID string, now time.Time) (domain.Review, error) {
existing, ok, err := e.store.GetReviewBySession(ctx, worker.ID)
if err != nil {
return domain.Review{}, err

247
backend/internal/review/review_test.go
View File

@ -61,6 +61,30 @@ func (f *fakeStore) UpdateReviewRunResult(_ context.Context, id string, status d
}
return false, nil
}
func (f *fakeStore) SupersedeReviewRun(_ context.Context, id, body string) (bool, error) {
for i := range f.runs {
if f.runs[i].ID == id {
if f.runs[i].Verdict != domain.VerdictNone || f.runs[i].Status == domain.ReviewRunFailed {
return false, nil
}
f.runs[i].Status = domain.ReviewRunFailed
f.runs[i].Body = body
return true, nil
}
}
return false, nil
}
func (f *fakeStore) SupersedeStaleRunningReviewRuns(_ context.Context, sessionID domain.SessionID, targetSHA, body string) (int64, error) {
var n int64
for i := range f.runs {
if f.runs[i].SessionID == sessionID && f.runs[i].TargetSHA != targetSHA && f.runs[i].Status == domain.ReviewRunRunning && f.runs[i].Verdict == domain.VerdictNone {
f.runs[i].Status = domain.ReviewRunFailed
f.runs[i].Body = body
n++
}
}
return n, nil
}
func (f *fakeStore) GetReviewRun(_ context.Context, id string) (domain.ReviewRun, bool, error) {
for _, r := range f.runs {
if r.ID == id {
@ -129,20 +153,8 @@ func (f *fakeLauncher) Notify(_ context.Context, handleID string, spec LaunchSpe
f.gotSpec = spec
return f.notifyErr
}
func (f *fakeLauncher) Alive(_ context.Context, _ string) (bool, error) { return f.alive, nil }
type fakeMessenger struct {
sends int
gotID domain.SessionID
gotMsg string
sendErr error
}
func (f *fakeMessenger) Send(_ context.Context, id domain.SessionID, msg string) error {
f.sends++
f.gotID = id
f.gotMsg = msg
return f.sendErr
func (f *fakeLauncher) Alive(_ context.Context, _ string) (bool, error) {
return f.alive || f.spawned, nil
}
func liveWorker() domain.SessionRecord {
@ -222,7 +234,6 @@ func TestTriggerConcurrentSameWorkerSpawnsOnce(t *testing.T) {
created++
}
}
// Exactly one trigger does the work; the rest reuse its run.
if created != 1 {
t.Errorf("Created=true count = %d, want exactly 1", created)
}
@ -259,9 +270,12 @@ func TestTriggerFallsBackToExistingRunOnUniqueConflict(t *testing.T) {
func TestTriggerIsIdempotentForSameCommit(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", TargetSHA: "sha1", Status: domain.ReviewRunRunning}},
runs: []domain.ReviewRun{{
ID: "run-1", SessionID: "mer-1", TargetSHA: "sha1",
Status: domain.ReviewRunComplete, Verdict: domain.VerdictApproved,
}},
}
launcher := &fakeLauncher{}
launcher := &fakeLauncher{alive: true}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, launcher)
res, err := eng.Trigger(context.Background(), "mer-1")
@ -279,6 +293,52 @@ func TestTriggerIsIdempotentForSameCommit(t *testing.T) {
}
}
func TestTriggerReusesRunningRowWithNoVerdict(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", TargetSHA: "sha1", Status: domain.ReviewRunRunning}},
}
launcher := &fakeLauncher{alive: false, handle: "review-mer-2"}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, launcher)
res, err := eng.Trigger(context.Background(), "mer-1")
if err != nil {
t.Fatalf("Trigger: %v", err)
}
if res.Created || res.Run.ID != "run-1" {
t.Fatalf("expected reuse of the running review for the same commit: %+v", res)
}
if launcher.spawned || launcher.notified {
t.Fatalf("running same-commit review should not relaunch: %+v", launcher)
}
if got := store.runs[0]; got.Status != domain.ReviewRunRunning {
t.Fatalf("running row should remain running, got %+v", got)
}
}
func TestTriggerSupersedesNonRunningRowWithNoVerdict(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", TargetSHA: "sha1", Status: domain.ReviewRunComplete}},
}
launcher := &fakeLauncher{alive: true, handle: "review-mer-1"}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, launcher)
res, err := eng.Trigger(context.Background(), "mer-1")
if err != nil {
t.Fatalf("Trigger: %v", err)
}
if !res.Created {
t.Fatalf("expected a fresh pass when prior non-running row has no verdict: %+v", res)
}
if !launcher.notified || launcher.spawned {
t.Fatalf("expected notify on live reviewer pane, not spawn: %+v", launcher)
}
if stale := store.runs[0]; stale.ID != "run-1" || stale.Status != domain.ReviewRunFailed {
t.Fatalf("expected stale run-1 marked failed, got %+v", stale)
}
}
func TestTriggerNotifiesLiveReviewerOnNewCommit(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
@ -302,6 +362,29 @@ func TestTriggerNotifiesLiveReviewerOnNewCommit(t *testing.T) {
}
}
func TestTriggerSupersedesOlderRunningRunOnNewCommit(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
runs: []domain.ReviewRun{{ID: "run-old", SessionID: "mer-1", TargetSHA: "sha0", Status: domain.ReviewRunRunning}},
}
launcher := &fakeLauncher{alive: true, handle: "review-mer-1"}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, launcher)
res, err := eng.Trigger(context.Background(), "mer-1")
if err != nil {
t.Fatalf("Trigger: %v", err)
}
if !res.Created || res.Run.TargetSHA != "sha1" {
t.Fatalf("expected new run for new commit, got %+v", res)
}
if old := store.runs[0]; old.ID != "run-old" || old.Status != domain.ReviewRunFailed {
t.Fatalf("expected older running run to be failed, got %+v", old)
}
if !launcher.notified || launcher.spawned {
t.Fatalf("expected live reviewer pane reused for new commit: %+v", launcher)
}
}
func TestTriggerSpawnsWhenReviewerDead(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},
@ -388,136 +471,6 @@ func TestTriggerRejectsBadWorkerState(t *testing.T) {
})
}
func TestSubmitRecordsVerdictAndBody(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", Status: domain.ReviewRunRunning}}}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, &fakeLauncher{})
run, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "please fix", "")
if err != nil {
t.Fatalf("Submit: %v", err)
}
if run.Status != domain.ReviewRunComplete || run.Verdict != domain.VerdictChangesRequested || run.Body != "please fix" {
t.Fatalf("run = %+v", run)
}
}
func newEngineWithMessenger(store Store, messenger ports.AgentMessenger) *Engine {
ids := 0
return New(Deps{
Store: store, Sessions: fakeSessions{rec: liveWorker(), ok: true}, PRs: prAt("sha1"),
Projects: fakeProjects{}, Launcher: &fakeLauncher{}, Messenger: messenger,
Clock: func() time.Time { return time.Unix(0, 0).UTC() },
NewID: func() string { ids++; return "id-" + string(rune('0'+ids)) },
})
}
func TestSubmitChangesRequestedMessagesWorker(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", Status: domain.ReviewRunRunning}}}
msgr := &fakeMessenger{}
eng := newEngineWithMessenger(store, msgr)
run, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix the bug", "98\x1b[2J765")
if err != nil {
t.Fatalf("Submit: %v", err)
}
if run.GithubReviewID != "98\x1b[2J765" || store.runs[0].GithubReviewID != "98\x1b[2J765" {
t.Fatalf("review id not persisted: run=%+v stored=%+v", run, store.runs[0])
}
if msgr.sends != 1 || msgr.gotID != "mer-1" {
t.Fatalf("expected one message to worker mer-1, got %+v", msgr)
}
if !strings.Contains(msgr.gotMsg, "fix the bug") || !strings.Contains(msgr.gotMsg, "98[2J765") {
t.Fatalf("message missing body or review id: %q", msgr.gotMsg)
}
if strings.Contains(msgr.gotMsg, "\x1b") {
t.Fatalf("message should sanitize review id before sending to worker: %q", msgr.gotMsg)
}
// The worker must be able to tell an AO internal review from external SCM
// reviewer feedback, and is asked to reply + resolve for an AO review.
if !strings.Contains(msgr.gotMsg, "AO code reviewer") {
t.Fatalf("message should identify the AO internal review: %q", msgr.gotMsg)
}
if !strings.Contains(msgr.gotMsg, "reply on that review") || !strings.Contains(msgr.gotMsg, "resolve") {
t.Fatalf("message should ask the worker to reply and resolve: %q", msgr.gotMsg)
}
}
func TestSubmitApprovedDoesNotMessageWorker(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", Status: domain.ReviewRunRunning}}}
msgr := &fakeMessenger{}
eng := newEngineWithMessenger(store, msgr)
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictApproved, "", "98765"); err != nil {
t.Fatalf("Submit: %v", err)
}
if msgr.sends != 0 {
t.Fatalf("approved review should not message the worker: %+v", msgr)
}
}
func TestSubmitChangesRequestedOmitsReviewIDWhenAbsent(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", Status: domain.ReviewRunRunning}}}
msgr := &fakeMessenger{}
eng := newEngineWithMessenger(store, msgr)
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", ""); err != nil {
t.Fatalf("Submit: %v", err)
}
if msgr.sends != 1 || !strings.Contains(msgr.gotMsg, "fix it") {
t.Fatalf("expected message with body: %+v", msgr)
}
// Still identified as an AO review, but with no id there is nothing to reply
// to or resolve, so that instruction is omitted.
if !strings.Contains(msgr.gotMsg, "AO code reviewer") {
t.Fatalf("message should identify the AO internal review: %q", msgr.gotMsg)
}
if strings.Contains(msgr.gotMsg, "reply on that review") {
t.Fatalf("message should not ask to reply/resolve when no review id was supplied: %q", msgr.gotMsg)
}
}
func TestSubmitPropagatesMessengerError(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "mer-1", Status: domain.ReviewRunRunning}}}
msgr := &fakeMessenger{sendErr: fmt.Errorf("dead pane")}
eng := newEngineWithMessenger(store, msgr)
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", "1"); err == nil {
t.Fatal("expected Submit to surface the messenger error")
}
// The run must stay running so a retried submit can try again, rather than
// being marked complete and then failing the status='running' guard.
if store.runs[0].Status != domain.ReviewRunRunning {
t.Fatalf("run should stay running after a failed send, got %q", store.runs[0].Status)
}
// A retry once the pane is back completes the run and messages the worker.
msgr.sendErr = nil
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", "1"); err != nil {
t.Fatalf("retry after recovered send: %v", err)
}
if store.runs[0].Status != domain.ReviewRunComplete || msgr.sends != 2 {
t.Fatalf("retry should complete the run and re-send: status=%q sends=%d", store.runs[0].Status, msgr.sends)
}
}
func TestSubmitValidationAndOwnership(t *testing.T) {
store := &fakeStore{runs: []domain.ReviewRun{{ID: "run-1", SessionID: "other", Status: domain.ReviewRunRunning}}}
eng := newEngineForTest(store, fakeSessions{rec: liveWorker(), ok: true}, prAt("sha1"), fakeProjects{}, &fakeLauncher{})
if _, err := eng.Submit(context.Background(), "mer-1", "", domain.VerdictApproved, "", ""); !errors.Is(err, ErrInvalid) {
t.Fatalf("missing run id err = %v", err)
}
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", "garbage", "b", ""); !errors.Is(err, ErrInvalid) {
t.Fatalf("bad verdict err = %v", err)
}
if _, err := eng.Submit(context.Background(), "mer-1", "missing", domain.VerdictApproved, "", ""); !errors.Is(err, ErrNotFound) {
t.Fatalf("unknown run err = %v", err)
}
if _, err := eng.Submit(context.Background(), "mer-1", "run-1", domain.VerdictApproved, "", ""); !errors.Is(err, ErrInvalid) {
t.Fatalf("ownership err = %v", err)
}
}
func TestListReturnsHandleAndRuns(t *testing.T) {
store := &fakeStore{
review: &domain.Review{ID: "rev-1", SessionID: "mer-1", ReviewerHandleID: "review-mer-1"},

124
backend/internal/service/review/review.go
View File

@ -6,8 +6,11 @@ package review
import (
"context"
"fmt"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/lifecycle"
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
reviewcore "github.com/aoagents/agent-orchestrator/backend/internal/review"
)
@ -30,13 +33,46 @@ type Manager interface {
// Service is the API-facing review service. It delegates to the core engine.
type Service struct {
engine *reviewcore.Engine
store Store
lifecycle Reducer
clock func() time.Time
}
var _ Manager = (*Service)(nil)
// Store is the review_run persistence surface owned by the service submit path.
type Store interface {
GetReviewRun(ctx context.Context, id string) (domain.ReviewRun, bool, error)
UpdateReviewRunResult(ctx context.Context, id string, status domain.ReviewRunStatus, verdict domain.ReviewVerdict, body, githubReviewID string) (bool, error)
MarkReviewRunDelivered(ctx context.Context, id string, deliveredAt time.Time) (bool, error)
}
// Reducer is the lifecycle reaction boundary used after a review result has
// been persisted.
type Reducer interface {
ApplyReviewResult(ctx context.Context, workerID domain.SessionID, result lifecycle.ReviewResult) (lifecycle.ReviewDeliveryOutcome, error)
}
// Option customizes the review service.
type Option func(*Service)
// WithLifecycleReducer wires post-submit review delivery through lifecycle.
func WithLifecycleReducer(r Reducer) Option {
return func(s *Service) { s.lifecycle = r }
}
// WithClock overrides the service clock for tests.
func WithClock(clock func() time.Time) Option {
return func(s *Service) { s.clock = clock }
}
// New wraps a core review engine as the API-facing service.
func New(engine *reviewcore.Engine) *Service {
return &Service{engine: engine}
func New(engine *reviewcore.Engine, store Store, opts ...Option) *Service {
s := &Service{engine: engine, store: store, clock: func() time.Time { return time.Now().UTC() }}
for _, opt := range opts {
opt(s)
}
return s
}
// Trigger starts (or reuses) a review pass for a worker's PR.
@ -46,7 +82,89 @@ func (s *Service) Trigger(ctx context.Context, workerID domain.SessionID) (revie
// Submit records a reviewer's result for a specific worker review pass.
func (s *Service) Submit(ctx context.Context, workerID domain.SessionID, runID string, verdict domain.ReviewVerdict, body, githubReviewID string) (domain.ReviewRun, error) {
return s.engine.Submit(ctx, workerID, runID, verdict, body, githubReviewID)
if workerID == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: worker session id is required", ErrInvalid)
}
if runID == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: review run id is required", ErrInvalid)
}
if !verdict.Valid() {
return domain.ReviewRun{}, fmt.Errorf("%w: verdict must be %q or %q", ErrInvalid, domain.VerdictApproved, domain.VerdictChangesRequested)
}
if verdict == domain.VerdictChangesRequested && body == "" {
return domain.ReviewRun{}, fmt.Errorf("%w: a changes_requested review requires a body", ErrInvalid)
}
if s.store == nil {
return domain.ReviewRun{}, fmt.Errorf("review service store is not configured")
}
run, ok, err := s.store.GetReviewRun(ctx, runID)
if err != nil {
return domain.ReviewRun{}, err
}
if !ok {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q", ErrNotFound, runID)
}
if run.SessionID != workerID {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q does not belong to worker %q", ErrInvalid, runID, workerID)
}
switch run.Status {
case domain.ReviewRunRunning:
updated, err := s.store.UpdateReviewRunResult(ctx, run.ID, domain.ReviewRunComplete, verdict, body, githubReviewID)
if err != nil {
return domain.ReviewRun{}, err
}
if !updated {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q is not running", ErrInvalid, runID)
}
run.Status = domain.ReviewRunComplete
run.Verdict = verdict
run.Body = body
run.GithubReviewID = githubReviewID
case domain.ReviewRunComplete:
if run.Verdict != verdict {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q already recorded verdict %q", ErrInvalid, runID, run.Verdict)
}
if body != "" && body != run.Body {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q already recorded a different body", ErrInvalid, runID)
}
if githubReviewID != "" && githubReviewID != run.GithubReviewID {
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q already recorded GitHub review id %q", ErrInvalid, runID, run.GithubReviewID)
}
case domain.ReviewRunDelivered:
return run, nil
default:
return domain.ReviewRun{}, fmt.Errorf("%w: review run %q is not running", ErrInvalid, runID)
}
if s.lifecycle == nil {
return run, nil
}
outcome, err := s.lifecycle.ApplyReviewResult(ctx, workerID, lifecycle.ReviewResult{
RunID: run.ID,
WorkerID: workerID,
PRURL: run.PRURL,
TargetSHA: run.TargetSHA,
Verdict: run.Verdict,
Body: run.Body,
GithubReviewID: run.GithubReviewID,
DeliveredAt: run.DeliveredAt,
})
if err != nil {
return domain.ReviewRun{}, err
}
if outcome == lifecycle.ReviewDeliverySent {
deliveredAt := s.clock()
updated, err := s.store.MarkReviewRunDelivered(ctx, run.ID, deliveredAt)
if err != nil {
return domain.ReviewRun{}, err
}
if updated {
run.Status = domain.ReviewRunDelivered
run.DeliveredAt = &deliveredAt
}
}
return run, nil
}
// List returns a worker's review state.

131
backend/internal/service/review/review_test.go Normal file
View File

@ -0,0 +1,131 @@
package review
import (
"context"
"errors"
"testing"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/lifecycle"
)
type fakeStore struct {
run domain.ReviewRun
ok bool
updateCalls int
markCalls int
}
func (f *fakeStore) GetReviewRun(context.Context, string) (domain.ReviewRun, bool, error) {
return f.run, f.ok, nil
}
func (f *fakeStore) UpdateReviewRunResult(_ context.Context, _ string, status domain.ReviewRunStatus, verdict domain.ReviewVerdict, body, githubReviewID string) (bool, error) {
if f.run.Status != domain.ReviewRunRunning {
return false, nil
}
f.updateCalls++
f.run.Status = status
f.run.Verdict = verdict
f.run.Body = body
f.run.GithubReviewID = githubReviewID
return true, nil
}
func (f *fakeStore) MarkReviewRunDelivered(_ context.Context, _ string, deliveredAt time.Time) (bool, error) {
f.markCalls++
if f.run.Status != domain.ReviewRunComplete || f.run.DeliveredAt != nil {
return false, nil
}
f.run.Status = domain.ReviewRunDelivered
f.run.DeliveredAt = &deliveredAt
return true, nil
}
type fakeReducer struct {
outcome lifecycle.ReviewDeliveryOutcome
err error
calls int
got lifecycle.ReviewResult
}
func (f *fakeReducer) ApplyReviewResult(_ context.Context, _ domain.SessionID, result lifecycle.ReviewResult) (lifecycle.ReviewDeliveryOutcome, error) {
f.calls++
f.got = result
return f.outcome, f.err
}
func TestSubmitPersistsThenAppliesThenStampsDelivered(t *testing.T) {
now := time.Unix(100, 0).UTC()
st := &fakeStore{ok: true, run: domain.ReviewRun{ID: "run-1", SessionID: "mer-1", PRURL: "pr1", TargetSHA: "sha1", Status: domain.ReviewRunRunning}}
reducer := &fakeReducer{outcome: lifecycle.ReviewDeliverySent}
svc := New(nil, st, WithLifecycleReducer(reducer), WithClock(func() time.Time { return now }))
run, err := svc.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", "987")
if err != nil {
t.Fatalf("Submit: %v", err)
}
if st.updateCalls != 1 || reducer.calls != 1 || st.markCalls != 1 {
t.Fatalf("calls update/reducer/mark = %d/%d/%d", st.updateCalls, reducer.calls, st.markCalls)
}
if reducer.got.Verdict != domain.VerdictChangesRequested || reducer.got.Body != "fix it" || reducer.got.GithubReviewID != "987" {
t.Fatalf("reducer saw wrong result: %+v", reducer.got)
}
if run.Status != domain.ReviewRunDelivered || run.DeliveredAt == nil || !run.DeliveredAt.Equal(now) {
t.Fatalf("run not stamped delivered: %+v", run)
}
}
func TestSubmitDeliveryFailureLeavesCompletedUndeliveredForRetry(t *testing.T) {
sendErr := errors.New("dead pane")
st := &fakeStore{ok: true, run: domain.ReviewRun{ID: "run-1", SessionID: "mer-1", PRURL: "pr1", TargetSHA: "sha1", Status: domain.ReviewRunRunning}}
reducer := &fakeReducer{err: sendErr}
svc := New(nil, st, WithLifecycleReducer(reducer))
if _, err := svc.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", "987"); !errors.Is(err, sendErr) {
t.Fatalf("err = %v, want sendErr", err)
}
if st.run.Status != domain.ReviewRunComplete || st.run.DeliveredAt != nil || st.markCalls != 0 {
t.Fatalf("failed delivery should leave completed/undelivered without stamp: %+v markCalls=%d", st.run, st.markCalls)
}
reducer.err = nil
reducer.outcome = lifecycle.ReviewDeliverySent
if _, err := svc.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, "fix it", "987"); err != nil {
t.Fatalf("retry Submit: %v", err)
}
if st.updateCalls != 1 || reducer.calls != 2 || st.run.Status != domain.ReviewRunDelivered || st.run.DeliveredAt == nil {
t.Fatalf("retry should not rewrite result and should stamp delivery: update=%d reducer=%d run=%+v", st.updateCalls, reducer.calls, st.run)
}
}
func TestSubmitCompletedRetryRejectsDifferentRecordedFields(t *testing.T) {
tests := []struct {
name string
body string
githubReviewID string
}{
{name: "different body", body: "different", githubReviewID: "987"},
{name: "different review id", body: "fix it", githubReviewID: "654"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
st := &fakeStore{ok: true, run: domain.ReviewRun{
ID: "run-1", SessionID: "mer-1", PRURL: "pr1", TargetSHA: "sha1",
Status: domain.ReviewRunComplete, Verdict: domain.VerdictChangesRequested,
Body: "fix it", GithubReviewID: "987",
}}
reducer := &fakeReducer{outcome: lifecycle.ReviewDeliverySent}
svc := New(nil, st, WithLifecycleReducer(reducer))
if _, err := svc.Submit(context.Background(), "mer-1", "run-1", domain.VerdictChangesRequested, tt.body, tt.githubReviewID); !errors.Is(err, ErrInvalid) {
t.Fatalf("err = %v, want ErrInvalid", err)
}
if st.updateCalls != 0 || st.markCalls != 0 || reducer.calls != 0 {
t.Fatalf("mismatched retry should not rewrite or deliver: update=%d mark=%d reducer=%d", st.updateCalls, st.markCalls, reducer.calls)
}
})
}
}

1
backend/internal/storage/sqlite/gen/models.go
View File

@ -146,6 +146,7 @@ type ReviewRun struct {
Body string
CreatedAt time.Time
GithubReviewID string
DeliveredAt sql.NullTime
}
type Session struct {

62
backend/internal/storage/sqlite/gen/review.sql.go
View File

@ -7,6 +7,7 @@ package gen
import (
"context"
"database/sql"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
@ -34,7 +35,7 @@ func (q *Queries) GetReviewBySession(ctx context.Context, sessionID domain.Sessi
}
const getReviewRun = `-- name: GetReviewRun :one
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE id = ?
`
@ -53,12 +54,13 @@ func (q *Queries) GetReviewRun(ctx context.Context, id string) (ReviewRun, error
&i.Body,
&i.CreatedAt,
&i.GithubReviewID,
&i.DeliveredAt,
)
return i, err
}
const getReviewRunBySessionAndSHA = `-- name: GetReviewRunBySessionAndSHA :one
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE session_id = ? AND target_sha = ? ORDER BY created_at DESC LIMIT 1
`
@ -82,6 +84,7 @@ func (q *Queries) GetReviewRunBySessionAndSHA(ctx context.Context, arg GetReview
&i.Body,
&i.CreatedAt,
&i.GithubReviewID,
&i.DeliveredAt,
)
return i, err
}
@ -123,7 +126,7 @@ func (q *Queries) InsertReviewRun(ctx context.Context, arg InsertReviewRunParams
}
const listReviewRunsBySession = `-- name: ListReviewRunsBySession :many
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE session_id = ? ORDER BY created_at DESC
`
@ -148,6 +151,7 @@ func (q *Queries) ListReviewRunsBySession(ctx context.Context, sessionID domain.
&i.Body,
&i.CreatedAt,
&i.GithubReviewID,
&i.DeliveredAt,
); err != nil {
return nil, err
}
@ -162,6 +166,58 @@ func (q *Queries) ListReviewRunsBySession(ctx context.Context, sessionID domain.
return items, nil
}
const markReviewRunDelivered = `-- name: MarkReviewRunDelivered :execrows
UPDATE review_run SET status = 'delivered', delivered_at = ? WHERE id = ? AND status = 'complete' AND delivered_at IS NULL
`
type MarkReviewRunDeliveredParams struct {
DeliveredAt sql.NullTime
ID string
}
func (q *Queries) MarkReviewRunDelivered(ctx context.Context, arg MarkReviewRunDeliveredParams) (int64, error) {
result, err := q.db.ExecContext(ctx, markReviewRunDelivered, arg.DeliveredAt, arg.ID)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
const supersedeReviewRun = `-- name: SupersedeReviewRun :execrows
UPDATE review_run SET status = 'failed', body = ? WHERE id = ? AND verdict = '' AND status != 'failed'
`
type SupersedeReviewRunParams struct {
Body string
ID string
}
func (q *Queries) SupersedeReviewRun(ctx context.Context, arg SupersedeReviewRunParams) (int64, error) {
result, err := q.db.ExecContext(ctx, supersedeReviewRun, arg.Body, arg.ID)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
const supersedeStaleRunningReviewRuns = `-- name: SupersedeStaleRunningReviewRuns :execrows
UPDATE review_run SET status = 'failed', body = ? WHERE session_id = ? AND target_sha != ? AND status = 'running' AND verdict = ''
`
type SupersedeStaleRunningReviewRunsParams struct {
Body string
SessionID domain.SessionID
TargetSha string
}
func (q *Queries) SupersedeStaleRunningReviewRuns(ctx context.Context, arg SupersedeStaleRunningReviewRunsParams) (int64, error) {
result, err := q.db.ExecContext(ctx, supersedeStaleRunningReviewRuns, arg.Body, arg.SessionID, arg.TargetSha)
if err != nil {
return 0, err
}
return result.RowsAffected()
}
const updateReviewRunResult = `-- name: UpdateReviewRunResult :execrows
UPDATE review_run SET status = ?, verdict = ?, body = ?, github_review_id = ? WHERE id = ? AND status = 'running'
`

14
backend/internal/storage/sqlite/migrations/0018_review_run_delivered_at.sql Normal file
View File

@ -0,0 +1,14 @@
-- AO-internal review changes-requested nudges are delivered through lifecycle
-- sendOnce after the review result is recorded. This nullable timestamp marks
-- passes whose worker nudge was durably delivered, so retries after a daemon
-- restart do not send the same review pass twice.
-- +goose Up
-- +goose StatementBegin
ALTER TABLE review_run ADD COLUMN delivered_at TIMESTAMP;
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
ALTER TABLE review_run DROP COLUMN delivered_at;
-- +goose StatementEnd

15
backend/internal/storage/sqlite/queries/review.sql
View File

@ -18,14 +18,23 @@ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
-- name: UpdateReviewRunResult :execrows
UPDATE review_run SET status = ?, verdict = ?, body = ?, github_review_id = ? WHERE id = ? AND status = 'running';
-- name: SupersedeReviewRun :execrows
UPDATE review_run SET status = 'failed', body = ? WHERE id = ? AND verdict = '' AND status != 'failed';
-- name: SupersedeStaleRunningReviewRuns :execrows
UPDATE review_run SET status = 'failed', body = ? WHERE session_id = ? AND target_sha != ? AND status = 'running' AND verdict = '';
-- name: MarkReviewRunDelivered :execrows
UPDATE review_run SET status = 'delivered', delivered_at = ? WHERE id = ? AND status = 'complete' AND delivered_at IS NULL;
-- name: GetReviewRun :one
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE id = ?;
-- name: GetReviewRunBySessionAndSHA :one
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE session_id = ? AND target_sha = ? ORDER BY created_at DESC LIMIT 1;
-- name: ListReviewRunsBySession :many
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id
SELECT id, review_id, session_id, harness, pr_url, target_sha, status, verdict, body, created_at, github_review_id, delivered_at
FROM review_run WHERE session_id = ? ORDER BY created_at DESC;

49
backend/internal/storage/sqlite/store/review_store.go
View File

@ -5,6 +5,7 @@ import (
"database/sql"
"errors"
"fmt"
"time"
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
"github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite/gen"
@ -82,6 +83,48 @@ func (s *Store) UpdateReviewRunResult(ctx context.Context, id string, status dom
return n > 0, nil
}
// SupersedeReviewRun marks an unverdicted non-failed pass failed so a new pass
// for the same commit can be recorded.
func (s *Store) SupersedeReviewRun(ctx context.Context, id, body string) (bool, error) {
s.writeMu.Lock()
defer s.writeMu.Unlock()
n, err := s.qw.SupersedeReviewRun(ctx, gen.SupersedeReviewRunParams{
Body: body,
ID: id,
})
if err != nil {
return false, err
}
return n > 0, nil
}
// SupersedeStaleRunningReviewRuns marks older running unverdicted passes for a
// worker failed before starting a review for a newer commit.
func (s *Store) SupersedeStaleRunningReviewRuns(ctx context.Context, sessionID domain.SessionID, targetSHA, body string) (int64, error) {
s.writeMu.Lock()
defer s.writeMu.Unlock()
return s.qw.SupersedeStaleRunningReviewRuns(ctx, gen.SupersedeStaleRunningReviewRunsParams{
Body: body,
SessionID: sessionID,
TargetSha: targetSHA,
})
}
// MarkReviewRunDelivered records that lifecycle delivered the worker nudge for
// a completed AO-internal review pass.
func (s *Store) MarkReviewRunDelivered(ctx context.Context, id string, deliveredAt time.Time) (bool, error) {
s.writeMu.Lock()
defer s.writeMu.Unlock()
n, err := s.qw.MarkReviewRunDelivered(ctx, gen.MarkReviewRunDeliveredParams{
DeliveredAt: sql.NullTime{Time: deliveredAt, Valid: true},
ID: id,
})
if err != nil {
return false, err
}
return n > 0, nil
}
// GetReviewRun returns one review pass by id.
func (s *Store) GetReviewRun(ctx context.Context, id string) (domain.ReviewRun, bool, error) {
row, err := s.qr.GetReviewRun(ctx, id)
@ -135,6 +178,11 @@ func reviewFromRow(r gen.Review) domain.Review {
}
func reviewRunFromRow(r gen.ReviewRun) domain.ReviewRun {
var deliveredAt *time.Time
if r.DeliveredAt.Valid {
t := r.DeliveredAt.Time
deliveredAt = &t
}
return domain.ReviewRun{
ID: r.ID,
ReviewID: r.ReviewID,
@ -147,5 +195,6 @@ func reviewRunFromRow(r gen.ReviewRun) domain.ReviewRun {
Body: r.Body,
GithubReviewID: r.GithubReviewID,
CreatedAt: r.CreatedAt,
DeliveredAt: deliveredAt,
}
}

2
frontend/src/api/schema.ts
View File

@ -680,6 +680,8 @@ export interface components {
body: string;
/** Format: date-time */
createdAt: string;
/** Format: date-time */
deliveredAt?: null | string;
githubReviewId: string;
harness: string;
id: string;