Merge app/v3 into workstreams/main (sync 5)

PERF.md

@@ -379,3 +379,25 @@ always coordinate through it, so no build leaves it out. On linux/amd64 it adds
 3,379,200 bytes to the stripped binary (75,755,785 to 79,134,985 on the same dev
 base), and the cap moves by the same amount, from 75,811,848 to 79,191,048, so
 the 56,063 bytes of headroom the base had are kept.
+
+## Usage writer identity startup
+
+`TestUsageMillionRowLedgerFlush` constructs 1,000,000 unique historical call
+identities in 185,000,000 bytes before measuring the public writer. Initialization
+streams valid full rows and retains their identities rather than a decoded row
+slice. Enqueue through flush has a 256 MiB allocation budget. The existing
+two-second flush contract must complete the new $0.002 call; replaying that call
+and historical identities adds no rows. An independent public ledger walk checks
+1,000,001 records, exact consumed bytes and $1000.002 total after the measurement.
+Elapsed times are diagnostic. The retained identity set still grows with history.
+
+
+## Usage writer overflow draining
+
+`BenchmarkLedgerWriter20K` times 20,000 distinct charges from `RecordUsage` to
+the last completed writer callback. Outside the timer it checks the flush,
+20,000 public ledger rows, $200 total and no drops. Compare the same harness
+on both revisions, with and without race instrumentation; run with `-run '^$'`.
+The writer batches at most 128 already admitted rows per ledger lock. A quiet
+queue flushes immediately, and flush/close barriers follow completed writes.
+Identity generation checks and duplicate suppression still run under that lock.

cmd/codeaf/app.go

@@ -180,6 +180,7 @@ func runApp(args []string) error {
 		backend.turnLong = *turnLong
 	}
 	defer backend.watchSpendRows()()
+	defer backend.watchAgentSpend()()
 	// The crews are watched from the start: a manager's start or a member
 	// woken by its manager must not wait for the window to ask for crews.
 	backend.memoryStore = func() (*store.Store, error) {
@@ -588,6 +589,9 @@ type appBackend struct {
 	// stopProgress the factory runner's progress news (appfactory_progress.go).
 	stopPhases   func()
 	stopProgress func()
+	// stopStages and stepStreams: the factory steps' events (appfactory_stepstream.go).
+	stopStages  func()
+	stepStreams map[string]*appStepStream
 	// briefs takes the progress news of this machine's factory's jobs as
 	// writes of their briefs (appfactory_brief_writer.go).
 	briefs *briefDebounce
@@ -629,8 +633,8 @@ type appBackend struct {
 	turns         atomic.Uint64
 	assembled     atomic.Bool
 
-	// unreadSay is held from working an agent's unread out to saying it,
-	// so the last unread the window hears is the last one worked out
+	// unreadSay is held from working an agent's unread and saved spend out
+	// to saying them, so the last readings the window hears are the latest
 	// (appunread.go). It is taken before mu and the app file's lock, never
 	// under them.
 	unreadSay sync.Mutex
@@ -730,6 +734,9 @@ type appBackend struct {
 	// spendRowsPending holds one read of today's figure for rows written
 	// together (watchSpendRows).
 	spendRowsPending atomic.Bool
+	// spendRestates hands saved-spend news off without taking mu or waiting
+	// for the window (appspendsaved.go).
+	spendRestates appSpendRestates
 	// turnLong is how long a reply runs before turn.long ([appTurnLong]).
 	turnLong time.Duration
 	// inbox is every open ask as one list (appinbox.go); emit is its tap.
@@ -776,6 +783,10 @@ type appAgent struct {
 	repliedAt      time.Time
 	askedElsewhere time.Time
 	saidUnread     bool
+	// saidSpend is the saved figure last said in a dressed open/list answer or
+	// agent.changed, never a cache for a later view. mu guards it; unreadSay
+	// orders sayings.
+	saidSpend float64
 
 	// listening is when this light record began watching crew traffic.
 	listening time.Time
@@ -1052,6 +1063,7 @@ func newAppBackend(emit appwire.Emitter, open appOpener) *appBackend {
 	b.stopPhases = b.listenPhases()
 	b.briefs = newBriefDebounce(appBriefDebounce, b.writeBriefLater)
 	b.stopProgress = b.listenProgress()
+	b.stopStages = b.listenStages()
 	guard.Go("app/sweep", b.sweep)
 	b.startLocalChatPublication()
 	return b
@@ -1738,6 +1750,7 @@ func (b *appBackend) openAgent(id string, warm bool) (appwire.Agent, error) {
 		}
 	})
 	if busy {
+		view.SpentUSD = b.savedSpendOf(view.ID)
 		return view, nil
 	}
 	a.gate.Lock()
@@ -3037,6 +3050,20 @@ func appUserEntryID(n int) string { return "u" + strconv.Itoa(n) }
 
 func appAnswerEntryID(n, m int) string { return "a" + strconv.Itoa(n) + "." + strconv.Itoa(m) }
 
+// appEntryMessage is the person's message an entry id of those two is under.
+func appEntryMessage(id string) (n int, ok bool) {
+	at, found := strings.CutPrefix(id, "u")
+	if !found {
+		answer, isAnswer := strings.CutPrefix(id, "a")
+		at, _, found = strings.Cut(answer, ".")
+		if !isAnswer {
+			return 0, false
+		}
+	}
+	n, err := strconv.Atoi(at)
+	return n, found && err == nil
+}
+
 func appToolEntryID(call string) string { return "t" + call }
 
 func appNoteEntryID(n, k int) string { return "n" + strconv.Itoa(n) + "." + strconv.Itoa(k) }
@@ -4580,6 +4607,9 @@ func (b *appBackend) Close() error {
 	if b.stopPhases != nil {
 		b.stopPhases()
 	}
+	if b.stopStages != nil {
+		b.stopStages()
+	}
 	if b.stopProgress != nil {
 		b.stopProgress()
 	}