fix(trace): stop crediting finished sessions on unrelated commits

app/cli/internal/trace/state/ailines.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/chainloop-dev/chainloop/pkg/attestation/crafter/materials/aicodingsession"
 )
@@ -30,11 +31,35 @@ import (
 type AILineAttribution struct {
 	SessionID string                                 `json:"session_id"`
 	Files     map[string][]aicodingsession.LineRange `json:"files"`
+	// Pending holds each file whose recorded ranges have not been committed
+	// yet, and is the only set a new commit may be matched against. Files keeps
+	// every range ever recorded because push-time enrichment needs the full
+	// history, but matching against it credits a finished session on every
+	// later commit that happens to touch a file it once edited.
+	Pending map[string]PendingEdit `json:"-"`
 }
 
-// newAILineAttribution returns an AILineAttribution with an initialized Files map.
+// PendingEdit is an uncommitted edit to one file.
+type PendingEdit struct {
+	// At is when the edit was made. Ownership of a file that two sessions both
+	// have pending goes to the later edit, which rewrote the earlier one's
+	// lines.
+	At time.Time
+	// Seq is the edit's position in its ledger, counting from one. Retiring an
+	// edit is keyed off this rather than At because wall-clock times tie on a
+	// host with a coarse clock, whereas an append-only ledger's positions are
+	// unique and strictly increasing. Positions from different ledgers are not
+	// comparable, which is why ownership still goes by At.
+	Seq int
+}
+
+// newAILineAttribution returns an AILineAttribution with initialized maps.
 func newAILineAttribution(sessionID string) *AILineAttribution {
-	return &AILineAttribution{SessionID: sessionID, Files: make(map[string][]aicodingsession.LineRange)}
+	return &AILineAttribution{
+		SessionID: sessionID,
+		Files:     make(map[string][]aicodingsession.LineRange),
+		Pending:   make(map[string]PendingEdit),
+	}
 }
 
 // aiLinesPath returns the path for a session's AI line attribution file (JSONL).
@@ -49,6 +74,22 @@ type aiLineEntry struct {
 	SessionID string                      `json:"session_id,omitempty"`
 	File      string                      `json:"file"`
 	Ranges    []aicodingsession.LineRange `json:"ranges"`
+	// RecordedAt is the RFC3339 timestamp of the edit, with sub-second
+	// precision so two sessions editing one file can be ordered. Entries
+	// written before this field existed have none, and are never treated as
+	// pending — those ledgers predate consumption tracking, so their files
+	// would otherwise keep crediting their session forever.
+	RecordedAt string `json:"recorded_at,omitempty"`
+	// ConsumedBy is the SHA of the commit that committed the file's pending
+	// ranges. Set only on marker entries appended by MarkConsumed, which
+	// clear the file from Pending instead of contributing ranges.
+	ConsumedBy string `json:"consumed_by,omitempty"`
+	// ConsumedSeq is the ledger position of the edit a marker retires. It
+	// bounds the marker to the snapshot MarkConsumed read, so an edit appended
+	// afterwards — which the commit cannot have contained — keeps its pending
+	// state. Zero on a marker written before this field existed, which retires
+	// the file outright.
+	ConsumedSeq int `json:"consumed_seq,omitempty"`
 }
 
 // LoadAILineAttribution loads the AI line attribution for a session by
@@ -63,16 +104,42 @@ func (s *Store) LoadAILineAttribution(sessionID string) *AILineAttribution {
 	}
 	defer func() { _ = f.Close() }()
 
+	// Entries are appended in order, so replaying them in order lets a file
+	// go pending -> consumed -> pending again across successive edits and
+	// commits.
+	seq := 0
 	dec := json.NewDecoder(f)
 	for dec.More() {
 		var entry aiLineEntry
 		if err := dec.Decode(&entry); err != nil {
 			break
 		}
+		seq++
 		if entry.SessionID != "" {
 			attr.SessionID = entry.SessionID
 		}
+
+		if entry.ConsumedBy != "" {
+			// Retire only the edit the commit actually contained. A concurrent
+			// RecordLineRanges can land between MarkConsumed reading the
+			// pending set and appending its markers, leaving a marker behind an
+			// edit it never saw; that edit sits past the marker's snapshot, so
+			// it keeps its pending state.
+			cur, pending := attr.Pending[entry.File]
+			if pending && (entry.ConsumedSeq == 0 || cur.Seq <= entry.ConsumedSeq) {
+				delete(attr.Pending, entry.File)
+			}
+
+			continue
+		}
+
 		attr.Files[entry.File] = append(attr.Files[entry.File], entry.Ranges...)
+		// An entry with no parseable timestamp is never pending: it either
+		// predates consumption tracking, in which case its session is long
+		// gone, or it cannot be ordered against a competing session's edit.
+		if recordedAt, err := time.Parse(time.RFC3339, entry.RecordedAt); err == nil {
+			attr.Pending[entry.File] = PendingEdit{At: recordedAt, Seq: seq}
+		}
 	}
 
 	return attr
@@ -81,6 +148,69 @@ func (s *Store) LoadAILineAttribution(sessionID string) *AILineAttribution {
 // RecordLineRanges appends line ranges for a file to a session's attribution JSONL.
 // An empty or nil ranges slice still records the file as touched (e.g., deletion-only edits).
 func (s *Store) RecordLineRanges(sessionID, filePath string, ranges []aicodingsession.LineRange) error {
+	return s.RecordLineRangesAt(sessionID, filePath, ranges, time.Now().UTC())
+}
+
+// RecordLineRangesAt records an edit made at a given time. Ownership of a file
+// two sessions both edited is decided by which edit came last, so tests that
+// exercise that need to set the times rather than hope the wall clock advances
+// between two calls — it does not on a host with a coarse clock.
+func (s *Store) RecordLineRangesAt(sessionID, filePath string, ranges []aicodingsession.LineRange, at time.Time) error {
+	return s.appendAILineEntries(sessionID, []aiLineEntry{{
+		SessionID:  sessionID,
+		File:       filePath,
+		Ranges:     ranges,
+		RecordedAt: at.UTC().Format(time.RFC3339Nano),
+	}})
+}
+
+// MarkConsumed records that the pending ranges of the given files landed in
+// commit sha, so they stop crediting the session on later commits. Files the
+// session has nothing pending for are ignored, which keeps the ledger from
+// growing a marker per commit per file the session never touched.
+func (s *Store) MarkConsumed(sessionID string, files []string, sha string) error {
+	if sha == "" || len(files) == 0 {
+		return nil
+	}
+
+	return s.markConsumedAt(s.LoadAILineAttribution(sessionID).Pending, sessionID, files, sha)
+}
+
+// markConsumedAt writes the retirement markers for a pending snapshot. The
+// snapshot is a parameter because it is the boundary this has to be correct
+// across: an edit recorded after it was taken is not part of the commit, and
+// pinning each marker to the position of the edit it retires is what keeps that
+// edit pending.
+func (s *Store) markConsumedAt(pending map[string]PendingEdit, sessionID string, files []string, sha string) error {
+	if len(pending) == 0 {
+		return nil
+	}
+
+	entries := make([]aiLineEntry, 0, len(files))
+	for _, file := range files {
+		edit, ok := pending[file]
+		if !ok {
+			continue
+		}
+		entries = append(entries, aiLineEntry{
+			SessionID:   sessionID,
+			File:        file,
+			RecordedAt:  edit.At.Format(time.RFC3339Nano),
+			ConsumedBy:  sha,
+			ConsumedSeq: edit.Seq,
+		})
+	}
+
+	return s.appendAILineEntries(sessionID, entries)
+}
+
+// appendAILineEntries appends entries to a session's attribution JSONL,
+// creating the file and its directory on first write.
+func (s *Store) appendAILineEntries(sessionID string, entries []aiLineEntry) error {
+	if len(entries) == 0 {
+		return nil
+	}
+
 	path := s.aiLinesPath(sessionID)
 	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
 		return fmt.Errorf("create ai-lines dir: %w", err)
@@ -91,10 +221,15 @@ func (s *Store) RecordLineRanges(sessionID, filePath string, ranges []aicodingse
 		return err
 	}
 
-	if err := json.NewEncoder(f).Encode(aiLineEntry{SessionID: sessionID, File: filePath, Ranges: ranges}); err != nil {
-		_ = f.Close()
-		return err
+	enc := json.NewEncoder(f)
+	for i := range entries {
+		if err := enc.Encode(entries[i]); err != nil {
+			_ = f.Close()
+
+			return err
+		}
 	}
+
 	return f.Close()
 }
 
@@ -106,6 +241,7 @@ func (s *Store) LoadAllAILineAttributions() ([]*AILineAttribution, error) {
 		if errors.Is(err, os.ErrNotExist) {
 			return nil, nil
 		}
+
 		return nil, err
 	}
 

app/cli/internal/trace/state/state.go

@@ -48,6 +48,12 @@ const (
 	commitRecordExt  = ".json"
 	sessionRecordExt = ".json"
 	aiLinesExt       = ".jsonl"
+
+	// staleSessionRetention is how long a session's records are kept once
+	// everything it produced has been attested. Reachability alone never
+	// expires them: a session referenced by one surviving commit pins itself
+	// into the repository for good.
+	staleSessionRetention = 7 * 24 * time.Hour
 )
 
 // InitTraceDir creates the <dir>/chainloop-trace/ directory structure.
@@ -89,7 +95,8 @@ func (s *Store) WipeTraceDir() error {
 //  1. Walk commits/, drop any record whose SHA isn't in liveSHAs (post-rebase
 //     orphan or commit on a deleted branch).
 //  2. Walk ai-lines/ and sessions/, drop any entry whose session ID is no
-//     longer referenced by a surviving CommitRecord.
+//     longer referenced by a surviving CommitRecord, or which is older than
+//     staleSessionRetention and has nothing left to attest.
 //
 // An empty liveSHAs is treated as a pathological signal (no local branches
 // readable, or the caller fed us garbage) and the GC short-circuits — wiping
@@ -107,6 +114,9 @@ func (s *Store) GCOrphans(liveSHAs map[string]bool) error {
 	}
 
 	liveSessions := make(map[string]struct{})
+	// Sessions with a surviving commit that has not been attested yet. Their
+	// state must survive the age check — the next push still needs it.
+	unattestedSessions := make(map[string]struct{})
 	for _, entry := range commitEntries {
 		if entry.IsDir() || !strings.HasSuffix(entry.Name(), commitRecordExt) {
 			continue
@@ -124,9 +134,14 @@ func (s *Store) GCOrphans(liveSHAs map[string]bool) error {
 		}
 		for _, sid := range rec.SessionIDs {
 			liveSessions[sanitizeID(sid)] = struct{}{}
+			if !rec.Tracked {
+				unattestedSessions[sanitizeID(sid)] = struct{}{}
+			}
 		}
 	}
 
+	stale := s.staleSessionIDs(unattestedSessions)
+
 	for _, child := range []struct {
 		sub string
 		ext string
@@ -148,7 +163,8 @@ func (s *Store) GCOrphans(liveSHAs map[string]bool) error {
 				continue
 			}
 			id := strings.TrimSuffix(entry.Name(), child.ext)
-			if _, alive := liveSessions[id]; !alive {
+			_, alive := liveSessions[id]
+			if _, expired := stale[id]; !alive || expired {
 				_ = os.Remove(filepath.Join(dir, entry.Name()))
 			}
 		}
@@ -157,6 +173,46 @@ func (s *Store) GCOrphans(liveSHAs map[string]bool) error {
 	return nil
 }
 
+// staleSessionIDs returns the sanitized IDs of session records older than
+// staleSessionRetention whose work has all been attested. Records with an
+// unparseable or missing StartedAt are never reported stale, so a bad
+// timestamp cannot cause data loss.
+//
+// Liveness is deliberately not consulted: session-end does not always run, so
+// a leaked record stays Active forever and would never expire.
+//
+// Uncommitted attribution is deliberately not spared either, so a week-old
+// edit that is committed later is attributed to nobody rather than to the
+// session that made it. That is the safe direction: keeping the entry means a
+// session nothing has heard from in a week can still claim a file and pull the
+// whole commit's diff into its evidence, and by then its transcript is usually
+// gone — raw/ is wiped on every push — so the attestation it wins yields
+// nothing anyway. Under-crediting beats crediting the wrong session.
+func (s *Store) staleSessionIDs(unattested map[string]struct{}) map[string]struct{} {
+	records, err := s.LoadAllSessionRecords()
+	if err != nil {
+		return nil
+	}
+
+	cutoff := time.Now().UTC().Add(-staleSessionRetention)
+	stale := make(map[string]struct{})
+	for _, rec := range records {
+		id := sanitizeID(rec.SessionID)
+		if _, ok := unattested[id]; ok {
+			continue
+		}
+		startedAt, err := time.Parse(time.RFC3339, rec.StartedAt)
+		if err != nil {
+			continue
+		}
+		if startedAt.Before(cutoff) {
+			stale[id] = struct{}{}
+		}
+	}
+
+	return stale
+}
+
 // RemoveTraceDir removes the entire <dir>/chainloop-trace/ directory.
 func (s *Store) RemoveTraceDir() error {
 	return os.RemoveAll(s.traceDirPath())