Notifier foundation: durable notification domain, storage, CDC, and lifecycle enqueue

[whoisasx]

Summary

Build the durable notification foundation for the Go rewrite. This issue owns the provider-neutral notification contract, rich notification payload rendering, SQLite persistence, CDC events, idempotent dedupe, and the first lifecycle integration that replaces the current noopNotifier with a store-backed notification enqueuer.

This issue does not implement dashboard UI, Electron desktop delivery, Slack, webhook, Discord, or any external notifier sink. Its job is to make human notifications durable and queryable.

Context

The current Go daemon already has:

• lifecycle reactions in backend/internal/lifecycle/reactions.go
• a small ports.Notifier interface in backend/internal/ports/outbound.go
• production wiring that still injects noopNotifier in backend/lifecycle_wiring.go
• SQLite persistence with trigger-written change_log CDC in backend/internal/storage/sqlite/migrations/0001_init.sql
• CDC event vocabulary in backend/internal/cdc/event.go

Legacy AO had useful notification concepts, especially its rich notification data shape, but it mixed notification routing into lifecycle and stored dashboard notifications separately in JSONL. The rewrite should instead persist notifications in SQLite as a first-class read model.

Goals

• Add a provider-neutral notification domain model.
• Extend lifecycle notification events so they carry stable identity, priority, reaction context, escalation context, and a deterministic dedupe key.
• Persist notifications in SQLite.
• Emit notification_created and notification_updated through existing change_log CDC triggers.
• Replace noopNotifier with a durable store-backed enqueuer.
• Preserve clean boundaries: lifecycle decides what should notify; notification package persists and renders the notification.
• Keep payloads rich enough for dashboard and AO app desktop to render meaningful notifications later.

Non-Goals

• No Slack implementation.
• No webhook implementation.
• No Discord implementation.
• No external notifier sink implementation.
• No Electron desktop delivery in this issue.
• No dashboard notification center UI in this issue.
• No shelling out to terminal-notifier, osascript, notify-send, or PowerShell.

Proposed Files

Add:

backend/internal/domain/notification.go
backend/internal/notification/payload.go
backend/internal/notification/renderer.go
backend/internal/notification/dedupe.go
backend/internal/notification/enqueuer.go
backend/internal/notification/enqueuer_test.go
backend/internal/notification/renderer_test.go
backend/internal/notification/dedupe_test.go
backend/internal/storage/sqlite/migrations/0002_notifications.sql
backend/internal/storage/sqlite/queries/notifications.sql
backend/internal/storage/sqlite/notification_store.go
backend/internal/storage/sqlite/notification_store_test.go

Modify:

backend/internal/ports/outbound.go
backend/internal/lifecycle/reactions.go
backend/lifecycle_wiring.go
backend/internal/cdc/event.go
backend/internal/storage/sqlite/store_test.go
backend/internal/integration/lifecycle_sqlite_test.go
backend/sqlc.yaml, only if needed by sqlc conventions
backend/internal/storage/sqlite/gen/*, regenerated by sqlc

Contract Changes

Update backend/internal/ports/outbound.go.

Add warning priority now so we do not need a later compatibility break:

type Priority string

const (
    PriorityUrgent  Priority = "urgent"
    PriorityAction  Priority = "action"
    PriorityWarning Priority = "warning"
    PriorityInfo    Priority = "info"
)

Extend the lifecycle-facing event:

type Event struct {
    Type       string
    Priority   Priority
    SessionID  domain.SessionID
    ProjectID  domain.ProjectID
    Message    string

    Reaction   *ReactionEvent
    Escalation *EscalationEvent
    DedupeKey  string
    CauseKey   string
    OccurredAt time.Time
}

type ReactionEvent struct {
    Key    string // agent-needs-input, approved-and-green, ci-failed, etc.
    Action string // notify | escalated
}

type EscalationEvent struct {
    Attempts   int
    Cause      string // max_retries | max_attempts | max_duration
    DurationMs int64
}

Add persisted/API domain types in backend/internal/domain/notification.go:

type NotificationID string

type Notification struct {
    Seq          int64
    ID           NotificationID
    ProjectID    ProjectID
    SessionID    SessionID
    Source       string
    EventType    string
    SemanticType string
    Priority     string
    Message      string
    Payload      json.RawMessage
    Actions      []NotificationAction
    DedupeKey    string
    CauseKey     string
    ReadAt       time.Time
    ArchivedAt   time.Time
    CreatedAt    time.Time
    UpdatedAt    time.Time
}

type NotificationAction struct {
    ID     string `json:"id"`
    Kind   string `json:"kind"`
    Label  string `json:"label"`
    Route  string `json:"route,omitempty"`
    URL    string `json:"url,omitempty"`
    Method string `json:"method,omitempty"`
}

Payload Shape

Use a Go equivalent of legacy AO's NotificationData V3 shape. This lets dashboard and AO app render useful notifications without provider-specific logic.

Target JSON shape:

{
  "schemaVersion": 3,
  "semanticType": "session.needs_input",
  "subject": {
    "session": { "id": "ao-7", "projectId": "ao" },
    "pr": { "number": 12, "url": "https://github.com/org/repo/pull/12" },
    "issue": { "id": "AO-12" },
    "branch": "feat/example"
  },
  "reaction": { "key": "agent-needs-input", "action": "notify" },
  "escalation": { "attempts": 3, "cause": "max_retries", "durationMs": 0 },
  "ci": { "status": "failing" },
  "review": { "decision": "changes_requested" },
  "merge": { "ready": true, "conflicts": false, "isBehind": false }
}

Initial renderer should fill only data that exists in the Go rewrite today:

• project id
• session id
• issue id
• branch from session metadata
• PR URL/number/draft from PR facts
• CI state
• review decision
• mergeability
• reaction key/action
• escalation details

Leave PR title, owner, repo, base branch, review thread details, and CI log tails omitted until SCM enrichment persists them.

Semantic Type Mapping

Map internal reaction keys to public semantic types:

approved-and-green  -> merge.ready
agent-stuck         -> session.stuck
agent-needs-input   -> session.needs_input
agent-exited        -> session.exited
pr-closed           -> pr.closed
pr-merged           -> pr.merged
ci-failed           -> ci.failing
review-comments     -> review.changes_requested
merge-conflicts     -> merge.conflicts

SQLite Migration

Add backend/internal/storage/sqlite/migrations/0002_notifications.sql.

Suggested schema:

-- +goose Up
-- +goose StatementBegin
CREATE TABLE notifications (
    seq           INTEGER PRIMARY KEY AUTOINCREMENT,
    id            TEXT NOT NULL UNIQUE DEFAULT ('ntf_' || lower(hex(randomblob(16)))),
    project_id    TEXT NOT NULL REFERENCES projects(id),
    session_id    TEXT NOT NULL REFERENCES sessions(id),
    source        TEXT NOT NULL DEFAULT 'lifecycle' CHECK (source IN ('lifecycle')),
    event_type    TEXT NOT NULL,
    semantic_type TEXT NOT NULL,
    priority      TEXT NOT NULL CHECK (priority IN ('urgent','action','warning','info')),
    message       TEXT NOT NULL,
    payload_json  TEXT NOT NULL CHECK (json_valid(payload_json)),
    actions_json  TEXT NOT NULL DEFAULT '[]' CHECK (json_valid(actions_json)),
    dedupe_key    TEXT NOT NULL UNIQUE,
    cause_key     TEXT NOT NULL DEFAULT '',
    read_at       TIMESTAMP,
    archived_at   TIMESTAMP,
    created_at    TIMESTAMP NOT NULL DEFAULT (datetime('now')),
    updated_at    TIMESTAMP NOT NULL DEFAULT (datetime('now'))
);

CREATE INDEX idx_notifications_project_seq ON notifications(project_id, seq DESC);
CREATE INDEX idx_notifications_session_seq ON notifications(session_id, seq DESC);
CREATE INDEX idx_notifications_unread ON notifications(seq DESC)
    WHERE read_at IS NULL AND archived_at IS NULL;
-- +goose StatementEnd

Add CDC triggers:

-- +goose StatementBegin
CREATE TRIGGER notifications_cdc_insert
AFTER INSERT ON notifications
BEGIN
    INSERT INTO change_log (project_id, session_id, event_type, payload, created_at)
    VALUES (
        NEW.project_id,
        NEW.session_id,
        'notification_created',
        json_object(
            'seq', NEW.seq,
            'id', NEW.id,
            'type', NEW.semantic_type,
            'priority', NEW.priority,
            'message', NEW.message,
            'data', json(NEW.payload_json),
            'actions', json(NEW.actions_json),
            'readAt', NEW.read_at,
            'archivedAt', NEW.archived_at
        ),
        NEW.created_at
    );
END;
-- +goose StatementEnd

-- +goose StatementBegin
CREATE TRIGGER notifications_cdc_update
AFTER UPDATE ON notifications
WHEN OLD.read_at IS NOT NEW.read_at
  OR OLD.archived_at IS NOT NEW.archived_at
BEGIN
    INSERT INTO change_log (project_id, session_id, event_type, payload, created_at)
    VALUES (
        NEW.project_id,
        NEW.session_id,
        'notification_updated',
        json_object(
            'seq', NEW.seq,
            'id', NEW.id,
            'readAt', NEW.read_at,
            'archivedAt', NEW.archived_at
        ),
        NEW.updated_at
    );
END;
-- +goose StatementEnd

Add down migration that drops triggers and table.

CDC Event Types

Update backend/internal/cdc/event.go:

const (
    EventNotificationCreated EventType = "notification_created"
    EventNotificationUpdated EventType = "notification_updated"
)

SQL Queries

Add backend/internal/storage/sqlite/queries/notifications.sql.

Required queries:

InsertNotification
GetNotification
GetNotificationByDedupeKey
ListNotifications
ListNotificationsByProject
ListNotificationsBySession
ListUnreadNotifications
MarkNotificationRead
MarkNotificationUnread
ArchiveNotification

InsertNotification should be idempotent. SQLite's RETURNING with ON CONFLICT DO NOTHING can return no row, so the store wrapper should handle duplicate by reading GetNotificationByDedupeKey.

Store Wrapper

Add backend/internal/storage/sqlite/notification_store.go.

Required methods:

func (s *Store) EnqueueNotification(ctx context.Context, row NotificationRow) (NotificationRow, bool, error)
func (s *Store) GetNotification(ctx context.Context, id string) (NotificationRow, bool, error)
func (s *Store) ListNotifications(ctx context.Context, filter NotificationFilter) ([]NotificationRow, error)
func (s *Store) MarkNotificationRead(ctx context.Context, id string, at time.Time) (NotificationRow, bool, error)
func (s *Store) MarkNotificationUnread(ctx context.Context, id string) (NotificationRow, bool, error)
func (s *Store) ArchiveNotification(ctx context.Context, id string, at time.Time) (NotificationRow, bool, error)

Idempotency rules:

• same dedupe_key returns existing row with created=false
• repeated mark-read should not create extra CDC if already read
• repeated archive should not create extra CDC if already archived

Dedupe Design

Use deterministic keys:

v1:lifecycle:{project_id}:{session_id}:{reaction_key}:{condition_hash}

Condition hash inputs:

• session-state notifications: session state + termination reason + session updated timestamp
• PR notifications: PR URL + CI + review + mergeability + latest relevant check/review signature when available
• escalations: underlying reaction key + same condition hash + escalation cause

This gives durable restart-safe dedupe without suppressing genuinely new episodes.

Lifecycle Integration

Update backend/internal/lifecycle/reactions.go so fireNotify provides:

• reaction key
• event type
• priority
• message
• occurred time
• cause key
• dedupe key or enough inputs for the enqueuer to compute it

Replace noopNotifier in backend/lifecycle_wiring.go:

renderer := notification.NewRenderer(store)
notifier := notification.NewEnqueuer(store, renderer, logger)
lcm := lifecycle.New(a, a, notifier, noopMessenger{})

Leave noopMessenger and runtime registry untouched.

Tests

Add renderer tests:

• every reaction key maps to expected semantic type
• payload includes session/project/issue/branch
• PR payload includes URL/number/CI/review/mergeability when available
• escalation payload includes attempts/cause/duration

Add dedupe tests:

• same reaction/condition produces same key
• changed condition produces new key
• escalation includes cause and does not collide with base reaction

Add SQLite tests:

• insert/list/get
• duplicate dedupe key returns existing row
• mark read/unread/archive
• repeated read/archive is idempotent
• JSON validity constraints reject invalid payload/actions
• CDC rows are generated for create/read/archive
• concurrent enqueue with same dedupe key creates one row

Add lifecycle integration tests:

• needs_input reaction creates durable notification
• approved-and-green reaction creates durable notification
• pr_merged creates durable notification
• notification row emits notification_created CDC

Update existing tests that assert old ports.Event shape.

Acceptance Criteria

• Lifecycle human notifications are no longer dropped into noopNotifier.
• Notifications are persisted in SQLite with rich payloads.
• Duplicate lifecycle dispatch after daemon restart does not create duplicate rows.
• Created/read/archive changes emit ordered CDC through change_log.
• No dashboard UI, Electron desktop, Slack, webhook, or Discord code is added.
• cd backend && go test ./... passes.
• cd backend && go vet ./... passes.
• sqlc-generated code is updated and committed.

Risks / Follow-Ups

• Dedupe can over-suppress if condition hash is too broad. Cover same-condition and new-episode tests.
• react() currently dispatches outside the lifecycle lock. Renderer should rehydrate current store state and avoid trusting stale context.
• Notifications and change_log will grow. Retention/pruning should be a later explicit issue.
• Payloads may eventually include sensitive CI/review text. Keep initial payload factual and avoid raw logs/comments until product requirements are explicit.