Skip to main content

SP3.5 — Cloud Realtime (cross-device live watch*) Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

EXPLORATION RULE (graphify-first): graphify-out/graph.json exists. Before grepping/reading around, run graphify query "<question>" (scoped subgraph), graphify path "<A>" "<B>", or graphify explain "<concept>". Read exact files only for exact signatures after orienting. Include this rule in any sub-exploration.

Goal: Layer Supabase Realtime (postgres_changes) onto SP3's cloud data path so an other-device commit drives the same five watch* emissions this device already gets from its own writes. On a realtime event, coarse-re-hydrate the affected aggregate from PostgREST (under RLS) into the in-memory cache; the cache's existing change-controller re-emits. This unblocks approval Beat 2 (a child sees the celebration when a parent approves remotely). Zero change to local DataMode.

Architecture: Realtime lives entirely behind the storage seam. A new RealtimePort (SupabaseRealtimePort in prod, FakeRealtimePort in tests) delivers a coarse "table changed" signal to CachedStorageAdapter, which coalesces/debounces the signal, re-reads that aggregate from its durable (the SupabaseStorageAdapter), and re-inserts into its cache (an InMemoryStorageAdapter) — firing the same broadcast StreamController the local write path fires. createClient injects the realtime port only on the cloud branch; local passes null. Client.dispose() fans out to the adapter's dispose() for channel teardown.

Tech Stack: Dart/Flutter (FVM), supabase 2.13.0 / realtime_client 2.8.0 (transitive), Supabase Realtime postgres_changes + RLS, flutter_test.

Spec (source of truth): docs/superpowers/specs/2026-07-20-sp35-cloud-realtime-design.md

Global Constraints​

  • Dart/Flutter monorepo, FVM only: fvm flutter / fvm dart — never bare flutter/dart, never node for Dart tooling.
  • ONE data path: Bloc → Repository → Client facade → Service → Adapter. Realtime lives inside the Adapter layer; presentation imports only the facade and NEVER supabase.
  • Migrations are FILE-ONLY in infra/supabase/migrations/ — the implementer writes the .sql; the controller applies it live after review. Never call a Supabase apply tool.
  • Anon/publishable key only in app code; service-role only inside Edge Functions.
  • DUAL GATE: RLS is the real guard — realtime must NOT widen visibility. Every event and re-read is RLS-scoped to the household (the household_id channel filter is belt-and-suspenders).
  • NO behaviour change for local DataMode (the default): realtime is cloud-only; the local branch passes realtime: null.
  • Explicit git add <files> — never -A; never stage graphify-out/, .superpowers/, .claude/.
  • Suite baselines: SDK 1114 / app 703 / DS 287 tests must not drop; new tests add on top.
  • After code changes, run graphify update . (final task).
  • No Drift table changes → NO build_runner run is required this phase. If one ever is, scope it with --build-filter and restore clobbered .g.dart siblings from HEAD.

Task 1: Realtime publication migration (file-only)​

Files:

  • Create: infra/supabase/migrations/20260720000100_realtime_publication.sql

Interfaces:

  • Consumes: the existing supabase_realtime publication (Supabase provisions it by default) and the five watched tables from 20260612000002_tier0_domain.sql and 20260624* authz migrations (household_members, chores, approvals, ledger_entries, redemptions).

  • Produces: those five tables published for postgres_changes + REPLICA IDENTITY FULL so the household_id=eq filter matches on UPDATE/DELETE payloads. Task 2's kRealtimeTables list mirrors these names exactly.

  • Step 1: Write the migration file

infra/supabase/migrations/20260720000100_realtime_publication.sql:

-- SP3.5 Cloud Realtime — publish the five WATCHED tables for postgres_changes
-- (design spec docs/superpowers/specs/2026-07-20-sp35-cloud-realtime-design.md).
--
-- These are exactly the tables backing the five watch* streams the app consumes
-- (watchMembers/Chores/Approvals/LedgerEntries/Redemptions). Nothing else is
-- watched, so nothing else is published — adding a table is a one-line follow-up
-- once a watch* stream exists for it.
--
-- RLS still gates delivery: the client subscribes with the member's JWT
-- (SupabaseClient wires realtime.setAuth on auth change), so postgres_changes
-- only delivers rows this member may see. The household_id channel filter is
-- belt-and-suspenders on top of RLS. REPLICA IDENTITY FULL is required so that
-- filter also matches on UPDATE/DELETE payloads (default replica identity ships
-- only the primary key in the change record).

-- ── Ensure the publication exists (Supabase creates it by default; guard for
--    a fresh/self-hosted project) ──────────────────────────────────────────
do $$
begin
  if not exists (select 1 from pg_publication where pubname = 'supabase_realtime')
  then
    create publication supabase_realtime;
  end if;
end $$;

-- ── Add the five watched tables to the publication (idempotent) ─────────────
do $$
declare
  t text;
begin
  foreach t in array array[
    'household_members', 'chores', 'approvals', 'ledger_entries', 'redemptions'
  ]
  loop
    if not exists (
      select 1 from pg_publication_tables
      where pubname = 'supabase_realtime'
        and schemaname = 'public'
        and tablename = t
    ) then
      execute format(
        'alter publication supabase_realtime add table public.%I', t
      );
    end if;
  end loop;
end $$;

-- ── REPLICA IDENTITY FULL so household_id filtering works on every event type
alter table public.household_members replica identity full;
alter table public.chores replica identity full;
alter table public.approvals replica identity full;
alter table public.ledger_entries replica identity full;
alter table public.redemptions replica identity full;
  • Step 2: Verify house invariants (file-level checks — the controller applies it live after review)

Run: grep -c "replica identity full" infra/supabase/migrations/20260720000100_realtime_publication.sql Expected: 5 (one per watched table).

Run: grep -n "add table public" infra/supabase/migrations/20260720000100_realtime_publication.sql Expected: 1 hit — the single idempotent alter publication ... add table inside the foreach loop (never an unconditional bare add table).

Run: grep -c "service_role\|security definer\|drop " infra/supabase/migrations/20260720000100_realtime_publication.sql Expected: 0 — this migration grants no roles, defines no functions, drops nothing.

Controller note (live apply): before applying, verify which tables are already published — select tablename from pg_publication_tables where pubname='supabase_realtime' and schemaname='public'; — the migration is idempotent so re-adding is a no-op, but this confirms the starting state.

  • Step 3: Commit
git add infra/supabase/migrations/20260720000100_realtime_publication.sql
git commit -m "feat: publish watched tables to supabase_realtime for cross-device realtime (file-only)"

Task 2: RealtimePort seam + FakeRealtimePort + SupabaseRealtimePort​

Files:

  • Create: packages/client_sdk/lib/src/adapters/realtime_port.dart
  • Create: packages/client_sdk/lib/src/adapters/cloud/supabase_realtime_port.dart
  • Create: packages/client_sdk/test/support/fake_realtime_port.dart
  • Modify: packages/client_sdk/lib/ports.dart (export RealtimePort)
  • Test: packages/client_sdk/test/support/fake_realtime_port_test.dart

Interfaces:

  • Consumes: package:supabase/supabase.dart (SupabaseClient, PostgresChangeEvent, PostgresChangeFilter, PostgresChangeFilterType, RealtimeChannel, RealtimeSubscribeStatus) — pure-Dart, already a transitive dep (2.13.0).

  • Produces (consumed by Tasks 3–5):

    • abstract interface class RealtimePort { void start({required String householdId, required void Function(String table) onChange, required void Function() onResync}); Future<void> dispose(); }
    • class SupabaseRealtimePort implements RealtimePort + const List<String> kRealtimeTables
    • class FakeRealtimePort implements RealtimePort with emit(String table) / resync() / bool disposed / String? householdId

  • Step 1: Write the failing test

packages/client_sdk/test/support/fake_realtime_port_test.dart:

import 'package:flutter_test/flutter_test.dart';

import 'fake_realtime_port.dart';

void main() {
  group('FakeRealtimePort', () {
    test('start records the household; emit/resync fan out to the callbacks',
        () {
      final port = FakeRealtimePort();
      final changed = <String>[];
      var resyncs = 0;

      port.start(
        householdId: 'h1',
        onChange: changed.add,
        onResync: () => resyncs++,
      );
      expect(port.householdId, 'h1');

      port.emit('approvals');
      port.emit('ledger_entries');
      port.resync();

      expect(changed, ['approvals', 'ledger_entries']);
      expect(resyncs, 1);
    });

    test('dispose marks disposed and silences later events', () async {
      final port = FakeRealtimePort();
      final changed = <String>[];
      port.start(
        householdId: 'h1',
        onChange: changed.add,
        onResync: () {},
      );
      await port.dispose();
      expect(port.disposed, isTrue);
      port.emit('chores'); // no-op after dispose
      expect(changed, isEmpty);
    });
  });
}
  • Step 2: Run test to verify it fails

Run (from packages/client_sdk/): fvm flutter test test/support/fake_realtime_port_test.dart Expected: FAIL — compile error: fake_realtime_port.dart does not exist.

  • Step 3: Write minimal implementation

packages/client_sdk/lib/src/adapters/realtime_port.dart (new file, complete):

/// Cross-device change notifier (SP3.5). Delivers a COARSE "table changed"
/// signal so the caller re-reads the affected aggregate under RLS and re-emits
/// from its cache. Pure Dart — NO `supabase` import here, so the cache decorator
/// that depends on this stays transport-agnostic (the Supabase implementation
/// lives in `cloud/supabase_realtime_port.dart`).
///
/// The row payload is intentionally NOT surfaced: the RLS-scoped re-read is the
/// authority, so a spoofed or stale event can never widen visibility or drift
/// the cache (design decisions R2 + R3).
abstract interface class RealtimePort {
  /// Begin delivering household-scoped row-change notifications for
  /// [householdId]. [onChange] receives the changed TABLE name only. [onResync]
  /// fires on every (re)subscription — initial subscribe AND post-reconnect —
  /// so the caller can re-hydrate to close any gap of events missed while
  /// disconnected. Idempotent: a second [start] while a channel is already open
  /// is a no-op.
  void start({
    required String householdId,
    required void Function(String table) onChange,
    required void Function() onResync,
  });

  /// Tears down the channel/socket. Safe to call when never started.
  Future<void> dispose();
}

packages/client_sdk/lib/src/adapters/cloud/supabase_realtime_port.dart (new file, complete):

import 'package:supabase/supabase.dart'
    show
        PostgresChangeEvent,
        PostgresChangeFilter,
        PostgresChangeFilterType,
        RealtimeChannel,
        RealtimeSubscribeStatus,
        SupabaseClient;

import '../realtime_port.dart';

/// The exact tables backing the five watch* streams (watchMembers / watchChores
/// / watchApprovals / watchLedgerEntries / watchRedemptions). These — and only
/// these — are in the `supabase_realtime` publication (migration
/// 20260720000100). Keep this list in lock-step with that migration.
const List<String> kRealtimeTables = <String>[
  'household_members',
  'chores',
  'approvals',
  'ledger_entries',
  'redemptions',
];

/// Production [RealtimePort] over Supabase Realtime `postgres_changes`.
///
/// ONE channel per household (`household:<id>`) carries one change registration
/// per watched table, each filtered `household_id=eq.<id>`. The SAME
/// [SupabaseClient] that holds the auth session is reused, so the socket already
/// carries the member's JWT (SupabaseClient calls `realtime.setAuth` on every
/// auth change): RLS gates delivery on the wire, and the household filter is
/// belt-and-suspenders on top (design decisions R1 + R3).
class SupabaseRealtimePort implements RealtimePort {
  SupabaseRealtimePort(this._client);

  final SupabaseClient _client;
  RealtimeChannel? _channel;

  @override
  void start({
    required String householdId,
    required void Function(String table) onChange,
    required void Function() onResync,
  }) {
    if (_channel != null) return; // already subscribed — idempotent

    var channel = _client.channel('household:$householdId');
    for (final table in kRealtimeTables) {
      channel = channel.onPostgresChanges(
        event: PostgresChangeEvent.all,
        schema: 'public',
        table: table,
        filter: PostgresChangeFilter(
          type: PostgresChangeFilterType.eq,
          column: 'household_id',
          value: householdId,
        ),
        // Coarse signal: hand up the table name only; the caller re-reads it
        // under RLS. The payload's newRecord is deliberately never trusted.
        callback: (payload) => onChange(payload.table),
      );
    }
    channel.subscribe((status, error) {
      // realtime_client's RetryTimer auto-reconnects with backoff; on every
      // (re)subscribe we resync so events missed while disconnected are caught.
      if (status == RealtimeSubscribeStatus.subscribed) onResync();
    });
    _channel = channel;
  }

  @override
  Future<void> dispose() async {
    final channel = _channel;
    _channel = null;
    if (channel != null) await _client.removeChannel(channel);
  }
}

packages/client_sdk/test/support/fake_realtime_port.dart (new file, complete):

import 'package:client_sdk/src/adapters/realtime_port.dart';

/// Test double for [RealtimePort]. Records the subscription and lets a test push
/// SYNTHETIC cross-device events: [emit] simulates another device committing a
/// row change on a table; [resync] simulates a (re)subscribe. No socket, no
/// timers — the whole event→re-hydrate→re-emit path is driven deterministically.
class FakeRealtimePort implements RealtimePort {
  String? householdId;
  bool disposed = false;
  void Function(String table)? _onChange;
  void Function()? _onResync;

  @override
  void start({
    required String householdId,
    required void Function(String table) onChange,
    required void Function() onResync,
  }) {
    this.householdId = householdId;
    _onChange = onChange;
    _onResync = onResync;
  }

  /// Simulate another device committing a row change on [table].
  void emit(String table) => _onChange?.call(table);

  /// Simulate a (re)subscribe — the gap-closing resync signal.
  void resync() => _onResync?.call();

  @override
  Future<void> dispose() async {
    disposed = true;
    _onChange = null;
    _onResync = null;
  }
}

Modify packages/client_sdk/lib/ports.dart: add the RealtimePort export next to the other adapter-implementer exports (it belongs on the ports.dart surface, not the main barrel — realtime is an adapter-provider concern, never a presentation import):

export 'src/adapters/realtime_port.dart' show RealtimePort;
  • Step 4: Run test to verify it passes

Run (from packages/client_sdk/): fvm flutter test test/support/fake_realtime_port_test.dart Expected: PASS (both tests green). Then confirm the new production file compiles under analysis: fvm dart analyze lib/src/adapters/cloud/supabase_realtime_port.dart lib/src/adapters/realtime_port.dart Expected: No issues found!

  • Step 5: Commit
git add packages/client_sdk/lib/src/adapters/realtime_port.dart packages/client_sdk/lib/src/adapters/cloud/supabase_realtime_port.dart packages/client_sdk/test/support/fake_realtime_port.dart packages/client_sdk/test/support/fake_realtime_port_test.dart packages/client_sdk/lib/ports.dart
git commit -m "feat: RealtimePort seam + SupabaseRealtimePort (postgres_changes) + fake"

Task 3: CachedStorageAdapter — event → re-hydrate → re-emit + debounce​

Files:

  • Modify: packages/client_sdk/lib/src/adapters/adapter.dart (add Disposable marker interface)
  • Modify: packages/client_sdk/lib/src/adapters/cached/cached_storage_adapter.dart
  • Test: packages/client_sdk/test/cached_realtime_test.dart

Interfaces:

  • Consumes: RealtimePort (Task 2), the existing _durable reads (getMembers / getChores / watchApprovals(...).first / getLedgerEntries / getRedemptions) and the cache insert* that fire the in-memory broadcast controllers (verified in in_memory_storage_adapter.dart: _memberChanges / _choreChanges / _approvalChanges / _ledgerChanges / _redemptionChanges).
  • Produces:
    • abstract interface class Disposable { Future<void> dispose(); } (in adapter.dart)
    • CachedStorageAdapter({required StoragePort durable, StoragePort? cache, RealtimePort? realtime, Duration realtimeDebounce = const Duration(milliseconds: 250)}) now implements StoragePort, ConsentPort, Disposable
    • Future<void> dispose() on CachedStorageAdapter

Design decision (documented): the re-hydrate is COARSE and insert-only (design decision R2) — it re-reads current rows and re-inserts, so a duplicate event is idempotent and the append-only invariants hold. It does NOT evict rows deleted on another device (approvals/ledger/redemptions are append/update-only; member/chore deletes reconcile on the next cold hydrate). The subscription starts inside _hydrate (the one place the household id is known) and is idempotent per household.

  • Step 1: Write the failing test

packages/client_sdk/test/cached_realtime_test.dart:

import 'package:client_sdk/client_sdk.dart';
import 'package:client_sdk/src/adapters/cached/cached_storage_adapter.dart';
import 'package:client_sdk/src/adapters/memory/in_memory_storage_adapter.dart';
import 'package:flutter_test/flutter_test.dart';

import 'support/fake_realtime_port.dart';

void main() {
  late InMemoryStorageAdapter durable; // the shared "cloud"
  late InMemoryStorageAdapter cache;
  late FakeRealtimePort realtime;
  late CachedStorageAdapter adapter;

  const household = Household(id: 'h1', name: 'Casa');
  const parent = HouseholdMember(
    id: 'p1',
    householdId: 'h1',
    displayName: 'Pat',
    kind: MemberKind.parent,
  );

  setUp(() async {
    durable = InMemoryStorageAdapter();
    cache = InMemoryStorageAdapter();
    realtime = FakeRealtimePort();
    await durable.insertHousehold(household);
    await durable.insertMember(parent);
    adapter = CachedStorageAdapter(
      durable: durable,
      cache: cache,
      realtime: realtime,
      realtimeDebounce: Duration.zero, // deterministic: flush next microtask
    );
  });

  test('subscribes to realtime for the household after first hydrate', () async {
    expect(realtime.householdId, isNull, reason: 'no subscription pre-hydrate');
    await adapter.getHousehold(); // triggers lazy hydrate
    expect(realtime.householdId, 'h1');
  });

  test('an other-device member insert re-emits watchMembers', () async {
    final seen = <List<HouseholdMember>>[];
    final sub = adapter.watchMembers('h1').listen(seen.add);
    await Future<void>.delayed(Duration.zero); // initial emission (just Pat)
    expect(seen.last.map((m) => m.id), ['p1']);

    // Another device commits a new member straight into the shared durable
    // (bypassing this adapter), then realtime signals the table changed.
    await durable.insertMember(const HouseholdMember(
      id: 'c1',
      householdId: 'h1',
      displayName: 'Sam',
      kind: MemberKind.child,
    ));
    realtime.emit('household_members');
    await Future<void>.delayed(Duration.zero); // debounce flush + re-emit

    expect(seen.last.map((m) => m.id), containsAll(['p1', 'c1']),
        reason: 'cross-device roster reached this device via re-hydrate');
    await sub.cancel();
  });

  test('resync (reconnect) re-hydrates all watched tables', () async {
    final approvals = <List<Approval>>[];
    final sub = adapter.watchApprovals('h1').listen(approvals.add);
    await Future<void>.delayed(Duration.zero);

    await durable.insertApproval(const Approval(
      id: 'a1',
      householdId: 'h1',
      memberId: 'c1',
      kind: ApprovalKind.completion,
      status: ApprovalStatus.approved,
      refId: 'sub1',
      tokenAmount: 3,
      bucket: Bucket.spend,
    ));
    realtime.resync(); // e.g. after a dropped socket reconnected
    await Future<void>.delayed(Duration.zero);

    expect(approvals.last.map((a) => a.id), contains('a1'));
    await sub.cancel();
  });

  test('dispose tears down realtime and is idempotent', () async {
    await adapter.getHousehold();
    expect(realtime.disposed, isFalse);
    await adapter.dispose();
    expect(realtime.disposed, isTrue);
    await adapter.dispose(); // safe twice
  });

  test('local mode (no realtime) never subscribes and behaves unchanged',
      () async {
    final localAdapter = CachedStorageAdapter(durable: durable, cache: cache);
    final members = await localAdapter.getMembers('h1');
    expect(members.map((m) => m.id), ['p1']);
    await localAdapter.dispose(); // no realtime — a clean no-op
  });
}
  • Step 2: Run test to verify it fails

Run (from packages/client_sdk/): fvm flutter test test/cached_realtime_test.dart Expected: FAIL — compile error: CachedStorageAdapter has no realtime / realtimeDebounce parameter and no dispose.

  • Step 3: Write minimal implementation

Add to packages/client_sdk/lib/src/adapters/adapter.dart (top-level, next to the StoragePort declaration — a tiny marker so the client can fan out teardown without importing the concrete cached adapter):

/// Optional teardown seam. A port that holds a live resource (a realtime
/// socket, a DB handle) implements this; [Client.dispose] fans out to it. Ports
/// without live resources (the in-memory fakes, the Drift store) simply do not
/// implement it, so teardown is a no-op there.
abstract interface class Disposable {
  Future<void> dispose();
}

Modify packages/client_sdk/lib/src/adapters/cached/cached_storage_adapter.dart:

Add import 'dart:async'; at the top of the import block, and import '../realtime_port.dart'; beside import '../adapter.dart';.

Change the class declaration and constructor (currently lines 49–55) to:

class CachedStorageAdapter implements StoragePort, ConsentPort, Disposable {
  CachedStorageAdapter({
    required StoragePort durable,
    StoragePort? cache,
    RealtimePort? realtime,
    Duration realtimeDebounce = const Duration(milliseconds: 250),
  }) : _durable = durable,
       _cache = cache ?? InMemoryStorageAdapter(),
       _realtime = realtime,
       _realtimeDebounce = realtimeDebounce;

  final StoragePort _durable;
  final StoragePort _cache;

  /// Cross-device change notifier. Null on the local tier — the whole realtime
  /// path then no-ops and behaviour is identical to SP3 (design decision R4/R5).
  final RealtimePort? _realtime;

  /// Debounce window coalescing an event burst (one commit can touch approvals
  /// AND ledger_entries) into one re-hydrate per affected table (decision R6).
  final Duration _realtimeDebounce;

  /// The household the realtime channel is scoped to; set once, post-hydrate.
  String? _realtimeHouseholdId;

  /// Tables changed since the last flush; drained by [_flushPending].
  final Set<String> _pendingTables = <String>{};
  Timer? _debounce;

In _hydrate (currently lines 79–163), after the household is loaded and its id is known — immediately after final id = household.id; (currently line 87) — insert the subscription kick-off:

    final id = household.id;
    _startRealtime(id);

Add the realtime machinery as a new section immediately BEFORE the // ── Households comment (currently line 165):

  // ── Realtime (SP3.5: cross-device live watch*) ────────────────
  //
  // On an other-device commit the [RealtimePort] hands up the changed TABLE
  // name; we COARSE re-read that aggregate from [_durable] (PostgREST, RLS-
  // scoped) and re-insert into [_cache], whose broadcast controller re-emits to
  // every watcher — the SAME machinery a local write already drives. The row
  // payload is never trusted; the RLS re-read is the authority (decisions
  // R2/R3). Insert-only merge: cross-device deletes reconcile on cold hydrate.

  /// Subscribes the realtime channel for [householdId] exactly once. No-op on
  /// the local tier (no [_realtime]) or when already live for this household.
  void _startRealtime(String householdId) {
    final realtime = _realtime;
    if (realtime == null) return;
    if (_realtimeHouseholdId == householdId) return;
    _realtimeHouseholdId = householdId;
    realtime.start(
      householdId: householdId,
      onChange: _onRealtimeChange,
      onResync: () => _resyncAll(householdId),
    );
  }

  /// Coalesce a change signal: remember the table and (re)arm the debounce.
  void _onRealtimeChange(String table) {
    _pendingTables.add(table);
    _debounce?.cancel();
    _debounce = Timer(_realtimeDebounce, _flushPending);
  }

  Future<void> _flushPending() async {
    final id = _realtimeHouseholdId;
    if (id == null) return;
    final tables = _pendingTables.toList(growable: false);
    _pendingTables.clear();
    for (final table in tables) {
      await _rehydrateTable(id, table);
    }
  }

  /// (Re)subscribe gap-closer: re-hydrate every watched table so events missed
  /// while the socket was down are caught. Fired on each (re)subscribe.
  Future<void> _resyncAll(String householdId) async {
    for (final table in const <String>[
      'household_members',
      'chores',
      'approvals',
      'ledger_entries',
      'redemptions',
    ]) {
      await _rehydrateTable(householdId, table);
    }
  }

  /// Coarse re-read of one aggregate from the durable into the cache. Each cache
  /// insert fires the matching broadcast controller, so the corresponding
  /// watch* stream re-emits. Approvals are watch-only on the port, so drain the
  /// durable watch's first emission (exactly as [_hydrate] does).
  Future<void> _rehydrateTable(String householdId, String table) async {
    switch (table) {
      case 'household_members':
        for (final member in await _durable.getMembers(householdId)) {
          await _cache.insertMember(member);
        }
      case 'chores':
        for (final chore in await _durable.getChores(householdId)) {
          await _cache.insertChore(chore);
        }
      case 'approvals':
        for (final approval
            in await _durable.watchApprovals(householdId).first) {
          await _cache.insertApproval(approval);
        }
      case 'ledger_entries':
        for (final entry in await _durable.getLedgerEntries(householdId)) {
          await _cache.insertLedgerEntry(entry);
        }
      case 'redemptions':
        for (final redemption in await _durable.getRedemptions(householdId)) {
          await _cache.insertRedemption(redemption);
        }
    }
  }

  /// Teardown: cancel the debounce and drop the realtime channel. Safe to call
  /// when realtime was never started (local tier) and safe to call twice.
  @override
  Future<void> dispose() async {
    _debounce?.cancel();
    _debounce = null;
    _pendingTables.clear();
    _realtimeHouseholdId = null;
    await _realtime?.dispose();
  }
  • Step 4: Run test to verify it passes

Run (from packages/client_sdk/): fvm flutter test test/cached_realtime_test.dart Expected: PASS (all five tests green). Then run the FULL SDK suite to confirm no regression from the constructor/class change: fvm flutter test Expected: PASS, ≥ 1114 tests (existing CachedStorageAdapter(...) call sites still compile — the new params are optional).

  • Step 5: Commit
git add packages/client_sdk/lib/src/adapters/adapter.dart packages/client_sdk/lib/src/adapters/cached/cached_storage_adapter.dart packages/client_sdk/test/cached_realtime_test.dart
git commit -m "feat: CachedStorageAdapter realtime event -> coarse re-hydrate -> re-emit + debounce"

Task 4: createClient lifecycle wiring + Client.dispose() teardown​

Files:

  • Modify: packages/client_sdk/lib/src/client/create_client.dart (inject SupabaseRealtimePort on the cloud branch)
  • Modify: packages/client_sdk/lib/src/client/client.dart (add Future<void> dispose(); to the Client interface)
  • Modify: packages/client_sdk/lib/src/client/client_impl.dart (capture storage; implement dispose())
  • Test: packages/client_sdk/test/client/create_client_realtime_test.dart

Interfaces:

  • Consumes: SupabaseRealtimePort (Task 2), Disposable (Task 3), the existing createClient cloud branch and clientFromPort.
  • Produces:
    • cloud data port now CachedStorageAdapter(durable: ..., cache: ..., realtime: SupabaseRealtimePort(supabaseClient))
    • Client.dispose() fanning out to a Disposable storage; local path storage is not Disposable, so dispose() is a no-op.

Design decision (documented): the SAME supabaseClient that carries the GoTrue session is reused for realtime — not a second client — so realtime.setAuth(userJwt) (wired by SupabaseClient._listenForAuthEvents) already scopes the socket to the member. clientFromPort gains a captured storage reference so ClientImpl.dispose() can fan out; MockClient (mocktail) auto-handles the new interface method — existing tests never call dispose(), so they are unaffected; new tests that call it stub it.

  • Step 1: Write the failing test

packages/client_sdk/test/client/create_client_realtime_test.dart:

import 'package:client_sdk/client_sdk.dart';
import 'package:client_sdk/ports.dart';
import 'package:flutter_test/flutter_test.dart';

/// A minimal Disposable storage so we can prove Client.dispose fans out without
/// standing up a real Supabase client.
class _DisposableSpyStore extends InMemoryStorageAdapter implements Disposable {
  bool disposed = false;
  @override
  Future<void> dispose() async => disposed = true;
}

void main() {
  test('Client.dispose fans out to a Disposable storage', () async {
    final store = _DisposableSpyStore();
    final client = clientFromPort(store);
    await client.dispose();
    expect(store.disposed, isTrue);
  });

  test('Client.dispose is a no-op for a non-Disposable storage', () async {
    final client = clientFromPort(InMemoryStorageAdapter());
    await client.dispose(); // must not throw
  });

  test('DataMode.local build never constructs realtime (unchanged behaviour)',
      () async {
    // A local build has no api and no realtime; dispose is a clean no-op.
    final client = createClient(config: const ClientConfig());
    await client.dispose();
  });
}
  • Step 2: Run test to verify it fails

Run (from packages/client_sdk/): fvm flutter test test/client/create_client_realtime_test.dart Expected: FAIL — compile error: Client has no dispose method / clientFromPort result has no dispose.

  • Step 3: Write minimal implementation

Modify packages/client_sdk/lib/src/client/client.dart — add to the abstract class Client interface (place it in the Auth/lifecycle section near the top, after the auth accessors):

  /// Releases resources held by the underlying storage (e.g. a cloud realtime
  /// channel). A no-op on the local/free tier. Call before dropping a cloud
  /// client (sign-out / data-mode switch) so the WebSocket is torn down.
  Future<void> dispose();

Modify packages/client_sdk/lib/src/client/client_impl.dart:

In clientFromPort (currently line 118 return ClientImpl._(), pass the storage through so the impl can fan out teardown — add storage: storage, to the ClientImpl._(...) call:

  return ClientImpl._(
    storage: storage,
    auth: resolvedAuth,
    householdService: HouseholdService(

In the ClientImpl._ constructor (currently line 173), add the parameter and field. Add required StoragePort storage, to the parameter list and : _storage = storage, as the first initializer (before _auth = auth):

  ClientImpl._({
    required StoragePort storage,
    required ClientAuth auth,
    required HouseholdService householdService,
    // ...existing params unchanged...
  }) : _storage = storage,
       _auth = auth,

Add the field beside the other _-fields:

  final StoragePort _storage;

Add the dispose implementation among the facade method overrides (anywhere in the class body):

  @override
  Future<void> dispose() async {
    // Fan out to the storage only when it holds a live resource (the cloud
    // CachedStorageAdapter is Disposable; the Drift + in-memory tiers are not).
    if (_storage case final Disposable disposable) {
      await disposable.dispose();
    }
  }

Modify packages/client_sdk/lib/src/client/create_client.dart — add the import and inject the realtime port on the cloud branch. Add to the import block:

import '../adapters/cloud/supabase_realtime_port.dart';

Change the cloud data-port construction (currently lines 75–80) to:

  final StoragePort dataPort = config.dataMode == DataMode.cloud
      ? CachedStorageAdapter(
          durable: SupabaseStorageAdapter(supabaseClient),
          cache: InMemoryStorageAdapter(),
          // SP3.5: cross-device live watch*. The SAME authed supabaseClient is
          // reused, so the socket carries the member JWT (RLS on the wire).
          realtime: SupabaseRealtimePort(supabaseClient),
        )
      : buildLocalStore(config.localStorageDirectory);

ClientImpl and clientFromPort already import ../adapters/adapter.dart (for StoragePort), which now also exports Disposable — no extra import is needed for the case final Disposable check.

  • Step 4: Run test to verify it passes

Run (from packages/client_sdk/): fvm flutter test test/client/create_client_realtime_test.dart Expected: PASS (all three tests green). Then the full SDK suite (the new Client.dispose() interface method must not break existing MockClient users, which never call it): fvm flutter test Expected: PASS, ≥ 1114 tests. Then the app suite (it consumes the Client facade + MockClient): from app/, fvm flutter test Expected: PASS, ≥ 703 tests. (If any app test now fails because a mock is exercised through a path that calls dispose, stub it: when(() => mockClient.dispose()).thenAnswer((_) async {}); — but no shipped app path calls dispose yet, so none should.)

  • Step 5: Commit
git add packages/client_sdk/lib/src/client/create_client.dart packages/client_sdk/lib/src/client/client.dart packages/client_sdk/lib/src/client/client_impl.dart packages/client_sdk/test/client/create_client_realtime_test.dart
git commit -m "feat: inject SupabaseRealtimePort on cloud build + Client.dispose teardown"

Task 5: Fake-driven end-to-end Beat 2 test (other-device approval → watch emission)​

Files:

  • Test: packages/client_sdk/test/cloud/realtime_beat2_test.dart

Interfaces:

  • Consumes: clientFromPort (via package:client_sdk/ports.dart), CachedStorageAdapter (Task 3), InMemoryStorageAdapter (shared "cloud" durable), FakeRealtimePort (Task 2). Exercises the REAL service graph over the cache — the highest-value proof.

Design decision (documented): two logical devices share ONE InMemoryStorageAdapter as the "cloud" durable. Device A writes an approved approval + its ledger entry straight into the shared durable (simulating a remote parent approval that already committed server-side); FakeRealtimePort.emit(...) then delivers the cross-device signal to device B's adapter, which re-hydrates and re-emits — proving Beat 2 with zero network.

  • Step 1: Write the failing test

packages/client_sdk/test/cloud/realtime_beat2_test.dart:

import 'package:client_sdk/client_sdk.dart';
import 'package:client_sdk/ports.dart';
import 'package:client_sdk/src/adapters/cached/cached_storage_adapter.dart';
import 'package:client_sdk/src/adapters/memory/in_memory_storage_adapter.dart';
import 'package:flutter_test/flutter_test.dart';

import '../support/fake_realtime_port.dart';

void main() {
  // The shared "cloud": both devices read/write the SAME durable, exactly as
  // two subscribed devices share one Supabase project.
  late InMemoryStorageAdapter cloud;
  late FakeRealtimePort realtimeB;
  late Client deviceB;

  const household = Household(id: 'h1', name: 'Casa');
  const child = HouseholdMember(
    id: 'c1',
    householdId: 'h1',
    displayName: 'Sam',
    kind: MemberKind.child,
  );

  setUp(() async {
    cloud = InMemoryStorageAdapter();
    await cloud.insertHousehold(household);
    await cloud.insertMember(child);

    // Device B (the child's device): real services over a cache-first adapter
    // whose durable IS the shared cloud, with a fake realtime port we drive.
    realtimeB = FakeRealtimePort();
    deviceB = clientFromPort(
      CachedStorageAdapter(
        durable: cloud,
        cache: InMemoryStorageAdapter(),
        realtime: realtimeB,
        realtimeDebounce: Duration.zero,
      ),
    );
  });

  test('Beat 2: a remote approval + ledger entry reach device B live', () async {
    // Device B is watching its own approvals + ledger (the celebration + wallet
    // surfaces). Prime the streams (triggers hydrate + realtime subscribe).
    final approvals = <List<Approval>>[];
    final ledger = <List<LedgerEntry>>[];
    final subA = deviceB.watchApprovals().listen(approvals.add);
    final subL = deviceB.watchLedgerEntries().listen(ledger.add);
    await Future<void>.delayed(Duration.zero);
    expect(realtimeB.householdId, 'h1', reason: 'B subscribed post-hydrate');
    expect(approvals.last, isEmpty);

    // Device A (the parent, elsewhere) approves the child's submission: the
    // approved approval + the earn ledger entry are now committed in the cloud.
    await cloud.insertApproval(const Approval(
      id: 'a1',
      householdId: 'h1',
      memberId: 'c1',
      kind: ApprovalKind.completion,
      status: ApprovalStatus.approved,
      refId: 'sub1',
      tokenAmount: 3,
      bucket: Bucket.spend,
    ));
    await cloud.insertLedgerEntry(const LedgerEntry(
      id: 'l1',
      householdId: 'h1',
      memberId: 'c1',
      delta: 3,
      bucket: Bucket.spend,
      kind: LedgerEntryKind.earn,
      refId: 'a1',
    ));

    // Realtime signals both changed tables (one commit, a coalesced burst).
    realtimeB.emit('approvals');
    realtimeB.emit('ledger_entries');
    await Future<void>.delayed(Duration.zero);

    // Beat 2: B sees the approved approval WITHOUT a manual refresh...
    expect(
      approvals.last.where((a) => a.id == 'a1' && a.status == ApprovalStatus.approved),
      isNotEmpty,
      reason: 'the child device sees the remote approval live',
    );
    // ...and the wallet reflects the earned tokens.
    expect(ledger.last.map((e) => e.id), contains('l1'));
    final wallet = await deviceB.walletOf('c1');
    expect(wallet.balanceFor(Bucket.spend), 3,
        reason: 'balance folded from the live ledger');

    await subA.cancel();
    await subL.cancel();
    await deviceB.dispose();
  });
}
  • Step 2: Run test to verify it fails (then passes)

Run (from packages/client_sdk/): fvm flutter test test/cloud/realtime_beat2_test.dart Expected: with Tasks 1–4 landed this PASSES immediately (it composes shipped seams). The load-bearing assertions are the two live emissions — approvals.last containing the approved a1 and ledger.last containing l1 — plus walletOf('c1').balanceFor(Bucket.spend) == 3 (facade accessors verified against client.dart: Future<Wallet> walletOf(String), Wallet.balanceFor(Bucket)).

  • Step 3: Commit
git add packages/client_sdk/test/cloud/realtime_beat2_test.dart
git commit -m "test: end-to-end Beat 2 — remote approval reaches device B via realtime re-hydrate"

Task 6: Guarded live realtime smoke + manual two-identity checklist​

Files:

  • Test: packages/client_sdk/test/cloud/realtime_live_test.dart
  • Create: docs/decisions/2026-07-20-sp35-cloud-realtime-manual-checklist.md

Interfaces:

  • Consumes: the real createClient(DataMode.cloud) path (Task 4) against project bgedvvmihygwxhjxlvfu, guarded by --dart-define creds exactly like live_smoke_test.dart.

  • Produces: a committed, runnable single-process live proof that postgres_changes delivers an event after a self-write; a manual checklist for true two-device cross-visibility (the publishable-key JWT limitation makes that un-headless).

  • Step 1: Write the guarded live smoke

packages/client_sdk/test/cloud/realtime_live_test.dart:

@Tags(['live'])
library;

/// SP3.5 cloud-realtime LIVE smoke (guarded). No-op without real creds.
///
/// Proves the real postgres_changes wire delivers an event to a subscribed
/// device after a write (something the in-memory fake cannot verify). It is a
/// SINGLE-process, SINGLE-identity smoke: the same signed-in user writes a row
/// and asserts its own subscription re-emits. True TWO-identity cross-visibility
/// (parent approves on A, child sees on B) is NOT headlessly automatable — a
/// publishable key cannot attach a second user's JWT — so that is the manual
/// checklist in docs/decisions/2026-07-20-sp35-cloud-realtime-manual-checklist.md.
///
/// Run (project bgedvvmihygwxhjxlvfu; email-confirm OFF; the realtime
/// publication migration 20260720000100 applied):
///   fvm flutter test packages/client_sdk/test/cloud/realtime_live_test.dart \
///     --dart-define-from-file=app/config/supabase.local.json \
///     --dart-define=SMOKE_RUN_ID=<unique-token>
import 'dart:async';

import 'package:client_sdk/client_sdk.dart';
import 'package:flutter_test/flutter_test.dart';

const _url = String.fromEnvironment('SUPABASE_URL');
const _anonKey = String.fromEnvironment('SUPABASE_ANON_KEY');
const _runId = String.fromEnvironment('SMOKE_RUN_ID', defaultValue: 'devsmoke');

class _MemSessionStore implements SessionStore {
  final _m = <String, String>{};
  @override
  Future<String?> read(String key) async => _m[key];
  @override
  Future<void> write(String key, String value) async => _m[key] = value;
  @override
  Future<void> clear(String key) async => _m.remove(key);
}

Client _cloudClient() => createClient(
      config: ClientConfig(
        api: ApiConfig(url: Uri.parse(_url), anonKey: _anonKey),
        dataMode: DataMode.cloud,
        sessionStore: _MemSessionStore(),
      ),
    );

Future<void> _ensureSignedIn(Client c, String email, String password) async {
  try {
    await c.auth.signInWithPassword(email: email, password: password);
  } on AuthFailure {
    await c.auth.signUpWithPassword(email: email, password: password);
  }
}

void main() {
  if (_url.isEmpty) {
    test(
      'realtime live smoke skipped — no SUPABASE_URL',
      () {},
      skip: 'pass --dart-define-from-file=app/config/supabase.local.json',
    );
    return;
  }

  test('a self-write re-emits through the live realtime subscription',
      () async {
    final a = _cloudClient();
    addTearDown(a.dispose);
    await _ensureSignedIn(a, '[email protected]', 'Passw0rd!$_runId');
    final uid = a.auth.currentUser?.id;
    expect(uid, isNotNull);

    await a.createHousehold(name: 'RT A $_runId');
    await a.addMember(
      displayName: 'A Parent',
      kind: MemberKind.parent,
      authUserId: uid,
    );

    // Subscribe: capture roster emissions. The first is the current roster;
    // a later one must arrive after we add a member (driven by the realtime
    // re-hydrate, not by this call — the write goes through the cache too, but
    // the realtime path re-fires the same stream, so at least one post-write
    // emission containing the new member is guaranteed).
    final rosters = <List<HouseholdMember>>[];
    final sub = a.watchMembers().listen(rosters.add);
    await Future<void>.delayed(const Duration(seconds: 2)); // socket connect

    await a.addMember(displayName: 'Kid $_runId', kind: MemberKind.child);

    // Give logical replication + the WS round-trip time to deliver the event.
    final gotKid = await _waitFor(
      () => rosters.isNotEmpty &&
          rosters.last.any((m) => m.displayName == 'Kid $_runId'),
      timeout: const Duration(seconds: 10),
    );
    expect(gotKid, isTrue,
        reason: 'the live subscription re-emitted the new member');

    await sub.cancel();
  }, timeout: const Timeout(Duration(seconds: 30)));
}

/// Polls [predicate] until true or [timeout] elapses.
Future<bool> _waitFor(
  bool Function() predicate, {
  required Duration timeout,
}) async {
  final deadline = DateTime.now().add(timeout);
  while (DateTime.now().isBefore(deadline)) {
    if (predicate()) return true;
    await Future<void>.delayed(const Duration(milliseconds: 200));
  }
  return predicate();
}
  • Step 2: Confirm the guard skips cleanly (no creds)

Run (from packages/client_sdk/): fvm flutter test test/cloud/realtime_live_test.dart Expected: PASS with 1 skipped test ("realtime live smoke skipped — no SUPABASE_URL"). The normal suite is never broken by this file.

  • Step 3: Write the manual two-identity checklist

docs/decisions/2026-07-20-sp35-cloud-realtime-manual-checklist.md:

# SP3.5 Cloud Realtime — manual two-identity cross-device checklist (2026-07-20)

Headless tests cannot attach a SECOND identity's JWT with a publishable key
(same limitation documented across `packages/client_sdk/test/cloud/*_live_test.dart`),
so true cross-device visibility — the whole point of SP3.5 — is proven manually.
The automated proofs cover everything else: the fake-port adapter path
(`test/cached_realtime_test.dart`), the end-to-end Beat 2 path
(`test/cloud/realtime_beat2_test.dart`), and the single-identity live wire
(`test/cloud/realtime_live_test.dart`).

## Preconditions
- Migration `20260720000100_realtime_publication.sql` applied live to
  `bgedvvmihygwxhjxlvfu` (the five watched tables published + REPLICA IDENTITY FULL).
- Two sessions signed in to the SAME household: a **parent** (device A) and a
  **child** member (device B) — two devices, or two app instances / profiles.

## Beat 2 (the headline)
1. Device B (child): submit a chore that needs approval; leave the app open on
   the child's Today/companion surface.
2. Device A (parent): approve the child's submission.
3. **Expect on device B, with NO manual refresh:** the approval flips to
   approved and the Beat 2 celebration fires; the wallet balance increases by the
   granted tokens within a couple of seconds.

## Roster / chores live
4. Device A: add a new member and create a chore.
5. **Expect on device B:** the roster and chore list update live.

## Isolation (dual gate holds on the wire)
6. Sign a THIRD identity into a DIFFERENT household on a third session.
7. Repeat steps 1–5. **Expect:** the third session sees NONE of household 1's
   events — RLS scopes the realtime delivery, not just the REST reads.

## Teardown
8. Sign out on device B. **Expect:** no further events arrive (the channel is
   torn down via `Client.dispose()` on the data-mode/auth transition).
  • Step 4: Commit
git add packages/client_sdk/test/cloud/realtime_live_test.dart docs/decisions/2026-07-20-sp35-cloud-realtime-manual-checklist.md
git commit -m "test: guarded live realtime smoke + manual two-identity cross-device checklist"

Task 7: Docs refresh + graphify update + baseline verification​

Files:

  • Modify: app/test-gallery/authored/developer/architecture/ — the auth/infrastructure page that describes the cloud data path (the SP3 page now gains a "realtime" note).
  • Modify: docs/superpowers/specs/2026-06-24-sp3-cloud-data-adapter-design.md (flip the out-of-scope "Realtime → SP3.5" line to "delivered by SP3.5, see spec 2026-07-20").
  • Run: graphify update .

Interfaces:

  • Consumes: the shipped SP3.5 surfaces (Tasks 1–6).

  • Produces: current architecture docs + a fresh knowledge graph.

  • Step 1: Note SP3.5 as delivered in the SP3 spec

In docs/superpowers/specs/2026-06-24-sp3-cloud-data-adapter-design.md, under "Out of scope (tracked follow-ups)", change the first bullet from:

- **Realtime / cross-device live updates → SP3.5** (Supabase Realtime on watch* streams).

to:

- **Realtime / cross-device live updates → DELIVERED by SP3.5**
  (Supabase Realtime on the five watch* streams — see
  `docs/superpowers/specs/2026-07-20-sp35-cloud-realtime-design.md`).
  • Step 2: Refresh the architecture page

Locate the cloud-data-path architecture page: graphify query "architecture developer page cloud data path infrastructure auth" then add one line to the cloud data-path description noting that in DataMode.cloud the five watch* streams are now driven cross-device by Supabase Realtime (postgres_changes, RLS-scoped), re-hydrating the cache on an other-device commit. Keep it to the existing page's voice; do NOT invent a new page.

  • Step 3: Update the knowledge graph

Run (from repo root): graphify update . Expected: graph.json + GRAPH_REPORT.md regenerate (AST-only, no API cost).

  • Step 4: Verify baselines across all three suites

Run (from packages/client_sdk/): fvm flutter test Expected: PASS, ≥ 1114 tests (SP3.5 added: fake-port suite, adapter realtime suite, create_client realtime suite, Beat 2 suite, +1 skipped live).

Run (from app/): fvm flutter test Expected: PASS, ≥ 703 tests (unchanged — no app code changed; the new Client.dispose() interface method is not called on any shipped app path).

Run (from packages/design_system/): fvm flutter test Expected: PASS, ≥ 287 tests (unchanged — SP3.5 touches no DS code).

  • Step 5: Commit
git add docs/superpowers/specs/2026-06-24-sp3-cloud-data-adapter-design.md app/test-gallery/authored/developer/architecture graphify-out/graph.json graphify-out/GRAPH_REPORT.md
git commit -m "docs: mark SP3.5 realtime delivered + refresh architecture page + graphify"

Reminder: git add the exact files only — never -A. graphify-out/graph.json and GRAPH_REPORT.md ARE committed (per repo CLAUDE.md); the graphify cache + HTML view are gitignored. Never stage .superpowers/ or .claude/.

Task summary​

  1. Realtime publication migration (file-only): publish the five watched tables + REPLICA IDENTITY FULL.
  2. RealtimePort seam + SupabaseRealtimePort (postgres_changes) + FakeRealtimePort.
  3. CachedStorageAdapter: realtime event → coarse re-hydrate → cache re-emit + debounce/coalesce + dispose.
  4. createClient cloud wiring injects SupabaseRealtimePort + Client.dispose() teardown (local unchanged).
  5. Fake-driven end-to-end Beat 2 test (remote approval → device B watchApprovals/wallet live).
  6. Guarded live realtime smoke (skips without creds) + manual two-identity cross-device checklist.
  7. Docs refresh + graphify update . + baseline verification (SDK 1114 / app 703 / DS 287).