Merge pull request #154 from bnema/fix/session-jsonl-concurrent-write

fix(core): serialize concurrent session writes

commit: edaaa7c5f9cfaf69df3dc15f304b1ae986f732f1
author: bnema · 2mo ago · 2026-08-21 02:11 UTC
committer: GitHub · 2mo ago · 2026-08-21 02:11 UTC
merge of: 7021aa7 + 0da46cd · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown
refs: tag: v0.23.0

2 files · +118 -21

M packages/core/session.go +43 -21


@@ -34,6 +34,7 @@ type Session struct {
34	34		ExtensionState map[string]json.RawMessage
35	35		writer *os.File
36	36		buf *bufio.Writer
	37	+	writerMu sync.Mutex
37	38
38	39		// freshFile is true when the file was created by NewSession (this
39	40		// process owns it) and false when OpenSession reopened an existing

@@ -1295,11 +1296,7 @@ func (s *Session) AppendMessage(m provider.Message) error {
1295	1296		if len(m.Content) == 0 {
1296	1297		return errors.New("message has no content")
1297	1298		}
1298		-	if err := s.writeLine(sessionLine{Type: "message", Message: &m}); err != nil {
1299		-	return err
1300		-	}
1301		-	s.messagesAppended++
1302		-	return nil
	1299	+	return s.writeCountedLine(sessionLine{Type: "message", Message: &m})
1303	1300		}
1304	1301
1305	1302		// Sync flushes the session file's bytes through the filesystem. Interactive

@@ -1309,6 +1306,8 @@ func (s *Session) Sync() error {
1309	1306		if s == nil \|\| s.writer == nil {
1310	1307		return nil
1311	1308		}
	1309	+	s.writerMu.Lock()
	1310	+	defer s.writerMu.Unlock()
1312	1311		if err := s.buf.Flush(); err != nil {
1313	1312		return err
1314	1313		}

@@ -1324,14 +1323,10 @@ func (s *Session) AppendCompaction(messages []provider.Message) error {
1324	1323		return nil
1325	1324		}
1326	1325		compactionMessages := messages
1327		-	if err := s.writeLine(sessionLine{Type: "compaction", Messages: &compactionMessages}); err != nil {
1328		-	return err
1329		-	}
1330	1326		// The compaction row itself is meaningful even when it replaces the
1331	1327		// transcript with an empty or nil slice. Count the append operation so a
1332	1328		// fresh session containing an empty checkpoint is not deleted on Close.
1333		-	s.messagesAppended++
1334		-	return nil
	1329	+	return s.writeCountedLine(sessionLine{Type: "compaction", Messages: &compactionMessages})
1335	1330		}
1336	1331
1337	1332		// AppendCompactionWithUsage atomically records a replacement transcript and

@@ -1343,16 +1338,12 @@ func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulat
1343	1338		return nil
1344	1339		}
1345	1340		compactionMessages := messages
1346		-	if err := s.writeUsageCheckpoint(sessionLine{
	1341	+	return s.writeUsageCheckpointCounted(sessionLine{
1347	1342		Type: "compaction",
1348	1343		Messages: &compactionMessages,
1349	1344		Usage: &cumulative,
1350	1345		Cumulative: &cumulative,
1351		-	}); err != nil {
1352		-	return err
1353		-	}
1354		-	s.messagesAppended++
1355		-	return nil
	1346	+	})
1356	1347		}
1357	1348
1358	1349		// UpdateModel records a provider/model switch in the session file.

@@ -1620,24 +1611,32 @@ func (s *Session) FlushRetryLifecycle(u, cum provider.Usage) error {
1620	1611		if len(s.pendingRetryLifecycle) == 0 {
1621	1612		return nil
1622	1613		}
1623		-	return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
	1614	+	return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}, false)
1624	1615		}
1625	1616
1626	1617		func (s *Session) writeUsageCheckpoint(row sessionLine) error {
1627	1618		s.retryMu.Lock()
1628	1619		defer s.retryMu.Unlock()
1629		-	return s.writeUsageCheckpointLocked(row)
	1620	+	return s.writeUsageCheckpointLocked(row, false)
1630	1621		}
1631	1622
1632		-	func (s *Session) writeUsageCheckpointLocked(row sessionLine) error {
	1623	+	func (s *Session) writeUsageCheckpointCounted(row sessionLine) error {
	1624	+	s.retryMu.Lock()
	1625	+	defer s.retryMu.Unlock()
	1626	+	return s.writeUsageCheckpointLocked(row, true)
	1627	+	}
	1628	+
	1629	+	func (s *Session) writeUsageCheckpointLocked(row sessionLine, counted bool) error {
1633	1630		if len(s.pendingRetryLifecycle) > 0 {
1634	1631		row.RetryLifecycle = append([]RetryLifecycleRecord(nil), s.pendingRetryLifecycle...)
1635	1632		}
	1633	+	s.writerMu.Lock()
	1634	+	defer s.writerMu.Unlock()
1636	1635		if row.Type == "usage" && len(row.RetryLifecycle) == 0 && row.Usage != nil && row.Cumulative != nil &&
1637	1636		s.lastWriteWasUsage && row.Usage == s.lastUsage && row.Cumulative == s.lastCumulativeUsage {
1638	1637		return nil
1639	1638		}
1640		-	if err := s.writeLine(row); err != nil {
	1639	+	if err := s.writeLineLocked(row); err != nil {
1641	1640		return err
1642	1641		}
1643	1642		if row.Type == "usage" && row.Usage != nil && row.Cumulative != nil {

@@ -1645,6 +1644,9 @@ func (s *Session) writeUsageCheckpointLocked(row sessionLine) error {
1645	1644		s.lastUsage = *row.Usage
1646	1645		s.lastCumulativeUsage = *row.Cumulative
1647	1646		}
	1647	+	if counted {
	1648	+	s.messagesAppended++
	1649	+	}
1648	1650		s.pendingRetryLifecycle = nil
1649	1651		return nil
1650	1652		}

@@ -1719,6 +1721,8 @@ func (s *Session) Flush() error {
1719	1721		if s == nil {
1720	1722		return nil
1721	1723		}
	1724	+	s.writerMu.Lock()
	1725	+	defer s.writerMu.Unlock()
1722	1726		return s.buf.Flush()
1723	1727		}
1724	1728

@@ -1731,7 +1735,9 @@ func (s *Session) Close() error {
1731	1735		if s == nil {
1732	1736		return nil
1733	1737		}
1734		-	flushErr := s.Flush()
	1738	+	s.writerMu.Lock()
	1739	+	defer s.writerMu.Unlock()
	1740	+	flushErr := s.buf.Flush()
1735	1741		closeErr := s.writer.Close()
1736	1742		if s.freshFile && s.messagesAppended == 0 && len(s.ExtensionState) == 0 && len(s.Meta.CompactHandoff) == 0 && s.Meta.Mission == nil && s.Meta.Goal == nil && len(s.Meta.GoalHistory) == 0 {
1737	1743		// Best-effort cleanup. We deliberately don't propagate the

@@ -1746,6 +1752,22 @@ func (s *Session) Close() error {
1746	1752		}
1747	1753
1748	1754		func (s *Session) writeLine(row sessionLine) error {
	1755	+	s.writerMu.Lock()
	1756	+	defer s.writerMu.Unlock()
	1757	+	return s.writeLineLocked(row)
	1758	+	}
	1759	+
	1760	+	func (s *Session) writeCountedLine(row sessionLine) error {
	1761	+	s.writerMu.Lock()
	1762	+	defer s.writerMu.Unlock()
	1763	+	if err := s.writeLineLocked(row); err != nil {
	1764	+	return err
	1765	+	}
	1766	+	s.messagesAppended++
	1767	+	return nil
	1768	+	}
	1769	+
	1770	+	func (s *Session) writeLineLocked(row sessionLine) error {
1749	1771		s.lastWriteWasUsage = false
1750	1772		b, err := json.Marshal(row)
1751	1773		if err != nil {

 	ExtensionState map[string]json.RawMessage
 	writer         *os.File
 	buf            *bufio.Writer
+	writerMu       sync.Mutex
 	// freshFile is true when the file was created by NewSession (this
 	// process owns it) and false when OpenSession reopened an existing
 	if len(m.Content) == 0 {
 		return errors.New("message has no content")
 	}
-	if err := s.writeLine(sessionLine{Type: "message", Message: &m}); err != nil {
+	return s.writeCountedLine(sessionLine{Type: "message", Message: &m})
-		return err
-	}
-	s.messagesAppended++
-	return nil
 }
 // Sync flushes the session file's bytes through the filesystem. Interactive
 	if s == nil || s.writer == nil {
 		return nil
 	}
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
 	if err := s.buf.Flush(); err != nil {
 		return err
 	}
 		return nil
 	}
 	compactionMessages := messages
-	if err := s.writeLine(sessionLine{Type: "compaction", Messages: &compactionMessages}); err != nil {
-		return err
-	}
 	// The compaction row itself is meaningful even when it replaces the
 	// transcript with an empty or nil slice. Count the append operation so a
 	// fresh session containing an empty checkpoint is not deleted on Close.
-	s.messagesAppended++
+	return s.writeCountedLine(sessionLine{Type: "compaction", Messages: &compactionMessages})
-	return nil
 }
 // AppendCompactionWithUsage atomically records a replacement transcript and
 		return nil
 	}
 	compactionMessages := messages
-	if err := s.writeUsageCheckpoint(sessionLine{
+	return s.writeUsageCheckpointCounted(sessionLine{
 		Type:       "compaction",
 		Messages:   &compactionMessages,
 		Usage:      &cumulative,
 		Cumulative: &cumulative,
-	}); err != nil {
+	})
-		return err
-	}
-	s.messagesAppended++
-	return nil
 }
 // UpdateModel records a provider/model switch in the session file.
 	if len(s.pendingRetryLifecycle) == 0 {
 		return nil
 	}
-	return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum})
+	return s.writeUsageCheckpointLocked(sessionLine{Type: "usage", Usage: &u, Cumulative: &cum}, false)
 }
 func (s *Session) writeUsageCheckpoint(row sessionLine) error {
 	s.retryMu.Lock()
 	defer s.retryMu.Unlock()
-	return s.writeUsageCheckpointLocked(row)
+	return s.writeUsageCheckpointLocked(row, false)
 }
-func (s *Session) writeUsageCheckpointLocked(row sessionLine) error {
+func (s *Session) writeUsageCheckpointCounted(row sessionLine) error {
+	s.retryMu.Lock()
+	defer s.retryMu.Unlock()
+	return s.writeUsageCheckpointLocked(row, true)
+}
+func (s *Session) writeUsageCheckpointLocked(row sessionLine, counted bool) error {
 	if len(s.pendingRetryLifecycle) > 0 {
 		row.RetryLifecycle = append([]RetryLifecycleRecord(nil), s.pendingRetryLifecycle...)
 	}
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
 	if row.Type == "usage" && len(row.RetryLifecycle) == 0 && row.Usage != nil && row.Cumulative != nil &&
 		s.lastWriteWasUsage && *row.Usage == s.lastUsage && *row.Cumulative == s.lastCumulativeUsage {
 		return nil
 	}
-	if err := s.writeLine(row); err != nil {
+	if err := s.writeLineLocked(row); err != nil {
 		return err
 	}
 	if row.Type == "usage" && row.Usage != nil && row.Cumulative != nil {
 		s.lastUsage = *row.Usage
 		s.lastCumulativeUsage = *row.Cumulative
 	}
+	if counted {
+		s.messagesAppended++
+	}
 	s.pendingRetryLifecycle = nil
 	return nil
 }
 	if s == nil {
 		return nil
 	}
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
 	return s.buf.Flush()
 }
 	if s == nil {
 		return nil
 	}
-	flushErr := s.Flush()
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
+	flushErr := s.buf.Flush()
 	closeErr := s.writer.Close()
 	if s.freshFile && s.messagesAppended == 0 && len(s.ExtensionState) == 0 && len(s.Meta.CompactHandoff) == 0 && s.Meta.Mission == nil && s.Meta.Goal == nil && len(s.Meta.GoalHistory) == 0 {
 		// Best-effort cleanup. We deliberately don't propagate the
 }
 func (s *Session) writeLine(row sessionLine) error {
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
+	return s.writeLineLocked(row)
+}
+func (s *Session) writeCountedLine(row sessionLine) error {
+	s.writerMu.Lock()
+	defer s.writerMu.Unlock()
+	if err := s.writeLineLocked(row); err != nil {
+		return err
+	}
+	s.messagesAppended++
+	return nil
+}
+func (s *Session) writeLineLocked(row sessionLine) error {
 	s.lastWriteWasUsage = false
 	b, err := json.Marshal(row)
 	if err != nil {

A packages/core/session_concurrency_test.go +75 -0

@@ -0,0 +1,75 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"encoding/json"
	6	+	"os"
	7	+	"strings"
	8	+	"sync"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/zut/packages/provider"
	12	+	)
	13	+
	14	+	func TestSessionConcurrentAppendsRemainValidJSONL(t *testing.T) {
	15	+	session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test")
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+
	20	+	const (
	21	+	writers = 8
	22	+	rows = 20
	23	+	)
	24	+	payload := strings.Repeat("x", 8*1024)
	25	+	start := make(chan struct{})
	26	+	errs := make(chan error, writers)
	27	+	var wg sync.WaitGroup
	28	+	for writer := 0; writer < writers; writer++ {
	29	+	wg.Add(1)
	30	+	go func() {
	31	+	defer wg.Done()
	32	+	<-start
	33	+	for i := 0; i < rows; i++ {
	34	+	if err := session.AppendMessage(provider.Message{
	35	+	Role: provider.RoleAssistant,
	36	+	Content: []provider.Content{provider.TextBlock{Text: payload}},
	37	+	}); err != nil {
	38	+	errs <- err
	39	+	return
	40	+	}
	41	+	}
	42	+	}()
	43	+	}
	44	+	close(start)
	45	+	wg.Wait()
	46	+	close(errs)
	47	+	for err := range errs {
	48	+	t.Errorf("append message: %v", err)
	49	+	}
	50	+	if err := session.Close(); err != nil {
	51	+	t.Fatal(err)
	52	+	}
	53	+
	54	+	file, err := os.Open(session.Path)
	55	+	if err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	defer file.Close()
	59	+
	60	+	lineCount := 0
	61	+	scanner := bufio.NewScanner(file)
	62	+	scanner.Buffer(make([]byte, 1024), 32*1024)
	63	+	for scanner.Scan() {
	64	+	lineCount++
	65	+	if !json.Valid(scanner.Bytes()) {
	66	+	t.Fatalf("line %d is not valid JSON", lineCount)
	67	+	}
	68	+	}
	69	+	if err := scanner.Err(); err != nil {
	70	+	t.Fatal(err)
	71	+	}
	72	+	if want := 1 + writers*rows; lineCount != want {
	73	+	t.Fatalf("line count = %d, want %d", lineCount, want)
	74	+	}
	75	+	}

@@ -0,0 +1,75 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"bufio"
	5	+	"encoding/json"
	6	+	"os"
	7	+	"strings"
	8	+	"sync"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/zut/packages/provider"
	12	+	)
	13	+
	14	+	func TestSessionConcurrentAppendsRemainValidJSONL(t *testing.T) {
	15	+	session, err := NewSession(t.TempDir(), "/workspace", "provider", "model", "test")
	16	+	if err != nil {
	17	+	t.Fatal(err)
	18	+	}
	19	+
	20	+	const (
	21	+	writers = 8
	22	+	rows = 20
	23	+	)
	24	+	payload := strings.Repeat("x", 8*1024)
	25	+	start := make(chan struct{})
	26	+	errs := make(chan error, writers)
	27	+	var wg sync.WaitGroup
	28	+	for writer := 0; writer < writers; writer++ {
	29	+	wg.Add(1)
	30	+	go func() {
	31	+	defer wg.Done()
	32	+	<-start
	33	+	for i := 0; i < rows; i++ {
	34	+	if err := session.AppendMessage(provider.Message{
	35	+	Role: provider.RoleAssistant,
	36	+	Content: []provider.Content{provider.TextBlock{Text: payload}},
	37	+	}); err != nil {
	38	+	errs <- err
	39	+	return
	40	+	}
	41	+	}
	42	+	}()
	43	+	}
	44	+	close(start)
	45	+	wg.Wait()
	46	+	close(errs)
	47	+	for err := range errs {
	48	+	t.Errorf("append message: %v", err)
	49	+	}
	50	+	if err := session.Close(); err != nil {
	51	+	t.Fatal(err)
	52	+	}
	53	+
	54	+	file, err := os.Open(session.Path)
	55	+	if err != nil {
	56	+	t.Fatal(err)
	57	+	}
	58	+	defer file.Close()
	59	+
	60	+	lineCount := 0
	61	+	scanner := bufio.NewScanner(file)
	62	+	scanner.Buffer(make([]byte, 1024), 32*1024)
	63	+	for scanner.Scan() {
	64	+	lineCount++
	65	+	if !json.Valid(scanner.Bytes()) {
	66	+	t.Fatalf("line %d is not valid JSON", lineCount)
	67	+	}
	68	+	}
	69	+	if err := scanner.Err(); err != nil {
	70	+	t.Fatal(err)
	71	+	}
	72	+	if want := 1 + writers*rows; lineCount != want {
	73	+	t.Fatalf("line count = %d, want %d", lineCount, want)
	74	+	}
	75	+	}