Merge pull request #30 from jameswei/fix/session-fork-after-compaction

fix(core): fork sessions from effective compacted transcript

commit: 7f954ceaa38f5c9ab8fbd742bfe42439cd45efaa
author: Patric Eckhart · 4mo ago · 2026-06-11 09:27 UTC
committer: GitHub · 4mo ago · 2026-06-11 09:27 UTC
merge of: 4685a30 + d1901d0 · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown
refs: tag: v0.2.29

2 files · +129 -18

M packages/core/session_portable.go +52 -18


@@ -13,6 +13,7 @@ import (
13	13		"time"
14	14
15	15		"github.com/google/uuid"
	16	+	"github.com/patriceckhart/zot/packages/provider"
16	17		)
17	18
18	19		// PortableExt is the filesystem extension used for exported sessions.

@@ -321,43 +322,76 @@ func BranchSession(parentPath, root, cwd, version string, upToMessageIdx int) (s
321	322		return "", err
322	323		}
323	324
324		-	// Copy message rows up to the cut point, plus all usage rows
325		-	// that land before the cut (they describe the cost of those
326		-	// messages). Rewind and use the large-row-safe JSONL reader.
	325	+	// Reconstruct the effective transcript the same way OpenSession
	326	+	// does: message rows append, and compaction rows replace everything
	327	+	// before them. The fork index is defined over that effective stream,
	328	+	// not over the raw audit rows kept on disk before a compaction.
327	329		if _, err := src.Seek(0, io.SeekStart); err != nil {
328	330		return "", fmt.Errorf("branch: rewind parent: %w", err)
329	331		}
330		-	msgCount := 0
	332	+	var effective []provider.Message
	333	+	var nonCompactedRows [][]byte
	334	+	effectiveCount := 0
	335	+	sawCompaction := false
331	336		if err := forEachJSONLLine(src, func(line []byte) error {
332		-	if msgCount >= upToMessageIdx {
333		-	return io.EOF
334		-	}
335	337		var h sessionLineHead
336	338		if err := json.Unmarshal(line, &h); err != nil {
337	339		return nil
338	340		}
339	341		switch h.Type {
340	342		case "message":
341		-	if _, err := bw.Write(line); err != nil {
342		-	return err
	343	+	if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 {
	344	+	effective = append(effective, msg)
	345	+	if !sawCompaction && effectiveCount < upToMessageIdx {
	346	+	raw := append([]byte(nil), line...)
	347	+	nonCompactedRows = append(nonCompactedRows, raw)
	348	+	}
	349	+	effectiveCount++
343	350		}
344		-	if err := bw.WriteByte('\n'); err != nil {
345		-	return err
	351	+	case "compaction":
	352	+	if compacted, err := hydrateCompaction(line); err == nil {
	353	+	effective = compacted
	354	+	effectiveCount = len(effective)
	355	+	sawCompaction = true
346	356		}
347		-	msgCount++
348	357		case "usage":
349		-	if _, err := bw.Write(line); err != nil {
350		-	return err
	358	+	if !sawCompaction && effectiveCount < upToMessageIdx {
	359	+	raw := append([]byte(nil), line...)
	360	+	nonCompactedRows = append(nonCompactedRows, raw)
351	361		}
352		-	if err := bw.WriteByte('\n'); err != nil {
353		-	return err
354		-	}
355		-	// don't increment msgCount for usage rows
356	362		}
357	363		return nil
358	364		}); err != nil && err != io.EOF {
359	365		return "", fmt.Errorf("branch: read parent: %w", err)
360	366		}
	367	+	if sawCompaction {
	368	+	limit := upToMessageIdx
	369	+	if limit > len(effective) {
	370	+	limit = len(effective)
	371	+	}
	372	+	for i := 0; i < limit; i++ {
	373	+	msg := effective[i]
	374	+	line, err := json.Marshal(sessionLine{Type: "message", Message: &msg})
	375	+	if err != nil {
	376	+	return "", fmt.Errorf("branch: marshal message: %w", err)
	377	+	}
	378	+	if _, err := bw.Write(line); err != nil {
	379	+	return "", err
	380	+	}
	381	+	if err := bw.WriteByte('\n'); err != nil {
	382	+	return "", err
	383	+	}
	384	+	}
	385	+	} else {
	386	+	for _, row := range nonCompactedRows {
	387	+	if _, err := bw.Write(row); err != nil {
	388	+	return "", err
	389	+	}
	390	+	if err := bw.WriteByte('\n'); err != nil {
	391	+	return "", err
	392	+	}
	393	+	}
	394	+	}
361	395		if err := bw.Flush(); err != nil {
362	396		return "", err
363	397		}

 	"time"
 	"github.com/google/uuid"
+	"github.com/patriceckhart/zot/packages/provider"
 )
 // PortableExt is the filesystem extension used for exported sessions.
 		return "", err
 	}
-	// Copy message rows up to the cut point, plus all usage rows
+	// Reconstruct the effective transcript the same way OpenSession
-	// that land before the cut (they describe the cost of those
+	// does: message rows append, and compaction rows replace everything
-	// messages). Rewind and use the large-row-safe JSONL reader.
+	// before them. The fork index is defined over that effective stream,
+	// not over the raw audit rows kept on disk before a compaction.
 	if _, err := src.Seek(0, io.SeekStart); err != nil {
 		return "", fmt.Errorf("branch: rewind parent: %w", err)
 	}
-	msgCount := 0
+	var effective []provider.Message
+	var nonCompactedRows [][]byte
+	effectiveCount := 0
+	sawCompaction := false
 	if err := forEachJSONLLine(src, func(line []byte) error {
-		if msgCount >= upToMessageIdx {
-			return io.EOF
-		}
 		var h sessionLineHead
 		if err := json.Unmarshal(line, &h); err != nil {
 			return nil
 		}
 		switch h.Type {
 		case "message":
-			if _, err := bw.Write(line); err != nil {
+			if msg, err := hydrateMessage(line); err == nil && len(msg.Content) > 0 {
-				return err
+				effective = append(effective, msg)
+				if !sawCompaction && effectiveCount < upToMessageIdx {
+					raw := append([]byte(nil), line...)
+					nonCompactedRows = append(nonCompactedRows, raw)
+				}
+				effectiveCount++
 			}
-			if err := bw.WriteByte('\n'); err != nil {
+		case "compaction":
-				return err
+			if compacted, err := hydrateCompaction(line); err == nil {
+				effective = compacted
+				effectiveCount = len(effective)
+				sawCompaction = true
 			}
-			msgCount++
 		case "usage":
-			if _, err := bw.Write(line); err != nil {
+			if !sawCompaction && effectiveCount < upToMessageIdx {
-				return err
+				raw := append([]byte(nil), line...)
+				nonCompactedRows = append(nonCompactedRows, raw)
 			}
-			if err := bw.WriteByte('\n'); err != nil {
-				return err
-			}
-			// don't increment msgCount for usage rows
 		}
 		return nil
 	}); err != nil && err != io.EOF {
 		return "", fmt.Errorf("branch: read parent: %w", err)
 	}
+	if sawCompaction {
+		limit := upToMessageIdx
+		if limit > len(effective) {
+			limit = len(effective)
+		}
+		for i := 0; i < limit; i++ {
+			msg := effective[i]
+			line, err := json.Marshal(sessionLine{Type: "message", Message: &msg})
+			if err != nil {
+				return "", fmt.Errorf("branch: marshal message: %w", err)
+			}
+			if _, err := bw.Write(line); err != nil {
+				return "", err
+			}
+			if err := bw.WriteByte('\n'); err != nil {
+				return "", err
+			}
+		}
+	} else {
+		for _, row := range nonCompactedRows {
+			if _, err := bw.Write(row); err != nil {
+				return "", err
+			}
+			if err := bw.WriteByte('\n'); err != nil {
+				return "", err
+			}
+		}
+	}
 	if err := bw.Flush(); err != nil {
 		return "", err
 	}

M packages/core/session_portable_test.go +77 -0


@@ -231,6 +231,71 @@ func TestBranchSessionCopiesPrefix(t *testing.T) {
231	231		}
232	232		}
233	233
	234	+	func TestBranchSessionUsesEffectiveTranscriptAfterCompaction(t *testing.T) {
	235	+	root := t.TempDir()
	236	+	cwd := "/project"
	237	+	parent, err := NewSession(root, cwd, "anthropic", "claude-opus-4-7", "0.0.0-test")
	238	+	if err != nil {
	239	+	t.Fatal(err)
	240	+	}
	241	+	for _, text := range []string{"raw-a", "raw-b", "raw-c", "raw-d"} {
	242	+	_ = parent.AppendMessage(provider.Message{
	243	+	Role: provider.RoleUser,
	244	+	Content: []provider.Content{provider.TextBlock{Text: text}},
	245	+	})
	246	+	}
	247	+	_ = parent.AppendCompaction([]provider.Message{
	248	+	{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "summary"}}},
	249	+	{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "tail-c"}}},
	250	+	{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "tail-d"}}},
	251	+	})
	252	+	_ = parent.AppendMessage(provider.Message{
	253	+	Role: provider.RoleUser,
	254	+	Content: []provider.Content{provider.TextBlock{Text: "after-compact"}},
	255	+	})
	256	+	_ = parent.Close()
	257	+
	258	+	opened, msgs, err := OpenSession(parent.Path)
	259	+	if err != nil {
	260	+	t.Fatalf("OpenSession parent: %v", err)
	261	+	}
	262	+	_ = opened.Close()
	263	+	assertMessageTexts(t, msgs, []string{"summary", "tail-c", "tail-d", "after-compact"})
	264	+
	265	+	branchPath, err := BranchSession(parent.Path, root, cwd, "0.0.0-test", 4)
	266	+	if err != nil {
	267	+	t.Fatalf("BranchSession: %v", err)
	268	+	}
	269	+	branch, branchMsgs, err := OpenSession(branchPath)
	270	+	if err != nil {
	271	+	t.Fatalf("OpenSession branch: %v", err)
	272	+	}
	273	+	defer branch.Close()
	274	+
	275	+	assertMessageTexts(t, branchMsgs, []string{"summary", "tail-c", "tail-d", "after-compact"})
	276	+	if branch.Meta.Parent != parent.Meta.ID {
	277	+	t.Errorf("parent id: want %q, got %q", parent.Meta.ID, branch.Meta.Parent)
	278	+	}
	279	+	if branch.Meta.ForkPoint != 4 {
	280	+	t.Errorf("fork_point: want 4, got %d", branch.Meta.ForkPoint)
	281	+	}
	282	+
	283	+	shortBranchPath, err := BranchSession(parent.Path, root, cwd, "0.0.0-test", 2)
	284	+	if err != nil {
	285	+	t.Fatalf("BranchSession short fork: %v", err)
	286	+	}
	287	+	shortBranch, shortBranchMsgs, err := OpenSession(shortBranchPath)
	288	+	if err != nil {
	289	+	t.Fatalf("OpenSession short branch: %v", err)
	290	+	}
	291	+	defer shortBranch.Close()
	292	+
	293	+	assertMessageTexts(t, shortBranchMsgs, []string{"summary", "tail-c"})
	294	+	if shortBranch.Meta.ForkPoint != 2 {
	295	+	t.Errorf("short branch fork_point: want 2, got %d", shortBranch.Meta.ForkPoint)
	296	+	}
	297	+	}
	298	+
234	299		// TestBuildSessionTree verifies parent/child edges are rebuilt
235	300		// from meta + sibling-scan.
236	301		func TestBuildSessionTree(t *testing.T) {

@@ -266,3 +331,15 @@ func TestBuildSessionTree(t *testing.T) {
266	331		t.Errorf("want 2 children, got %d", len(rootNode.Children))
267	332		}
268	333		}
	334	+
	335	+	func assertMessageTexts(t *testing.T, msgs []provider.Message, want []string) {
	336	+	t.Helper()
	337	+	if len(msgs) != len(want) {
	338	+	t.Fatalf("message count: want %d, got %d", len(want), len(msgs))
	339	+	}
	340	+	for i, msg := range msgs {
	341	+	if got := extractText(msg); got != want[i] {
	342	+	t.Errorf("message %d: want %q, got %q", i, want[i], got)
	343	+	}
	344	+	}
	345	+	}


@@ -231,6 +231,71 @@ func TestBranchSessionCopiesPrefix(t *testing.T) {
231	}	231		}
232	}	232		}
233		233
		234	+	func TestBranchSessionUsesEffectiveTranscriptAfterCompaction(t *testing.T) {
		235	+	root := t.TempDir()
		236	+	cwd := "/project"
		237	+	parent, err := NewSession(root, cwd, "anthropic", "claude-opus-4-7", "0.0.0-test")
		238	+	if err != nil {
		239	+	t.Fatal(err)
		240	+	}
		241	+	for _, text := range []string{"raw-a", "raw-b", "raw-c", "raw-d"} {
		242	+	_ = parent.AppendMessage(provider.Message{
		243	+	Role: provider.RoleUser,
		244	+	Content: []provider.Content{provider.TextBlock{Text: text}},
		245	+	})
		246	+	}
		247	+	_ = parent.AppendCompaction([]provider.Message{
		248	+	{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "summary"}}},
		249	+	{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "tail-c"}}},
		250	+	{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "tail-d"}}},
		251	+	})
		252	+	_ = parent.AppendMessage(provider.Message{
		253	+	Role: provider.RoleUser,
		254	+	Content: []provider.Content{provider.TextBlock{Text: "after-compact"}},
		255	+	})
		256	+	_ = parent.Close()
		257	+
		258	+	opened, msgs, err := OpenSession(parent.Path)
		259	+	if err != nil {
		260	+	t.Fatalf("OpenSession parent: %v", err)
		261	+	}
		262	+	_ = opened.Close()
		263	+	assertMessageTexts(t, msgs, []string{"summary", "tail-c", "tail-d", "after-compact"})
		264	+
		265	+	branchPath, err := BranchSession(parent.Path, root, cwd, "0.0.0-test", 4)
		266	+	if err != nil {
		267	+	t.Fatalf("BranchSession: %v", err)
		268	+	}
		269	+	branch, branchMsgs, err := OpenSession(branchPath)
		270	+	if err != nil {
		271	+	t.Fatalf("OpenSession branch: %v", err)
		272	+	}
		273	+	defer branch.Close()
		274	+
		275	+	assertMessageTexts(t, branchMsgs, []string{"summary", "tail-c", "tail-d", "after-compact"})
		276	+	if branch.Meta.Parent != parent.Meta.ID {
		277	+	t.Errorf("parent id: want %q, got %q", parent.Meta.ID, branch.Meta.Parent)
		278	+	}
		279	+	if branch.Meta.ForkPoint != 4 {
		280	+	t.Errorf("fork_point: want 4, got %d", branch.Meta.ForkPoint)
		281	+	}
		282	+
		283	+	shortBranchPath, err := BranchSession(parent.Path, root, cwd, "0.0.0-test", 2)
		284	+	if err != nil {
		285	+	t.Fatalf("BranchSession short fork: %v", err)
		286	+	}
		287	+	shortBranch, shortBranchMsgs, err := OpenSession(shortBranchPath)
		288	+	if err != nil {
		289	+	t.Fatalf("OpenSession short branch: %v", err)
		290	+	}
		291	+	defer shortBranch.Close()
		292	+
		293	+	assertMessageTexts(t, shortBranchMsgs, []string{"summary", "tail-c"})
		294	+	if shortBranch.Meta.ForkPoint != 2 {
		295	+	t.Errorf("short branch fork_point: want 2, got %d", shortBranch.Meta.ForkPoint)
		296	+	}
		297	+	}
		298	+
234	// TestBuildSessionTree verifies parent/child edges are rebuilt	299		// TestBuildSessionTree verifies parent/child edges are rebuilt
235	// from meta + sibling-scan.	300		// from meta + sibling-scan.
236	func TestBuildSessionTree(t *testing.T) {	301		func TestBuildSessionTree(t *testing.T) {

@@ -266,3 +331,15 @@ func TestBuildSessionTree(t *testing.T) {
266	t.Errorf("want 2 children, got %d", len(rootNode.Children))	331		t.Errorf("want 2 children, got %d", len(rootNode.Children))
267	}	332		}
268	}	333		}
		334	+
		335	+	func assertMessageTexts(t *testing.T, msgs []provider.Message, want []string) {
		336	+	t.Helper()
		337	+	if len(msgs) != len(want) {
		338	+	t.Fatalf("message count: want %d, got %d", len(want), len(msgs))
		339	+	}
		340	+	for i, msg := range msgs {
		341	+	if got := extractText(msg); got != want[i] {
		342	+	t.Errorf("message %d: want %q, got %q", i, want[i], got)
		343	+	}
		344	+	}
		345	+	}