refactor(core): centralize session log decoding (#161)

* refactor(core): centralize session log decoding

* fix(core): preserve branch read cancellation

commit: ddb26d4f053ac34bf17c8e71f1c9fd8f20ca6fd3
author: bnema · 53d ago · 2026-08-30 15:28 UTC
committer: GitHub · 53d ago · 2026-08-30 15:28 UTC
parent: 560e11b
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

4 files · +256 -239

M packages/core/session.go +30 -205


@@ -406,7 +406,6 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
406	406		defer f.Close()
407	407
408	408		var snapshot SessionSnapshot
409		-	var sawMeta bool
410	409		generation := 0
411	410		type rawCheckpoint struct {
412	411		messageCount int

@@ -417,114 +416,48 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
417	416		titleFromRename := false
418	417		extensionState := map[string]json.RawMessage{}
419	418
420		-	err = forEachStrictJSONLLineContext(ctx, f, func(line []byte, lineNo int) error {
421		-	var head sessionLineHead
422		-	if err := json.Unmarshal(line, &head); err != nil {
423		-	return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
424		-	}
425		-	if head.Type == "" {
426		-	return fmt.Errorf("line %d: missing row type", lineNo)
427		-	}
428		-	if !sawMeta && head.Type != "meta" {
429		-	return fmt.Errorf("line %d: first row is not meta", lineNo)
430		-	}
431		-
432		-	switch head.Type {
	419	+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
	420	+	switch row.typeName {
433	421		case "meta":
434		-	var row sessionLine
435		-	if err := json.Unmarshal(line, &row); err != nil {
436		-	return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)
	422	+	snapshot.Meta = *row.meta
	423	+	if !titleFromRename && row.meta.Title != "" {
	424	+	snapshot.Title = row.meta.Title
437	425		}
438		-	if row.Meta == nil \|\| row.Meta.ID == "" {
439		-	return fmt.Errorf("line %d: meta row has no session id", lineNo)
440		-	}
441		-	snapshot.Meta = *row.Meta
442		-	if !titleFromRename && row.Meta.Title != "" {
443		-	snapshot.Title = row.Meta.Title
444		-	}
445		-	sawMeta = true
446		-
447	426		case "message":
448		-	msg, err := hydrateMessage(line)
449		-	if err != nil {
450		-	return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
451		-	}
452		-	snapshot.Messages = append(snapshot.Messages, msg)
453		-
	427	+	snapshot.Messages = append(snapshot.Messages, *row.message)
454	428		case "compaction":
455		-	compacted, err := hydrateCompaction(line)
456		-	if err != nil {
457		-	return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
458		-	}
459		-	snapshot.Messages = compacted
	429	+	snapshot.Messages = row.messages
460	430		generation++
461		-	cumulative, err := hydrateCompactionUsage(line)
462		-	if err != nil {
463		-	return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
464		-	}
465		-	if cumulative != nil {
	431	+	if row.cumulative != nil {
466	432		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
467	433		messageCount: len(snapshot.Messages),
468		-	cumulative: *cumulative,
	434	+	cumulative: *row.cumulative,
469	435		generation: generation,
470	436		})
471	437		}
472		-
473	438		case "usage":
474		-	var row struct {
475		-	Usage *provider.Usage `json:"usage"`
476		-	Cumulative *provider.Usage `json:"cumulative"`
477		-	}
478		-	if err := json.Unmarshal(line, &row); err != nil {
479		-	return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
480		-	}
481		-	if row.Cumulative == nil {
482		-	return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
483		-	}
484	439		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
485	440		messageCount: len(snapshot.Messages),
486		-	cumulative: *row.Cumulative,
	441	+	cumulative: *row.cumulative,
487	442		generation: generation,
488	443		})
489		-
490	444		case "rename":
491		-	var row struct {
492		-	Title *string `json:"title"`
493		-	}
494		-	if err := json.Unmarshal(line, &row); err != nil {
495		-	return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
496		-	}
497		-	if row.Title == nil {
498		-	return fmt.Errorf("line %d: rename row has no title", lineNo)
499		-	}
500	445		titleFromRename = true
501		-	snapshot.Title = *row.Title
502		-
	446	+	snapshot.Title = *row.title
503	447		case "extension_state":
504		-	var row sessionLine
505		-	if err := json.Unmarshal(line, &row); err != nil {
506		-	return fmt.Errorf("line %d: invalid extension state row: %w", lineNo, err)
507		-	}
508		-	if row.Extension != "" && len(row.State) <= maxExtensionStateBytes {
509		-	if len(row.State) == 0 \|\| strings.TrimSpace(string(row.State)) == "null" {
510		-	delete(extensionState, row.Extension)
511		-	} else if json.Valid(row.State) {
512		-	extensionState[row.Extension] = append(json.RawMessage(nil), row.State...)
	448	+	if row.extension != "" && len(row.state) <= maxExtensionStateBytes {
	449	+	if len(row.state) == 0 \|\| strings.TrimSpace(string(row.state)) == "null" {
	450	+	delete(extensionState, row.extension)
	451	+	} else if json.Valid(row.state) {
	452	+	extensionState[row.extension] = append(json.RawMessage(nil), row.state...)
513	453		}
514	454		}
515		-
516		-	default:
517		-	return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
518	455		}
519	456		return nil
520	457		})
521	458		if err != nil {
522	459		return SessionSnapshot{}, fmt.Errorf("session snapshot %q: %w", path, err)
523	460		}
524		-	if !sawMeta {
525		-	return SessionSnapshot{}, fmt.Errorf("session snapshot %q: file is empty", path)
526		-	}
527		-
528	461		normalizeSessionGoalMeta(&snapshot.Meta)
529	462		snapshot.CompactionGeneration = generation
530	463		snapshot.ExtensionState = extensionState

@@ -581,7 +514,6 @@ func readSessionHistory(ctx context.Context, path string) (SessionHistory, error
581	514		defer f.Close()
582	515
583	516		var history SessionHistory
584		-	var sawMeta bool
585	517		type rawCheckpoint struct {
586	518		messageCount int
587	519		cumulative provider.Usage

@@ -611,108 +543,41 @@ func readSessionHistory(ctx context.Context, path string) (SessionHistory, error
611	543		current = rawSegment{}
612	544		}
613	545
614		-	err = forEachStrictJSONLLineContext(ctx, f, func(line []byte, lineNo int) error {
615		-	var head sessionLineHead
616		-	if err := json.Unmarshal(line, &head); err != nil {
617		-	return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
618		-	}
619		-	if head.Type == "" {
620		-	return fmt.Errorf("line %d: missing row type", lineNo)
621		-	}
622		-	if !sawMeta && head.Type != "meta" {
623		-	return fmt.Errorf("line %d: first row is not meta", lineNo)
624		-	}
625		-
626		-	switch head.Type {
	546	+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
	547	+	switch row.typeName {
627	548		case "meta":
628		-	var row sessionLine
629		-	if err := json.Unmarshal(line, &row); err != nil {
630		-	return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)
631		-	}
632		-	if row.Meta == nil \|\| row.Meta.ID == "" {
633		-	return fmt.Errorf("line %d: meta row has no session id", lineNo)
634		-	}
635		-	history.Meta = *row.Meta
636		-	sawMeta = true
637		-
	549	+	history.Meta = *row.meta
638	550		case "message":
639		-	message, err := hydrateMessage(line)
640		-	if err != nil {
641		-	return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
642		-	}
643		-	current.messages = append(current.messages, message)
644		-
	551	+	current.messages = append(current.messages, *row.message)
645	552		case "compaction":
646		-	compacted, err := hydrateCompaction(line)
647		-	if err != nil {
648		-	return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
649		-	}
650	553		appendCurrent()
651		-	current = rawSegment{compacted: true, messages: compacted}
652		-	cumulative, err := hydrateCompactionUsage(line)
653		-	if err != nil {
654		-	return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
655		-	}
656		-	if cumulative != nil {
	554	+	current = rawSegment{compacted: true, messages: row.messages}
	555	+	if row.cumulative != nil {
657	556		current.checkpoints = append(current.checkpoints, rawCheckpoint{
658	557		messageCount: len(current.messages),
659		-	cumulative: *cumulative,
	558	+	cumulative: *row.cumulative,
660	559		})
661	560		}
662		-
663	561		case "usage":
664		-	var row struct {
665		-	Cumulative *provider.Usage `json:"cumulative"`
666		-	}
667		-	if err := json.Unmarshal(line, &row); err != nil {
668		-	return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
669		-	}
670		-	if row.Cumulative == nil {
671		-	return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
672		-	}
673	562		current.checkpoints = append(current.checkpoints, rawCheckpoint{
674	563		messageCount: len(current.messages),
675		-	cumulative: *row.Cumulative,
	564	+	cumulative: *row.cumulative,
676	565		})
677		-
678		-	case "extension_state":
679		-	// Extension state is session metadata rather than transcript history.
680		-	// The snapshot reader restores it for resume; tree history only needs
681		-	// provider messages and usage checkpoints.
682		-
683		-	case "rename":
684		-	var row struct {
685		-	Title *string `json:"title"`
686		-	}
687		-	if err := json.Unmarshal(line, &row); err != nil {
688		-	return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
689		-	}
690		-	if row.Title == nil {
691		-	return fmt.Errorf("line %d: rename row has no title", lineNo)
692		-	}
693		-
694		-	default:
695		-	return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
	566	+	case "extension_state", "rename":
	567	+	// Extension state and titles are session metadata rather than
	568	+	// transcript history. The decoder still validates their row shape.
696	569		}
697	570		return nil
698	571		})
699	572		if err != nil {
700	573		return SessionHistory{}, fmt.Errorf("session history %q: %w", path, err)
701	574		}
702		-	if !sawMeta {
703		-	return SessionHistory{}, fmt.Errorf("session history %q: file is empty", path)
704		-	}
705	575		appendCurrent()
706	576		return history, nil
707	577		}
708	578
709		-	// forEachStrictJSONLLine is the complete-read counterpart to
710		-	// forEachJSONLLine. It rejects malformed and blank rows and reports the
711		-	// line number so callers cannot accidentally proceed with a partial file.
712		-	func forEachStrictJSONLLine(r io.Reader, fn func([]byte, int) error) error {
713		-	return forEachStrictJSONLLineContext(context.Background(), r, fn)
714		-	}
715		-
	579	+	// forEachStrictJSONLLineContext rejects malformed and blank rows and reports
	580	+	// the line number so callers cannot accidentally proceed with a partial file.
716	581		func forEachStrictJSONLLineContext(ctx context.Context, r io.Reader, fn func([]byte, int) error) error {
717	582		br := bufio.NewReader(r)
718	583		lineNo := 0

@@ -1788,16 +1653,6 @@ func (s *Session) writeLineLocked(row sessionLine) error {
1788	1653		// We persist messages by reading the raw "message" object back and
1789	1654		// rebuilding Content from discriminated fields.
1790	1655
1791		-	func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1792		-	var row struct {
1793		-	Cumulative *provider.Usage `json:"cumulative"`
1794		-	}
1795		-	if err := json.Unmarshal(lineBytes, &row); err != nil {
1796		-	return nil, err
1797		-	}
1798		-	return row.Cumulative, nil
1799		-	}
1800		-
1801	1656		func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1802	1657		var row struct {
1803	1658		Messages json.RawMessage `json:"messages"`

@@ -1805,37 +1660,7 @@ func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1805	1660		if err := json.Unmarshal(lineBytes, &row); err != nil {
1806	1661		return nil, err
1807	1662		}
1808		-	if len(row.Messages) == 0 \|\| bytes.Equal(bytes.TrimSpace(row.Messages), []byte("null")) {
1809		-	// Older writers omitted the field for an empty compaction and some
1810		-	// wrote it as null. Both represent a valid empty replacement.
1811		-	return []provider.Message{}, nil
1812		-	}
1813		-	var rawMessages []json.RawMessage
1814		-	if err := json.Unmarshal(row.Messages, &rawMessages); err != nil {
1815		-	return nil, fmt.Errorf("invalid messages: %w", err)
1816		-	}
1817		-	messages := make([]provider.Message, 0, len(rawMessages))
1818		-	for idx, raw := range rawMessages {
1819		-	msg, err := hydrateMessageObject(raw)
1820		-	if err != nil {
1821		-	return nil, fmt.Errorf("message %d: %w", idx, err)
1822		-	}
1823		-	messages = append(messages, msg)
1824		-	}
1825		-	return messages, nil
1826		-	}
1827		-
1828		-	func hydrateMessage(lineBytes []byte) (provider.Message, error) {
1829		-	var row struct {
1830		-	Message json.RawMessage `json:"message"`
1831		-	}
1832		-	if err := json.Unmarshal(lineBytes, &row); err != nil {
1833		-	return provider.Message{}, err
1834		-	}
1835		-	if len(row.Message) == 0 \|\| bytes.Equal(bytes.TrimSpace(row.Message), []byte("null")) {
1836		-	return provider.Message{}, fmt.Errorf("message row has no message")
1837		-	}
1838		-	return hydrateMessageObject(row.Message)
	1663	+	return hydrateCompactionMessages(row.Messages)
1839	1664		}
1840	1665
1841	1666		func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {


@@ -406,7 +406,6 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
406		defer f.Close()	406		defer f.Close()
407			407
408		var snapshot SessionSnapshot	408		var snapshot SessionSnapshot
409	-	var sawMeta bool
410		generation := 0	409		generation := 0
411		type rawCheckpoint struct {	410		type rawCheckpoint struct {
412		messageCount int	411		messageCount int

@@ -417,114 +416,48 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
417		titleFromRename := false	416		titleFromRename := false
418		extensionState := map[string]json.RawMessage{}	417		extensionState := map[string]json.RawMessage{}
419			418
420	-	err = forEachStrictJSONLLineContext(ctx, f, func(line []byte, lineNo int) error {	419	+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
421	-	var head sessionLineHead	420	+	switch row.typeName {
422	-	if err := json.Unmarshal(line, &head); err != nil {
423	-	return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
424	-	}
425	-	if head.Type == "" {
426	-	return fmt.Errorf("line %d: missing row type", lineNo)
427	-	}
428	-	if !sawMeta && head.Type != "meta" {
429	-	return fmt.Errorf("line %d: first row is not meta", lineNo)
430	-	}
431	-
432	-	switch head.Type {
433		case "meta":	421		case "meta":
434	-	var row sessionLine	422	+	snapshot.Meta = *row.meta
435	-	if err := json.Unmarshal(line, &row); err != nil {	423	+	if !titleFromRename && row.meta.Title != "" {
436	-	return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)	424	+	snapshot.Title = row.meta.Title
437		}	425		}
438	-	if row.Meta == nil \|\| row.Meta.ID == "" {
439	-	return fmt.Errorf("line %d: meta row has no session id", lineNo)
440	-	}
441	-	snapshot.Meta = *row.Meta
442	-	if !titleFromRename && row.Meta.Title != "" {
443	-	snapshot.Title = row.Meta.Title
444	-	}
445	-	sawMeta = true
446	-
447		case "message":	426		case "message":
448	-	msg, err := hydrateMessage(line)	427	+	snapshot.Messages = append(snapshot.Messages, *row.message)
449	-	if err != nil {
450	-	return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
451	-	}
452	-	snapshot.Messages = append(snapshot.Messages, msg)
453	-
454		case "compaction":	428		case "compaction":
455	-	compacted, err := hydrateCompaction(line)	429	+	snapshot.Messages = row.messages
456	-	if err != nil {
457	-	return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
458	-	}
459	-	snapshot.Messages = compacted
460		generation++	430		generation++
461	-	cumulative, err := hydrateCompactionUsage(line)	431	+	if row.cumulative != nil {
462	-	if err != nil {
463	-	return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
464	-	}
465	-	if cumulative != nil {
466		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{	432		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
467		messageCount: len(snapshot.Messages),	433		messageCount: len(snapshot.Messages),
468	-	cumulative: *cumulative,	434	+	cumulative: *row.cumulative,
469		generation: generation,	435		generation: generation,
470		})	436		})
471		}	437		}
472	-
473		case "usage":	438		case "usage":
474	-	var row struct {
475	-	Usage *provider.Usage `json:"usage"`
476	-	Cumulative *provider.Usage `json:"cumulative"`
477	-	}
478	-	if err := json.Unmarshal(line, &row); err != nil {
479	-	return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
480	-	}
481	-	if row.Cumulative == nil {
482	-	return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
483	-	}
484		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{	439		rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
485		messageCount: len(snapshot.Messages),	440		messageCount: len(snapshot.Messages),
486	-	cumulative: *row.Cumulative,	441	+	cumulative: *row.cumulative,
487		generation: generation,	442		generation: generation,
488		})	443		})
489	-
490		case "rename":	444		case "rename":
491	-	var row struct {
492	-	Title *string `json:"title"`
493	-	}
494	-	if err := json.Unmarshal(line, &row); err != nil {
495	-	return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
496	-	}
497	-	if row.Title == nil {
498	-	return fmt.Errorf("line %d: rename row has no title", lineNo)
499	-	}
500		titleFromRename = true	445		titleFromRename = true
501	-	snapshot.Title = *row.Title	446	+	snapshot.Title = *row.title
502	-
503		case "extension_state":	447		case "extension_state":
504	-	var row sessionLine	448	+	if row.extension != "" && len(row.state) <= maxExtensionStateBytes {
505	-	if err := json.Unmarshal(line, &row); err != nil {	449	+	if len(row.state) == 0 \|\| strings.TrimSpace(string(row.state)) == "null" {
506	-	return fmt.Errorf("line %d: invalid extension state row: %w", lineNo, err)	450	+	delete(extensionState, row.extension)
507	-	}	451	+	} else if json.Valid(row.state) {
508	-	if row.Extension != "" && len(row.State) <= maxExtensionStateBytes {	452	+	extensionState[row.extension] = append(json.RawMessage(nil), row.state...)
509	-	if len(row.State) == 0 \|\| strings.TrimSpace(string(row.State)) == "null" {
510	-	delete(extensionState, row.Extension)
511	-	} else if json.Valid(row.State) {
512	-	extensionState[row.Extension] = append(json.RawMessage(nil), row.State...)
513		}	453		}
514		}	454		}
515	-
516	-	default:
517	-	return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
518		}	455		}
519		return nil	456		return nil
520		})	457		})
521		if err != nil {	458		if err != nil {
522		return SessionSnapshot{}, fmt.Errorf("session snapshot %q: %w", path, err)	459		return SessionSnapshot{}, fmt.Errorf("session snapshot %q: %w", path, err)
523		}	460		}
524	-	if !sawMeta {
525	-	return SessionSnapshot{}, fmt.Errorf("session snapshot %q: file is empty", path)
526	-	}
527	-
528		normalizeSessionGoalMeta(&snapshot.Meta)	461		normalizeSessionGoalMeta(&snapshot.Meta)
529		snapshot.CompactionGeneration = generation	462		snapshot.CompactionGeneration = generation
530		snapshot.ExtensionState = extensionState	463		snapshot.ExtensionState = extensionState

@@ -581,7 +514,6 @@ func readSessionHistory(ctx context.Context, path string) (SessionHistory, error
581		defer f.Close()	514		defer f.Close()
582			515
583		var history SessionHistory	516		var history SessionHistory
584	-	var sawMeta bool
585		type rawCheckpoint struct {	517		type rawCheckpoint struct {
586		messageCount int	518		messageCount int
587		cumulative provider.Usage	519		cumulative provider.Usage

@@ -611,108 +543,41 @@ func readSessionHistory(ctx context.Context, path string) (SessionHistory, error
611		current = rawSegment{}	543		current = rawSegment{}
612		}	544		}
613			545
614	-	err = forEachStrictJSONLLineContext(ctx, f, func(line []byte, lineNo int) error {	546	+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
615	-	var head sessionLineHead	547	+	switch row.typeName {
616	-	if err := json.Unmarshal(line, &head); err != nil {
617	-	return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
618	-	}
619	-	if head.Type == "" {
620	-	return fmt.Errorf("line %d: missing row type", lineNo)
621	-	}
622	-	if !sawMeta && head.Type != "meta" {
623	-	return fmt.Errorf("line %d: first row is not meta", lineNo)
624	-	}
625	-
626	-	switch head.Type {
627		case "meta":	548		case "meta":
628	-	var row sessionLine	549	+	history.Meta = *row.meta
629	-	if err := json.Unmarshal(line, &row); err != nil {
630	-	return fmt.Errorf("line %d: invalid meta row: %w", lineNo, err)
631	-	}
632	-	if row.Meta == nil \|\| row.Meta.ID == "" {
633	-	return fmt.Errorf("line %d: meta row has no session id", lineNo)
634	-	}
635	-	history.Meta = *row.Meta
636	-	sawMeta = true
637	-
638		case "message":	550		case "message":
639	-	message, err := hydrateMessage(line)	551	+	current.messages = append(current.messages, *row.message)
640	-	if err != nil {
641	-	return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
642	-	}
643	-	current.messages = append(current.messages, message)
644	-
645		case "compaction":	552		case "compaction":
646	-	compacted, err := hydrateCompaction(line)
647	-	if err != nil {
648	-	return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
649	-	}
650		appendCurrent()	553		appendCurrent()
651	-	current = rawSegment{compacted: true, messages: compacted}	554	+	current = rawSegment{compacted: true, messages: row.messages}
652	-	cumulative, err := hydrateCompactionUsage(line)	555	+	if row.cumulative != nil {
653	-	if err != nil {
654	-	return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
655	-	}
656	-	if cumulative != nil {
657		current.checkpoints = append(current.checkpoints, rawCheckpoint{	556		current.checkpoints = append(current.checkpoints, rawCheckpoint{
658		messageCount: len(current.messages),	557		messageCount: len(current.messages),
659	-	cumulative: *cumulative,	558	+	cumulative: *row.cumulative,
660		})	559		})
661		}	560		}
662	-
663		case "usage":	561		case "usage":
664	-	var row struct {
665	-	Cumulative *provider.Usage `json:"cumulative"`
666	-	}
667	-	if err := json.Unmarshal(line, &row); err != nil {
668	-	return fmt.Errorf("line %d: invalid usage row: %w", lineNo, err)
669	-	}
670	-	if row.Cumulative == nil {
671	-	return fmt.Errorf("line %d: usage row has no cumulative usage", lineNo)
672	-	}
673		current.checkpoints = append(current.checkpoints, rawCheckpoint{	562		current.checkpoints = append(current.checkpoints, rawCheckpoint{
674		messageCount: len(current.messages),	563		messageCount: len(current.messages),
675	-	cumulative: *row.Cumulative,	564	+	cumulative: *row.cumulative,
676		})	565		})
677	-		566	+	case "extension_state", "rename":
678	-	case "extension_state":	567	+	// Extension state and titles are session metadata rather than
679	-	// Extension state is session metadata rather than transcript history.	568	+	// transcript history. The decoder still validates their row shape.
680	-	// The snapshot reader restores it for resume; tree history only needs
681	-	// provider messages and usage checkpoints.
682	-
683	-	case "rename":
684	-	var row struct {
685	-	Title *string `json:"title"`
686	-	}
687	-	if err := json.Unmarshal(line, &row); err != nil {
688	-	return fmt.Errorf("line %d: invalid rename row: %w", lineNo, err)
689	-	}
690	-	if row.Title == nil {
691	-	return fmt.Errorf("line %d: rename row has no title", lineNo)
692	-	}
693	-
694	-	default:
695	-	return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
696		}	569		}
697		return nil	570		return nil
698		})	571		})
699		if err != nil {	572		if err != nil {
700		return SessionHistory{}, fmt.Errorf("session history %q: %w", path, err)	573		return SessionHistory{}, fmt.Errorf("session history %q: %w", path, err)
701		}	574		}
702	-	if !sawMeta {
703	-	return SessionHistory{}, fmt.Errorf("session history %q: file is empty", path)
704	-	}
705		appendCurrent()	575		appendCurrent()
706		return history, nil	576		return history, nil
707		}	577		}
708			578
709	-	// forEachStrictJSONLLine is the complete-read counterpart to	579	+	// forEachStrictJSONLLineContext rejects malformed and blank rows and reports
710	-	// forEachJSONLLine. It rejects malformed and blank rows and reports the	580	+	// the line number so callers cannot accidentally proceed with a partial file.
711	-	// line number so callers cannot accidentally proceed with a partial file.
712	-	func forEachStrictJSONLLine(r io.Reader, fn func([]byte, int) error) error {
713	-	return forEachStrictJSONLLineContext(context.Background(), r, fn)
714	-	}
715	-
716		func forEachStrictJSONLLineContext(ctx context.Context, r io.Reader, fn func([]byte, int) error) error {	581		func forEachStrictJSONLLineContext(ctx context.Context, r io.Reader, fn func([]byte, int) error) error {
717		br := bufio.NewReader(r)	582		br := bufio.NewReader(r)
718		lineNo := 0	583		lineNo := 0

@@ -1788,16 +1653,6 @@ func (s *Session) writeLineLocked(row sessionLine) error {
1788		// We persist messages by reading the raw "message" object back and	1653		// We persist messages by reading the raw "message" object back and
1789		// rebuilding Content from discriminated fields.	1654		// rebuilding Content from discriminated fields.
1790			1655
1791	-	func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1792	-	var row struct {
1793	-	Cumulative *provider.Usage `json:"cumulative"`
1794	-	}
1795	-	if err := json.Unmarshal(lineBytes, &row); err != nil {
1796	-	return nil, err
1797	-	}
1798	-	return row.Cumulative, nil
1799	-	}
1800	-
1801		func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {	1656		func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1802		var row struct {	1657		var row struct {
1803		Messages json.RawMessage `json:"messages"`	1658		Messages json.RawMessage `json:"messages"`

@@ -1805,37 +1660,7 @@ func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1805		if err := json.Unmarshal(lineBytes, &row); err != nil {	1660		if err := json.Unmarshal(lineBytes, &row); err != nil {
1806		return nil, err	1661		return nil, err
1807		}	1662		}
1808	-	if len(row.Messages) == 0 \|\| bytes.Equal(bytes.TrimSpace(row.Messages), []byte("null")) {	1663	+	return hydrateCompactionMessages(row.Messages)
1809	-	// Older writers omitted the field for an empty compaction and some
1810	-	// wrote it as null. Both represent a valid empty replacement.
1811	-	return []provider.Message{}, nil
1812	-	}
1813	-	var rawMessages []json.RawMessage
1814	-	if err := json.Unmarshal(row.Messages, &rawMessages); err != nil {
1815	-	return nil, fmt.Errorf("invalid messages: %w", err)
1816	-	}
1817	-	messages := make([]provider.Message, 0, len(rawMessages))
1818	-	for idx, raw := range rawMessages {
1819	-	msg, err := hydrateMessageObject(raw)
1820	-	if err != nil {
1821	-	return nil, fmt.Errorf("message %d: %w", idx, err)
1822	-	}
1823	-	messages = append(messages, msg)
1824	-	}
1825	-	return messages, nil
1826	-	}
1827	-
1828	-	func hydrateMessage(lineBytes []byte) (provider.Message, error) {
1829	-	var row struct {
1830	-	Message json.RawMessage `json:"message"`
1831	-	}
1832	-	if err := json.Unmarshal(lineBytes, &row); err != nil {
1833	-	return provider.Message{}, err
1834	-	}
1835	-	if len(row.Message) == 0 \|\| bytes.Equal(bytes.TrimSpace(row.Message), []byte("null")) {
1836	-	return provider.Message{}, fmt.Errorf("message row has no message")
1837	-	}
1838	-	return hydrateMessageObject(row.Message)
1839		}	1664		}
1840			1665
1841		func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {	1666		func hydrateMessageObject(rawMessage []byte) (provider.Message, error) {

A packages/core/session_log.go +136 -0

@@ -0,0 +1,136 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"io"
	9	+
	10	+	"github.com/bnema/zut/packages/provider"
	11	+	)
	12	+
	13	+	// sessionLogRow is the validated, hydrated representation shared by complete
	14	+	// session readers. It keeps the JSONL format private while giving each reader
	15	+	// one typed stream to reduce into its own projection.
	16	+	type sessionLogRow struct {
	17	+	typeName string
	18	+	meta *SessionMeta
	19	+	message *provider.Message
	20	+	messages []provider.Message
	21	+	cumulative *provider.Usage
	22	+	title *string
	23	+	extension string
	24	+	state json.RawMessage
	25	+	}
	26	+
	27	+	// forEachSessionLogRowContext owns the strict persisted-row contract: JSONL
	28	+	// framing, the leading meta row, known row types, required fields, and message
	29	+	// content hydration. Callers only implement projection-specific reduction.
	30	+	func forEachSessionLogRowContext(ctx context.Context, r io.Reader, fn func(sessionLogRow) error) error {
	31	+	sawMeta := false
	32	+	err := forEachStrictJSONLLineContext(ctx, r, func(line []byte, lineNo int) error {
	33	+	row, err := decodeSessionLogRow(line)
	34	+	if err != nil {
	35	+	return fmt.Errorf("line %d: %w", lineNo, err)
	36	+	}
	37	+	if !sawMeta && row.typeName != "meta" {
	38	+	return fmt.Errorf("line %d: first row is not meta", lineNo)
	39	+	}
	40	+	if row.typeName == "meta" {
	41	+	sawMeta = true
	42	+	}
	43	+	return fn(row)
	44	+	})
	45	+	if err != nil {
	46	+	return err
	47	+	}
	48	+	if !sawMeta {
	49	+	return fmt.Errorf("file is empty")
	50	+	}
	51	+	return nil
	52	+	}
	53	+
	54	+	func decodeSessionLogRow(line []byte) (sessionLogRow, error) {
	55	+	var wire struct {
	56	+	Type string `json:"type"`
	57	+	Meta *SessionMeta `json:"meta"`
	58	+	Message json.RawMessage `json:"message"`
	59	+	Messages json.RawMessage `json:"messages"`
	60	+	Cumulative *provider.Usage `json:"cumulative"`
	61	+	Title *string `json:"title"`
	62	+	Extension string `json:"extension"`
	63	+	State json.RawMessage `json:"state"`
	64	+	}
	65	+	if err := json.Unmarshal(line, &wire); err != nil {
	66	+	return sessionLogRow{}, fmt.Errorf("invalid JSON: %w", err)
	67	+	}
	68	+	if wire.Type == "" {
	69	+	return sessionLogRow{}, fmt.Errorf("missing row type")
	70	+	}
	71	+
	72	+	row := sessionLogRow{
	73	+	typeName: wire.Type,
	74	+	cumulative: wire.Cumulative,
	75	+	title: wire.Title,
	76	+	extension: wire.Extension,
	77	+	state: wire.State,
	78	+	}
	79	+	switch wire.Type {
	80	+	case "meta":
	81	+	if wire.Meta == nil \|\| wire.Meta.ID == "" {
	82	+	return sessionLogRow{}, fmt.Errorf("meta row has no session id")
	83	+	}
	84	+	row.meta = wire.Meta
	85	+	case "message":
	86	+	if len(wire.Message) == 0 \|\| bytes.Equal(bytes.TrimSpace(wire.Message), []byte("null")) {
	87	+	return sessionLogRow{}, fmt.Errorf("invalid message row: message row has no message")
	88	+	}
	89	+	message, err := hydrateMessageObject(wire.Message)
	90	+	if err != nil {
	91	+	return sessionLogRow{}, fmt.Errorf("invalid message row: %w", err)
	92	+	}
	93	+	row.message = &message
	94	+	case "compaction":
	95	+	messages, err := hydrateCompactionMessages(wire.Messages)
	96	+	if err != nil {
	97	+	return sessionLogRow{}, fmt.Errorf("invalid compaction row: %w", err)
	98	+	}
	99	+	row.messages = messages
	100	+	case "usage":
	101	+	if wire.Cumulative == nil {
	102	+	return sessionLogRow{}, fmt.Errorf("usage row has no cumulative usage")
	103	+	}
	104	+	case "rename":
	105	+	if wire.Title == nil {
	106	+	return sessionLogRow{}, fmt.Errorf("rename row has no title")
	107	+	}
	108	+	case "extension_state":
	109	+	// Invalid or oversized extension snapshots are ignored by reducers for
	110	+	// compatibility with existing sessions; transcript rows stay strict.
	111	+	default:
	112	+	return sessionLogRow{}, fmt.Errorf("unknown row type %q", wire.Type)
	113	+	}
	114	+	return row, nil
	115	+	}
	116	+
	117	+	func hydrateCompactionMessages(raw json.RawMessage) ([]provider.Message, error) {
	118	+	if len(raw) == 0 \|\| bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
	119	+	// Older writers omitted the field for an empty compaction and some
	120	+	// wrote it as null. Both represent a valid empty replacement.
	121	+	return []provider.Message{}, nil
	122	+	}
	123	+	var rawMessages []json.RawMessage
	124	+	if err := json.Unmarshal(raw, &rawMessages); err != nil {
	125	+	return nil, fmt.Errorf("invalid messages: %w", err)
	126	+	}
	127	+	messages := make([]provider.Message, 0, len(rawMessages))
	128	+	for idx, rawMessage := range rawMessages {
	129	+	message, err := hydrateMessageObject(rawMessage)
	130	+	if err != nil {
	131	+	return nil, fmt.Errorf("message %d: %w", idx, err)
	132	+	}
	133	+	messages = append(messages, message)
	134	+	}
	135	+	return messages, nil
	136	+	}

@@ -0,0 +1,136 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"io"
	9	+
	10	+	"github.com/bnema/zut/packages/provider"
	11	+	)
	12	+
	13	+	// sessionLogRow is the validated, hydrated representation shared by complete
	14	+	// session readers. It keeps the JSONL format private while giving each reader
	15	+	// one typed stream to reduce into its own projection.
	16	+	type sessionLogRow struct {
	17	+	typeName string
	18	+	meta *SessionMeta
	19	+	message *provider.Message
	20	+	messages []provider.Message
	21	+	cumulative *provider.Usage
	22	+	title *string
	23	+	extension string
	24	+	state json.RawMessage
	25	+	}
	26	+
	27	+	// forEachSessionLogRowContext owns the strict persisted-row contract: JSONL
	28	+	// framing, the leading meta row, known row types, required fields, and message
	29	+	// content hydration. Callers only implement projection-specific reduction.
	30	+	func forEachSessionLogRowContext(ctx context.Context, r io.Reader, fn func(sessionLogRow) error) error {
	31	+	sawMeta := false
	32	+	err := forEachStrictJSONLLineContext(ctx, r, func(line []byte, lineNo int) error {
	33	+	row, err := decodeSessionLogRow(line)
	34	+	if err != nil {
	35	+	return fmt.Errorf("line %d: %w", lineNo, err)
	36	+	}
	37	+	if !sawMeta && row.typeName != "meta" {
	38	+	return fmt.Errorf("line %d: first row is not meta", lineNo)
	39	+	}
	40	+	if row.typeName == "meta" {
	41	+	sawMeta = true
	42	+	}
	43	+	return fn(row)
	44	+	})
	45	+	if err != nil {
	46	+	return err
	47	+	}
	48	+	if !sawMeta {
	49	+	return fmt.Errorf("file is empty")
	50	+	}
	51	+	return nil
	52	+	}
	53	+
	54	+	func decodeSessionLogRow(line []byte) (sessionLogRow, error) {
	55	+	var wire struct {
	56	+	Type string `json:"type"`
	57	+	Meta *SessionMeta `json:"meta"`
	58	+	Message json.RawMessage `json:"message"`
	59	+	Messages json.RawMessage `json:"messages"`
	60	+	Cumulative *provider.Usage `json:"cumulative"`
	61	+	Title *string `json:"title"`
	62	+	Extension string `json:"extension"`
	63	+	State json.RawMessage `json:"state"`
	64	+	}
	65	+	if err := json.Unmarshal(line, &wire); err != nil {
	66	+	return sessionLogRow{}, fmt.Errorf("invalid JSON: %w", err)
	67	+	}
	68	+	if wire.Type == "" {
	69	+	return sessionLogRow{}, fmt.Errorf("missing row type")
	70	+	}
	71	+
	72	+	row := sessionLogRow{
	73	+	typeName: wire.Type,
	74	+	cumulative: wire.Cumulative,
	75	+	title: wire.Title,
	76	+	extension: wire.Extension,
	77	+	state: wire.State,
	78	+	}
	79	+	switch wire.Type {
	80	+	case "meta":
	81	+	if wire.Meta == nil \|\| wire.Meta.ID == "" {
	82	+	return sessionLogRow{}, fmt.Errorf("meta row has no session id")
	83	+	}
	84	+	row.meta = wire.Meta
	85	+	case "message":
	86	+	if len(wire.Message) == 0 \|\| bytes.Equal(bytes.TrimSpace(wire.Message), []byte("null")) {
	87	+	return sessionLogRow{}, fmt.Errorf("invalid message row: message row has no message")
	88	+	}
	89	+	message, err := hydrateMessageObject(wire.Message)
	90	+	if err != nil {
	91	+	return sessionLogRow{}, fmt.Errorf("invalid message row: %w", err)
	92	+	}
	93	+	row.message = &message
	94	+	case "compaction":
	95	+	messages, err := hydrateCompactionMessages(wire.Messages)
	96	+	if err != nil {
	97	+	return sessionLogRow{}, fmt.Errorf("invalid compaction row: %w", err)
	98	+	}
	99	+	row.messages = messages
	100	+	case "usage":
	101	+	if wire.Cumulative == nil {
	102	+	return sessionLogRow{}, fmt.Errorf("usage row has no cumulative usage")
	103	+	}
	104	+	case "rename":
	105	+	if wire.Title == nil {
	106	+	return sessionLogRow{}, fmt.Errorf("rename row has no title")
	107	+	}
	108	+	case "extension_state":
	109	+	// Invalid or oversized extension snapshots are ignored by reducers for
	110	+	// compatibility with existing sessions; transcript rows stay strict.
	111	+	default:
	112	+	return sessionLogRow{}, fmt.Errorf("unknown row type %q", wire.Type)
	113	+	}
	114	+	return row, nil
	115	+	}
	116	+
	117	+	func hydrateCompactionMessages(raw json.RawMessage) ([]provider.Message, error) {
	118	+	if len(raw) == 0 \|\| bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
	119	+	// Older writers omitted the field for an empty compaction and some
	120	+	// wrote it as null. Both represent a valid empty replacement.
	121	+	return []provider.Message{}, nil
	122	+	}
	123	+	var rawMessages []json.RawMessage
	124	+	if err := json.Unmarshal(raw, &rawMessages); err != nil {
	125	+	return nil, fmt.Errorf("invalid messages: %w", err)
	126	+	}
	127	+	messages := make([]provider.Message, 0, len(rawMessages))
	128	+	for idx, rawMessage := range rawMessages {
	129	+	message, err := hydrateMessageObject(rawMessage)
	130	+	if err != nil {
	131	+	return nil, fmt.Errorf("message %d: %w", idx, err)
	132	+	}
	133	+	messages = append(messages, message)
	134	+	}
	135	+	return messages, nil
	136	+	}

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

@@ -0,0 +1,75 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"errors"
	7	+	"os"
	8	+	"path/filepath"
	9	+	"strings"
	10	+	"testing"
	11	+	"time"
	12	+	)
	13	+
	14	+	func TestSessionLogProjectionsRejectInvalidRowsConsistently(t *testing.T) {
	15	+	meta := SessionMeta{ID: "session-1", CWD: "/workspace", Started: time.Now().UTC()}
	16	+	metaLine, err := json.Marshal(sessionLine{Type: "meta", Meta: &meta})
	17	+	if err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+
	21	+	tests := []struct {
	22	+	name string
	23	+	contents string
	24	+	want string
	25	+	}{
	26	+	{name: "missing meta", contents: `{"type":"message","message":{"role":"user","content":[{"text":"hello"}]}}` + "\n", want: "first row is not meta"},
	27	+	{name: "invalid message", contents: string(metaLine) + "\n" + `{"type":"message","message":{"role":"user"}}` + "\n", want: "invalid message row"},
	28	+	{name: "invalid usage", contents: string(metaLine) + "\n" + `{"type":"usage","cumulative":null}` + "\n", want: "usage row has no cumulative usage"},
	29	+	{name: "invalid rename", contents: string(metaLine) + "\n" + `{"type":"rename"}` + "\n", want: "rename row has no title"},
	30	+	{name: "unknown type", contents: string(metaLine) + "\n" + `{"type":"future-row"}` + "\n", want: `unknown row type "future-row"`},
	31	+	}
	32	+
	33	+	for _, test := range tests {
	34	+	t.Run(test.name, func(t *testing.T) {
	35	+	path := filepath.Join(t.TempDir(), "session.jsonl")
	36	+	if err := os.WriteFile(path, []byte(test.contents), 0o644); err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+
	40	+	_, snapshotErr := ReadSessionSnapshot(path)
	41	+	_, historyErr := ReadSessionHistory(path)
	42	+	_, branchErr := readExtensionStateAtFork(context.Background(), path, 0)
	43	+	for name, err := range map[string]error{
	44	+	"snapshot": snapshotErr,
	45	+	"history": historyErr,
	46	+	"branch": branchErr,
	47	+	} {
	48	+	if err == nil \|\| !strings.Contains(err.Error(), test.want) {
	49	+	t.Errorf("%s error = %v, want substring %q", name, err, test.want)
	50	+	}
	51	+	}
	52	+	})
	53	+	}
	54	+	}
	55	+
	56	+	func TestBranchSessionThreadsCancellationToExtensionStateRead(t *testing.T) {
	57	+	root := t.TempDir()
	58	+	session, err := NewSession(root, "/workspace", "provider", "model", "test")
	59	+	if err != nil {
	60	+	t.Fatal(err)
	61	+	}
	62	+	if err := session.AppendExtensionState("extension", json.RawMessage(`{"version":1}`)); err != nil {
	63	+	t.Fatal(err)
	64	+	}
	65	+	path := session.Path
	66	+	if err := session.Close(); err != nil {
	67	+	t.Fatal(err)
	68	+	}
	69	+
	70	+	ctx, cancel := context.WithCancel(context.Background())
	71	+	cancel()
	72	+	if _, err := branchSession(ctx, path, root, "/workspace", "test", 0, false); !errors.Is(err, context.Canceled) {
	73	+	t.Fatalf("branchSession error = %v, want context.Canceled", err)
	74	+	}
	75	+	}

@@ -0,0 +1,75 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"errors"
	7	+	"os"
	8	+	"path/filepath"
	9	+	"strings"
	10	+	"testing"
	11	+	"time"
	12	+	)
	13	+
	14	+	func TestSessionLogProjectionsRejectInvalidRowsConsistently(t *testing.T) {
	15	+	meta := SessionMeta{ID: "session-1", CWD: "/workspace", Started: time.Now().UTC()}
	16	+	metaLine, err := json.Marshal(sessionLine{Type: "meta", Meta: &meta})
	17	+	if err != nil {
	18	+	t.Fatal(err)
	19	+	}
	20	+
	21	+	tests := []struct {
	22	+	name string
	23	+	contents string
	24	+	want string
	25	+	}{
	26	+	{name: "missing meta", contents: `{"type":"message","message":{"role":"user","content":[{"text":"hello"}]}}` + "\n", want: "first row is not meta"},
	27	+	{name: "invalid message", contents: string(metaLine) + "\n" + `{"type":"message","message":{"role":"user"}}` + "\n", want: "invalid message row"},
	28	+	{name: "invalid usage", contents: string(metaLine) + "\n" + `{"type":"usage","cumulative":null}` + "\n", want: "usage row has no cumulative usage"},
	29	+	{name: "invalid rename", contents: string(metaLine) + "\n" + `{"type":"rename"}` + "\n", want: "rename row has no title"},
	30	+	{name: "unknown type", contents: string(metaLine) + "\n" + `{"type":"future-row"}` + "\n", want: `unknown row type "future-row"`},
	31	+	}
	32	+
	33	+	for _, test := range tests {
	34	+	t.Run(test.name, func(t *testing.T) {
	35	+	path := filepath.Join(t.TempDir(), "session.jsonl")
	36	+	if err := os.WriteFile(path, []byte(test.contents), 0o644); err != nil {
	37	+	t.Fatal(err)
	38	+	}
	39	+
	40	+	_, snapshotErr := ReadSessionSnapshot(path)
	41	+	_, historyErr := ReadSessionHistory(path)
	42	+	_, branchErr := readExtensionStateAtFork(context.Background(), path, 0)
	43	+	for name, err := range map[string]error{
	44	+	"snapshot": snapshotErr,
	45	+	"history": historyErr,
	46	+	"branch": branchErr,
	47	+	} {
	48	+	if err == nil \|\| !strings.Contains(err.Error(), test.want) {
	49	+	t.Errorf("%s error = %v, want substring %q", name, err, test.want)
	50	+	}
	51	+	}
	52	+	})
	53	+	}
	54	+	}
	55	+
	56	+	func TestBranchSessionThreadsCancellationToExtensionStateRead(t *testing.T) {
	57	+	root := t.TempDir()
	58	+	session, err := NewSession(root, "/workspace", "provider", "model", "test")
	59	+	if err != nil {
	60	+	t.Fatal(err)
	61	+	}
	62	+	if err := session.AppendExtensionState("extension", json.RawMessage(`{"version":1}`)); err != nil {
	63	+	t.Fatal(err)
	64	+	}
	65	+	path := session.Path
	66	+	if err := session.Close(); err != nil {
	67	+	t.Fatal(err)
	68	+	}
	69	+
	70	+	ctx, cancel := context.WithCancel(context.Background())
	71	+	cancel()
	72	+	if _, err := branchSession(ctx, path, root, "/workspace", "test", 0, false); !errors.Is(err, context.Canceled) {
	73	+	t.Fatalf("branchSession error = %v, want context.Canceled", err)
	74	+	}
	75	+	}

M packages/core/session_portable.go +15 -34


@@ -284,16 +284,16 @@ func ImportSession(srcPath, root, cwd, version string) (string, error) {
284	284		//
285	285		// Returns the path of the new session file, ready for OpenSession.
286	286		func BranchSession(parentPath, root, cwd, version string, upToMessageIdx int) (string, error) {
287		-	return branchSession(parentPath, root, cwd, version, upToMessageIdx, false)
	287	+	return branchSession(context.Background(), parentPath, root, cwd, version, upToMessageIdx, false)
288	288		}
289	289
290	290		// BranchSessionHidden creates a branch that participates in /session tree but
291	291		// is hidden from the flat /sessions picker. Used for in-place tree navigation.
292	292		func BranchSessionHidden(parentPath, root, cwd, version string, upToMessageIdx int) (string, error) {
293		-	return branchSession(parentPath, root, cwd, version, upToMessageIdx, true)
	293	+	return branchSession(context.Background(), parentPath, root, cwd, version, upToMessageIdx, true)
294	294		}
295	295
296		-	func branchSession(parentPath, root, cwd, version string, upToMessageIdx int, hideFromSessions bool) (string, error) {
	296	+	func branchSession(ctx context.Context, parentPath, root, cwd, version string, upToMessageIdx int, hideFromSessions bool) (string, error) {
297	297		if parentPath == "" {
298	298		return "", errors.New("branch: parent path is empty")
299	299		}

@@ -318,7 +318,7 @@ func branchSession(parentPath, root, cwd, version string, upToMessageIdx int, hi
318	318		limit++
319	319		}
320	320
321		-	extensionState, err := readExtensionStateAtFork(parentPath, limit)
	321	+	extensionState, err := readExtensionStateAtFork(ctx, parentPath, limit)
322	322		if err != nil {
323	323		return "", err
324	324		}

@@ -350,7 +350,7 @@ func BranchSessionHiddenFromHistory(parentPath, root, cwd, version string, segme
350	350		if limit > 0 && limit < len(segment.Messages) && messageHasToolCalls(segment.Messages[limit-1]) && segment.Messages[limit].Role == provider.RoleTool {
351	351		limit++
352	352		}
353		-	extensionState, err := readExtensionStateAtFork(parentPath, limit)
	353	+	extensionState, err := readExtensionStateAtFork(context.Background(), parentPath, limit)
354	354		if err != nil {
355	355		return "", err
356	356		}

@@ -361,7 +361,7 @@ func BranchSessionHiddenFromHistory(parentPath, root, cwd, version string, segme
361	361		// persisted at or before a branch boundary. Extension state is independent of
362	362		// provider messages, but its timeline still follows the effective message
363	363		// stream across compaction rows.
364		-	func readExtensionStateAtFork(path string, limit int) (map[string]json.RawMessage, error) {
	364	+	func readExtensionStateAtFork(ctx context.Context, path string, limit int) (map[string]json.RawMessage, error) {
365	365		f, err := os.Open(path)
366	366		if err != nil {
367	367		return nil, fmt.Errorf("branch: open parent for extension state: %w", err)

@@ -377,54 +377,35 @@ func readExtensionStateAtFork(path string, limit int) (map[string]json.RawMessag
377	377		}
378	378		return clone
379	379		}
380		-	applyState := func(target map[string]json.RawMessage, row sessionLine) {
381		-	if len(row.State) == 0 \|\| strings.TrimSpace(string(row.State)) == "null" {
382		-	delete(target, row.Extension)
	380	+	applyState := func(target map[string]json.RawMessage, row sessionLogRow) {
	381	+	if len(row.state) == 0 \|\| strings.TrimSpace(string(row.state)) == "null" {
	382	+	delete(target, row.extension)
383	383		return
384	384		}
385		-	if json.Valid(row.State) {
386		-	target[row.Extension] = append(json.RawMessage(nil), row.State...)
	385	+	if json.Valid(row.state) {
	386	+	target[row.extension] = append(json.RawMessage(nil), row.state...)
387	387		}
388	388		}
389	389		effectiveCount := 0
390		-	err = forEachStrictJSONLLine(f, func(line []byte, lineNo int) error {
391		-	var head sessionLineHead
392		-	if err := json.Unmarshal(line, &head); err != nil {
393		-	return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
394		-	}
395		-	switch head.Type {
396		-	case "meta", "usage", "rename":
397		-	return nil
	390	+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
	391	+	switch row.typeName {
398	392		case "message":
399		-	if _, err := hydrateMessage(line); err != nil {
400		-	return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
401		-	}
402	393		effectiveCount++
403	394		case "compaction":
404		-	messages, err := hydrateCompaction(line)
405		-	if err != nil {
406		-	return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
407		-	}
408		-	effectiveCount = len(messages)
	395	+	effectiveCount = len(row.messages)
409	396		// A compaction replaces the effective transcript. State that was
410	397		// recorded anywhere before this boundary belongs to every fork
411	398		// inside the replacement transcript, even when its old message
412	399		// index was after the requested fork point.
413	400		forkState = cloneState(state)
414	401		case "extension_state":
415		-	var row sessionLine
416		-	if err := json.Unmarshal(line, &row); err != nil {
417		-	return fmt.Errorf("line %d: invalid extension state row: %w", lineNo, err)
418		-	}
419		-	if row.Extension == "" \|\| len(row.State) > maxExtensionStateBytes {
	402	+	if row.extension == "" \|\| len(row.state) > maxExtensionStateBytes {
420	403		return nil
421	404		}
422	405		applyState(state, row)
423	406		if effectiveCount <= limit {
424	407		applyState(forkState, row)
425	408		}
426		-	default:
427		-	return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
428	409		}
429	410		return nil
430	411		})

 //
 // Returns the path of the new session file, ready for OpenSession.
 func BranchSession(parentPath, root, cwd, version string, upToMessageIdx int) (string, error) {
-	return branchSession(parentPath, root, cwd, version, upToMessageIdx, false)
+	return branchSession(context.Background(), parentPath, root, cwd, version, upToMessageIdx, false)
 }
 // BranchSessionHidden creates a branch that participates in /session tree but
 // is hidden from the flat /sessions picker. Used for in-place tree navigation.
 func BranchSessionHidden(parentPath, root, cwd, version string, upToMessageIdx int) (string, error) {
-	return branchSession(parentPath, root, cwd, version, upToMessageIdx, true)
+	return branchSession(context.Background(), parentPath, root, cwd, version, upToMessageIdx, true)
 }
-func branchSession(parentPath, root, cwd, version string, upToMessageIdx int, hideFromSessions bool) (string, error) {
+func branchSession(ctx context.Context, parentPath, root, cwd, version string, upToMessageIdx int, hideFromSessions bool) (string, error) {
 	if parentPath == "" {
 		return "", errors.New("branch: parent path is empty")
 	}
 		limit++
 	}
-	extensionState, err := readExtensionStateAtFork(parentPath, limit)
+	extensionState, err := readExtensionStateAtFork(ctx, parentPath, limit)
 	if err != nil {
 		return "", err
 	}
 	if limit > 0 && limit < len(segment.Messages) && messageHasToolCalls(segment.Messages[limit-1]) && segment.Messages[limit].Role == provider.RoleTool {
 		limit++
 	}
-	extensionState, err := readExtensionStateAtFork(parentPath, limit)
+	extensionState, err := readExtensionStateAtFork(context.Background(), parentPath, limit)
 	if err != nil {
 		return "", err
 	}
 // persisted at or before a branch boundary. Extension state is independent of
 // provider messages, but its timeline still follows the effective message
 // stream across compaction rows.
-func readExtensionStateAtFork(path string, limit int) (map[string]json.RawMessage, error) {
+func readExtensionStateAtFork(ctx context.Context, path string, limit int) (map[string]json.RawMessage, error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return nil, fmt.Errorf("branch: open parent for extension state: %w", err)
 		}
 		return clone
 	}
-	applyState := func(target map[string]json.RawMessage, row sessionLine) {
+	applyState := func(target map[string]json.RawMessage, row sessionLogRow) {
-		if len(row.State) == 0 || strings.TrimSpace(string(row.State)) == "null" {
+		if len(row.state) == 0 || strings.TrimSpace(string(row.state)) == "null" {
-			delete(target, row.Extension)
+			delete(target, row.extension)
 			return
 		}
-		if json.Valid(row.State) {
+		if json.Valid(row.state) {
-			target[row.Extension] = append(json.RawMessage(nil), row.State...)
+			target[row.extension] = append(json.RawMessage(nil), row.state...)
 		}
 	}
 	effectiveCount := 0
-	err = forEachStrictJSONLLine(f, func(line []byte, lineNo int) error {
+	err = forEachSessionLogRowContext(ctx, f, func(row sessionLogRow) error {
-		var head sessionLineHead
+		switch row.typeName {
-		if err := json.Unmarshal(line, &head); err != nil {
-			return fmt.Errorf("line %d: invalid JSON: %w", lineNo, err)
-		}
-		switch head.Type {
-		case "meta", "usage", "rename":
-			return nil
 		case "message":
-			if _, err := hydrateMessage(line); err != nil {
-				return fmt.Errorf("line %d: invalid message row: %w", lineNo, err)
-			}
 			effectiveCount++
 		case "compaction":
-			messages, err := hydrateCompaction(line)
+			effectiveCount = len(row.messages)
-			if err != nil {
-				return fmt.Errorf("line %d: invalid compaction row: %w", lineNo, err)
-			}
-			effectiveCount = len(messages)
 			// A compaction replaces the effective transcript. State that was
 			// recorded anywhere before this boundary belongs to every fork
 			// inside the replacement transcript, even when its old message
 			// index was after the requested fork point.
 			forkState = cloneState(state)
 		case "extension_state":
-			var row sessionLine
+			if row.extension == "" || len(row.state) > maxExtensionStateBytes {
-			if err := json.Unmarshal(line, &row); err != nil {
-				return fmt.Errorf("line %d: invalid extension state row: %w", lineNo, err)
-			}
-			if row.Extension == "" || len(row.State) > maxExtensionStateBytes {
 				return nil
 			}
 			applyState(state, row)
 			if effectiveCount <= limit {
 				applyState(forkState, row)
 			}
-		default:
-			return fmt.Errorf("line %d: unknown row type %q", lineNo, head.Type)
 		}
 		return nil
 	})