Skip to content
Original file line number Diff line number Diff line change
Expand Up @@ -241,15 +241,15 @@ struct HomeLiveBook {
if ended.count > Self.endedLimit { ended.removeFirst(ended.count - Self.endedLimit) }
}

/// What each conversation shows: its most recently touched run with reply
/// text or an activity line.
var shown: [ConversationID: HomeLiveRun] {
var best: [ConversationID: (Key, Run)] = [:]
for (key, run) in runs where run.activity != nil || run.hasTalk {
if let current = best[key.conversation], current.1.touched >= run.touched { continue }
best[key.conversation] = (key, run)
/// What `conversation` shows: its most recently touched run with reply text or an activity
/// line, nil when none. Only that conversation's runs are read and joined.
func shown(in conversation: ConversationID) -> HomeLiveRun? {
var best: (Key, Run)?
for (key, run) in runs where key.conversation == conversation && (run.activity != nil || run.hasTalk) {
if let current = best, current.1.touched >= run.touched { continue }
best = (key, run)
}
return best.mapValues { liveRun($0.0, $0.1) }
return best.map { liveRun($0.0, $0.1) }
}

/// Every run of `conversation` with reply text, in the order the runs began: each is one
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import Foundation

/// What a whole-message Home view draws of one conversation's running replies: who writes each
/// one (its typing dots until its message posts), and the Chief run's activity line and thought
/// headline. Never the text: the view shows a reply only when it is posted whole
/// (`HomeReplyDisplay.wholeMessages`), so a talk token leaves this value as it was and wakes no
/// such view (HOT PATH AUDIT area F, handoff/hot-path-audit/stream/AUDIT.md row H1).
public struct HomeRunShape: Hashable, Sendable {
/// One running reply: the key its posted message will carry, and its author.
public struct Reply: Hashable, Sendable {
public var turn: IdempotencyKey
public var author: ParticipantID

public init(turn: IdempotencyKey, author: ParticipantID) {
self.turn = turn
self.author = author
}
}

/// The running replies in the order their turns began (``HomeStore/streams(in:)`` without text).
public var replies: [Reply]
/// The author of the run ``HomeStore/liveRun(in:)`` shows, nil when none.
public var runner: ParticipantID?
/// That run's activity line.
public var activity: HomeLiveActivity?
/// That run's thought headline.
public var thought: String?

public init(replies: [Reply] = [], runner: ParticipantID? = nil, activity: HomeLiveActivity? = nil,
thought: String? = nil) {
self.replies = replies
self.runner = runner
self.activity = activity
self.thought = thought
}

/// The shape of `run` and `streams`.
public init(run: HomeLiveRun?, streams: [HomeStreamRow]) {
self.init(replies: streams.map { Reply(turn: $0.turn, author: $0.author) }, runner: run?.participant,
activity: run?.activity, thought: run?.thought)
}

/// Nothing runs.
public var isEmpty: Bool { replies.isEmpty && runner == nil }
}
Original file line number Diff line number Diff line change
Expand Up @@ -45,15 +45,22 @@ extension HomeStore {
if changed { publishLive(conversation) }
}

/// After the live or draft book changed for `conversation`: `live`, `drafts` and that
/// conversation's `streams` take the books' values (each assigned only when it changed, so an
/// unchanged one wakes no observer).
/// After the live or draft book changed for `conversation`: that conversation's `live`,
/// `streams` and `runShapes`, and `drafts`, take the books' values (each assigned only when it
/// changed, so an unchanged one wakes no observer). Every change to either book is to one
/// conversation, so the others are left as they are: a token costs its own conversation's
/// reply, not every running reply of every conversation.
// perf-hot: 0.1ms home-stream-token (every streamed token of a Chief reply)
func publishLive(_ conversation: ConversationID) {
let shown = liveBook.shown
if live != shown { live = shown }
let run = liveBook.shown(in: conversation)
if live[conversation] != run { live[conversation] = run }
if drafts != draftBook.live { drafts = draftBook.live }
let rows = streamRows(conversation)
if streams[conversation, default: []] != rows { streams[conversation] = rows.isEmpty ? nil : rows }
let shape = HomeRunShape(run: run, streams: rows)
// Compared as stored: no run is no entry (an empty shape over none would wake every view).
let stored = shape.isEmpty ? nil : shape
if runShapes[conversation] != stored { runShapes[conversation] = stored }
}

/// Every running reply of `conversation` in begin order: the Chief live runs, then a reply
Expand All @@ -72,6 +79,9 @@ extension HomeStore {
/// The running Chief run `id` shows (`live`), nil when none.
public func liveRun(in id: ConversationID) -> HomeLiveRun? { live[id] }

/// What a whole-message view of `id` draws of its running replies (``runShapes``).
public func runShape(in id: ConversationID) -> HomeRunShape { runShapes[id] ?? HomeRunShape() }

/// Who shows typing dots in `id`: the owner's typing frames, plus the author of a run still
/// running there. One rule for every view: typing frames do not outlive a connection
/// (`handle(.connection)`), a running turn's dots come from its live run.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,10 @@ public final class HomeStore {
public internal(set) var live: [ConversationID: HomeLiveRun] = [:]
/// Every running reply per conversation, in the order its turn began (``HomeTimeline``).
public internal(set) var streams: [ConversationID: [HomeStreamRow]] = [:]
/// What a whole-message view draws of each conversation's running replies (``HomeRunShape``):
/// who writes, the activity line and the thought headline, never the text. A talk token changes
/// `live` and `streams` and leaves this as it was, so a view that does not draw the text sleeps.
public internal(set) var runShapes: [ConversationID: HomeRunShape] = [:]
/// `transcript(for:)` of each conversation at a `transcriptVersion` (a streamed chunk does
/// not map the window again).
@ObservationIgnored var transcriptCache: [ConversationID: (version: Int, me: ParticipantID, items: [TranscriptItem])] = [:]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,19 +8,31 @@ extension HomeProjection {
/// streaming): a change to another conversation costs one ``HomeProjectionInputs`` compare,
/// and a pass that falls in a keystroke's display frame waits for the next frame (one pass
/// then, with the whole current value: `refresh` reads it all, so nothing is lost).
///
/// Whole messages (the default) draw no reply text while it streams: the view reads the
/// replies' shape (``HomeRunShape``: who writes, the activity line and the thought headline),
/// which a talk token does not change, so a token wakes no whole-message view (HOT PATH AUDIT
/// area F, row H1: one O(transcript) pass per token in every open Home view before).
// perf-hot: 0.1ms home-stream-token (each store change of the open conversation; no wake for a talk token)
func observe() {
guard !stopped else { return }
let id = conversation
let text = core.replies == .streaming
withObservationTracking {
_ = homeStore.transcriptVersion[id, default: 0] // // crash program: a dictionary read the ratchet can type
_ = homeStore.typing[id, default: []]
_ = homeStore.streams(in: id) // the running replies grow their bubbles
_ = homeStore.liveRun(in: id) // a Chief's live run: its reply so far and activity line
if text {
_ = homeStore.streams(in: id) // the running replies grow their bubbles
_ = homeStore.liveRun(in: id) // a Chief's live run: its reply so far and activity line
} else {
_ = homeStore.runShape(in: id) // who writes, the activity line and the thought headline
}
_ = homeStore.isConfirmed(id) // a send held for the owner goes once it names this conversation
_ = homeStore.rows
} onChange: { [weak self] in
Task { @MainActor [weak self] in
guard let self, !self.stopped else { return }
self.observedWakes += 1
self.observedPass()
self.observe()
}
Expand All @@ -30,7 +42,7 @@ extension HomeProjection {
/// The observed store changed: refresh when this conversation's inputs did, after a keystroke's frame.
func observedPass() {
guard !heldPass else { return } // the held pass reads the newest value
let now = HomeProjectionInputs(homeStore, conversation)
let now = HomeProjectionInputs(homeStore, conversation, replies: core.replies)
guard now != observedInputs else { return }
if controller.inKeystrokeFrame {
heldPass = true
Expand All @@ -47,6 +59,6 @@ extension HomeProjection {

/// The inputs `start`'s first refresh read (the first observed change compares against them).
func noteObservedInputs() {
observedInputs = HomeProjectionInputs(homeStore, conversation)
observedInputs = HomeProjectionInputs(homeStore, conversation, replies: core.replies)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,8 @@ final class HomeProjection: @preconcurrency ChatIntents {
private(set) var appliedUpdates = 0
/// Store reads of this conversation (`refresh`), changed or not (tests: traffic for other conversations costs none).
private(set) var refreshPasses = 0
/// Store observation wakes (HomeProjection+Observe.swift), for tests: a talk token wakes no whole-message view.
var observedWakes = 0
/// Bench: main-thread ms of each step of the projection passes since it was set to `[:]`,
/// summed per step name (`passStep`, HomeStateChangePerfTests); nil, the default, records nothing.
var passLog: [String: Double]?
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,29 @@ struct HomeProjectionInputs: Equatable {
var summary: ConversationSummary?
var row: InboxRow?
var typing: Set<ParticipantID>?
/// The running replies as drawn: their text when the view grows stream bubbles, else only their
/// shape (a whole-message view draws no reply text, so a talk token changes nothing it reads).
var streams: [HomeStreamRow]
var live: HomeLiveRun?
var shape: HomeRunShape?
var confirmed: Bool
var hasOlder: Bool

/// Reads `conversation`'s inputs; every read is a constant-time lookup except the row (one pass over the inbox).
@MainActor
init(_ store: HomeStore, _ conversation: ConversationID) {
init(_ store: HomeStore, _ conversation: ConversationID, replies: HomeReplyDisplay) {
me = store.me?.id
transcriptVersion = store.transcriptVersion[conversation, default: 0]
summary = store.summary(conversation)
row = store.rows.first { $0.summary.id == conversation }
typing = store.typing.value(for: conversation)
streams = store.streams(in: conversation)
live = store.liveRun(in: conversation)
if replies == .streaming {
streams = store.streams(in: conversation)
live = store.liveRun(in: conversation)
} else {
streams = []
shape = store.runShape(in: conversation)
}
confirmed = store.isConfirmed(conversation)
hasOlder = store.hasOlderMessages(in: conversation)
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ final class HomeChatRig {
let me: ParticipantID
let chief: ParticipantID

init(history: Int = 40, width: CGFloat = 700, height: CGFloat = 900) async throws {
init(history: Int = 40, width: CGFloat = 700, height: CGFloat = 900, replies: HomeReplyDisplay = .streaming) async throws {
// The mock owner's own answer to a send comes after the test ends.
source = MockHomeSource(options: MockHomeSource.Options(chiefHistory: history, latency: .zero, replyDelay: .seconds(3600)))
store = HomeStore(source: source)
Expand All @@ -36,8 +36,9 @@ final class HomeChatRig {
chief = await source.chief.id
window = NSWindow(contentRect: NSRect(x: 0, y: 0, width: width, height: height), styleMask: [.titled], backing: .buffered, defer: true)
window.isReleasedWhenClosed = false
// The order, property and stress tests check the stream bubbles (HomeReplyDisplay.streaming).
view = MessagesLabHomeView(store: store, conversation: Self.conversation, me: me, wake: NoWake(), replies: .streaming)
// The order, property and stress tests check the stream bubbles (HomeReplyDisplay.streaming);
// the token cost tests read the app's default (whole messages).
view = MessagesLabHomeView(store: store, conversation: Self.conversation, me: me, wake: NoWake(), replies: replies)
view.frame = NSRect(x: 0, y: 0, width: width, height: height)
window.contentView = view
view.layoutSubtreeIfNeeded()
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import AppKit
@testable import CmuxHomeCore
import Foundation
import Testing
@testable import MessagesLabHome

/// A streamed token costs Home's main thread nothing it cannot show (HOT PATH AUDIT, area F,
/// handoff/hot-path-audit/stream/AUDIT.md). Home's default is whole messages
/// (`HomeReplyDisplay.wholeMessages`, 0c0b107df48): while the Chief's reply streams the person
/// sees the Chief's typing dots and its activity line, and the reply only when it is posted. So a
/// talk token changes nothing on screen, and the floor of its main-thread work is the store's
/// append of the token: no projection pass, in any open Home view, however long the transcript.
///
/// The real view (`MessagesLabHomeView`) over a real `HomeStore` on the mock owner with a
/// 2,000-message history, the Chief's live run driven through the store's live path
/// (`chief-live-v1` deltas, `HomeLiveBook`), one view and then four views of the conversation open
/// (four Home tabs). Each token is the store's event, then each open view's observed pass (what
/// its store observation runs on the next main turn). `HOMETOKEN` lines carry the main-thread
/// time per token.
@MainActor @Suite(.serialized) struct HomeWholeMessageTokenCostTests {
static let tokens = 1_500

private static func line(_ name: String, _ xs: [Double]) -> String {
String(format: "HOMETOKEN %@ n %d p50 %.4f p95 %.4f p99 %.4f max %.3f ms", name, xs.count,
HomeStreamingPerfTests.p(xs, 0.5), HomeStreamingPerfTests.p(xs, 0.95),
HomeStreamingPerfTests.p(xs, 0.99), xs.max() ?? 0)
}

@Test(arguments: [1, 4])
func aTalkTokenRunsNoProjectionPassInWholeMessageMode(views count: Int) async throws {
let rig = try await HomeChatRig(history: 2_000, replies: .wholeMessages)
defer { rig.stop() }
var extra: [MessagesLabHomeView] = []
for _ in 1..<count {
let view = MessagesLabHomeView(store: rig.store, conversation: HomeChatRig.conversation, me: rig.me,
wake: HomeChatRig.NoWake(), replies: .wholeMessages)
view.frame = NSRect(x: 0, y: 0, width: 700, height: 900)
view.layoutSubtreeIfNeeded()
extra.append(view)
}
defer { for view in extra { view.stop() } }
let views = [rig.view] + extra
// The run starts: its first token puts up the Chief's typing dots (one pass per view).
rig.delta("token_cost", seq: 0, "\(HomeChatRig.tag)Here is the plan.")
for view in views { view.projection.observedPass() }
rig.frame()
// The views' store observations run in their own main-actor tasks.
for _ in 0..<4 { await Task.yield() }
let before = views.map(\.projection.refreshPasses)
let wakesBefore = views.map(\.projection.observedWakes)
let words = HomeStreamingPerfTests.replyTokens(count: Self.tokens)
var store: [Double] = [], passes: [Double] = [], total: [Double] = []
for (index, word) in words.enumerated() {
let t0 = CACurrentMediaTime()
rig.delta("token_cost", seq: UInt64(index + 1), word)
let t1 = CACurrentMediaTime()
// What a wake costs each view (its observation runs this pass when the token wakes it).
for view in views { view.projection.observedPass() }
let t2 = CACurrentMediaTime()
store.append((t1 - t0) * 1_000)
passes.append((t2 - t1) * 1_000)
total.append((t2 - t0) * 1_000)
// The real path: each view's store observation, if the token woke it.
for _ in 0..<3 { await Task.yield() }
}
let after = views.map(\.projection.refreshPasses)
let ran = zip(after, before).map { $0 - $1 }.reduce(0, +)
let woke = zip(views.map(\.projection.observedWakes), wakesBefore).map { $0 - $1 }.reduce(0, +)
let reply = rig.store.liveRun(in: HomeChatRig.conversation)?.text.utf8.count ?? 0
print(Self.line("views=\(count) store", store))
print(Self.line("views=\(count) passes", passes))
print(Self.line("views=\(count) total", total))
print("HOMETOKEN views=\(count) tokens \(Self.tokens) refreshPasses \(ran) wakes \(woke) replyBytes \(reply) history \(rig.store.transcript(for: HomeChatRig.conversation).count)")
#expect(woke == 0, "\(woke) store observation wakes for \(Self.tokens) talk tokens in \(count) whole-message view(s)")
#expect(ran == 0, "\(ran) projection passes for \(Self.tokens) talk tokens in \(count) whole-message view(s): none shows a token")
}

/// A turn that starts with thinking (most Chief turns): before the run has reply text or an
/// activity line nothing shows for it, so its thought tokens wake no view either
/// (review of #33945: an empty shape was written over no shape on every thought token).
@Test func thoughtTokensBeforeAnyReplyWakeNoWholeMessageView() async throws {
let rig = try await HomeChatRig(history: 200, replies: .wholeMessages)
defer { rig.stop() }
rig.frame()
for _ in 0..<4 { await Task.yield() }
let wakesBefore = rig.view.projection.observedWakes
for seq in 0..<200 {
rig.store.handle(.live(.delta(HomeLiveDelta(
conversation: HomeChatRig.conversation, author: rig.chief, run: "thinking_first", step: "thought",
isTalk: false, seq: UInt64(seq), text: seq == 0 ? "Reading the branch" : " and more", isFinal: false))))
for _ in 0..<3 { await Task.yield() }
}
let woke = rig.view.projection.observedWakes - wakesBefore
print("HOMETOKEN thought tokens 200 wakes \(woke)")
#expect(woke == 0, "\(woke) store observation wakes for 200 thought tokens before any reply text")
}
}
5 changes: 5 additions & 0 deletions scripts/cmux-next/perf-hot-benches.json
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,11 @@
"what": "one acpmux hub record seen by an open _acpmux/wait: the session filter and the streamed-chunk skip; a chunk never evaluates the wait's rows",
"bench": "cmux-tui-rust-check.sh test a_streamed_token_does_not_evaluate_an_open_wait (WAIT-STREAM line: evaluations per 500 tokens), then scripts/measure/acpmux-overhead.sh bench=daemon (STRESS token to client) on the fleet",
"owner": "PL-acpmux (HOT PATH AUDIT area F: hpa-stream)"
},
"home-stream-token": {
"what": "one streamed token of a Chief reply on the main thread: HomeLiveBook append, publishLive of that conversation, and each open Home view's store observation (whole-message views are not woken by a talk token)",
"bench": "HomeWholeMessageTokenCostTests (HOMETOKEN lines: ms per token, wakes and passes, 1 and 4 views, 2,000 messages) and HomeStreamingPerfTests (streaming mode), run on the fleet (package-test-lane.sh suite Packages/Shared/CmuxMessagesLab)",
"owner": "hpa-stream (HOT PATH AUDIT area F)"
}
}
}
Loading