Merge nucleic/plucky-opal-weasel into dev

2026-07-11 02:28:09 -07:00
parent c594923940
commit 2425f6ffdd
5 changed files with 289 additions and 3 deletions
@@ -0,0 +1,47 @@
+import Foundation
+import NucleicProtocol
+
+/// The on-device snapshot of the phone's mesh-cast ledger (`CastLedger`) — the activity feed,
+/// mesh messages, and runner-presence cards it has received, plus (implicitly) the cursors a
+/// reconnect subscribes with. Like `SessionCache`, a *projection cache*, never canonical: a
+/// live host always supersedes it; it exists so a relaunch renders the feed instantly and
+/// resubscribes O(gap) instead of re-pulling every channel from zero.
+///
+/// One JSON file under Application Support (`NucleicCastCache/casts.json`) — the ledger's
+/// per-channel retention already bounds it. Reads/writes hop onto a utility `Task.detached`.
+enum CastCache {
+    private static var root: URL? {
+        guard let base = try? FileManager.default.url(
+            for: .applicationSupportDirectory, in: .userDomainMask, appropriateFor: nil, create: true)
+        else { return nil }
+        return base.appendingPathComponent("NucleicCastCache", isDirectory: true)
+    }
+    private static var castsURL: URL? { root?.appendingPathComponent("casts.json") }
+
+    /// Load the persisted ledger — called once at launch, before any host connects. The ledger's
+    /// init re-applies retention, so a policy change between builds re-caps on load.
+    static func loadLedger() -> CastLedger {
+        guard let url = castsURL, let data = try? Data(contentsOf: url),
+              let casts = try? JSONDecoder().decode([WireCast].self, from: data)
+        else { return CastLedger() }
+        return CastLedger(casts: casts)
+    }
+
+    /// Persist the ledger's snapshot. Off the main actor; callers debounce bursts.
+    static func save(_ ledger: CastLedger) async {
+        let casts = ledger.allCasts
+        await Task.detached(priority: .utility) {
+            guard let root, let url = castsURL else { return }
+            try? FileManager.default.createDirectory(at: root, withIntermediateDirectories: true)
+            if let data = try? JSONEncoder().encode(casts) {
+                try? data.write(to: url, options: .atomic)
+            }
+        }.value
+    }
+
+    /// Drop the cache — on full unpair (no Mac left whose casts this device should hold).
+    static func clear() {
+        guard let root else { return }
+        try? FileManager.default.removeItem(at: root)
+    }
+}
@@ -116,6 +116,14 @@ final class HostConnection {
        /// The outcome of a `createProject` this phone sent (CLOUD_RUNTIME §4.3), correlated by
        /// requestID — settle the Add Project sheet (the project row rides the dashboard push).
        var projectCreated: (WireProjectCreated) -> Void = { _ in }
+        /// The cursors this connection's `castSubscribe` carries (mesh casting) — the merged
+        /// ledger's tips across EVERY origin, so any connected Mac tops up any origin's gap.
+        var castCursors: () -> [CastCursor] = { [] }
+        /// A live batch of mesh casts from this host — RemoteStore applies it to the merged
+        /// ledger (identity dedup absorbs the same cast arriving via N Macs).
+        var castsReceived: ([WireCast]) -> Void = { _ in }
+        /// One catch-up batch for one (origin, channel) of the cast feed.
+        var castCatchUp: (CastCatchUp) -> Void = { _ in }
    }
    private let callbacks: Callbacks

@@ -402,7 +410,9 @@ final class HostConnection {
            clientCaps: WireClientCapabilities(
                mesh: 1, canSyncRoster: true,
                canProvideIntelligence: PhoneIntelligenceExecutor.isSupported,
-                intelligenceProfile: PhoneIntelligenceExecutor.profile))
+                intelligenceProfile: PhoneIntelligenceExecutor.profile,
+                // Mesh casting: receive the activity feed / messages / runner presence.
+                canCast: true))
        self.client = client
        consume(client, pairingPayload: pairingPayload)
    }
@@ -504,6 +514,12 @@ final class HostConnection {
            callbacks.didUpdate()
            send(.listSessions)
            send(.listDashboard)
+            // Mesh casting: subscribe with the merged ledger's cursors (every origin we hold) —
+            // this host serves the gaps for its own casts AND any peer's casts it mirrors, so a
+            // phone converges even when a cast's origin is offline.
+            if welcome.capabilities.canCast {
+                send(.castSubscribe(CastSubscribe(cursors: callbacks.castCursors())))
+            }
            // Warm-resubscribe from what we already have, so a reconnect on a long transcript replays
            // only the gap rather than snapping back to the host's 200-event tail.
            if let id = openSessionID {
@@ -661,6 +677,16 @@ final class HostConnection {
                let result = await PhoneIntelligenceExecutor.execute(request)
                self?.send(.intelligenceResult(result))
            }
+        case .chatStarted:
+            // Mesh-dispatch ack — the phone doesn't drive mesh dispatch yet (the Mac composer
+            // does); inert until the iOS "Auto (Mesh)" destination lands.
+            break
+        case .casts(let batch):
+            // Live mesh casts — up to the merged ledger (identity dedup across N connections).
+            callbacks.castsReceived(batch.casts)
+        case .castCatchUp(let catchUp):
+            // One catch-up batch for one (origin, channel) — `reset` handling lives in the ledger.
+            callbacks.castCatchUp(catchUp)
        case .credentialNeeded, .credentialUpdate,
             // The owner's runner-pool credential (item 4) — inert until the phone grows a
             // pool-management surface; Macs are the managers today.
@@ -58,6 +58,42 @@ final class RemoteStore: ObservableObject {
    private var cachedSummaries: [WireSessionSummary] = SessionCache.loadSummaries()
    /// Debounced disk-write tasks for the offline cache (coalesce a burst of updates into one write).
    private var summaryPersistTask: Task<Void, Never>?
+
+    /// The merged mesh-cast ledger (mesh casting) — the activity feed / messages / runner
+    /// presence from every origin, deduped on cast identity across all connected Macs. Seeded
+    /// from disk at launch (renders offline, and its tips are the reconnect cursors), persisted
+    /// debounced on change. Feed/dispatch surfaces read projections of this.
+    @Published private(set) var castLedger: CastLedger = CastCache.loadLedger()
+    private var castPersistTask: Task<Void, Never>?
+
+    /// The merged fleet feed, oldest-first — the "activity" channel decoded across every origin.
+    /// Seeded from the ledger on appear (renders offline), appended by `applyCasts`.
+    @Published private(set) var activityFeed: [ActivityFeedItem] = []
+    /// When the user last viewed the feed — rows newer than this count as unread.
+    @Published private(set) var feedLastSeenAt: Date =
+        UserDefaults.standard.object(forKey: "nucleic.feed.lastSeenAt") as? Date ?? .distantPast
+    var feedUnreadCount: Int {
+        activityFeed.reduce(0) { $0 + ($1.event.at > feedLastSeenAt ? 1 : 0) }
+    }
+    func markFeedSeen() {
+        feedLastSeenAt = Date()
+        UserDefaults.standard.set(feedLastSeenAt, forKey: "nucleic.feed.lastSeenAt")
+    }
+
+    /// Apply freshly received casts (live or catch-up) and persist if anything changed.
+    private func applyCasts(_ apply: (inout CastLedger) -> [WireCast]) {
+        var ledger = castLedger
+        let fresh = apply(&ledger)
+        guard !fresh.isEmpty else { return }
+        castLedger = ledger
+        activityFeed = ActivityFeedItem.merged(activityFeed, adding: fresh)
+        castPersistTask?.cancel()
+        castPersistTask = Task { [ledger] in
+            try? await Task.sleep(for: .milliseconds(500))
+            guard !Task.isCancelled else { return }
+            await CastCache.save(ledger)
+        }
+    }
    private var transcriptPersistTask: Task<Void, Never>?
    @Published private(set) var capabilities = WireCapabilities(canModifyToolInput: false, allowAlwaysScopes: [])
    @Published private(set) var grantedScope: DeviceScope = .approve
@@ -460,6 +496,10 @@ final class RemoteStore: ObservableObject {
        if isPaired {
            // Show the saved chat history immediately, before any host connects.
            if sessions.isEmpty { sessions = cachedSummaries }
+            // …and the cached fleet feed (mesh casting) — live catch-up merges on top.
+            if activityFeed.isEmpty {
+                activityFeed = ActivityFeedItem.merged([], adding: castLedger.allCasts)
+            }
            // Assume connected while the first handshake completes, so launch doesn't open on a
            // grayed-out "Disconnected" shell for the second it takes to come up.
            beginOptimisticWindow()
@@ -924,6 +964,17 @@ final class RemoteStore: ObservableObject {
            IdentityStore.savePairedHost(host)
            self.rebuildAggregate()
        }
+        cb.castCursors = { [weak self] in
+            // Mesh casting: subscribe with the MERGED ledger's tips (every origin), so this host
+            // serves only the gaps — including gaps for origins it merely mirrors.
+            self?.castLedger.cursors ?? []
+        }
+        cb.castsReceived = { [weak self] casts in
+            self?.applyCasts { $0.apply(casts) }
+        }
+        cb.castCatchUp = { [weak self] batch in
+            self?.applyCasts { $0.applyCatchUp(batch) }
+        }
        cb.meshRosterChanged = { [weak self] in
            // Mesh "join": a Mac was learned or revoked via gossip. Reconnect to every paired Mac
            // (connecting the newcomer) and drop any that left — without switching the active host.
@@ -1318,6 +1369,9 @@ final class RemoteStore: ObservableObject {
        // No Mac left whose history to hold — drop the offline cache too.
        cachedSummaries = []
        SessionCache.clear()
+        castLedger = CastLedger()
+        activityFeed = []
+        CastCache.clear()
        sessions = []
        dashboard = .empty
        // Nothing left to dial — spin the embedded Tailscale node down if it was running.
@@ -1833,7 +1887,10 @@ final class RemoteStore: ObservableObject {
             .registerLiveActivity, .endLiveActivity, .registerPushToStartToken, .setForeground,
             .transferOffer, .transferChunk, .transferCommit, .transferCancel, .fetchTranscript,
             .requestPairingCode, .cancelPairingCode, .respondMacPair,
-             .intelligenceResult, .credentialManifest, .credentialProvision, .createProject:
+             .intelligenceResult, .credentialManifest, .credentialProvision, .createProject,
+             // Cast subscriptions are per-connection (each `HostConnection` subscribes on its
+             // own ready, with the merged ledger's cursors) — nothing routes them here.
+             .castSubscribe:
            break
        }
    }
@@ -2024,7 +2081,9 @@ final class RemoteStore: ObservableObject {
             // Antimatter runner verbs (ANTIMATTER_RUNNER §5–6) — demo has no runner host.
             .intelligenceResult, .credentialManifest, .credentialProvision,
             // Remote project creation (CLOUD_RUNTIME §4.3) — demo has no host to clone on.
-             .createProject:
+             .createProject,
+             // Mesh casting (demo has no host to cast).
+             .castSubscribe:
            break  // passive / already handled by the seeded fixtures (demo has no mesh peers)
        }
    }
@@ -0,0 +1,150 @@
+import SwiftUI
+import NucleicProtocol
+
+/// The phone's fleet activity feed (mesh casting): notable events from every Mac and runner —
+/// finished/needs-input turns, approvals, ships, transfers, dispatches — merged across origins
+/// and surviving late joins (an hour-later app open replays the gap). A passive history
+/// surface: it never alerts; rows deep-link to their session via the notification route.
+///
+/// `RecentActivitySection` is the Home dashboard's three-row peek; `ActivityFeedListView` is
+/// the pushed full list.
+struct RecentActivitySection: View {
+    @EnvironmentObject var store: RemoteStore
+
+    var body: some View {
+        if !store.activityFeed.isEmpty {
+            VStack(alignment: .leading, spacing: 10) {
+                HStack {
+                    Label("Recent activity", systemImage: "bell")
+                        .font(.headline)
+                    Spacer()
+                    NavigationLink {
+                        ActivityFeedListView()
+                    } label: {
+                        Text("See all")
+                            .font(.subheadline)
+                    }
+                }
+                VStack(spacing: 0) {
+                    ForEach(store.activityFeed.suffix(3).reversed()) { item in
+                        ActivityFeedRow(item: item)
+                        if item.id != store.activityFeed.suffix(3).first?.id {
+                            Divider().padding(.leading, 30)
+                        }
+                    }
+                }
+            }
+            .frame(maxWidth: .infinity, alignment: .leading)
+            .card()
+            .onAppear { store.markFeedSeen() }
+        }
+    }
+}
+
+struct ActivityFeedListView: View {
+    @EnvironmentObject var store: RemoteStore
+
+    var body: some View {
+        List(store.activityFeed.reversed()) { item in
+            ActivityFeedRow(item: item)
+                .listRowInsets(EdgeInsets(top: 8, leading: 16, bottom: 8, trailing: 16))
+        }
+        .listStyle(.plain)
+        .navigationTitle("Activity")
+        .navigationBarTitleDisplayMode(.inline)
+        .overlay {
+            if store.activityFeed.isEmpty {
+                ContentUnavailableView(
+                    "No fleet activity yet", systemImage: "bell.slash",
+                    description: Text("Finished chats, approvals, ships, and transfers from every Mac and runner land here."))
+            }
+        }
+        .onAppear { store.markFeedSeen() }
+    }
+}
+
+private struct ActivityFeedRow: View {
+    @EnvironmentObject var store: RemoteStore
+    let item: ActivityFeedItem
+
+    private var event: FeedEvent { item.event }
+
+    var body: some View {
+        Button {
+            // Route like a notification tap: jump to the Sessions tab and push the session on
+            // its owning host (works for any mesh host's session the phone mirrors).
+            store.route(to: event.sessionID)
+        } label: {
+            HStack(alignment: .top, spacing: 10) {
+                Image(systemName: Self.icon(for: event.kind))
+                    .foregroundStyle(Self.tint(for: event.kind))
+                    .font(.subheadline)
+                    .frame(width: 20)
+                    .padding(.top, 2)
+                VStack(alignment: .leading, spacing: 2) {
+                    HStack(spacing: 6) {
+                        Text(event.sessionTitle.isEmpty ? "Untitled chat" : event.sessionTitle)
+                            .font(.subheadline.weight(.medium))
+                            .foregroundStyle(.primary)
+                            .lineLimit(1)
+                        Spacer(minLength: 4)
+                        Text(event.at, format: .relative(presentation: .named))
+                            .font(.caption2)
+                            .foregroundStyle(.tertiary)
+                    }
+                    Text(subtitle)
+                        .font(.caption)
+                        .foregroundStyle(.secondary)
+                        .lineLimit(2)
+                }
+            }
+            .padding(.vertical, 6)
+            .contentShape(Rectangle())
+        }
+        .buttonStyle(.plain)
+    }
+
+    private var subtitle: String {
+        var parts: [String] = [Self.headline(for: event)]
+        if !event.projectName.isEmpty { parts.append(event.projectName) }
+        if !event.hostName.isEmpty { parts.append(event.hostName) }
+        return parts.joined(separator: " · ")
+    }
+
+    static func headline(for event: FeedEvent) -> String {
+        switch event.kind {
+        case .finished: "Finished"
+        case .needsInput: "Waiting for you"
+        case .needsApproval: event.detail.map { "Needs approval: \($0)" } ?? "Needs approval"
+        case .shipped: event.detail.map { "Shipped · \($0)" } ?? "Shipped"
+        case .transferredOut: "Moved to \(event.counterpartHostName ?? "another Mac")"
+        case .transferredIn: "Arrived from \(event.counterpartHostName ?? "another Mac")"
+        case .error: event.detail.map { "Error: \($0)" } ?? "Error"
+        case .dispatched: "Dispatched here via Mesh"
+        default: event.kind.rawValue  // a newer build's kind — render generically, never drop
+        }
+    }
+
+    static func icon(for kind: FeedEvent.Kind) -> String {
+        switch kind {
+        case .finished: "checkmark.circle"
+        case .needsInput: "bubble.left.and.exclamationmark.bubble.right"
+        case .needsApproval: "hand.raised"
+        case .shipped: "arrow.triangle.merge"
+        case .transferredOut: "arrow.up.right.square"
+        case .transferredIn: "arrow.down.left.square"
+        case .error: "exclamationmark.triangle"
+        case .dispatched: "paperplane"
+        default: "circle"
+        }
+    }
+
+    static func tint(for kind: FeedEvent.Kind) -> Color {
+        switch kind {
+        case .finished, .shipped: .green
+        case .needsInput, .needsApproval: .orange
+        case .error: .red
+        default: .secondary
+        }
+    }
+}
@@ -97,6 +97,10 @@ struct HomeView: View {

                    if !running.isEmpty { InProgressSessions(sessions: running) }

+                    // Fleet activity (mesh casting): the merged cross-host feed's newest three
+                    // rows, with "See all" pushing the full list. Renders offline from the cache.
+                    RecentActivitySection()
+
                    QuickTodos()
                }
                .padding()