Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions cmd/codeaf/app.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -630,8 +631,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
Expand Down Expand Up @@ -731,6 +732,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.
Expand Down Expand Up @@ -776,6 +780,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
Expand Down Expand Up @@ -1737,6 +1745,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()
Expand Down
5 changes: 3 additions & 2 deletions cmd/codeaf/appinbox.go
Original file line number Diff line number Diff line change
Expand Up @@ -78,11 +78,12 @@ func (t *appInboxTap) Emit(event string, data any) {
}
// EVERY agent.changed IS DRESSED ON ITS WAY OUT, whichever road said it:
// its crews, its live line, working while a turn runs, and unread. Its
// unread is worked out and said in one step, so two roads saying the
// agent at once cannot leave the window with the older reading.
// unread and saved spend are worked out and said in one step, so two roads
// saying the agent at once cannot leave the window with an older reading.
if changed, ok := data.(appwire.AgentChanged); ok {
t.b.dressState(&changed.Agent)
t.b.sayingUnread(func() {
t.b.dressSpendLocked(&changed.Agent)
t.b.dressUnreadLocked(&changed.Agent)
t.next.Emit(event, changed)
})
Expand Down
143 changes: 143 additions & 0 deletions cmd/codeaf/appspendsaved.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
package main

import (
"path/filepath"
"strings"
"sync"

"github.com/Agent-Field/codeaf/internal/appwire"
"github.com/Agent-Field/codeaf/internal/offpath"
"github.com/Agent-Field/codeaf/internal/session"
)

// appSpendRestates coalesces news per conversation. Its lock protects only
// handoffs, never disk reads, the backend lock or saying an agent to the window.
type appSpendRestates struct {
mu sync.Mutex
of map[string]*offpath.Write
stopped bool
}

// watchAgentSpend follows saved conversation totals separately from the
// machine ledger: a late receipt can save spend after its turn has ended.
func (b *appBackend) watchAgentSpend() (stop func()) {
unwatch := session.OnSpendSaved(b.agentSpendSaved)
return func() {
unwatch()
for _, restate := range b.spendRestates.stop() {
restate.Settle()
}
}
}

// stop refuses later news and answers the restates already handed off, which
// the caller settles outside the lock.
func (s *appSpendRestates) stop() []*offpath.Write {
s.mu.Lock()
defer s.mu.Unlock()
s.stopped = true
return s.pendingLocked()
}

// settle joins the restates already handed off, outside the lock.
func (s *appSpendRestates) settle() {
for _, restate := range s.pending() {
restate.Settle()
}
}

func (s *appSpendRestates) pending() []*offpath.Write {
s.mu.Lock()
defer s.mu.Unlock()
return s.pendingLocked()
}

func (s *appSpendRestates) pendingLocked() []*offpath.Write {
pending := make([]*offpath.Write, 0, len(s.of))
for _, restate := range s.of {
pending = append(pending, restate)
}
return pending
}

// agentSpendSaved only hands off identity. Even delayed news asks what is on
// disk now; neither its figure nor a producer's copied view is a spend cache.
func (b *appBackend) agentSpendSaved(saved session.SpendSaved) {
dir := strings.TrimSpace(saved.Dir)
if dir == "" {
return
}
dir = filepath.Clean(dir)
s := &b.spendRestates
s.mu.Lock()
defer s.mu.Unlock()
if s.stopped {
return
}
if s.of == nil {
s.of = map[string]*offpath.Write{}
}
restate := s.of[dir]
if restate == nil {
restate = offpath.Deferred(func() { b.restateSpend(dir) })
s.of[dir] = restate
}
restate.Owe()
}

// restateSpend mirrors restateUnread: compare with what was last said under
// the saying lock, then use the ordinary door, which dresses afresh at delivery.
func (b *appBackend) restateSpend(dir string) {
id := filepath.Base(dir)
var a *appAgent
stale := false
b.sayingUnread(func() {
var said float64
b.locked(func() {
if b.closing {
return
}
if have := b.agents[id]; have != nil && have.transcript != "" && session.DirOf(have.transcript) == dir {
a, said = have, have.saidSpend
}
})
if a != nil {
spent, err := session.SavedSpend(dir)
stale = err == nil && spent != said
}
})
if stale {
b.locked(func() { stale = !b.closing })
if stale {
b.sayListed(a, id)
}
}
}

// savedSpendOf reads only the saved figure, without LoadMeta's journal and
// workspace overlays. It never takes unreadSay or records the figure as said,
// so an already-opening answer need not wait for the window's event writer.
func (b *appBackend) savedSpendOf(id string) float64 {
var transcript string
b.locked(func() {
if a := b.agents[id]; a != nil {
transcript = a.transcript
}
})
if transcript == "" {
return 0
}
spent, _ := session.SavedSpend(session.DirOf(transcript))
return spent
}

// dressSpendLocked fills the saved figure and remembers it as said. Callers
// hold unreadSay through remembering and saying it.
func (b *appBackend) dressSpendLocked(view *appwire.Agent) {
view.SpentUSD = b.savedSpendOf(view.ID)
b.locked(func() {
if a := b.agents[view.ID]; a != nil {
a.saidSpend = view.SpentUSD
}
})
}
Loading
Loading