detached d79d4ec

fix(subagents): harden projection persistence

commit
d79d4ec785f3b6846696a12b02ae298135b60794
author
brice · · 2026-08-18 15:46 UTC
parent
81538e1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +145 -37
Unified
UnifiedSplit
M packages/agent/subagents/resident_journal.go +55 -37
@@ -2,14 +2,12 @@ package subagents
22
33 import (
44 "bufio"
5- "bytes"
65 "encoding/json"
76 "errors"
87 "fmt"
98 "io"
109 "os"
1110 "path/filepath"
12- "runtime"
1311 "sort"
1412 "strings"
1513 "sync"
@@ -165,21 +163,38 @@ func (j *ResidentJournal) RecordAgentEvent(event core.AgentEvent) error {
165163 j.mu.Unlock()
166164 return errors.New("resident journal: closed")
167165 }
168- if record.Type == residentRecordUsage && j.usageRecordAlreadyDurable(record) {
169- j.mu.Unlock()
170- j.publishAgentEvent(event)
171- return nil
166+ if record.Type == residentRecordUsage {
167+ durable, metadataPersisted := j.usageRecordPersistence(record)
168+ if durable {
169+ var err error
170+ if !metadataPersisted {
171+ err = j.projectUsageMetadata()
172+ if err == nil {
173+ j.markUsageMetadataPersisted(record)
174+ }
175+ }
176+ j.mu.Unlock()
177+ if err != nil {
178+ return err
179+ }
180+ j.publishAgentEvent(event)
181+ return nil
182+ }
172183 }
173184 err := j.appendSync(record)
174185 if usageEvent, ok := event.(core.EvUsage); err == nil && ok {
175186 j.usageMu.Lock()
176187 j.usage = usageEvent.Cumulative
177- j.usagePersisted = true
188+ j.usageRecordPersisted = true
189+ j.usageMetadataPersisted = false
178190 if current := usageEvent.Usage.PromptTokens(); current > 0 {
179191 j.contextUsed = current
180192 }
181193 j.usageMu.Unlock()
182194 err = j.projectUsageMetadata()
195+ if err == nil {
196+ j.markUsageMetadataPersisted(record)
197+ }
183198 }
184199 j.mu.Unlock()
185200 if err == nil {
@@ -232,20 +247,21 @@ type ResidentResult struct {
232247 // ResidentJournal serializes durable child records. It is intentionally a
233248 // narrow storage primitive; the manager owns scheduling and provider work.
234249 type ResidentJournal struct {
235- mu sync.Mutex
236- eventMu sync.RWMutex
237- eventObserver func(core.AgentEvent)
238- summaryMu sync.RWMutex
239- latestSummary string
240- usageMu sync.RWMutex
241- usage provider.Usage
242- usagePersisted bool
243- contextUsed int
244- contextMax int
245- subscription bool
246- dir string
247- file *os.File
248- lease *residentLease
250+ mu sync.Mutex
251+ eventMu sync.RWMutex
252+ eventObserver func(core.AgentEvent)
253+ summaryMu sync.RWMutex
254+ latestSummary string
255+ usageMu sync.RWMutex
256+ usage provider.Usage
257+ usageRecordPersisted bool
258+ usageMetadataPersisted bool
259+ contextUsed int
260+ contextMax int
261+ subscription bool
262+ dir string
263+ file *os.File
264+ lease *residentLease
249265 }
250266
251267 // ResidentUsageSnapshot is the bounded usage projection shared by durable and
@@ -284,17 +300,31 @@ func (j *ResidentJournal) usageSnapshot() ResidentUsageSnapshot {
284300 return ResidentUsageSnapshot{Usage: j.usage, ContextUsed: j.contextUsed, ContextMax: j.contextMax, Subscription: j.subscription}
285301 }
286302
287-func (j *ResidentJournal) usageRecordAlreadyDurable(record residentRecord) bool {
303+func (j *ResidentJournal) usageRecordPersistence(record residentRecord) (durable, metadataPersisted bool) {
288304 if record.Usage == nil {
289- return false
305+ return false, false
290306 }
291307 j.usageMu.RLock()
292308 defer j.usageMu.RUnlock()
293- return j.usagePersisted &&
294- j.usage == *record.Usage &&
309+ matches := j.usage == *record.Usage &&
295310 j.contextUsed == record.ContextUsed &&
296311 j.contextMax == record.ContextMax &&
297312 j.subscription == record.Subscription
313+ return j.usageRecordPersisted && matches, j.usageMetadataPersisted && matches
314+}
315+
316+func (j *ResidentJournal) markUsageMetadataPersisted(record residentRecord) {
317+ if record.Usage == nil {
318+ return
319+ }
320+ j.usageMu.Lock()
321+ if j.usage == *record.Usage &&
322+ j.contextUsed == record.ContextUsed &&
323+ j.contextMax == record.ContextMax &&
324+ j.subscription == record.Subscription {
325+ j.usageMetadataPersisted = true
326+ }
327+ j.usageMu.Unlock()
298328 }
299329
300330 // SetEventObserver publishes in-memory agent events without creating another
@@ -1128,15 +1158,3 @@ func writeResidentProjection(dir, name string, value any) error {
11281158 }
11291159 return nil
11301160 }
1131-
1132-func residentProjectionCurrent(path string, data []byte) bool {
1133- info, err := os.Lstat(path)
1134- if err != nil || !info.Mode().IsRegular() {
1135- return false
1136- }
1137- if runtime.GOOS != "windows" && info.Mode().Perm() != 0o600 {
1138- return false
1139- }
1140- current, err := os.ReadFile(path)
1141- return err == nil && bytes.Equal(current, data)
1142-}
@@ -2,14 +2,12 @@ package subagents
2 2
3 import (3 import (
4 "bufio"4 "bufio"
5- "bytes"
6 "encoding/json"5 "encoding/json"
7 "errors"6 "errors"
8 "fmt"7 "fmt"
9 "io"8 "io"
10 "os"9 "os"
11 "path/filepath"10 "path/filepath"
12- "runtime"
13 "sort"11 "sort"
14 "strings"12 "strings"
15 "sync"13 "sync"
@@ -165,21 +163,38 @@ func (j *ResidentJournal) RecordAgentEvent(event core.AgentEvent) error {
165 j.mu.Unlock()163 j.mu.Unlock()
166 return errors.New("resident journal: closed")164 return errors.New("resident journal: closed")
167 }165 }
168- if record.Type == residentRecordUsage && j.usageRecordAlreadyDurable(record) {166+ if record.Type == residentRecordUsage {
169- j.mu.Unlock()167+ durable, metadataPersisted := j.usageRecordPersistence(record)
170- j.publishAgentEvent(event)168+ if durable {
171- return nil169+ var err error
170+ if !metadataPersisted {
171+ err = j.projectUsageMetadata()
172+ if err == nil {
173+ j.markUsageMetadataPersisted(record)
174+ }
175+ }
176+ j.mu.Unlock()
177+ if err != nil {
178+ return err
179+ }
180+ j.publishAgentEvent(event)
181+ return nil
182+ }
172 }183 }
173 err := j.appendSync(record)184 err := j.appendSync(record)
174 if usageEvent, ok := event.(core.EvUsage); err == nil && ok {185 if usageEvent, ok := event.(core.EvUsage); err == nil && ok {
175 j.usageMu.Lock()186 j.usageMu.Lock()
176 j.usage = usageEvent.Cumulative187 j.usage = usageEvent.Cumulative
177- j.usagePersisted = true188+ j.usageRecordPersisted = true
189+ j.usageMetadataPersisted = false
178 if current := usageEvent.Usage.PromptTokens(); current > 0 {190 if current := usageEvent.Usage.PromptTokens(); current > 0 {
179 j.contextUsed = current191 j.contextUsed = current
180 }192 }
181 j.usageMu.Unlock()193 j.usageMu.Unlock()
182 err = j.projectUsageMetadata()194 err = j.projectUsageMetadata()
195+ if err == nil {
196+ j.markUsageMetadataPersisted(record)
197+ }
183 }198 }
184 j.mu.Unlock()199 j.mu.Unlock()
185 if err == nil {200 if err == nil {
@@ -232,20 +247,21 @@ type ResidentResult struct {
232 // ResidentJournal serializes durable child records. It is intentionally a247 // ResidentJournal serializes durable child records. It is intentionally a
233 // narrow storage primitive; the manager owns scheduling and provider work.248 // narrow storage primitive; the manager owns scheduling and provider work.
234 type ResidentJournal struct {249 type ResidentJournal struct {
235- mu sync.Mutex250+ mu sync.Mutex
236- eventMu sync.RWMutex251+ eventMu sync.RWMutex
237- eventObserver func(core.AgentEvent)252+ eventObserver func(core.AgentEvent)
238- summaryMu sync.RWMutex253+ summaryMu sync.RWMutex
239- latestSummary string254+ latestSummary string
240- usageMu sync.RWMutex255+ usageMu sync.RWMutex
241- usage provider.Usage256+ usage provider.Usage
242- usagePersisted bool257+ usageRecordPersisted bool
243- contextUsed int258+ usageMetadataPersisted bool
244- contextMax int259+ contextUsed int
245- subscription bool260+ contextMax int
246- dir string261+ subscription bool
247- file *os.File262+ dir string
248- lease *residentLease263+ file *os.File
264+ lease *residentLease
249 }265 }
250 266
251 // ResidentUsageSnapshot is the bounded usage projection shared by durable and267 // ResidentUsageSnapshot is the bounded usage projection shared by durable and
@@ -284,17 +300,31 @@ func (j *ResidentJournal) usageSnapshot() ResidentUsageSnapshot {
284 return ResidentUsageSnapshot{Usage: j.usage, ContextUsed: j.contextUsed, ContextMax: j.contextMax, Subscription: j.subscription}300 return ResidentUsageSnapshot{Usage: j.usage, ContextUsed: j.contextUsed, ContextMax: j.contextMax, Subscription: j.subscription}
285 }301 }
286 302
287-func (j *ResidentJournal) usageRecordAlreadyDurable(record residentRecord) bool {303+func (j *ResidentJournal) usageRecordPersistence(record residentRecord) (durable, metadataPersisted bool) {
288 if record.Usage == nil {304 if record.Usage == nil {
289- return false305+ return false, false
290 }306 }
291 j.usageMu.RLock()307 j.usageMu.RLock()
292 defer j.usageMu.RUnlock()308 defer j.usageMu.RUnlock()
293- return j.usagePersisted &&309+ matches := j.usage == *record.Usage &&
294- j.usage == *record.Usage &&
295 j.contextUsed == record.ContextUsed &&310 j.contextUsed == record.ContextUsed &&
296 j.contextMax == record.ContextMax &&311 j.contextMax == record.ContextMax &&
297 j.subscription == record.Subscription312 j.subscription == record.Subscription
313+ return j.usageRecordPersisted && matches, j.usageMetadataPersisted && matches
314+}
315+
316+func (j *ResidentJournal) markUsageMetadataPersisted(record residentRecord) {
317+ if record.Usage == nil {
318+ return
319+ }
320+ j.usageMu.Lock()
321+ if j.usage == *record.Usage &&
322+ j.contextUsed == record.ContextUsed &&
323+ j.contextMax == record.ContextMax &&
324+ j.subscription == record.Subscription {
325+ j.usageMetadataPersisted = true
326+ }
327+ j.usageMu.Unlock()
298 }328 }
299 329
300 // SetEventObserver publishes in-memory agent events without creating another330 // SetEventObserver publishes in-memory agent events without creating another
@@ -1128,15 +1158,3 @@ func writeResidentProjection(dir, name string, value any) error {
1128 }1158 }
1129 return nil1159 return nil
1130 }1160 }
1131-
1132-func residentProjectionCurrent(path string, data []byte) bool {
1133- info, err := os.Lstat(path)
1134- if err != nil || !info.Mode().IsRegular() {
1135- return false
1136- }
1137- if runtime.GOOS != "windows" && info.Mode().Perm() != 0o600 {
1138- return false
1139- }
1140- current, err := os.ReadFile(path)
1141- return err == nil && bytes.Equal(current, data)
1142-}
M packages/agent/subagents/resident_journal_test.go +52 -0
@@ -123,6 +123,55 @@ func TestResidentJournalSkipsDuplicateUsageRecord(t *testing.T) {
123123 }
124124 }
125125
126+func TestResidentJournalDuplicateUsageRetriesFailedMetadataProjection(t *testing.T) {
127+ root := t.TempDir()
128+ journal, err := OpenResidentJournal(root, "retry-usage-projection")
129+ if err != nil {
130+ t.Fatal(err)
131+ }
132+ t.Cleanup(func() { _ = journal.Close() })
133+ spec := ResidentChildSpec{ID: "retry-usage-projection", SessionID: "usage-session", Provider: "openai", Model: "gpt-test"}
134+ if err := journal.Accept(spec, "task"); err != nil {
135+ t.Fatal(err)
136+ }
137+ journal.ConfigureUsage(128_000, false)
138+ metadataPath := filepath.Join(journal.Dir(), residentMetadataName)
139+ originalMetadata, err := os.ReadFile(metadataPath)
140+ if err != nil {
141+ t.Fatal(err)
142+ }
143+ if err := os.WriteFile(metadataPath, []byte("not-json"), 0o600); err != nil {
144+ t.Fatal(err)
145+ }
146+
147+ usage := provider.Usage{InputTokens: 10_000, OutputTokens: 500}
148+ event := core.EvUsage{Usage: usage, Cumulative: usage}
149+ if err := journal.RecordAgentEvent(event); err == nil {
150+ t.Fatal("first usage projection succeeded with malformed metadata")
151+ }
152+ if err := os.WriteFile(metadataPath, originalMetadata, 0o600); err != nil {
153+ t.Fatal(err)
154+ }
155+ if err := journal.RecordAgentEvent(event); err != nil {
156+ t.Fatalf("retry identical usage: %v", err)
157+ }
158+
159+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
160+ if err != nil {
161+ t.Fatal(err)
162+ }
163+ if got := len(records); got != 2 {
164+ t.Fatalf("journal records = %d, want accepted record plus one usage record", got)
165+ }
166+ metadata, err := ReadResidentMetadata(metadataPath)
167+ if err != nil {
168+ t.Fatal(err)
169+ }
170+ if metadata.Usage != usage || metadata.ContextUsed != usage.PromptTokens() {
171+ t.Fatalf("metadata usage = %+v, context = %d; want %+v, %d", metadata.Usage, metadata.ContextUsed, usage, usage.PromptTokens())
172+ }
173+}
174+
126175 func TestReconcileResidentJournalRetainsRootCacheID(t *testing.T) {
127176 root := t.TempDir()
128177 journal, err := OpenResidentJournal(root, "cache-child")
@@ -177,6 +226,9 @@ func TestResidentJournalProjectsUsageEmittedAfterInterruption(t *testing.T) {
177226 }
178227
179228 func TestResidentProjectionSkipsIdenticalRewrite(t *testing.T) {
229+ if runtime.GOOS == "windows" {
230+ t.Skip("no-follow projection reads are unavailable on Windows")
231+ }
180232 dir := t.TempDir()
181233 metadata := ResidentMetadata{
182234 Version: residentJournalVersion,
@@ -123,6 +123,55 @@ func TestResidentJournalSkipsDuplicateUsageRecord(t *testing.T) {
123 }123 }
124 }124 }
125 125
126+func TestResidentJournalDuplicateUsageRetriesFailedMetadataProjection(t *testing.T) {
127+ root := t.TempDir()
128+ journal, err := OpenResidentJournal(root, "retry-usage-projection")
129+ if err != nil {
130+ t.Fatal(err)
131+ }
132+ t.Cleanup(func() { _ = journal.Close() })
133+ spec := ResidentChildSpec{ID: "retry-usage-projection", SessionID: "usage-session", Provider: "openai", Model: "gpt-test"}
134+ if err := journal.Accept(spec, "task"); err != nil {
135+ t.Fatal(err)
136+ }
137+ journal.ConfigureUsage(128_000, false)
138+ metadataPath := filepath.Join(journal.Dir(), residentMetadataName)
139+ originalMetadata, err := os.ReadFile(metadataPath)
140+ if err != nil {
141+ t.Fatal(err)
142+ }
143+ if err := os.WriteFile(metadataPath, []byte("not-json"), 0o600); err != nil {
144+ t.Fatal(err)
145+ }
146+
147+ usage := provider.Usage{InputTokens: 10_000, OutputTokens: 500}
148+ event := core.EvUsage{Usage: usage, Cumulative: usage}
149+ if err := journal.RecordAgentEvent(event); err == nil {
150+ t.Fatal("first usage projection succeeded with malformed metadata")
151+ }
152+ if err := os.WriteFile(metadataPath, originalMetadata, 0o600); err != nil {
153+ t.Fatal(err)
154+ }
155+ if err := journal.RecordAgentEvent(event); err != nil {
156+ t.Fatalf("retry identical usage: %v", err)
157+ }
158+
159+ records, err := ReadResidentJournal(filepath.Join(journal.Dir(), residentTranscriptName))
160+ if err != nil {
161+ t.Fatal(err)
162+ }
163+ if got := len(records); got != 2 {
164+ t.Fatalf("journal records = %d, want accepted record plus one usage record", got)
165+ }
166+ metadata, err := ReadResidentMetadata(metadataPath)
167+ if err != nil {
168+ t.Fatal(err)
169+ }
170+ if metadata.Usage != usage || metadata.ContextUsed != usage.PromptTokens() {
171+ t.Fatalf("metadata usage = %+v, context = %d; want %+v, %d", metadata.Usage, metadata.ContextUsed, usage, usage.PromptTokens())
172+ }
173+}
174+
126 func TestReconcileResidentJournalRetainsRootCacheID(t *testing.T) {175 func TestReconcileResidentJournalRetainsRootCacheID(t *testing.T) {
127 root := t.TempDir()176 root := t.TempDir()
128 journal, err := OpenResidentJournal(root, "cache-child")177 journal, err := OpenResidentJournal(root, "cache-child")
@@ -177,6 +226,9 @@ func TestResidentJournalProjectsUsageEmittedAfterInterruption(t *testing.T) {
177 }226 }
178 227
179 func TestResidentProjectionSkipsIdenticalRewrite(t *testing.T) {228 func TestResidentProjectionSkipsIdenticalRewrite(t *testing.T) {
229+ if runtime.GOOS == "windows" {
230+ t.Skip("no-follow projection reads are unavailable on Windows")
231+ }
180 dir := t.TempDir()232 dir := t.TempDir()
181 metadata := ResidentMetadata{233 metadata := ResidentMetadata{
182 Version: residentJournalVersion,234 Version: residentJournalVersion,
A packages/agent/subagents/resident_projection_unix.go +31 -0
@@ -0,0 +1,31 @@
1+//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
2+
3+package subagents
4+
5+import (
6+ "bytes"
7+ "io"
8+ "os"
9+
10+ "golang.org/x/sys/unix"
11+)
12+
13+func residentProjectionCurrent(path string, data []byte) bool {
14+ fd, err := unix.Open(path, unix.O_RDONLY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
15+ if err != nil {
16+ return false
17+ }
18+ file := os.NewFile(uintptr(fd), path)
19+ if file == nil {
20+ _ = unix.Close(fd)
21+ return false
22+ }
23+ defer file.Close()
24+
25+ info, err := file.Stat()
26+ if err != nil || !info.Mode().IsRegular() || info.Mode().Perm() != 0o600 {
27+ return false
28+ }
29+ current, err := io.ReadAll(file)
30+ return err == nil && bytes.Equal(current, data)
31+}
@@ -0,0 +1,31 @@
1+//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
2+
3+package subagents
4+
5+import (
6+ "bytes"
7+ "io"
8+ "os"
9+
10+ "golang.org/x/sys/unix"
11+)
12+
13+func residentProjectionCurrent(path string, data []byte) bool {
14+ fd, err := unix.Open(path, unix.O_RDONLY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
15+ if err != nil {
16+ return false
17+ }
18+ file := os.NewFile(uintptr(fd), path)
19+ if file == nil {
20+ _ = unix.Close(fd)
21+ return false
22+ }
23+ defer file.Close()
24+
25+ info, err := file.Stat()
26+ if err != nil || !info.Mode().IsRegular() || info.Mode().Perm() != 0o600 {
27+ return false
28+ }
29+ current, err := io.ReadAll(file)
30+ return err == nil && bytes.Equal(current, data)
31+}
A packages/agent/subagents/resident_projection_windows.go +7 -0
@@ -0,0 +1,7 @@
1+package subagents
2+
3+// Windows does not expose a no-follow regular-file read through os. Disable the
4+// no-rewrite fast path so atomic replacement remains the only projection path.
5+func residentProjectionCurrent(string, []byte) bool {
6+ return false
7+}
@@ -0,0 +1,7 @@
1+package subagents
2+
3+// Windows does not expose a no-follow regular-file read through os. Disable the
4+// no-rewrite fast path so atomic replacement remains the only projection path.
5+func residentProjectionCurrent(string, []byte) bool {
6+ return false
7+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 476

↑↓ moveenter openesc close