Persist compaction checkpoints in sessions
- commit
- c98e701843a7c2961fba08dc69d9a65f3ed054ef
- author
- patriceckhart · · 2026-05-09 21:02 UTC
- parent
- facc709
- signature
- unsigned
4 files · +99 -12
M internal/agent/cli.go +11 -0
| @@ -575,12 +575,23 @@ func runInteractive(ctx context.Context, args Args, version string) error { | |||
| 575 | 575 | } | |
| 576 | 576 | _ = sess.AppendUsage(cum, cum) | |
| 577 | 577 | } | |
| 578 | + | persistCompaction := func(messages []provider.Message) { | |
| 579 | + | persistMu.Lock() | |
| 580 | + | defer persistMu.Unlock() | |
| 581 | + | if sess == nil { | |
| 582 | + | return | |
| 583 | + | } | |
| 584 | + | if err := sess.AppendCompaction(messages); err == nil { | |
| 585 | + | sessBaselineMsgs = len(messages) | |
| 586 | + | } | |
| 587 | + | } | |
| 578 | 588 | wireAgentPersist := func(a *core.Agent) *core.Agent { | |
| 579 | 589 | if a == nil { | |
| 580 | 590 | return a | |
| 581 | 591 | } | |
| 582 | 592 | a.OnMessageAppended = persistMessage | |
| 583 | 593 | a.OnUsage = persistUsage | |
| 594 | + | a.OnTranscriptCompacted = persistCompaction | |
| 584 | 595 | return a | |
| 585 | 596 | } | |
| 586 | 597 | wireAgentPersist(ag) | |
| @@ -575,12 +575,23 @@ func runInteractive(ctx context.Context, args Args, version string) error { | |||||
| 575 | } | 575 | } | ||
| 576 | _ = sess.AppendUsage(cum, cum) | 576 | _ = sess.AppendUsage(cum, cum) | ||
| 577 | } | 577 | } | ||
| 578 | + | persistCompaction := func(messages []provider.Message) { | |||
| 579 | + | persistMu.Lock() | |||
| 580 | + | defer persistMu.Unlock() | |||
| 581 | + | if sess == nil { | |||
| 582 | + | return | |||
| 583 | + | } | |||
| 584 | + | if err := sess.AppendCompaction(messages); err == nil { | |||
| 585 | + | sessBaselineMsgs = len(messages) | |||
| 586 | + | } | |||
| 587 | + | } | |||
| 578 | wireAgentPersist := func(a *core.Agent) *core.Agent { | 588 | wireAgentPersist := func(a *core.Agent) *core.Agent { | ||
| 579 | if a == nil { | 589 | if a == nil { | ||
| 580 | return a | 590 | return a | ||
| 581 | } | 591 | } | ||
| 582 | a.OnMessageAppended = persistMessage | 592 | a.OnMessageAppended = persistMessage | ||
| 583 | a.OnUsage = persistUsage | 593 | a.OnUsage = persistUsage | ||
| 594 | + | a.OnTranscriptCompacted = persistCompaction | |||
| 584 | return a | 595 | return a | ||
| 585 | } | 596 | } | ||
| 586 | wireAgentPersist(ag) | 597 | wireAgentPersist(ag) | ||
M internal/core/agent.go +7 -0
| @@ -66,6 +66,13 @@ type Agent struct { | |||
| 66 | 66 | // current and a crash recovers the right cost figure. | |
| 67 | 67 | OnUsage func(cumulative provider.Usage) | |
| 68 | 68 | ||
| 69 | + | // OnTranscriptCompacted, if set, fires after Compact replaces the | |
| 70 | + | // in-memory transcript with the synthetic summary plus kept tail. | |
| 71 | + | // Hosts wire this to append an explicit compaction checkpoint to | |
| 72 | + | // the session log; per-message append hooks do not fire for this | |
| 73 | + | // wholesale transcript replacement. | |
| 74 | + | OnTranscriptCompacted func(messages []provider.Message) | |
| 75 | + | ||
| 69 | 76 | mu sync.Mutex | |
| 70 | 77 | messages []provider.Message | |
| 71 | 78 | // rev increments whenever the transcript slice is replaced or a | |
| @@ -66,6 +66,13 @@ type Agent struct { | |||||
| 66 | // current and a crash recovers the right cost figure. | 66 | // current and a crash recovers the right cost figure. | ||
| 67 | OnUsage func(cumulative provider.Usage) | 67 | OnUsage func(cumulative provider.Usage) | ||
| 68 | 68 | ||||
| 69 | + | // OnTranscriptCompacted, if set, fires after Compact replaces the | |||
| 70 | + | // in-memory transcript with the synthetic summary plus kept tail. | |||
| 71 | + | // Hosts wire this to append an explicit compaction checkpoint to | |||
| 72 | + | // the session log; per-message append hooks do not fire for this | |||
| 73 | + | // wholesale transcript replacement. | |||
| 74 | + | OnTranscriptCompacted func(messages []provider.Message) | |||
| 75 | + | ||||
| 69 | mu sync.Mutex | 76 | mu sync.Mutex | ||
| 70 | messages []provider.Message | 77 | messages []provider.Message | ||
| 71 | // rev increments whenever the transcript slice is replaced or a | 78 | // rev increments whenever the transcript slice is replaced or a | ||
M internal/core/compact.go +7 -0
| @@ -113,8 +113,15 @@ func (a *Agent) Compact(ctx context.Context, keepTail int, sink func(delta strin | |||
| 113 | 113 | ||
| 114 | 114 | a.mu.Lock() | |
| 115 | 115 | a.messages = next | |
| 116 | + | a.rev++ | |
| 117 | + | onCompacted := a.OnTranscriptCompacted | |
| 118 | + | persisted := append([]provider.Message(nil), next...) | |
| 116 | 119 | a.mu.Unlock() | |
| 117 | 120 | ||
| 121 | + | if onCompacted != nil { | |
| 122 | + | onCompacted(persisted) | |
| 123 | + | } | |
| 124 | + | ||
| 118 | 125 | return summary, nil | |
| 119 | 126 | } | |
| 120 | 127 | ||
| @@ -113,8 +113,15 @@ func (a *Agent) Compact(ctx context.Context, keepTail int, sink func(delta strin | |||||
| 113 | 113 | ||||
| 114 | a.mu.Lock() | 114 | a.mu.Lock() | ||
| 115 | a.messages = next | 115 | a.messages = next | ||
| 116 | + | a.rev++ | |||
| 117 | + | onCompacted := a.OnTranscriptCompacted | |||
| 118 | + | persisted := append([]provider.Message(nil), next...) | |||
| 116 | a.mu.Unlock() | 119 | a.mu.Unlock() | ||
| 117 | 120 | ||||
| 121 | + | if onCompacted != nil { | |||
| 122 | + | onCompacted(persisted) | |||
| 123 | + | } | |||
| 124 | + | ||||
| 118 | return summary, nil | 125 | return summary, nil | ||
| 119 | } | 126 | } | ||
| 120 | 127 | ||||
M internal/core/session.go +74 -12
| @@ -66,11 +66,12 @@ type SessionMeta struct { | |||
| 66 | 66 | // the default unmarshaler cannot reconstruct); it is written with a | |
| 67 | 67 | // regular provider.Message value. | |
| 68 | 68 | type sessionLine struct { | |
| 69 | - | Type string `json:"type"` | |
| 70 | - | Meta *SessionMeta `json:"meta,omitempty"` | |
| 71 | - | Message *provider.Message `json:"message,omitempty"` | |
| 72 | - | Usage *provider.Usage `json:"usage,omitempty"` | |
| 73 | - | Cumulative *provider.Usage `json:"cumulative,omitempty"` | |
| 69 | + | Type string `json:"type"` | |
| 70 | + | Meta *SessionMeta `json:"meta,omitempty"` | |
| 71 | + | Message *provider.Message `json:"message,omitempty"` | |
| 72 | + | Messages []provider.Message `json:"messages,omitempty"` | |
| 73 | + | Usage *provider.Usage `json:"usage,omitempty"` | |
| 74 | + | Cumulative *provider.Usage `json:"cumulative,omitempty"` | |
| 74 | 75 | } | |
| 75 | 76 | ||
| 76 | 77 | type sessionLineHead struct { | |
| @@ -190,6 +191,10 @@ func OpenSession(path string) (*Session, []provider.Message, error) { | |||
| 190 | 191 | if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 { | |
| 191 | 192 | messages = append(messages, msg) | |
| 192 | 193 | } | |
| 194 | + | case "compaction": | |
| 195 | + | if compacted, err := hydrateCompaction(line); err == nil { | |
| 196 | + | messages = compacted | |
| 197 | + | } | |
| 193 | 198 | } | |
| 194 | 199 | return nil | |
| 195 | 200 | }); err != nil { | |
| @@ -373,6 +378,13 @@ func describeSession(path string) SessionSummary { | |||
| 373 | 378 | if s.FirstUserText == "" { | |
| 374 | 379 | s.FirstUserText = firstUserText(line) | |
| 375 | 380 | } | |
| 381 | + | case "compaction": | |
| 382 | + | if compacted, err := hydrateCompaction(line); err == nil { | |
| 383 | + | s.MessageCount = len(compacted) | |
| 384 | + | if s.FirstUserText == "" && len(compacted) > 0 { | |
| 385 | + | s.FirstUserText = firstTextFromMessage(compacted[0]) | |
| 386 | + | } | |
| 387 | + | } | |
| 376 | 388 | case "rename": | |
| 377 | 389 | var row struct { | |
| 378 | 390 | Title string `json:"title"` | |
| @@ -416,6 +428,15 @@ func firstUserText(line []byte) string { | |||
| 416 | 428 | return "" | |
| 417 | 429 | } | |
| 418 | 430 | ||
| 431 | + | func firstTextFromMessage(msg provider.Message) string { | |
| 432 | + | for _, c := range msg.Content { | |
| 433 | + | if tb, ok := c.(provider.TextBlock); ok && tb.Text != "" { | |
| 434 | + | return tb.Text | |
| 435 | + | } | |
| 436 | + | } | |
| 437 | + | return "" | |
| 438 | + | } | |
| 439 | + | ||
| 419 | 440 | // PruneEmptySessions deletes session files in cwd's session directory | |
| 420 | 441 | // that contain only a meta line (no messages were ever appended). | |
| 421 | 442 | // Cleans up the backlog of empty stubs created by old zot versions | |
| @@ -517,6 +538,21 @@ func (s *Session) AppendMessage(m provider.Message) error { | |||
| 517 | 538 | return nil | |
| 518 | 539 | } | |
| 519 | 540 | ||
| 541 | + | // AppendCompaction writes a checkpoint that replaces all earlier | |
| 542 | + | // transcript rows when the session is resumed. The old rows remain in | |
| 543 | + | // the JSONL file for audit/export, while loaders use the latest | |
| 544 | + | // compaction row as the effective transcript. | |
| 545 | + | func (s *Session) AppendCompaction(messages []provider.Message) error { | |
| 546 | + | if s == nil { | |
| 547 | + | return nil | |
| 548 | + | } | |
| 549 | + | if err := s.writeLine(sessionLine{Type: "compaction", Messages: messages}); err != nil { | |
| 550 | + | return err | |
| 551 | + | } | |
| 552 | + | s.messagesAppended = len(messages) | |
| 553 | + | return nil | |
| 554 | + | } | |
| 555 | + | ||
| 520 | 556 | // UpdateModel records a provider/model switch in the session file. | |
| 521 | 557 | // The reader keeps the most recent meta entry, so the session resumes | |
| 522 | 558 | // with the updated model. | |
| @@ -580,19 +616,45 @@ func (s *Session) writeLine(row sessionLine) error { | |||
| 580 | 616 | // We persist messages by reading the raw "message" object back and | |
| 581 | 617 | // rebuilding Content from discriminated fields. | |
| 582 | 618 | ||
| 619 | + | func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) { | |
| 620 | + | var row struct { | |
| 621 | + | Messages []json.RawMessage `json:"messages"` | |
| 622 | + | } | |
| 623 | + | if err := json.Unmarshal(lineBytes, &row); err != nil { | |
| 624 | + | return nil, err | |
| 625 | + | } | |
| 626 | + | messages := make([]provider.Message, 0, len(row.Messages)) | |
| 627 | + | for _, raw := range row.Messages { | |
| 628 | + | msg, err := hydrateMessageObject(raw) | |
| 629 | + | if err == nil && len(msg.Content) > 0 { | |
| 630 | + | messages = append(messages, msg) | |
| 631 | + | } | |
| 632 | + | } | |
| 633 | + | return messages, nil | |
| 634 | + | } | |
| 635 | + | ||
| 583 | 636 | func hydrateMessage(lineBytes []byte) (provider.Message, error) { | |
| 584 | 637 | var row struct { | |
| 585 | - | Message struct { | |
| 586 | - | Role provider.Role `json:"role"` | |
| 587 | - | Content []json.RawMessage `json:"content"` | |
| 588 | - | Time time.Time `json:"time"` | |
| 589 | - | } `json:"message"` | |
| 638 | + | Message json.RawMessage `json:"message"` | |
| 590 | 639 | } | |
| 591 | 640 | if err := json.Unmarshal(lineBytes, &row); err != nil { | |
| 592 | 641 | return provider.Message{}, err | |
| 593 | 642 | } | |
| 594 | - | msg := provider.Message{Role: row.Message.Role, Time: row.Message.Time} | |
| 595 | - | for _, raw := range row.Message.Content { | |
| 643 | + | return hydrateMessageObject(row.Message) | |
| 644 | + | } | |
| 645 | + | ||
| 646 | + | func hydrateMessageObject(rawMessage []byte) (provider.Message, error) { | |
| 647 | + | var row struct { | |
| 648 | + | Role provider.Role `json:"role"` | |
| 649 | + | Content []json.RawMessage `json:"content"` | |
| 650 | + | Time time.Time `json:"time"` | |
| 651 | + | Meta map[string]string `json:"meta,omitempty"` | |
| 652 | + | } | |
| 653 | + | if err := json.Unmarshal(rawMessage, &row); err != nil { | |
| 654 | + | return provider.Message{}, err | |
| 655 | + | } | |
| 656 | + | msg := provider.Message{Role: row.Role, Time: row.Time, Meta: row.Meta} | |
| 657 | + | for _, raw := range row.Content { | |
| 596 | 658 | var head struct { | |
| 597 | 659 | Text string `json:"text"` | |
| 598 | 660 | MimeType string `json:"mime_type"` | |
| @@ -66,11 +66,12 @@ type SessionMeta struct { | |||||
| 66 | // the default unmarshaler cannot reconstruct); it is written with a | 66 | // the default unmarshaler cannot reconstruct); it is written with a | ||
| 67 | // regular provider.Message value. | 67 | // regular provider.Message value. | ||
| 68 | type sessionLine struct { | 68 | type sessionLine struct { | ||
| 69 | - | Type string `json:"type"` | 69 | + | Type string `json:"type"` |
| 70 | - | Meta *SessionMeta `json:"meta,omitempty"` | 70 | + | Meta *SessionMeta `json:"meta,omitempty"` |
| 71 | - | Message *provider.Message `json:"message,omitempty"` | 71 | + | Message *provider.Message `json:"message,omitempty"` |
| 72 | - | Usage *provider.Usage `json:"usage,omitempty"` | 72 | + | Messages []provider.Message `json:"messages,omitempty"` |
| 73 | - | Cumulative *provider.Usage `json:"cumulative,omitempty"` | 73 | + | Usage *provider.Usage `json:"usage,omitempty"` |
| 74 | + | Cumulative *provider.Usage `json:"cumulative,omitempty"` | |||
| 74 | } | 75 | } | ||
| 75 | 76 | ||||
| 76 | type sessionLineHead struct { | 77 | type sessionLineHead struct { | ||
| @@ -190,6 +191,10 @@ func OpenSession(path string) (*Session, []provider.Message, error) { | |||||
| 190 | if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 { | 191 | if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 { | ||
| 191 | messages = append(messages, msg) | 192 | messages = append(messages, msg) | ||
| 192 | } | 193 | } | ||
| 194 | + | case "compaction": | |||
| 195 | + | if compacted, err := hydrateCompaction(line); err == nil { | |||
| 196 | + | messages = compacted | |||
| 197 | + | } | |||
| 193 | } | 198 | } | ||
| 194 | return nil | 199 | return nil | ||
| 195 | }); err != nil { | 200 | }); err != nil { | ||
| @@ -373,6 +378,13 @@ func describeSession(path string) SessionSummary { | |||||
| 373 | if s.FirstUserText == "" { | 378 | if s.FirstUserText == "" { | ||
| 374 | s.FirstUserText = firstUserText(line) | 379 | s.FirstUserText = firstUserText(line) | ||
| 375 | } | 380 | } | ||
| 381 | + | case "compaction": | |||
| 382 | + | if compacted, err := hydrateCompaction(line); err == nil { | |||
| 383 | + | s.MessageCount = len(compacted) | |||
| 384 | + | if s.FirstUserText == "" && len(compacted) > 0 { | |||
| 385 | + | s.FirstUserText = firstTextFromMessage(compacted[0]) | |||
| 386 | + | } | |||
| 387 | + | } | |||
| 376 | case "rename": | 388 | case "rename": | ||
| 377 | var row struct { | 389 | var row struct { | ||
| 378 | Title string `json:"title"` | 390 | Title string `json:"title"` | ||
| @@ -416,6 +428,15 @@ func firstUserText(line []byte) string { | |||||
| 416 | return "" | 428 | return "" | ||
| 417 | } | 429 | } | ||
| 418 | 430 | ||||
| 431 | + | func firstTextFromMessage(msg provider.Message) string { | |||
| 432 | + | for _, c := range msg.Content { | |||
| 433 | + | if tb, ok := c.(provider.TextBlock); ok && tb.Text != "" { | |||
| 434 | + | return tb.Text | |||
| 435 | + | } | |||
| 436 | + | } | |||
| 437 | + | return "" | |||
| 438 | + | } | |||
| 439 | + | ||||
| 419 | // PruneEmptySessions deletes session files in cwd's session directory | 440 | // PruneEmptySessions deletes session files in cwd's session directory | ||
| 420 | // that contain only a meta line (no messages were ever appended). | 441 | // that contain only a meta line (no messages were ever appended). | ||
| 421 | // Cleans up the backlog of empty stubs created by old zot versions | 442 | // Cleans up the backlog of empty stubs created by old zot versions | ||
| @@ -517,6 +538,21 @@ func (s *Session) AppendMessage(m provider.Message) error { | |||||
| 517 | return nil | 538 | return nil | ||
| 518 | } | 539 | } | ||
| 519 | 540 | ||||
| 541 | + | // AppendCompaction writes a checkpoint that replaces all earlier | |||
| 542 | + | // transcript rows when the session is resumed. The old rows remain in | |||
| 543 | + | // the JSONL file for audit/export, while loaders use the latest | |||
| 544 | + | // compaction row as the effective transcript. | |||
| 545 | + | func (s *Session) AppendCompaction(messages []provider.Message) error { | |||
| 546 | + | if s == nil { | |||
| 547 | + | return nil | |||
| 548 | + | } | |||
| 549 | + | if err := s.writeLine(sessionLine{Type: "compaction", Messages: messages}); err != nil { | |||
| 550 | + | return err | |||
| 551 | + | } | |||
| 552 | + | s.messagesAppended = len(messages) | |||
| 553 | + | return nil | |||
| 554 | + | } | |||
| 555 | + | ||||
| 520 | // UpdateModel records a provider/model switch in the session file. | 556 | // UpdateModel records a provider/model switch in the session file. | ||
| 521 | // The reader keeps the most recent meta entry, so the session resumes | 557 | // The reader keeps the most recent meta entry, so the session resumes | ||
| 522 | // with the updated model. | 558 | // with the updated model. | ||
| @@ -580,19 +616,45 @@ func (s *Session) writeLine(row sessionLine) error { | |||||
| 580 | // We persist messages by reading the raw "message" object back and | 616 | // We persist messages by reading the raw "message" object back and | ||
| 581 | // rebuilding Content from discriminated fields. | 617 | // rebuilding Content from discriminated fields. | ||
| 582 | 618 | ||||
| 619 | + | func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) { | |||
| 620 | + | var row struct { | |||
| 621 | + | Messages []json.RawMessage `json:"messages"` | |||
| 622 | + | } | |||
| 623 | + | if err := json.Unmarshal(lineBytes, &row); err != nil { | |||
| 624 | + | return nil, err | |||
| 625 | + | } | |||
| 626 | + | messages := make([]provider.Message, 0, len(row.Messages)) | |||
| 627 | + | for _, raw := range row.Messages { | |||
| 628 | + | msg, err := hydrateMessageObject(raw) | |||
| 629 | + | if err == nil && len(msg.Content) > 0 { | |||
| 630 | + | messages = append(messages, msg) | |||
| 631 | + | } | |||
| 632 | + | } | |||
| 633 | + | return messages, nil | |||
| 634 | + | } | |||
| 635 | + | ||||
| 583 | func hydrateMessage(lineBytes []byte) (provider.Message, error) { | 636 | func hydrateMessage(lineBytes []byte) (provider.Message, error) { | ||
| 584 | var row struct { | 637 | var row struct { | ||
| 585 | - | Message struct { | 638 | + | Message json.RawMessage `json:"message"` |
| 586 | - | Role provider.Role `json:"role"` | |||
| 587 | - | Content []json.RawMessage `json:"content"` | |||
| 588 | - | Time time.Time `json:"time"` | |||
| 589 | - | } `json:"message"` | |||
| 590 | } | 639 | } | ||
| 591 | if err := json.Unmarshal(lineBytes, &row); err != nil { | 640 | if err := json.Unmarshal(lineBytes, &row); err != nil { | ||
| 592 | return provider.Message{}, err | 641 | return provider.Message{}, err | ||
| 593 | } | 642 | } | ||
| 594 | - | msg := provider.Message{Role: row.Message.Role, Time: row.Message.Time} | 643 | + | return hydrateMessageObject(row.Message) |
| 595 | - | for _, raw := range row.Message.Content { | 644 | + | } |
| 645 | + | ||||
| 646 | + | func hydrateMessageObject(rawMessage []byte) (provider.Message, error) { | |||
| 647 | + | var row struct { | |||
| 648 | + | Role provider.Role `json:"role"` | |||
| 649 | + | Content []json.RawMessage `json:"content"` | |||
| 650 | + | Time time.Time `json:"time"` | |||
| 651 | + | Meta map[string]string `json:"meta,omitempty"` | |||
| 652 | + | } | |||
| 653 | + | if err := json.Unmarshal(rawMessage, &row); err != nil { | |||
| 654 | + | return provider.Message{}, err | |||
| 655 | + | } | |||
| 656 | + | msg := provider.Message{Role: row.Role, Time: row.Time, Meta: row.Meta} | |||
| 657 | + | for _, raw := range row.Content { | |||
| 596 | var head struct { | 658 | var head struct { | ||
| 597 | Text string `json:"text"` | 659 | Text string `json:"text"` | ||
| 598 | MimeType string `json:"mime_type"` | 660 | MimeType string `json:"mime_type"` | ||