diff --git a/pkg/runtime/event.go b/pkg/runtime/event.go index 56cc24f95..51ee4f5b7 100644 --- a/pkg/runtime/event.go +++ b/pkg/runtime/event.go @@ -506,6 +506,11 @@ type SessionSummaryEvent struct { Cost float64 `json:"cost,omitempty"` Model string `json:"model,omitempty"` Usage *chat.Usage `json:"usage,omitempty"` + + // persisted is set when the compaction path wrote this summary directly. + // The persistence observer uses it to avoid writing the same summary again + // on normal observed RunStream executions. It is intentionally not serialized. + persisted bool } // SessionSummary builds the event announcing an applied compaction summary. diff --git a/pkg/runtime/persistence_observer.go b/pkg/runtime/persistence_observer.go index b86d1dfc7..a663a9fc7 100644 --- a/pkg/runtime/persistence_observer.go +++ b/pkg/runtime/persistence_observer.go @@ -119,6 +119,9 @@ func (p *PersistenceObserver) OnEvent(ctx context.Context, sess *session.Session } case *SessionSummaryEvent: + if e.persisted { + break + } item := session.Item{Summary: e.Summary, FirstKeptEntry: e.FirstKeptEntry, Cost: e.Cost, Model: e.Model} if e.Usage != nil { // Copy so the persisted item doesn't alias the event's pointer. diff --git a/pkg/runtime/remote_runtime.go b/pkg/runtime/remote_runtime.go index bc515982a..6fa513943 100644 --- a/pkg/runtime/remote_runtime.go +++ b/pkg/runtime/remote_runtime.go @@ -863,6 +863,10 @@ func (s *RemoteSessionStore) AddSubSession(context.Context, string, *session.Ses return fmt.Errorf("add sub session: %w", ErrUnsupported) } +func (s *RemoteSessionStore) PersistCompaction(context.Context, *session.Session, int64, int64, session.Item) error { + return fmt.Errorf("persist compaction: %w", ErrUnsupported) +} + func (s *RemoteSessionStore) AddSummary(context.Context, string, session.Item) error { return fmt.Errorf("add summary: %w", ErrUnsupported) } diff --git a/pkg/runtime/session_compaction.go b/pkg/runtime/session_compaction.go index 1ba8f6b0a..3628e8bc0 100644 --- a/pkg/runtime/session_compaction.go +++ b/pkg/runtime/session_compaction.go @@ -2,6 +2,7 @@ package runtime import ( "context" + "fmt" "log/slog" "github.com/docker/docker-agent/pkg/agent" @@ -127,20 +128,33 @@ func (r *LocalRuntime) doCompact(ctx context.Context, sess *session.Session, a * // the new summary's estimated size. preInputTokens, preOutputTokens := sess.Usage() - // Apply the summary to the session. This is intrinsically - // runtime-private: it mutates session-internal state and persists - // through the runtime's session store. - sess.ApplyCompaction(result.InputTokens, 0, session.Item{ + // Apply and persist the summary as one intrinsic successful-compaction step. + // Manual Summarize calls do not pass through the RunStream observer chain, + // while normal observed runs do; marking the emitted event as persisted + // keeps the observer from appending a duplicate row. + item := session.Item{ Summary: result.Summary, FirstKeptEntry: result.FirstKeptEntry, Cost: result.Cost, Model: result.Model, Usage: summaryUsage(result), - }) - _ = r.sessionStore.UpdateSession(ctx, sess) + } + // Atomically persist the metadata and summary before mutating the live + // session. A failed write is a failed compaction: no success summary event + // is emitted and the in-memory continuation remains unchanged. + if err := r.sessionStore.PersistCompaction(ctx, sess, result.InputTokens, 0, item); err != nil { + slog.ErrorContext(ctx, "Failed to persist session compaction", "session_id", sess.ID, "error", err) + events.Emit(ErrorForSession(sess.ID, fmt.Sprintf("Failed to persist session compaction: %v", err))) + outcome = CompactionOutcomeFailed + return + } slog.DebugContext(ctx, "Generated session summary", "session_id", sess.ID, "summary_length", len(result.Summary)) - events.Emit(SessionSummary(sess.ID, result.Summary, a.Name(), result.FirstKeptEntry, result.Cost, result.Model, summaryUsage(result))) + summaryEvent := SessionSummary(sess.ID, result.Summary, a.Name(), result.FirstKeptEntry, result.Cost, result.Model, summaryUsage(result)) + if e, ok := summaryEvent.(*SessionSummaryEvent); ok { + e.persisted = true + } + events.Emit(summaryEvent) // after_compaction: observational. Fired only when a summary was // actually applied to the session. The hook receives the diff --git a/pkg/runtime/session_compaction_test.go b/pkg/runtime/session_compaction_test.go index f1f6e40f4..14421cbeb 100644 --- a/pkg/runtime/session_compaction_test.go +++ b/pkg/runtime/session_compaction_test.go @@ -1,8 +1,12 @@ package runtime import ( + "context" + "errors" "os" + "path/filepath" "runtime" + "strings" "testing" "github.com/stretchr/testify/assert" @@ -13,6 +17,7 @@ import ( "github.com/docker/docker-agent/pkg/config/latest" "github.com/docker/docker-agent/pkg/modelsdev" "github.com/docker/docker-agent/pkg/session" + "github.com/docker/docker-agent/pkg/session/sqlitestore" "github.com/docker/docker-agent/pkg/team" ) @@ -301,6 +306,179 @@ func TestDoCompactAfterHookFires(t *testing.T) { "after_compaction must receive the produced summary and the *pre-compaction* token counts") } +type failingCompactionStore struct { + session.Store + + err error +} + +func (s failingCompactionStore) PersistCompaction(context.Context, *session.Session, int64, int64, session.Item) error { + return s.err +} + +func TestDoCompactInMemoryStoreAppendsSummaryOnce(t *testing.T) { + store := session.NewInMemorySessionStore() + summaryStream := newStreamBuilder().AddContent("one summary").AddStopWithUsage(1, 1).Build() + prov := &queueProvider{id: "test/mock-model", streams: []chat.MessageStream{summaryStream}} + root := agent.New("root", "test", agent.WithModel(prov)) + rt, err := NewLocalRuntime(t.Context(), team.New(team.WithAgents(root)), + WithSessionCompaction(false), WithSessionStore(store), + WithModelStore(mockModelStoreWithLimit{limit: 100_000})) + require.NoError(t, err) + + sess := session.New(session.WithID("compact-memory"), session.WithMessages([]session.Item{ + session.NewMessageItem(session.UserMessage("hi")), + session.NewMessageItem(&session.Message{Message: chat.Message{Role: chat.MessageRoleAssistant, Content: "hello"}}), + })) + require.NoError(t, store.AddSession(t.Context(), sess)) + + events := make(chan Event, 32) + rt.Summarize(t.Context(), sess, "", NewChannelSink(events)) + close(events) + for range events { + } + + reloaded, err := store.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + var summaries int + for _, item := range reloaded.Messages { + if item.Summary != "" { + summaries++ + } + } + assert.Equal(t, 1, summaries) + assert.Same(t, sess, reloaded, "aliasing store must mutate the live session exactly once") +} + +func TestDoCompactPersistenceFailureReportsFailedWithoutApplying(t *testing.T) { + base := session.NewInMemorySessionStore() + persistErr := errors.New("disk full") + store := failingCompactionStore{Store: base, err: persistErr} + summaryStream := newStreamBuilder().AddContent("lost summary").AddStopWithUsage(1, 1).Build() + prov := &queueProvider{id: "test/mock-model", streams: []chat.MessageStream{summaryStream}} + root := agent.New("root", "test", agent.WithModel(prov)) + rt, err := NewLocalRuntime(t.Context(), team.New(team.WithAgents(root)), + WithSessionCompaction(false), WithSessionStore(store), + WithModelStore(mockModelStoreWithLimit{limit: 100_000})) + require.NoError(t, err) + + sess := session.New(session.WithID("compact-fails"), session.WithMessages([]session.Item{ + session.NewMessageItem(session.UserMessage("hi")), + session.NewMessageItem(&session.Message{Message: chat.Message{Role: chat.MessageRoleAssistant, Content: "hello"}}), + })) + require.NoError(t, base.AddSession(t.Context(), sess)) + before := len(sess.Messages) + + events := make(chan Event, 32) + rt.Summarize(t.Context(), sess, "", NewChannelSink(events)) + close(events) + var outcome string + var sawError, sawSummary bool + for ev := range events { + switch e := ev.(type) { + case *SessionCompactionEvent: + if e.Status == "completed" { + outcome = e.Outcome + } + case *ErrorEvent: + sawError = true + case *SessionSummaryEvent: + sawSummary = true + } + } + assert.Equal(t, CompactionOutcomeFailed, outcome) + assert.True(t, sawError) + assert.False(t, sawSummary, "an unpersisted summary must not be announced as successful") + assert.Len(t, sess.Messages, before, "failed persistence must leave live continuation unchanged") +} + +func TestDoCompactPersistsSummaryForSQLiteReload(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "sessions.db") + store, err := sqlitestore.New(t.Context(), dbPath) + require.NoError(t, err) + + summaryStream := newStreamBuilder().AddContent("persisted summary").AddStopWithUsage(12, 3).Build() + prov := &queueProvider{id: "test/mock-model", streams: []chat.MessageStream{summaryStream}} + root := agent.New("root", "test", agent.WithModel(prov)) + rt, err := NewLocalRuntime(t.Context(), team.New(team.WithAgents(root)), + WithSessionCompaction(false), + WithSessionStore(store), + WithModelStore(mockModelStoreWithLimit{limit: 100_000}), + ) + require.NoError(t, err) + + sess := session.New( + session.WithID("compact-reload"), + session.WithMessages([]session.Item{ + session.NewMessageItem(session.UserMessage("old question")), + session.NewMessageItem(&session.Message{Message: chat.Message{Role: chat.MessageRoleAssistant, Content: "old answer"}}), + }), + ) + require.NoError(t, store.AddSession(t.Context(), sess)) + + events := make(chan Event, 32) + rt.Summarize(t.Context(), sess, "", NewChannelSink(events)) + close(events) + for range events { + } + require.NoError(t, store.Close()) + + reopened, err := sqlitestore.New(t.Context(), dbPath) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, reopened.Close()) }) + reloaded, err := reopened.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + require.Len(t, reloaded.Messages, 3) + assert.Equal(t, "persisted summary", reloaded.Messages[2].Summary) + + messages := reloaded.GetMessages(root) + var continuation strings.Builder + for _, msg := range messages { + continuation.WriteString(msg.Content) + } + assert.Contains(t, continuation.String(), "persisted summary") + assert.NotContains(t, continuation.String(), "old question", "reload must continue from the compacted view") +} + +func TestDoCompactObservedRunDoesNotDuplicateSummary(t *testing.T) { + dbPath := filepath.Join(t.TempDir(), "sessions.db") + store, err := sqlitestore.New(t.Context(), dbPath) + require.NoError(t, err) + summaryStream := newStreamBuilder().AddContent("one summary").AddStopWithUsage(1, 1).Build() + prov := &queueProvider{id: "test/mock-model", streams: []chat.MessageStream{summaryStream}} + root := agent.New("root", "test", agent.WithModel(prov)) + rt, err := NewLocalRuntime(t.Context(), team.New(team.WithAgents(root)), + WithSessionCompaction(false), + WithSessionStore(store), + WithModelStore(mockModelStoreWithLimit{limit: 100_000}), + ) + require.NoError(t, err) + + sess := session.New(session.WithID("compact-dedup"), session.WithMessages([]session.Item{ + session.NewMessageItem(session.UserMessage("hi")), + session.NewMessageItem(&session.Message{Message: chat.Message{Role: chat.MessageRoleAssistant, Content: "hello"}}), + })) + require.NoError(t, store.AddSession(t.Context(), sess)) + + inner := make(chan Event, 32) + observed := rt.observe(t.Context(), sess, inner) + rt.compactWithReason(t.Context(), sess, "", compactionReasonManual, NewChannelSink(inner)) + close(inner) + for range observed { + } + + reloaded, err := store.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + var summaries int + for _, item := range reloaded.Messages { + if item.Summary != "" { + summaries++ + } + } + assert.Equal(t, 1, summaries, "intrinsic persistence and the observer must not both append the summary") +} + // TestDoCompactNoHooksMatchesPriorBehavior is a regression guard: with // no compaction-related hooks configured, compactWithReason must still // emit the same SessionCompaction started/completed pair that all diff --git a/pkg/session/session.go b/pkg/session/session.go index 982630e9e..0ea600f42 100644 --- a/pkg/session/session.go +++ b/pkg/session/session.go @@ -914,15 +914,25 @@ func (s *Session) TitleSnapshot() string { return s.Title } -// ApplyCompaction atomically resets the session's cumulative token -// counts and appends a summary item under s.mu so concurrent readers -// (e.g. the persistence observer's UpdateSession snapshot) cannot -// observe the new tokens without the matching summary item. +// ApplyCompaction atomically resets the session's cumulative token counts, +// derives scalar cost from canonical item history including the new summary, +// and appends that summary under s.mu. func (s *Session) ApplyCompaction(inputTokens, outputTokens int64, item Item) { s.mu.Lock() defer s.mu.Unlock() + s.applyCompactionLocked(inputTokens, outputTokens, s.totalCostLocked()+item.Cost, item) +} + +func (s *Session) applyCompaction(inputTokens, outputTokens int64, resultingCost float64, item Item) { + s.mu.Lock() + defer s.mu.Unlock() + s.applyCompactionLocked(inputTokens, outputTokens, resultingCost, item) +} + +func (s *Session) applyCompactionLocked(inputTokens, outputTokens int64, resultingCost float64, item Item) { s.InputTokens = inputTokens s.OutputTokens = outputTokens + s.Cost = resultingCost s.Messages = append(s.Messages, item) } @@ -1666,7 +1676,10 @@ func (s *Session) MessagesSnapshot() []Item { func (s *Session) TotalCost() float64 { s.mu.RLock() defer s.mu.RUnlock() + return s.totalCostLocked() +} +func (s *Session) totalCostLocked() float64 { var cost float64 for _, item := range s.Messages { switch { diff --git a/pkg/session/store.go b/pkg/session/store.go index 72ee4ed21..0e9006383 100644 --- a/pkg/session/store.go +++ b/pkg/session/store.go @@ -119,6 +119,12 @@ type Store interface { // The sub-session is stored as a separate session row with parent_id set. AddSubSession(ctx context.Context, parentSessionID string, subSession *Session) error + // PersistCompaction atomically upserts session metadata and its summary item. + // A missing row is created (matching UpdateSession); an existing row is only + // updated when its origin matches. Implementations must apply resulting cost + // and must not append twice when session aliases the stored live object. + PersistCompaction(ctx context.Context, session *Session, inputTokens, outputTokens int64, item Item) error + // AddSummary adds a summary item to a session at the next position. // item.FirstKeptEntry is the index of the first message kept verbatim during // compaction; item.Cost/Model/Usage attribute the summary's spend (zero @@ -362,6 +368,60 @@ func (s *InMemorySessionStore) AddSubSession(_ context.Context, parentSessionID return nil } +func compactionSessionSnapshot(session *Session, inputTokens, outputTokens int64, item Item) (*Session, float64) { + session.mu.RLock() + defer session.mu.RUnlock() + resultingCost := session.totalCostLocked() + item.Cost + return &Session{ + ID: session.ID, + Origin: session.Origin, + Title: session.Title, + CreatedAt: session.CreatedAt, + ToolsApproved: session.ToolsApproved, + SafetyPolicy: session.SafetyPolicy, + HideToolResults: session.HideToolResults, + WorkingDir: session.WorkingDir, + SendUserMessage: session.SendUserMessage, + MaxIterations: session.MaxIterations, + Starred: session.Starred, + InputTokens: inputTokens, + OutputTokens: outputTokens, + Cost: resultingCost, + Permissions: session.Permissions.Clone(), + Attributes: maps.Clone(session.Attributes), + AgentModelOverrides: cloneStringMap(session.AgentModelOverrides), + CustomModelsUsed: cloneStringSlice(session.CustomModelsUsed), + InstructionContext: cloneInstructionContext(session.InstructionContext), + ParentID: session.ParentID, + }, resultingCost +} + +// PersistCompaction atomically reflects a successful compaction in the stored +// session and applies it to compacted. The common in-memory case stores the +// live session pointer, so the operation must append exactly once. +func (s *InMemorySessionStore) PersistCompaction(_ context.Context, compacted *Session, inputTokens, outputTokens int64, item Item) error { + if compacted.ID == "" { + return ErrEmptyID + } + snapshot, resultingCost := compactionSessionSnapshot(compacted, inputTokens, outputTokens, item) + stored, exists := s.sessions.Load(snapshot.ID) + if !exists { + compacted.applyCompaction(inputTokens, outputTokens, resultingCost, item) + s.sessions.Store(snapshot.ID, compacted) + return nil + } + if stored.Origin != snapshot.Origin { + return fmt.Errorf("persist compaction %q: %w", snapshot.ID, ErrOriginMismatch) + } + if stored == compacted { + compacted.applyCompaction(inputTokens, outputTokens, resultingCost, item) + return nil + } + stored.applyCompaction(inputTokens, outputTokens, resultingCost, item) + compacted.applyCompaction(inputTokens, outputTokens, resultingCost, item) + return nil +} + // AddSummary adds a summary item to a session at the next position. func (s *InMemorySessionStore) AddSummary(_ context.Context, sessionID string, item Item) error { if sessionID == "" { @@ -1276,6 +1336,68 @@ func (s *SQLiteSessionStore) addItemTx(ctx context.Context, tx *sql.Tx, sessionI } } +// PersistCompaction commits the compaction metadata and summary row in one +// transaction so a reload cannot observe only half of the continuation state. +func (s *SQLiteSessionStore) PersistCompaction(ctx context.Context, compacted *Session, inputTokens, outputTokens int64, item Item) error { + if compacted.ID == "" { + return ErrEmptyID + } + usageJSON, err := summaryUsageJSON(item.Usage) + if err != nil { + return err + } + snapshot, resultingCost := compactionSessionSnapshot(compacted, inputTokens, outputTokens, item) + fields, err := sessionPersistedFieldsOf(snapshot) + if err != nil { + return err + } + + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + + result, err := tx.ExecContext(ctx, + `INSERT INTO sessions ( + id, origin, tools_approved, safety_policy, input_tokens, output_tokens, title, cost, send_user_message, + max_iterations, working_dir, created_at, starred, permissions, agent_model_overrides, + custom_models_used, thinking, parent_id, instruction_context, attributes + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + input_tokens = excluded.input_tokens, + output_tokens = excluded.output_tokens, + cost = excluded.cost + WHERE sessions.origin = excluded.origin`, + snapshot.ID, snapshot.Origin, snapshot.ToolsApproved, string(snapshot.SafetyPolicy), snapshot.InputTokens, snapshot.OutputTokens, + snapshot.Title, snapshot.Cost, snapshot.SendUserMessage, snapshot.MaxIterations, snapshot.WorkingDir, + snapshot.CreatedAt.Format(time.RFC3339), snapshot.Starred, fields.PermissionsJSON, fields.AgentModelOverridesJSON, + fields.CustomModelsUsedJSON, false, fields.ParentID, fields.InstructionContextJSON, fields.AttributesJSON) + if err != nil { + return err + } + rowsAffected, err := result.RowsAffected() + if err != nil { + return err + } + if rowsAffected == 0 { + return fmt.Errorf("persist compaction %q: %w", snapshot.ID, ErrOriginMismatch) + } + + _, err = tx.ExecContext(ctx, + `INSERT INTO session_items (session_id, position, item_type, summary_text, first_kept_entry, cost, model, usage_json) + VALUES (?, (SELECT COALESCE(MAX(position), -1) + 1 FROM session_items WHERE session_id = ?), 'summary', ?, ?, ?, ?, ?)`, + snapshot.ID, snapshot.ID, item.Summary, item.FirstKeptEntry, item.Cost, item.Model, usageJSON) + if err != nil { + return err + } + if err := tx.Commit(); err != nil { + return err + } + compacted.applyCompaction(inputTokens, outputTokens, resultingCost, item) + return nil +} + // AddSummary adds a summary item to a session at the next position. func (s *SQLiteSessionStore) AddSummary(ctx context.Context, sessionID string, item Item) error { if sessionID == "" { diff --git a/pkg/session/store_compaction_test.go b/pkg/session/store_compaction_test.go new file mode 100644 index 000000000..cb3c9addd --- /dev/null +++ b/pkg/session/store_compaction_test.go @@ -0,0 +1,142 @@ +package session + +import ( + "path/filepath" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/chat" + "github.com/docker/docker-agent/pkg/sqliteutil" +) + +func newConcreteSQLiteStoreForCompactionTest(t *testing.T) *SQLiteSessionStore { + t.Helper() + db, err := sqliteutil.OpenDB(t.Context(), filepath.Join(t.TempDir(), "sessions.db")) + require.NoError(t, err) + store, err := NewSQLiteSessionStoreFromDB(t.Context(), db) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, store.Close()) }) + return store +} + +func newInMemoryStoreForCompactionTest(t *testing.T) Store { + t.Helper() + return NewInMemorySessionStore() +} + +func newSQLiteStoreForCompactionTest(t *testing.T) Store { + t.Helper() + return newConcreteSQLiteStoreForCompactionTest(t) +} + +func TestPersistCompactionStoresResultingCost(t *testing.T) { + tests := []struct { + name string + store func(*testing.T) Store + }{ + {name: "memory", store: newInMemoryStoreForCompactionTest}, + {name: "sqlite", store: newSQLiteStoreForCompactionTest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := tt.store(t) + sess := New(WithID("cost"), WithMessages([]Item{ + NewMessageItem(&Message{Message: chat.Message{Role: chat.MessageRoleAssistant, Content: "prior", Cost: 1.25}}), + })) + // Deliberately stale legacy scalar: canonical cost comes from items. + sess.SetTokensAndCost(100, 20, 99) + require.NoError(t, store.AddSession(t.Context(), sess)) + + require.NoError(t, store.PersistCompaction(t.Context(), sess, 7, 0, Item{Summary: "summary", Cost: 0.75})) + reloaded, err := store.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + input, output, cost := reloaded.TokensAndCost() + assert.Equal(t, int64(7), input) + assert.Zero(t, output) + assert.InDelta(t, 2.0, cost, 1e-9) + require.Len(t, reloaded.Messages, 2) + assert.InDelta(t, 0.75, reloaded.Messages[1].Cost, 1e-9) + }) + } +} + +func TestPersistCompactionMissingRowUpsertsAcrossStores(t *testing.T) { + tests := []struct { + name string + store func(*testing.T) Store + }{ + {name: "memory", store: newInMemoryStoreForCompactionTest}, + {name: "sqlite", store: newSQLiteStoreForCompactionTest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := tt.store(t) + sess := New(WithID("missing")) + sess.SetTokensAndCost(50, 10, 0.5) + require.NoError(t, store.PersistCompaction(t.Context(), sess, 5, 0, Item{Summary: "summary", Cost: 0.25})) + + reloaded, err := store.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + input, output, cost := reloaded.TokensAndCost() + assert.Equal(t, int64(5), input) + assert.Zero(t, output) + assert.InDelta(t, 0.25, cost, 1e-9) + require.Len(t, reloaded.Messages, 1) + assert.Equal(t, "summary", reloaded.Messages[0].Summary) + }) + } +} + +func TestSQLitePersistCompactionSnapshotsConcurrentMetadata(t *testing.T) { + store := newConcreteSQLiteStoreForCompactionTest(t) + sess := New(WithID("race")) + require.NoError(t, store.AddSession(t.Context(), sess)) + + var wg sync.WaitGroup + wg.Go(func() { + for i := range 200 { + sess.SetTitle("title") + sess.SetToolsApproved(i%2 == 0) + sess.SetSafetyPolicy(SafetyPolicy("standard")) + sess.SetAttribute("iteration", "value") + sess.SetPermissions(&PermissionsConfig{}) + } + }) + for range 50 { + err := store.PersistCompaction(t.Context(), sess, 7, 0, Item{Summary: "summary"}) + require.NoError(t, err) + } + wg.Wait() +} + +func TestSQLitePersistCompactionRollsBackMetadataWhenSummaryInsertFails(t *testing.T) { + store := newConcreteSQLiteStoreForCompactionTest(t) + sess := New(WithID("rollback")) + sess.SetTokensAndCost(100, 20, 1.25) + require.NoError(t, store.AddSession(t.Context(), sess)) + require.NoError(t, func() error { + _, err := store.db.ExecContext(t.Context(), ` + CREATE TRIGGER fail_compaction_summary + BEFORE INSERT ON session_items + WHEN NEW.item_type = 'summary' + BEGIN + SELECT RAISE(ABORT, 'injected summary failure'); + END`) + return err + }()) + + err := store.PersistCompaction(t.Context(), sess, 7, 0, Item{Summary: "must roll back", Cost: 0.75}) + require.ErrorContains(t, err, "injected summary failure") + + reloaded, err := store.GetSession(t.Context(), sess.ID) + require.NoError(t, err) + input, output, cost := reloaded.TokensAndCost() + assert.Equal(t, int64(100), input) + assert.Equal(t, int64(20), output) + assert.InDelta(t, 1.25, cost, 1e-9) + assert.Empty(t, reloaded.Messages) + assert.Empty(t, sess.Messages, "live state must not mutate before transaction commit") +}