feat(backend): atomic PR-observation write + CDC on check status updates

Addresses review on PR-observation persistence:

- pr_checks now has an AFTER UPDATE CDC trigger (guarded on status change), so a
  check flipping in_progress->failed on the same commit emits change_log instead
  of updating silently. Restores symmetry with the sessions/pr triggers.
- writePR persists scalar facts + checks + comments in ONE transaction via
  Store.WritePRObservation, so a mid-write failure can't leave the pr row (and
  its CDC event) committed while checks/comments are partial. Collapses the
  PRWriter port's three write methods into one WritePR.
- db.go: record why modernc.org/sqlite (pure-Go, CGO-free static binary) at the
  import site.

Regression tests for both the update-trigger (emit on change, suppress no-op
re-poll) and the transactional write. go test -race ./... green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/internal/lifecycle/manager.go

@@ -176,27 +176,31 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o
 	return m.runReactions(ctx, id, prContent(o))
 }
 
-// writePR upserts the scalar facts, records each check run, and replaces the
-// comment set. PR-table CDC is emitted by the DB triggers.
+// writePR persists the observation's scalar facts, check runs, and comment set
+// in one atomic store call. PR-table CDC is emitted by the DB triggers.
 func (m *Manager) writePR(ctx context.Context, id domain.SessionID, o ports.PRObservation) error {
 	now := m.clock()
-	if err := m.pr.UpsertPR(ctx, ports.PRRow{
+	row := ports.PRRow{
 		URL: o.URL, SessionID: string(id), Number: o.Number,
 		Draft: o.Draft, Merged: o.Merged, Closed: o.Closed,
 		CI: o.CI, Review: o.Review, Mergeability: o.Mergeability, UpdatedAt: now,
-	}); err != nil {
-		return err
 	}
-	for _, c := range o.Checks {
+	checks := make([]ports.PRCheckRow, len(o.Checks))
+	for i, c := range o.Checks {
 		c.PRURL = o.URL
 		if c.CreatedAt.IsZero() {
 			c.CreatedAt = now
 		}
-		if err := m.pr.RecordCheck(ctx, c); err != nil {
-			return err
-		}
+		checks[i] = c
 	}
-	return m.pr.ReplacePRComments(ctx, o.URL, o.Comments)
+	comments := make([]ports.PRComment, len(o.Comments))
+	for i, c := range o.Comments {
+		if c.CreatedAt.IsZero() {
+			c.CreatedAt = now
+		}
+		comments[i] = c
+	}
+	return m.pr.WritePR(ctx, row, checks, comments)
 }
 
 // ---- mutation commands from the Session Manager ----

backend/internal/lifecycle/manager_test.go

@@ -82,12 +82,10 @@ func (f *fakeStore) PRFactsForSession(_ context.Context, id domain.SessionID) (d
 	}
 	return facts, nil
 }
-func (f *fakeStore) UpsertPR(_ context.Context, r ports.PRRow) error {
-	f.pr[domain.SessionID(r.SessionID)] = r
-	return nil
-}
-func (f *fakeStore) RecordCheck(_ context.Context, r ports.PRCheckRow) error {
-	f.checks = append(f.checks, r)
+func (f *fakeStore) WritePR(_ context.Context, pr ports.PRRow, checks []ports.PRCheckRow, comments []ports.PRComment) error {
+	f.pr[domain.SessionID(pr.SessionID)] = pr
+	f.checks = append(f.checks, checks...)
+	f.comments[pr.URL] = comments
 	return nil
 }
 func (f *fakeStore) RecentCheckStatuses(_ context.Context, url, name string, limit int) ([]string, error) {
@@ -99,10 +97,6 @@ func (f *fakeStore) RecentCheckStatuses(_ context.Context, url, name string, lim
 	}
 	return out, nil
 }
-func (f *fakeStore) ReplacePRComments(_ context.Context, url string, cs []ports.PRComment) error {
-	f.comments[url] = cs
-	return nil
-}
 
 type fakeNotifier struct{ events []ports.Event }
 

backend/internal/ports/outbound.go

@@ -24,10 +24,12 @@ type SessionStore interface {
 // PRWriter records the PR facts a PR observation carries. The pr table's own DB
 // triggers emit the CDC; this just writes the rows.
 type PRWriter interface {
-	UpsertPR(ctx context.Context, r PRRow) error
-	RecordCheck(ctx context.Context, r PRCheckRow) error
+	// WritePR persists a full PR observation — scalar facts, check runs, and the
+	// replacement comment set — in one transaction, so the rows and the CDC
+	// events they emit are all-or-nothing.
+	WritePR(ctx context.Context, pr PRRow, checks []PRCheckRow, comments []PRComment) error
+	// RecentCheckStatuses reads the last `limit` runs of a check (the CI brake).
 	RecentCheckStatuses(ctx context.Context, prURL, name string, limit int) ([]string, error)
-	ReplacePRComments(ctx context.Context, prURL string, comments []PRComment) error
 }
 
 // Notifier delivers an event to the human (desktop/Slack later). Push, never poll.

backend/internal/storage/sqlite/db.go

@@ -11,6 +11,9 @@ import (
 	"path/filepath"
 
 	"github.com/pressly/goose/v3"
+	// modernc.org/sqlite is the pure-Go (CGO-free) SQLite driver — chosen so the
+	// daemon cross-compiles and ships as a static binary with no libsqlite/CGO
+	// toolchain dependency, at the cost of some raw throughput vs a C-backed driver.
 	_ "modernc.org/sqlite"
 )
 

backend/internal/storage/sqlite/migrations/0001_init.sql

@@ -202,6 +202,25 @@ BEGIN
 END;
 -- +goose StatementEnd
 
+-- A re-polled check can change status on the same commit (in_progress -> failed)
+-- via UpsertPRCheck's ON CONFLICT DO UPDATE. Without this trigger that status
+-- transition would update the row silently, so CDC consumers would never see it.
+-- Guarded on the status so a no-op re-poll emits nothing.
+-- +goose StatementBegin
+CREATE TRIGGER pr_checks_cdc_update
+AFTER UPDATE ON pr_checks
+WHEN OLD.status <> NEW.status
+BEGIN
+    INSERT INTO change_log (project_id, session_id, event_type, payload, created_at)
+    VALUES (
+        (SELECT s.project_id FROM pr p JOIN sessions s ON s.id = p.session_id WHERE p.url = NEW.pr_url),
+        (SELECT session_id FROM pr WHERE url = NEW.pr_url),
+        'pr_check_recorded',
+        json_object('pr', NEW.pr_url, 'name', NEW.name, 'commit', NEW.commit_hash, 'status', NEW.status),
+        NEW.created_at);
+END;
+-- +goose StatementEnd
+
 -- +goose Down
 -- +goose StatementBegin
 DROP TABLE change_log;

backend/internal/storage/sqlite/pr_cdc_test.go

@@ -0,0 +1,86 @@
+package sqlite
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// A check can change status on the same commit (in_progress -> failed) via
+// UpsertPRCheck's ON CONFLICT DO UPDATE. CDC must emit on that transition, not
+// only on the first insert — otherwise live clients never see the status change.
+func TestPRChecksCDC_EmitsOnInsertAndStatusUpdate(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	seedProject(t, s, "mer")
+	rec, err := s.CreateSession(ctx, sampleRecord("mer"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	url := "https://example/pr/1"
+	if err := s.UpsertPR(ctx, PRRow{URL: url, SessionID: string(rec.ID), Number: 1}); err != nil {
+		t.Fatal(err)
+	}
+
+	now := time.Now()
+	mustCheck := func(status string) {
+		if err := s.RecordCheck(ctx, PRCheckRow{PRURL: url, Name: "build", CommitHash: "c1", Status: status, CreatedAt: now}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mustCheck("in_progress") // insert  -> event
+	mustCheck("failed")      // status change on same commit (update) -> event
+	mustCheck("failed")      // no-op re-poll (status unchanged) -> NO event
+
+	rows, err := s.ReadChangeLogAfter(ctx, 0, 100)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var checkEvents []ChangeLogRow
+	for _, r := range rows {
+		if r.EventType == "pr_check_recorded" {
+			checkEvents = append(checkEvents, r)
+		}
+	}
+	if len(checkEvents) != 2 {
+		t.Fatalf("want 2 check CDC events (insert + status change, no-op suppressed), got %d", len(checkEvents))
+	}
+	if !strings.Contains(checkEvents[1].Payload, `"status":"failed"`) {
+		t.Fatalf("the update event should carry the new status, got %q", checkEvents[1].Payload)
+	}
+}
+
+// WritePRObservation persists scalar facts, checks, and comments in one tx; all
+// three should be queryable afterward.
+func TestWritePRObservation_PersistsScalarsChecksAndComments(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	seedProject(t, s, "mer")
+	rec, err := s.CreateSession(ctx, sampleRecord("mer"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	url := "https://example/pr/7"
+	now := time.Now()
+
+	err = s.WritePRObservation(ctx,
+		PRRow{URL: url, SessionID: string(rec.ID), Number: 7, CIState: "failing", UpdatedAt: now},
+		[]PRCheckRow{{PRURL: url, Name: "build", CommitHash: "c1", Status: "failed", CreatedAt: now}},
+		[]PRCommentRow{{PRURL: url, CommentID: "1", Author: "reviewer", Body: "use a const", CreatedAt: now}},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pr, ok, err := s.GetPR(ctx, url)
+	if err != nil || !ok || pr.CIState != "failing" {
+		t.Fatalf("scalar facts not persisted: ok=%v ci=%q err=%v", ok, pr.CIState, err)
+	}
+	if checks, _ := s.ListChecks(ctx, url); len(checks) != 1 || checks[0].Status != "failed" {
+		t.Fatalf("check not persisted: %+v", checks)
+	}
+	if comments, _ := s.ListPRComments(ctx, url); len(comments) != 1 || comments[0].Body != "use a const" {
+		t.Fatalf("comment not persisted: %+v", comments)
+	}
+}

backend/internal/storage/sqlite/pr_store.go

@@ -27,18 +27,7 @@ type PRRow struct {
 // fields default to their "nothing known yet" value so a partial row is valid
 // against the CHECK constraints (matches the domain zero values none/unknown).
 func (s *Store) UpsertPR(ctx context.Context, r PRRow) error {
-	if r.State == "" {
-		r.State = "open"
-	}
-	if r.ReviewDecision == "" {
-		r.ReviewDecision = "none"
-	}
-	if r.CIState == "" {
-		r.CIState = "unknown"
-	}
-	if r.Mergeability == "" {
-		r.Mergeability = "unknown"
-	}
+	r = r.withDefaults()
 	s.writeMu.Lock()
 	defer s.writeMu.Unlock()
 	return s.qw.UpsertPR(ctx, gen.UpsertPRParams{
@@ -53,6 +42,67 @@ func (s *Store) UpsertPR(ctx context.Context, r PRRow) error {
 	})
 }
 
+// WritePRObservation persists a full PR observation — scalar facts, check runs,
+// and the replacement comment set — in one write transaction, so the rows and
+// the change_log events their triggers emit are committed all-or-nothing. The
+// scalar PR upsert runs first so the checks'/comments' CDC triggers can resolve
+// the session id from the pr row within the same transaction.
+func (s *Store) WritePRObservation(ctx context.Context, pr PRRow, checks []PRCheckRow, comments []PRCommentRow) error {
+	pr = pr.withDefaults()
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+	return s.inTx(ctx, "write pr observation", func(q *gen.Queries) error {
+		if err := q.UpsertPR(ctx, gen.UpsertPRParams{
+			Url: pr.URL, SessionID: pr.SessionID, Number: pr.Number,
+			PrState: pr.State, ReviewDecision: pr.ReviewDecision,
+			CiState: pr.CIState, Mergeability: pr.Mergeability, UpdatedAt: pr.UpdatedAt,
+		}); err != nil {
+			return err
+		}
+		for _, c := range checks {
+			if c.Status == "" {
+				c.Status = "unknown"
+			}
+			if err := q.UpsertPRCheck(ctx, gen.UpsertPRCheckParams{
+				PrUrl: c.PRURL, Name: c.Name, CommitHash: c.CommitHash,
+				Status: c.Status, Url: c.URL, LogTail: c.LogTail, CreatedAt: c.CreatedAt,
+			}); err != nil {
+				return err
+			}
+		}
+		if err := q.DeletePRComments(ctx, pr.URL); err != nil {
+			return err
+		}
+		for _, cm := range comments {
+			if err := q.UpsertPRComment(ctx, gen.UpsertPRCommentParams{
+				PrUrl: pr.URL, CommentID: cm.CommentID, Author: cm.Author, File: cm.File,
+				Line: cm.Line, Body: cm.Body, Resolved: boolToInt(cm.Resolved), CreatedAt: cm.CreatedAt,
+			}); err != nil {
+				return fmt.Errorf("comment %q: %w", cm.CommentID, err)
+			}
+		}
+		return nil
+	})
+}
+
+// withDefaults fills empty enum fields with their "nothing known yet" value so a
+// partial row satisfies the CHECK constraints (matches UpsertPR).
+func (r PRRow) withDefaults() PRRow {
+	if r.State == "" {
+		r.State = "open"
+	}
+	if r.ReviewDecision == "" {
+		r.ReviewDecision = "none"
+	}
+	if r.CIState == "" {
+		r.CIState = "unknown"
+	}
+	if r.Mergeability == "" {
+		r.Mergeability = "unknown"
+	}
+	return r
+}
+
 // GetPR returns the PR facts for a URL, or ok=false if absent.
 func (s *Store) GetPR(ctx context.Context, url string) (PRRow, bool, error) {
 	p, err := s.qr.GetPR(ctx, url)

backend/lifecycle_wiring.go

@@ -87,33 +87,30 @@ func (a storeAdapter) PRFactsForSession(ctx context.Context, id domain.SessionID
 	return facts, nil
 }
 
-func (a storeAdapter) UpsertPR(ctx context.Context, r ports.PRRow) error {
-	return a.Store.UpsertPR(ctx, sqlite.PRRow{
-		URL: r.URL, SessionID: r.SessionID, Number: int64(r.Number),
-		State:          prState(r),
-		ReviewDecision: string(r.Review),
-		CIState:        string(r.CI),
-		Mergeability:   string(r.Mergeability),
-		UpdatedAt:      r.UpdatedAt,
-	})
-}
-
-func (a storeAdapter) RecordCheck(ctx context.Context, r ports.PRCheckRow) error {
-	return a.Store.RecordCheck(ctx, sqlite.PRCheckRow{
-		PRURL: r.PRURL, Name: r.Name, CommitHash: r.CommitHash,
-		Status: r.Status, URL: r.URL, LogTail: r.LogTail, CreatedAt: r.CreatedAt,
-	})
-}
-
-func (a storeAdapter) ReplacePRComments(ctx context.Context, prURL string, comments []ports.PRComment) error {
-	rows := make([]sqlite.PRCommentRow, len(comments))
+func (a storeAdapter) WritePR(ctx context.Context, pr ports.PRRow, checks []ports.PRCheckRow, comments []ports.PRComment) error {
+	row := sqlite.PRRow{
+		URL: pr.URL, SessionID: pr.SessionID, Number: int64(pr.Number),
+		State:          prState(pr),
+		ReviewDecision: string(pr.Review),
+		CIState:        string(pr.CI),
+		Mergeability:   string(pr.Mergeability),
+		UpdatedAt:      pr.UpdatedAt,
+	}
+	checkRows := make([]sqlite.PRCheckRow, len(checks))
+	for i, c := range checks {
+		checkRows[i] = sqlite.PRCheckRow{
+			PRURL: c.PRURL, Name: c.Name, CommitHash: c.CommitHash,
+			Status: c.Status, URL: c.URL, LogTail: c.LogTail, CreatedAt: c.CreatedAt,
+		}
+	}
+	commentRows := make([]sqlite.PRCommentRow, len(comments))
 	for i, c := range comments {
-		rows[i] = sqlite.PRCommentRow{
-			PRURL: prURL, CommentID: c.ID, Author: c.Author, File: c.File,
+		commentRows[i] = sqlite.PRCommentRow{
+			PRURL: pr.URL, CommentID: c.ID, Author: c.Author, File: c.File,
 			Line: int64(c.Line), Body: c.Body, Resolved: c.Resolved, CreatedAt: c.CreatedAt,
 		}
 	}
-	return a.Store.ReplacePRComments(ctx, prURL, rows)
+	return a.Store.WritePRObservation(ctx, row, checkRows, commentRows)
 }
 
 // prState collapses the PR's bools into the single pr.state column value.