Skip to content

fix(importer): preserve repository attribution across import and watch #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tobilg merged 1 commit into from
Sep 15, 2026
Merged

fix(importer): preserve repository attribution across import and watch #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,8 @@ ai-observer watch [claude-code|codex|gemini|all] [options]
- **First start with `--backfill`:** Loads all existing session data, then watches for new changes.
- **Restart:** Resumes from where it left off, importing any data written while the watcher was stopped.

Claude Code and Codex logs and token/cost metrics receive repository attribution in both import and watch mode. The watcher retains repository context across restarts and recovers it from existing session files when needed. See [repository attribution](docs/import.md#repository-attribution-in-import-and-watch) for evidence precedence, reference-mining scope, and limits on updating historical records.

At startup, the watcher detects which tools are installed and reports their status:

```
Expand Down
28 changes: 6 additions & 22 deletions backend/internal/importer/claude.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,11 +184,8 @@ type claudeWorktreeSession struct {

// claudeSessionMeta holds session-level metadata collected in a first pass
type claudeSessionMeta struct {
GitBranch string
Cwd string
OriginalCwd string // from worktree-state.worktreeSession.originalCwd
CandidateRepos []string // owner/repo references from pr-link entries (most-frequent first)
PRLinks []claudePRLink
RepositoryMetadata
PRLinks []claudePRLink
}

type claudePRLink struct {
Expand All @@ -202,24 +199,14 @@ type claudePRLink struct {
func (p *ClaudeParser) collectSessionMeta(lines []string) claudeSessionMeta {
meta := claudeSessionMeta{}
seenPRs := make(map[string]bool)
seenRepos := make(map[string]bool)

for _, line := range lines {
var entry ClaudeJSONLEntry
if err := json.Unmarshal([]byte(line), &entry); err != nil {
continue
}
if entry.GitBranch != "" && meta.GitBranch == "" {
meta.GitBranch = entry.GitBranch
}
if entry.Cwd != "" && meta.Cwd == "" {
meta.Cwd = entry.Cwd
}
if entry.Type == "worktree-state" && entry.WorktreeSession != nil {
if oc := entry.WorktreeSession.OriginalCwd; oc != "" && meta.OriginalCwd == "" {
meta.OriginalCwd = oc
}
}
meta.ObserveClaude(entry)

if entry.Type == "pr-link" && entry.PRNumber > 0 {
ts, err := time.Parse(time.RFC3339Nano, entry.Timestamp)
if err != nil {
Expand All @@ -237,10 +224,7 @@ func (p *ClaudeParser) collectSessionMeta(lines []string) claudeSessionMeta {
continue
}
seenPRs[key] = true
if r := repo; !seenRepos[r] {
seenRepos[r] = true
meta.CandidateRepos = append(meta.CandidateRepos, r)
}

meta.PRLinks = append(meta.PRLinks, claudePRLink{Number: entry.PRNumber, URL: entry.PRUrl, Repository: repo, Timestamp: ts})
}
}
Expand All @@ -251,7 +235,7 @@ func (p *ClaudeParser) collectSessionMeta(lines []string) claudeSessionMeta {
//
// A linked PR is only a candidate: its repository may differ from the working tree.
func extractRepository(meta claudeSessionMeta) string {
return resolveSessionRepository(meta.CandidateRepos, "", meta.OriginalCwd, meta.Cwd)
return meta.Repository()
}

// countLines returns the number of lines in s (at least 1 for non-empty strings)
Expand Down
60 changes: 18 additions & 42 deletions backend/internal/importer/codex.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,7 @@ type CodexReasoningSummary struct {
// codexGitInfo contains git metadata from session_meta
type codexGitInfo struct {
RepositoryURL string `json:"repository_url"`
Branch string `json:"branch"`
}

// CodexSessionMeta represents session metadata
Expand Down Expand Up @@ -182,7 +183,7 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult
var sessionMeta *CodexSessionMeta
var currentModel string
var lastTokenCount *CodexTokenCount
var repository string
var attribution RepositoryMetadata
messageIndex := 0 // Track message order for transcripts

for scanner.Scan() {
Expand All @@ -209,6 +210,8 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult
}
}

attribution.ObserveCodex(entry)

// Update time range
if result.FirstTime.IsZero() || ts.Before(result.FirstTime) {
result.FirstTime = ts
Expand All @@ -229,13 +232,6 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult
currentModel = meta.Model
}

// Resolve repository using git URL from session metadata (authoritative for Codex).
gitRepoURL := ""
if meta.Git != nil {
gitRepoURL = meta.Git.RepositoryURL
}
repository = resolveSessionRepository(nil, gitRepoURL, "", meta.Cwd)

// Create session start log
logRecord := api.LogRecord{
Timestamp: ts,
Expand All @@ -255,9 +251,6 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult
if meta.Cwd != "" {
logRecord.LogAttributes["cwd"] = meta.Cwd
}
if repository != "" {
logRecord.LogAttributes["repository"] = repository
}
result.Logs = append(result.Logs, logRecord)
result.RecordCount++
}
Expand Down Expand Up @@ -305,29 +298,29 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult

// Create metrics for non-zero deltas
if deltaInput > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "input", float64(deltaInput), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "input", float64(deltaInput)))
}
if deltaOutput > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "output", float64(deltaOutput), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "output", float64(deltaOutput)))
}
if deltaCacheCreation > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "cache_creation", float64(deltaCacheCreation), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "cache_creation", float64(deltaCacheCreation)))
}
if deltaCacheRead > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "cache_read", float64(deltaCacheRead), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "cache_read", float64(deltaCacheRead)))
}
if deltaReasoning > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "reasoning", float64(deltaReasoning), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "reasoning", float64(deltaReasoning)))
}
if deltaTool > 0 {
result.Metrics = append(result.Metrics, createCodexTokenMetric(ts, currentModel, "tool", float64(deltaTool), repository))
result.Metrics = append(result.Metrics, CreateCodexTokenMetric(ts, currentModel, "tool", float64(deltaTool)))
}

// Calculate and add cost metric
// Note: cache_read is used for cost calculation (cache_creation tokens are billed at input rate)
cost := pricing.CalculateCodexCost(currentModel, int64(deltaInput), int64(deltaCacheRead), int64(deltaOutput))
if cost != nil && *cost > 0 {
result.Metrics = append(result.Metrics, createCodexCostMetric(ts, currentModel, *cost, repository))
result.Metrics = append(result.Metrics, CreateCodexCostMetric(ts, currentModel, *cost))
}

lastTokenCount = tokenCount
Expand Down Expand Up @@ -532,19 +525,18 @@ func (p *CodexParser) ParseFile(ctx context.Context, path string) (*ImportResult
return nil, fmt.Errorf("reading file: %w", err)
}

attribution.Enrich(result.Metrics, result.Logs)
return result, nil
}

// createCodexTokenMetric creates a token usage metric for Codex with optional repository attribution.
func createCodexTokenMetric(ts time.Time, model, tokenType string, value float64, repository string) api.MetricDataPoint {
// CreateCodexTokenMetric creates a token usage metric. Importers and watchers
// attach repository context to the completed batch.
func CreateCodexTokenMetric(ts time.Time, model, tokenType string, value float64) api.MetricDataPoint {
attrs := map[string]string{
"type": tokenType,
"model": model,
"import_source": "local_jsonl",
}
if repository != "" {
attrs["repository"] = repository
}
return api.MetricDataPoint{
Timestamp: ts,
ServiceName: SourceCodex.ServiceName(),
Expand All @@ -555,22 +547,13 @@ func createCodexTokenMetric(ts time.Time, model, tokenType string, value float64
}
}

// CreateCodexTokenMetric creates a token usage metric for Codex.
// This public form is used by the file watcher; ParseFile uses createCodexTokenMetric
// (private) which also attaches repository attribution.
func CreateCodexTokenMetric(ts time.Time, model, tokenType string, value float64) api.MetricDataPoint {
return createCodexTokenMetric(ts, model, tokenType, value, "")
}

// createCodexCostMetric creates a cost usage metric for Codex with optional repository attribution.
func createCodexCostMetric(ts time.Time, model string, cost float64, repository string) api.MetricDataPoint {
// CreateCodexCostMetric creates a cost metric. Importers and watchers attach
// repository context to the completed batch.
func CreateCodexCostMetric(ts time.Time, model string, cost float64) api.MetricDataPoint {
attrs := map[string]string{
"model": model,
"import_source": "local_jsonl",
}
if repository != "" {
attrs["repository"] = repository
}
return api.MetricDataPoint{
Timestamp: ts,
ServiceName: SourceCodex.ServiceName(),
Expand All @@ -580,10 +563,3 @@ func createCodexCostMetric(ts time.Time, model string, cost float64, repository
Attributes: attrs,
}
}

// CreateCodexCostMetric creates a cost usage metric for Codex.
// This public form is used by the file watcher; ParseFile uses createCodexCostMetric
// (private) which also attaches repository attribution.
func CreateCodexCostMetric(ts time.Time, model string, cost float64) api.MetricDataPoint {
return createCodexCostMetric(ts, model, cost, "")
}
Loading
Loading