kenn-io--agentsview/internal/parser/codex_provider.go

package parser

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"strings"
)

var _ Provider = (*codexProvider)(nil)

type codexProviderFactory struct {
	def         AgentDef
	cursorCache *codexCursorCache
}

func newCodexProviderFactory(def AgentDef) ProviderFactory {
	return &codexProviderFactory{
		def:         cloneAgentDef(def),
		cursorCache: newProductionCodexCursorCache(),
	}
}

func (f *codexProviderFactory) Definition() AgentDef {
	return cloneAgentDef(f.def)
}

func (f *codexProviderFactory) Capabilities() Capabilities {
	return codexProviderCapabilities()
}

func (f *codexProviderFactory) NewProvider(cfg ProviderConfig) Provider {
	cfg = cfg.Clone()
	return &codexProvider{
		ProviderBase: ProviderBase{
			Def:    cloneAgentDef(f.def),
			Caps:   codexProviderCapabilities(),
			Config: cfg,
		},
		sources:     newCodexSourceSet(cfg.Roots),
		cursorCache: f.cursorCache,
	}
}

type codexProvider struct {
	ProviderBase
	sources     codexSourceSet
	cursorCache *codexCursorCache
}

func (p *codexProvider) Discover(ctx context.Context) ([]SourceRef, error) {
	return p.sources.Discover(ctx)
}

func (p *codexProvider) WatchPlan(ctx context.Context) (WatchPlan, error) {
	return p.sources.WatchPlan(ctx)
}

func (p *codexProvider) SourcesForChangedPath(
	ctx context.Context,
	req ChangedPathRequest,
) ([]SourceRef, error) {
	return p.sources.SourcesForChangedPath(ctx, req)
}

func (p *codexProvider) FindSource(
	ctx context.Context,
	req FindSourceRequest,
) (SourceRef, bool, error) {
	req = ProviderFindRequestWithRawSessionID(p.Def, req)
	return p.sources.FindSource(ctx, req)
}

// AllSourcePathsForUUID returns every on-disk Codex transcript path under the
// provider's roots whose filename carries the given session UUID, without the
// live-over-archived deduplication Discover applies. A UUID can exist as both a
// live dated copy and a flat archived copy under the same root; the sync engine
// uses the full set so an mtime cutoff can judge each copy independently.
func (p *codexProvider) AllSourcePathsForUUID(uuid string) []string {
	if uuid == "" {
		return nil
	}
	seen := make(map[string]struct{})
	var paths []string
	for _, root := range p.sources.roots {
		for _, path := range p.sources.discoverSessionPaths(root) {
			if CodexSessionUUIDFromFilename(filepath.Base(path)) != uuid {
				continue
			}
			clean := filepath.Clean(path)
			if _, ok := seen[clean]; ok {
				continue
			}
			seen[clean] = struct{}{}
			paths = append(paths, path)
		}
	}
	return paths
}

// SourceRefForPath builds a SourceRef pinned to the exact transcript path,
// without live-over-archived canonicalization. Discovery, raw-ID lookup, and
// fresh stored-source lookup still prefer the live dated transcript, but
// changed-path events and non-fresh stored paths preserve the already-selected
// on-disk copy. The sync engine uses this when its DB-aware or mtime-aware
// logic has chosen a duplicated Codex UUID source (e.g. a stored archived copy
// or the cutoff-newer copy), so that choice is honored instead of being flipped
// back to the preferred dated layout. Returns false when the path is not a
// recognizable Codex source.
func (p *codexProvider) SourceRefForPath(path string) (SourceRef, bool) {
	for _, root := range p.sources.roots {
		if source, ok := p.sources.sourceRef(root, path, true); ok {
			return source, true
		}
		if source, ok := p.sources.directPathSource(root, path, true); ok {
			return source, true
		}
	}
	return SourceRef{}, false
}

func (p *codexProvider) Fingerprint(
	ctx context.Context,
	source SourceRef,
) (SourceFingerprint, error) {
	return p.sources.Fingerprint(ctx, source)
}

func (p *codexProvider) Parse(
	ctx context.Context,
	req ParseRequest,
) (ParseOutcome, error) {
	if err := ctx.Err(); err != nil {
		return ParseOutcome{}, err
	}
	path, ok := p.sources.pathFromSource(req.Source)
	if !ok {
		return ParseOutcome{}, fmt.Errorf("codex source path unavailable")
	}
	if req.ForceParse {
		EvictCodexSessionIndexForSession(path)
	}
	machine := firstNonEmptyJSONLString(req.Machine, p.Config.Machine)
	sess, msgs, err := p.parseSession(path, machine, false)
	if err != nil {
		return ParseOutcome{}, err
	}
	if sess == nil {
		return ParseOutcome{
			ResultSetComplete: true,
			SkipReason:        SkipNoSession,
		}, nil
	}
	if req.Fingerprint.Hash != "" {
		sess.File.Hash = req.Fingerprint.Hash
	}
	return ParseOutcome{
		Results: []ParseResultOutcome{{
			Result: ParseResult{
				Session:  *sess,
				Messages: msgs,
			},
			DataVersion: DataVersionCurrent,
		}},
		ResultSetComplete: true,
		// A requested full parse is the authoritative message set, so
		// force-replace the stored rows; this remains distinct from the provider's
		// incremental append path and preserves the legacy behavior where a late
		// token_count line appended to
		// an existing turn rewrites the stored message instead of being dropped by
		// an append-only write.
		ForceReplace: true,
	}, nil
}

func (p *codexProvider) ParseIncremental(
	ctx context.Context,
	req IncrementalRequest,
) (IncrementalOutcome, IncrementalStatus, error) {
	if err := ctx.Err(); err != nil {
		return IncrementalOutcome{}, IncrementalUnsupported, err
	}
	path, ok := p.sources.pathFromSource(req.Source)
	if !ok {
		return IncrementalOutcome{}, IncrementalUnsupported,
			fmt.Errorf("codex source path unavailable")
	}
	if req.Offset < 0 || req.Fingerprint.Size < req.Offset {
		return IncrementalOutcome{ForceReplace: true},
			IncrementalNeedsFullParse, nil
	}
	f, err := os.Open(path)
	if err != nil {
		return IncrementalOutcome{}, IncrementalNeedsFullParse, err
	}
	defer f.Close()
	info, err := f.Stat()
	if err != nil {
		return IncrementalOutcome{}, IncrementalNeedsFullParse, err
	}
	inode, device := sourceFileIdentity(info)
	if (req.Fingerprint.Inode != 0 && req.Fingerprint.Inode != inode) ||
		(req.Fingerprint.Device != 0 && req.Fingerprint.Device != device) ||
		info.Size() < req.Fingerprint.Size {
		return IncrementalOutcome{ForceReplace: true},
			IncrementalNeedsFullParse, nil
	}
	safe, err := codexSafeResumeOffsetFile(f, req.Offset)
	if err != nil {
		return IncrementalOutcome{}, IncrementalNeedsFullParse, err
	}
	if !safe {
		return IncrementalOutcome{ForceReplace: true},
			IncrementalNeedsFullParse, nil
	}
	if req.Fingerprint.Size == req.Offset {
		return IncrementalOutcome{}, IncrementalNoNewData, nil
	}

	result, err := p.parseSessionFromSnapshot(
		path,
		req.Offset,
		req.StartOrdinal,
		false,
		f,
		info,
		req.Fingerprint.Size,
	)
	if err != nil {
		if IsIncrementalFullParseFallback(err) {
			return IncrementalOutcome{ForceReplace: true},
				IncrementalNeedsFullParse, nil
		}
		return IncrementalOutcome{}, IncrementalNeedsFullParse, err
	}
	if !result.initialCursor.firstUserSeen && result.cursor.firstUserSeen {
		// The stored session has no genuine first prompt yet, so appending this
		// tail would leave its persisted FirstMessage preview stale. A full parse
		// can rebuild both the messages and the session summary atomically.
		return IncrementalOutcome{ForceReplace: true},
			IncrementalNeedsFullParse, nil
	}
	p.cursorCache.Put(
		path,
		req.Offset,
		result.inode,
		result.device,
		result.initialCursor,
	)
	if result.consumedBytes == 0 {
		return IncrementalOutcome{}, IncrementalNoNewData, nil
	}
	p.cursorCache.Put(
		path,
		req.Offset+result.consumedBytes,
		result.inode,
		result.device,
		result.cursor,
	)

	totalOut, peakCtx, hasTotalOut, hasPeakCtx :=
		codexProviderTokenTotals(result.messages)
	termination := codexIncrementalTermination(result.cursor.lastTaskEvent)
	return IncrementalOutcome{
		SessionID:            req.SessionID,
		Messages:             result.messages,
		EndedAt:              result.endedAt,
		ConsumedBytes:        result.consumedBytes,
		MessageCount:         len(result.messages),
		UserMessageCount:     codexProviderUserMessageCount(result.messages),
		TotalOutputTokens:    totalOut,
		PeakContextTokens:    peakCtx,
		HasTotalOutputTokens: hasTotalOut,
		HasPeakContextTokens: hasPeakCtx,
		TerminationStatus:    termination,
	}, IncrementalApplied, nil
}

type codexSource struct {
	Root   string
	Path   string
	UUID   string
	Layout CodexLayout
}

type codexSourceSet struct {
	roots []string
}

func newCodexSourceSet(roots []string) codexSourceSet {
	return codexSourceSet{roots: cleanJSONLRoots(roots)}
}

func (s codexSourceSet) Discover(ctx context.Context) ([]SourceRef, error) {
	return s.discover(ctx, func(string) bool { return true })
}

func (s codexSourceSet) discover(
	ctx context.Context,
	includeRoot func(string) bool,
) ([]SourceRef, error) {
	var sources []SourceRef
	byKey := make(map[string]SourceRef)
	for _, root := range s.roots {
		if err := ctx.Err(); err != nil {
			return nil, err
		}
		if !includeRoot(root) {
			continue
		}
		if strings.HasPrefix(root, "s3://") {
			// s3:// roots have no local layout to walk: enumerate the objects
			// directly and carry each one's durable metadata in the Opaque
			// payload. Each object is its own session keyed by URI, so the
			// live-over-archived preference (which inspects a local codexSource
			// layout) does not apply here.
			for _, file := range discoverCodexS3(root) {
				source := s3SourceRefFromDiscoveredFile(file)
				if _, ok := byKey[source.Key]; ok {
					continue
				}
				byKey[source.Key] = source
			}
			continue
		}
		for _, path := range s.discoverSessionPaths(root) {
			source, ok := s.sourceRef(root, path, true)
			if !ok {
				source, ok = s.directPathSource(root, path, true)
			}
			if !ok {
				continue
			}
			if current, ok := byKey[source.Key]; ok &&
				!preferCodexSource(source, current) {
				continue
			}
			byKey[source.Key] = source
		}
	}
	for _, source := range byKey {
		sources = append(sources, source)
	}
	sortJSONLSources(sources)
	return sources, nil
}

// discoverSessionPaths finds all Codex JSONL session file paths under
// sessionsDir, covering both the standard year/month/day layout and a flat
// archived directory. Paths are returned sorted for deterministic discovery,
// matching the behavior the package-level entrypoint provided before the fold.
func (s codexSourceSet) discoverSessionPaths(sessionsDir string) []string {
	var paths []string

	entries, err := os.ReadDir(sessionsDir)
	if err != nil {
		return nil
	}
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		if !isCodexSessionFilename(entry.Name()) {
			continue
		}
		paths = append(paths, filepath.Join(sessionsDir, entry.Name()))
	}

	walkCodexDayDirs(sessionsDir, func(dayPath string) bool {
		dayEntries, err := os.ReadDir(dayPath)
		if err != nil {
			return true
		}
		for _, sf := range dayEntries {
			if sf.IsDir() {
				continue
			}
			if !isCodexSessionFilename(sf.Name()) {
				continue
			}
			paths = append(paths, filepath.Join(dayPath, sf.Name()))
		}
		return true
	})

	slices.Sort(paths)
	return paths
}

// findSourceFile resolves a Codex session file by UUID under sessionsDir.
// It prefers the standard year/month/day live path when present, then falls
// back to a flat archived directory entry, matching the lookup precedence the
// package-level entrypoint provided before the fold.
func (s codexSourceSet) findSourceFile(sessionsDir, sessionID string) string {
	if !IsValidSessionID(sessionID) {
		return ""
	}

	var archived string
	entries, err := os.ReadDir(sessionsDir)
	if err == nil {
		for _, f := range entries {
			if f.IsDir() {
				continue
			}
			name := f.Name()
			if !isCodexSessionFilename(name) {
				continue
			}
			if extractUUIDFromRollout(name) == sessionID {
				archived = filepath.Join(sessionsDir, name)
				break
			}
		}
	}

	var live string
	walkCodexDayDirs(sessionsDir, func(dayPath string) bool {
		if live != "" {
			return false
		}
		dayEntries, err := os.ReadDir(dayPath)
		if err != nil {
			return true
		}
		for _, f := range dayEntries {
			if f.IsDir() {
				continue
			}
			name := f.Name()
			if !isCodexSessionFilename(name) {
				continue
			}
			if extractUUIDFromRollout(name) == sessionID {
				live = filepath.Join(dayPath, name)
				return false
			}
		}
		return true
	})
	if live != "" {
		return live
	}
	return archived
}

func (s codexSourceSet) WatchPlan(context.Context) (WatchPlan, error) {
	roots := make([]WatchRoot, 0, len(s.roots)*2)
	seenShallow := make(map[string]struct{})
	for _, root := range s.roots {
		roots = append(roots, WatchRoot{
			Path:         root,
			Recursive:    true,
			IncludeGlobs: []string{"*.jsonl"},
			DebounceKey:  string(AgentCodex) + ":sessions:" + root,
		})
		for _, shallow := range ResolveCodexShallowWatchRoots(root) {
			shallow = filepath.Clean(shallow)
			if _, ok := seenShallow[shallow]; ok {
				continue
			}
			seenShallow[shallow] = struct{}{}
			roots = append(roots, WatchRoot{
				Path:         shallow,
				Recursive:    false,
				IncludeGlobs: []string{CodexSessionIndexFilename},
				DebounceKey:  string(AgentCodex) + ":index:" + shallow,
			})
		}
	}
	return WatchPlan{Roots: roots}, nil
}

func (s codexSourceSet) SourcesForChangedPath(
	ctx context.Context,
	req ChangedPathRequest,
) ([]SourceRef, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if filepath.Base(req.Path) == CodexSessionIndexFilename {
		return s.sourcesForIndexPath(ctx, req.Path)
	}
	for _, root := range s.roots {
		source, ok := s.sourceRef(root, req.Path, true)
		if !ok {
			source, ok = s.directPathSource(root, req.Path, true)
		}
		if ok {
			return []SourceRef{source}, nil
		}
		if !jsonlMissingPathFallbackAllowed(req) {
			continue
		}
		source, ok = s.sourceRef(root, req.Path, false)
		if !ok {
			source, ok = s.directPathSource(root, req.Path, false)
		}
		if ok {
			return []SourceRef{source}, nil
		}
	}
	return nil, nil
}

func (s codexSourceSet) FindSource(
	ctx context.Context,
	req FindSourceRequest,
) (SourceRef, bool, error) {
	if err := ctx.Err(); err != nil {
		return SourceRef{}, false, err
	}
	for _, path := range []string{req.StoredFilePath, req.FingerprintKey} {
		if path == "" {
			continue
		}
		for _, root := range s.roots {
			if source, ok := s.sourceRef(root, path, true); ok {
				if !req.RequireFreshSource || req.PreferStoredSource {
					return source, true, nil
				}
				return s.canonicalSource(ctx, source)
			}
			if source, ok := s.directPathSource(root, path, true); ok {
				if !req.RequireFreshSource || req.PreferStoredSource {
					return source, true, nil
				}
				return s.canonicalSource(ctx, source)
			}
		}
	}
	if req.RawSessionID == "" {
		return SourceRef{}, false, nil
	}
	for _, root := range s.roots {
		path := s.findSourceFile(root, req.RawSessionID)
		if path == "" {
			continue
		}
		if source, ok := s.sourceRef(root, path, true); ok {
			return s.canonicalSource(ctx, source)
		}
	}
	return SourceRef{}, false, nil
}

func (s codexSourceSet) Fingerprint(
	ctx context.Context,
	source SourceRef,
) (SourceFingerprint, error) {
	if err := ctx.Err(); err != nil {
		return SourceFingerprint{}, err
	}
	path, ok := s.pathFromSource(source)
	if !ok {
		return SourceFingerprint{}, fmt.Errorf("codex source path unavailable")
	}
	info, err := os.Stat(path)
	if err != nil {
		return SourceFingerprint{}, fmt.Errorf("stat %s: %w", path, err)
	}
	if info.IsDir() {
		return SourceFingerprint{}, fmt.Errorf("stat %s: source is a directory", path)
	}
	hash, err := hashJSONLSourceFile(path)
	if err != nil {
		return SourceFingerprint{}, err
	}
	inode, device := sourceFileIdentity(info)
	return SourceFingerprint{
		Key:     firstNonEmptyJSONLString(source.FingerprintKey, source.Key, path),
		Size:    info.Size(),
		MTimeNS: CodexEffectiveMtime(path, info.ModTime().UnixNano()),
		Inode:   inode,
		Device:  device,
		Hash:    hash,
	}, nil
}

func (s codexSourceSet) pathFromSource(source SourceRef) (string, bool) {
	switch src := source.Opaque.(type) {
	case codexSource:
		return src.Path, src.Path != ""
	case *codexSource:
		if src != nil && src.Path != "" {
			return src.Path, true
		}
	case MaterializedFileSource:
		return src.Path, src.Path != ""
	}
	for _, candidate := range []string{
		source.DisplayPath,
		source.FingerprintKey,
		source.Key,
	} {
		for _, root := range s.roots {
			if ref, ok := s.sourceRef(root, candidate, true); ok {
				src := ref.Opaque.(codexSource)
				return src.Path, true
			}
			if ref, ok := s.directPathSource(root, candidate, true); ok {
				src := ref.Opaque.(codexSource)
				return src.Path, true
			}
		}
	}
	return "", false
}

func (s codexSourceSet) sourcesForIndexPath(
	ctx context.Context,
	indexPath string,
) ([]SourceRef, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	indexDir := filepath.Dir(indexPath)
	return s.discover(ctx, func(root string) bool {
		return filepath.Dir(root) == indexDir
	})
}

func (s codexSourceSet) sourceRef(
	root string,
	path string,
	requireRegular bool,
) (SourceRef, bool) {
	root = filepath.Clean(root)
	path = filepath.Clean(path)
	layout, uuid, ok := CodexSessionPathInfo(root, path)
	if !ok || uuid == "" {
		return SourceRef{}, false
	}
	if requireRegular && !IsRegularFile(path) {
		return SourceRef{}, false
	}
	return SourceRef{
		Provider:       AgentCodex,
		Key:            codexSourceKey(uuid),
		DisplayPath:    path,
		FingerprintKey: path,
		Opaque: codexSource{
			Root:   root,
			Path:   path,
			UUID:   uuid,
			Layout: layout,
		},
	}, true
}

func (s codexSourceSet) directPathSource(
	root string,
	path string,
	requireRegular bool,
) (SourceRef, bool) {
	root = filepath.Clean(root)
	path = filepath.Clean(path)
	if !strings.HasSuffix(path, ".jsonl") || !pathUnderRoot(root, path) {
		return SourceRef{}, false
	}
	if requireRegular && !IsRegularFile(path) {
		return SourceRef{}, false
	}
	return SourceRef{
		Provider:       AgentCodex,
		Key:            path,
		DisplayPath:    path,
		FingerprintKey: path,
		Opaque: codexSource{
			Root: root,
			Path: path,
		},
	}, true
}

func (s codexSourceSet) canonicalSource(
	ctx context.Context,
	source SourceRef,
) (SourceRef, bool, error) {
	src, ok := source.Opaque.(codexSource)
	if !ok || src.UUID == "" {
		return source, true, nil
	}
	best := source
	for _, root := range s.roots {
		if err := ctx.Err(); err != nil {
			return SourceRef{}, false, err
		}
		path := s.findSourceFile(root, src.UUID)
		if path == "" {
			continue
		}
		candidate, ok := s.sourceRef(root, path, true)
		if !ok {
			continue
		}
		if preferCodexSource(candidate, best) {
			best = candidate
		}
	}
	return best, true, nil
}

func codexSourceKey(uuid string) string {
	return string(AgentCodex) + ":" + uuid
}

func preferCodexSource(candidate, current SourceRef) bool {
	cand := candidate.Opaque.(codexSource)
	curr := current.Opaque.(codexSource)
	if cand.Layout != curr.Layout {
		return cand.Layout == CodexLayoutDated
	}
	return candidate.DisplayPath < current.DisplayPath
}

func codexProviderUserMessageCount(messages []ParsedMessage) int {
	count := 0
	for _, message := range messages {
		if message.Role == RoleUser && message.Content != "" {
			count++
		}
	}
	return count
}

func codexProviderTokenTotals(
	messages []ParsedMessage,
) (totalOut int, peakCtx int, hasTotalOut bool, hasPeakCtx bool) {
	for _, message := range messages {
		if message.HasOutputTokens {
			totalOut += message.OutputTokens
			hasTotalOut = true
		}
		if message.HasContextTokens &&
			(!hasPeakCtx || message.ContextTokens > peakCtx) {
			peakCtx = message.ContextTokens
			hasPeakCtx = true
		}
	}
	return totalOut, peakCtx, hasTotalOut, hasPeakCtx
}

func codexIncrementalTermination(
	lastTaskEvent string,
) *TerminationStatus {
	status := classifyCodexTermination(lastTaskEvent)
	if status == "" {
		return nil
	}
	return &status
}

func codexProviderCapabilities() Capabilities {
	return Capabilities{
		Source: SourceCapabilities{
			DiscoverSources:      CapabilitySupported,
			WatchSources:         CapabilitySupported,
			ClassifyChangedPath:  CapabilitySupported,
			FindSource:           CapabilitySupported,
			CompositeFingerprint: CapabilitySupported,
			IncrementalAppend:    CapabilitySupported,
			MultiSessionSource:   CapabilityNotApplicable,
			PerSessionErrors:     CapabilityNotApplicable,
			ExcludedSessions:     CapabilityNotApplicable,
			ForceReplaceOnParse:  CapabilitySupported,
			VerifiedLocalStat:    CapabilitySupported,
		},
		Content: ContentCapabilities{
			FirstMessage:         CapabilitySupported,
			SessionName:          CapabilitySupported,
			Cwd:                  CapabilitySupported,
			GitBranch:            CapabilitySupported,
			Relationships:        CapabilitySupported,
			Subagents:            CapabilitySupported,
			Thinking:             CapabilitySupported,
			ToolCalls:            CapabilitySupported,
			ToolResults:          CapabilitySupported,
			ToolResultEvents:     CapabilitySupported,
			PerMessageTokenUsage: CapabilitySupported,
			TerminationStatus:    CapabilitySupported,
			Model:                CapabilitySupported,
		},
	}
}