From 7b3a9ad5320159f307ac8cbe1cdf4f6008ca0ea6 Mon Sep 17 00:00:00 2001 From: tobilg Date: Tue, 15 Sep 2026 10:56:10 +0200 Subject: [PATCH] fix(importer): preserve session repository attribution in watch mode --- README.md | 2 + backend/internal/importer/claude.go | 28 +- backend/internal/importer/codex.go | 60 ++--- backend/internal/importer/repos.go | 192 +++++++------ backend/internal/importer/repos_test.go | 172 ++++++++++++ .../internal/importer/repository_metadata.go | 218 +++++++++++++++ .../internal/watcher/claude_incremental.go | 26 +- backend/internal/watcher/codex_incremental.go | 54 +++- backend/internal/watcher/repository.go | 31 +++ backend/internal/watcher/repository_test.go | 253 ++++++++++++++++++ docs/import.md | 21 +- 11 files changed, 896 insertions(+), 161 deletions(-) create mode 100644 backend/internal/importer/repos_test.go create mode 100644 backend/internal/importer/repository_metadata.go create mode 100644 backend/internal/watcher/repository.go create mode 100644 backend/internal/watcher/repository_test.go diff --git a/README.md b/README.md index 955f32c..5f83471 100644 --- a/README.md +++ b/README.md @@ -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: ``` diff --git a/backend/internal/importer/claude.go b/backend/internal/importer/claude.go index 45d11ef..7f3edbc 100644 --- a/backend/internal/importer/claude.go +++ b/backend/internal/importer/claude.go @@ -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 { @@ -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 { @@ -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}) } } @@ -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) diff --git a/backend/internal/importer/codex.go b/backend/internal/importer/codex.go index 07f8860..a73a373 100644 --- a/backend/internal/importer/codex.go +++ b/backend/internal/importer/codex.go @@ -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 @@ -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() { @@ -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 @@ -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, @@ -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++ } @@ -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 @@ -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(), @@ -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(), @@ -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, "") -} diff --git a/backend/internal/importer/repos.go b/backend/internal/importer/repos.go index ccd73eb..8f356dc 100644 --- a/backend/internal/importer/repos.go +++ b/backend/internal/importer/repos.go @@ -5,9 +5,10 @@ import ( "os/exec" "path/filepath" "regexp" - "strconv" "strings" "sync" + + "mvdan.cc/sh/v3/syntax" ) // Repository attribution logic. @@ -21,19 +22,16 @@ import ( // 2. On-disk git remote from cwd (real remote, authoritative) // 3. Candidate repos whose name matches cwd name or path segment // 4. Bare cwd directory name -// 5. First candidate repository reference +// 5. A unique candidate when no cwd is available var ( githubURLRe = regexp.MustCompile(`(?i)(?:^|://|@)(?:[^/@]+@)?github\.com[:/]+([^/]+/[^/?#]+)`) schemeRe = regexp.MustCompile(`(?i)^[a-z][a-z0-9+.-]*://`) userAtRe = regexp.MustCompile(`^[^@/]+@`) - ghPRRe = regexp.MustCompile(`(?i)\bgh\s+pr\s+(create|merge|close|reopen|checkout|view|diff|review|list|edit|comment|ready|status)\b`) - repoFlagRe = regexp.MustCompile(`(?i)--repo[ =]\s*['"]?([^\s'"]+)`) - pullURLRe = regexp.MustCompile(`(?i)https?://github\.com/([^/\s]+/[^/\s]+?)/pull/(\d+)`) - // repoURLRe catches any github.com/owner/repo reference not already matched by pullURLRe. - // The lazy second component combined with the required terminator stops at path separators. - repoURLRe = regexp.MustCompile(`(?i)github\.com[:/]+([^/\s]+/[^/\s]+?)(?:\.git)?(?:[/\s)"',` + "`" + `]|$)`) + // Match standalone GitHub references, excluding lookalike hosts and dynamic paths. + repoURLRe = regexp.MustCompile(`(?i)(?:^|[^a-z0-9_.@/-])(?:https?://|ssh://git@|git@)?github\.com[:/]([a-z0-9_.-]+/[a-z0-9_.-]+)`) + repoNameRe = regexp.MustCompile(`^[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+$`) ) var placeholderRepos = map[string]bool{ @@ -130,11 +128,11 @@ func cwdGitRemote(path string) string { // resolveSessionRepository returns the best repository id for a session. // // candidateRepos are owner/repo references seen for the session - from pr-link -// entries and text-mined references - ideally most-frequent first. +// entries and text-mined references - deduplicated by repository identity. // // The working directory says which repo we're in; candidates supply the owner/ // prefix the cwd alone can't. A candidate whose repo-part matches the cwd name -// or a path segment is the highest-confidence answer. A candidate that doesn't +// or a path segment can supply an owner when Git metadata is absent. A candidate that doesn't // match is treated as a stray and used only when there's no cwd at all. func resolveSessionRepository(candidateRepos []string, gitRepoURL, originalCwd, cwd string) string { // Codex carries the cwd's actual git remote - authoritative. @@ -162,29 +160,44 @@ func resolveSessionRepository(candidateRepos []string, gitRepoURL, originalCwd, } } + // Resolve only an unambiguous candidate. An exact cwd-name match outranks + // ancestor matches; competing owners of the same name fall back to cwd. if len(candidates) > 0 && path != "" { segments := make(map[string]bool) for _, seg := range strings.Split(filepath.ToSlash(path), "/") { - if seg != "" { - segments[strings.ToLower(seg)] = true - } + segments[strings.ToLower(seg)] = true } - for _, r := range candidates { - parts := strings.Split(r, "/") - name := strings.ToLower(parts[len(parts)-1]) - if name == strings.ToLower(cwdName) || segments[name] { - return r + for _, exact := range []bool{true, false} { + var matches []string + for _, repo := range candidates { + name := strings.ToLower(repo[strings.LastIndex(repo, "/")+1:]) + if (exact && name == strings.ToLower(cwdName)) || (!exact && segments[name]) { + matches = append(matches, repo) + } + } + if len(matches) > 0 { + if repo := uniqueRepository(matches); repo != "" { + return repo + } + return cwdName } } } - if cwdName != "" { return cwdName } - if len(candidates) > 0 { - return candidates[0] + return uniqueRepository(candidates) +} + +func uniqueRepository(candidates []string) string { + var result string + for _, repo := range candidates { + if result != "" && !strings.EqualFold(result, repo) { + return "" + } + result = repo } - return "" + return result } // isPlaceholder returns true if repo is a documentation/example placeholder. @@ -195,87 +208,88 @@ func isPlaceholder(repo string) bool { func normalizeRepoFlag(val string) string { if gh := githubRepoFromURL(val); gh != "" { - if isPlaceholder(gh) { - return "" - } - return gh - } - val = strings.TrimRight(strings.TrimSpace(val), "/") - var parts []string - for _, p := range strings.Split(val, "/") { - if p != "" { - parts = append(parts, p) - } + val = gh } - if len(parts) >= 2 && !strings.Contains(parts[len(parts)-2], ".") { - repo := strings.TrimSuffix(parts[len(parts)-2]+"/"+parts[len(parts)-1], ".git") - if isPlaceholder(repo) { - return "" - } - return repo + val = strings.TrimSuffix(strings.TrimRight(strings.TrimSpace(val), "/"), ".git") + if !repoNameRe.MatchString(val) || isPlaceholder(val) { + return "" } - return "" + return val } -// extractedRefs holds PR/repo references mined from free text. -type extractedRefs struct { - PRActions []string - PRURLs []string - PRNumbers []int - Repos []string -} - -// extractRefs scans one or more strings for gh pr commands, PR URLs, -// --repo flags, and GitHub repo references. -func extractRefs(texts ...string) extractedRefs { - result := extractedRefs{} - seenActions := map[string]bool{} - seenURLs := map[string]bool{} - seenNumbers := map[int]bool{} - seenRepos := map[string]bool{} - +// extractRepositoryRefs mines attribution candidates only. References never +// create activity metrics, and commands/expansions are parsed, never executed. +func extractRepositoryRefs(texts ...string) []string { + var repos []string for _, text := range texts { - if text == "" { + repos = append(repos, extractGitHubRefs(text)...) + // Parse literal gh pr commands, including quoted --repo/-R values. + // Markdown/prose that is not valid shell still contributes URL references. + if !strings.Contains(text, "gh") { continue } - for _, m := range ghPRRe.FindAllStringSubmatch(text, -1) { - action := "gh pr " + strings.ToLower(m[1]) - if !seenActions[action] { - seenActions[action] = true - result.PRActions = append(result.PRActions, action) - } + file, err := syntax.NewParser().Parse(strings.NewReader(text), "") + if err != nil { + continue } - for _, m := range pullURLRe.FindAllStringSubmatch(text, -1) { - repo := strings.TrimSuffix(m[1], ".git") - if isPlaceholder(repo) { - continue + syntax.Walk(file, func(node syntax.Node) bool { + if _, ok := node.(*syntax.FuncDecl); ok { + return false } - if !seenRepos[repo] { - seenRepos[repo] = true - result.Repos = append(result.Repos, repo) + call, ok := node.(*syntax.CallExpr) + if !ok { + return true } - if num, err := strconv.Atoi(m[2]); err == nil && !seenNumbers[num] { - seenNumbers[num] = true - result.PRNumbers = append(result.PRNumbers, num) + args := make([]string, len(call.Args)) + for i, arg := range call.Args { + args[i], _ = literalShellWord(arg.Parts) } - if url := m[0]; !seenURLs[url] { - seenURLs[url] = true - result.PRURLs = append(result.PRURLs, url) + repos = append(repos, repositoryFromGHArgs(args)...) + return true + }) + } + return repos +} + +func repositoryFromGHArgs(args []string) []string { + if len(args) < 3 || args[0] != "gh" || args[1] != "pr" { + return nil + } + var repos []string + for i := 2; i < len(args); i++ { + arg := args[i] + var value string + switch { + case arg == "--": + return repos + case arg == "--repo" || arg == "-R": + i++ + if i < len(args) { + value = args[i] } + case strings.HasPrefix(arg, "--repo="): + value = strings.TrimPrefix(arg, "--repo=") + case strings.HasPrefix(arg, "-R"): + value = strings.TrimPrefix(arg, "-R") } - for _, m := range repoFlagRe.FindAllStringSubmatch(text, -1) { - if r := normalizeRepoFlag(m[1]); r != "" && !seenRepos[r] { - seenRepos[r] = true - result.Repos = append(result.Repos, r) - } + if repo := normalizeRepoFlag(value); repo != "" { + repos = append(repos, repo) } - for _, m := range repoURLRe.FindAllStringSubmatch(text, -1) { - r := strings.TrimSuffix(m[1], ".git") - if !isPlaceholder(r) && !seenRepos[r] { - seenRepos[r] = true - result.Repos = append(result.Repos, r) - } + } + return repos +} + +func extractGitHubRefs(text string) []string { + var repos []string + for _, match := range repoURLRe.FindAllStringSubmatchIndex(text, -1) { + // Check the terminator without consuming it: adjacent references can + // share a separator. Reject partial matches before shell expansions. + if end := match[3]; end < len(text) && !strings.ContainsRune("/ \t\r\n?#)\"',;`>]}", rune(text[end])) { + continue + } + if repo := normalizeRepoFlag(strings.TrimRight(text[match[2]:match[3]], ".")); repo != "" { + repos = append(repos, repo) } } - return result + return repos } diff --git a/backend/internal/importer/repos_test.go b/backend/internal/importer/repos_test.go new file mode 100644 index 0000000..6221927 --- /dev/null +++ b/backend/internal/importer/repos_test.go @@ -0,0 +1,172 @@ +package importer + +import ( + "encoding/json" + "os" + "os/exec" + "path/filepath" + "reflect" + "testing" +) + +func TestRepositoryReferenceMining(t *testing.T) { + for _, tc := range []struct { + name, text string + want []string + }{ + {"https", "See https://github.com/acme/project.git", []string{"acme/project"}}, + {"pull_url", "(https://github.com/acme/project/pull/42)", []string{"acme/project"}}, + {"query", "https://github.com/acme/project?tab=readme", []string{"acme/project"}}, + {"ssh", "git@github.com:acme/project.git", []string{"acme/project"}}, + {"ssh_scheme", "ssh://git@github.com/acme/project.git", []string{"acme/project"}}, + {"bare_url", "github.com/acme/project", []string{"acme/project"}}, + {"repo_flag", "gh pr view 42 --repo acme/project", []string{"acme/project"}}, + {"quoted_flag", "gh pr view --repo 'acme/project'", []string{"acme/project"}}, + {"equals_flag", "gh pr list --repo=acme/project", []string{"acme/project"}}, + {"short_flag", "gh pr list -Racme/project", []string{"acme/project"}}, + {"short_separate", "gh pr list -R acme/project", []string{"acme/project"}}, + {"deduplicate", "https://github.com/acme/project https://github.com/ACME/PROJECT/pull/2", []string{"acme/project"}}, + {"adjacent_references", "https://github.com/acme/project https://github.com/other/project", []string{"acme/project", "other/project"}}, + {"other_command", "other --repo acme/project", nil}, + {"echoed_command", "echo 'gh pr list --repo acme/project'", nil}, + {"dynamic_flag", "gh pr list --repo \"$ORG/project\"", nil}, + {"dynamic_url", "https://github.com/acme/project-$SUFFIX", nil}, + {"placeholder", "https://github.com/owner/repo gh pr list --repo '/'", nil}, + {"lookalike_host", "https://notgithub.com/acme/project https://github.com.evil.test/acme/project", nil}, + {"url_in_other_path", "https://example.test/github.com/acme/project", nil}, + {"flag_after_separator", "gh pr view -- --repo acme/project", nil}, + } { + t.Run(tc.name, func(t *testing.T) { + var meta RepositoryMetadata + meta.observeText(tc.text) + if !reflect.DeepEqual(meta.CandidateRepos, tc.want) { + t.Fatalf("candidates = %q, want %q", meta.CandidateRepos, tc.want) + } + }) + } +} + +func TestRepositoryResolutionPrecedence(t *testing.T) { + missing := filepath.Join(t.TempDir(), "project") + for _, tc := range []struct { + name string + meta RepositoryMetadata + want string + }{ + {"git_metadata", RepositoryMetadata{GitRepository: "acme/actual", Cwd: missing, CandidateRepos: []string{"other/project"}}, "acme/actual"}, + {"matching_reference", RepositoryMetadata{Cwd: missing, CandidateRepos: []string{"acme/project"}}, "acme/project"}, + {"unrelated_reference", RepositoryMetadata{Cwd: missing, CandidateRepos: []string{"other/dependency"}}, "project"}, + {"ambiguous_owner", RepositoryMetadata{Cwd: missing, CandidateRepos: []string{"acme/project", "other/project"}}, "project"}, + {"unique_without_cwd", RepositoryMetadata{CandidateRepos: []string{"acme/project"}}, "acme/project"}, + {"ambiguous_without_cwd", RepositoryMetadata{CandidateRepos: []string{"acme/project", "other/dependency"}}, ""}, + {"cwd_before_ancestor", RepositoryMetadata{Cwd: filepath.Join(missing, "child"), CandidateRepos: []string{"acme/project", "other/child"}}, "other/child"}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := tc.meta.Repository(); got != tc.want { + t.Fatalf("repository = %q, want %q", got, tc.want) + } + }) + } +} + +func repositoryGit(t *testing.T, args ...string) { + t.Helper() + if output, err := exec.Command("git", args...).CombinedOutput(); err != nil { + t.Fatalf("git %v: %s: %v", args, output, err) + } +} + +func TestRepositoryRemoteAndWorktree(t *testing.T) { + dir := filepath.Join(t.TempDir(), "different-local-name") + repositoryGit(t, "init", dir) + repositoryGit(t, "-C", dir, "remote", "add", "origin", "https://github.com/acme/project.git") + meta := RepositoryMetadata{Cwd: dir, CandidateRepos: []string{"other/dependency"}} + if got := meta.Repository(); got != "acme/project" { + t.Fatalf("remote repository = %q", got) + } + repositoryGit(t, "-C", dir, "-c", "user.name=Test", "-c", "user.email=test@example.invalid", "-c", "commit.gpgsign=false", "commit", "--allow-empty", "-m", "fixture") + worktree := filepath.Join(t.TempDir(), "feature-worktree") + repositoryGit(t, "-C", dir, "worktree", "add", "--detach", worktree) + meta.Cwd = worktree + if got := meta.Repository(); got != "acme/project" { + t.Fatalf("worktree repository = %q", got) + } + meta.GitRepository = "recorded/original" + if got := meta.Repository(); got != "recorded/original" { + t.Fatalf("recorded remote did not win: %q", got) + } +} + +func TestRepositoryWithoutGit(t *testing.T) { + dir := filepath.Join(t.TempDir(), "project") + if err := os.Mkdir(dir, 0700); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", t.TempDir()) + meta := RepositoryMetadata{Cwd: dir, CandidateRepos: []string{"acme/project"}} + if got := meta.Repository(); got != "acme/project" { + t.Fatalf("repository without git = %q", got) + } +} + +func TestRepositoryRetainsRemoteAfterRestart(t *testing.T) { + dir := t.TempDir() + repositoryGit(t, "init", dir) + repositoryGit(t, "-C", dir, "remote", "add", "origin", "https://secret@github.com/acme/project.git") + meta := RepositoryMetadata{Cwd: dir} + meta.Enrich(nil, nil) + data, err := json.Marshal(meta) + if err != nil { + t.Fatal(err) + } + if err := os.RemoveAll(filepath.Join(dir, ".git")); err != nil { + t.Fatal(err) + } + gitRemoteCache.Delete(dir) + var restored RepositoryMetadata + if err := json.Unmarshal(data, &restored); err != nil { + t.Fatal(err) + } + if restored.LocalRepository != "acme/project" || restored.Repository() != "acme/project" { + t.Fatalf("lost previously discovered remote: %+v", restored) + } +} + +func TestRepositoryMetadataScope(t *testing.T) { + for _, tc := range []struct { + name, entry string + want string + }{ + {"claude_command", `{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Bash","input":{"command":"gh pr view -R acme/project"}}]}}`, "acme/project"}, + {"claude_result", `{"type":"user","message":{"content":[{"type":"tool_result","content":[{"type":"text","text":"https://github.com/acme/project/pull/1"}]}]}}`, "acme/project"}, + {"claude_write_excluded", `{"type":"assistant","message":{"content":[{"type":"tool_use","name":"Write","input":{"content":"https://github.com/acme/project"}}]}}`, ""}, + {"codex_command", `{"type":"response_item","payload":{"type":"function_call","name":"exec_command","arguments":"{\"cmd\":\"gh pr view --repo acme/project\"}"}}`, "acme/project"}, + {"codex_argv", `{"type":"response_item","payload":{"type":"function_call","name":"shell","arguments":"{\"command\":[\"gh\",\"pr\",\"view\",\"--repo\",\"acme/project\"]}"}}`, "acme/project"}, + {"codex_shell_argv", `{"type":"response_item","payload":{"type":"function_call","name":"shell","arguments":"{\"command\":[\"bash\",\"-lc\",\"gh pr view --repo acme/project\"]}"}}`, "acme/project"}, + {"codex_echo_argv", `{"type":"response_item","payload":{"type":"function_call","name":"shell","arguments":"{\"command\":[\"echo\",\"gh pr view --repo acme/project\"]}"}}`, ""}, + {"codex_result", `{"type":"response_item","payload":{"type":"function_call_output","output":"https://github.com/acme/project/pull/1"}}`, "acme/project"}, + {"codex_event", `{"type":"event_msg","payload":{"type":"agent_message","message":"https://github.com/acme/project"}}`, "acme/project"}, + {"codex_system_excluded", `{"type":"response_item","payload":{"type":"message","role":"developer","content":[{"type":"input_text","text":"https://github.com/acme/project"}]}}`, ""}, + {"codex_reasoning_excluded", `{"type":"response_item","payload":{"type":"reasoning","summary":[{"type":"summary_text","text":"https://github.com/acme/project"}]}}`, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + var meta RepositoryMetadata + if tc.name[:6] == "claude" { + var entry ClaudeJSONLEntry + if err := json.Unmarshal([]byte(tc.entry), &entry); err != nil { + t.Fatal(err) + } + meta.ObserveClaude(entry) + } else { + var entry CodexJSONLEntry + if err := json.Unmarshal([]byte(tc.entry), &entry); err != nil { + t.Fatal(err) + } + meta.ObserveCodex(entry) + } + if got := meta.Repository(); got != tc.want { + t.Fatalf("repository = %q, want %q", got, tc.want) + } + }) + } +} diff --git a/backend/internal/importer/repository_metadata.go b/backend/internal/importer/repository_metadata.go new file mode 100644 index 0000000..7b36d47 --- /dev/null +++ b/backend/internal/importer/repository_metadata.go @@ -0,0 +1,218 @@ +package importer + +import ( + "encoding/json" + "path/filepath" + "strings" + "time" + + "github.com/tobilg/ai-observer/internal/api" +) + +// RepositoryMetadata is shared by full imports and incremental parsers. Watchers +// persist it alongside their offsets, so later batches retain session context. +// GitRepository stores the normalized identifier, never URL credentials. +type RepositoryMetadata struct { + Cwd string `json:"cwd,omitempty"` + OriginalCwd string `json:"originalCwd,omitempty"` + GitBranch string `json:"gitBranch,omitempty"` + GitRepository string `json:"gitRepository,omitempty"` + LocalRepository string `json:"localRepository,omitempty"` + LocalPath string `json:"localPath,omitempty"` + CandidateRepos []string `json:"candidateRepos,omitempty"` +} + +func (m *RepositoryMetadata) addCandidates(repos ...string) { + for _, repo := range repos { + if repo == "" { + continue + } + seen := false + for _, existing := range m.CandidateRepos { + if strings.EqualFold(existing, repo) { + seen = true + break + } + } + if !seen { + m.CandidateRepos = append(m.CandidateRepos, repo) + } + } +} + +func (m *RepositoryMetadata) observeText(texts ...string) { + m.addCandidates(extractRepositoryRefs(texts...)...) +} + +// ObserveClaude collects attribution evidence without emitting any telemetry. +func (m *RepositoryMetadata) ObserveClaude(entry ClaudeJSONLEntry) { + if m.Cwd == "" { + m.Cwd = entry.Cwd + } + if m.GitBranch == "" { + m.GitBranch = entry.GitBranch + } + if entry.Type == "worktree-state" && entry.WorktreeSession != nil && m.OriginalCwd == "" { + m.OriginalCwd = entry.WorktreeSession.OriginalCwd + } + if entry.Type == "pr-link" && entry.PRNumber > 0 { + if _, err := time.Parse(time.RFC3339Nano, entry.Timestamp); err == nil { + repo := strings.TrimSpace(entry.PRRepository) + if repo == "" { + repo = githubRepoFromURL(entry.PRUrl) + } + m.addCandidates(repo) + } + } + if entry.Message == nil || (entry.Type != "user" && entry.Type != "assistant") { + return + } + for _, content := range entry.Message.Content { + switch content.Type { + case "text": + m.observeText(content.Text) + case "tool_result": + m.observeText(content.Content) + case "tool_use": + if content.Name == "Bash" { + data, err := json.Marshal(content.Input) + if err == nil { + m.observeCommandArguments(data) + } + } + } + } +} + +// ObserveCodex accepts metadata, visible messages/results, and shell tool inputs. +// Arbitrary tool arguments and reasoning summaries are not attribution evidence. +func (m *RepositoryMetadata) ObserveCodex(entry CodexJSONLEntry) { + switch entry.Type { + case "session_meta": + var meta CodexSessionMeta + if json.Unmarshal(entry.Payload, &meta) != nil { + return + } + if m.Cwd == "" { + m.Cwd = meta.Cwd + } + if meta.Git != nil { + if m.GitRepository == "" { + m.GitRepository = normalizeGitURL(meta.Git.RepositoryURL) + } + if m.GitBranch == "" { + m.GitBranch = meta.Git.Branch + } + } + case "response_item": + var item CodexResponseItem + if json.Unmarshal(entry.Payload, &item) != nil { + return + } + switch item.Type { + case "message": + if item.Role != "user" && item.Role != "assistant" { + return + } + for _, content := range item.Content { + if content.Type == "input_text" || content.Type == "output_text" { + m.observeText(content.Text) + } + } + case "function_call": + switch item.Name { + case "exec_command", "shell_command", "shell": + m.observeCommandArguments([]byte(item.Arguments)) + } + case "function_call_output": + var output string + if json.Unmarshal(item.Output, &output) == nil { + m.observeText(output) + } + } + case "event_msg": + var event struct{ Type, Message string } + if json.Unmarshal(entry.Payload, &event) == nil && (event.Type == "user_message" || event.Type == "agent_message") { + m.observeText(event.Message) + } + } +} + +func (m *RepositoryMetadata) observeCommandArguments(data []byte) { + var args struct { + Command json.RawMessage `json:"command"` + Cmd string `json:"cmd"` + } + if json.Unmarshal(data, &args) != nil { + return + } + m.observeText(args.Cmd) + var command string + if json.Unmarshal(args.Command, &command) == nil { + m.observeText(command) + return + } + var argv []string + if json.Unmarshal(args.Command, &argv) == nil { + for _, arg := range argv { + m.addCandidates(extractGitHubRefs(arg)...) + } + m.addCandidates(repositoryFromGHArgs(argv)...) + if len(argv) >= 3 { + switch filepath.Base(argv[0]) { + case "bash", "sh", "zsh": + if strings.HasPrefix(argv[1], "-") && !strings.HasPrefix(argv[1], "--") && strings.Contains(argv[1], "c") { + m.observeText(argv[2]) + } + } + } + } +} + +// Repository returns the best identifier supported by the evidence seen so far. +func (m *RepositoryMetadata) Repository() string { + if m.GitRepository != "" { + return m.GitRepository + } + path := m.OriginalCwd + if path == "" { + path = m.Cwd + } + if path != m.LocalPath { + m.LocalPath, m.LocalRepository = path, "" + } + if m.LocalRepository == "" { + m.LocalRepository = cwdGitRemote(path) + } + // Persist a discovered remote so a removed worktree or a restart without + // Git does not discard evidence that was available in an earlier batch. + return resolveSessionRepository(m.CandidateRepos, m.LocalRepository, m.OriginalCwd, m.Cwd) +} + +// Enrich applies session attribution to a batch of session logs and usage +// metrics. Do not use it on activity metrics attributed to a different repo, +// such as structured PR-link events. +func (m *RepositoryMetadata) Enrich(metrics []api.MetricDataPoint, logs []api.LogRecord) { + repository := m.Repository() + apply := func(attrs map[string]string) map[string]string { + if repository == "" && m.GitBranch == "" { + return attrs + } + if attrs == nil { + attrs = make(map[string]string) + } + if repository != "" { + attrs["repository"] = repository + } + if m.GitBranch != "" { + attrs["git_branch"] = m.GitBranch + } + return attrs + } + for i := range metrics { + metrics[i].Attributes = apply(metrics[i].Attributes) + } + for i := range logs { + logs[i].LogAttributes = apply(logs[i].LogAttributes) + } +} diff --git a/backend/internal/watcher/claude_incremental.go b/backend/internal/watcher/claude_incremental.go index f7add43..20e7df9 100644 --- a/backend/internal/watcher/claude_incremental.go +++ b/backend/internal/watcher/claude_incremental.go @@ -18,8 +18,9 @@ import ( ) type claudeParserState struct { - MessageIndex int `json:"messageIndex"` - SeenRequests map[string]bool `json:"seenRequests"` + Repository *importer.RepositoryMetadata `json:"repository,omitempty"` + MessageIndex int `json:"messageIndex"` + SeenRequests map[string]bool `json:"seenRequests"` } type claudeIncrementalParser struct { @@ -58,6 +59,23 @@ func (p *claudeIncrementalParser) ParseIncremental(ctx context.Context, filePath parserState.MessageIndex = workingState.MessageCount } + if parserState.Repository == nil { + parserState.Repository = &importer.RepositoryMetadata{} + if err := replayRepositoryPrefix(ctx, file, workingState.ByteOffset, func(line []byte) { + var entry importer.ClaudeJSONLEntry + if json.Unmarshal(line, &entry) != nil { + return + } + parserState.Repository.ObserveClaude(entry) + // Empty legacy state also lacks the full import's usage dedup keys. + if entry.Type == "assistant" && entry.Message != nil && entry.Message.Usage != nil && entry.Message.ID != "" && entry.RequestID != "" { + parserState.SeenRequests[entry.Message.ID+":"+entry.RequestID] = true + } + }); err != nil { + return nil, err + } + } + result := &IncrementalResult{} reader := bufio.NewReaderSize(file, 1024*1024) committedBytes := int64(0) @@ -107,6 +125,8 @@ func (p *claudeIncrementalParser) ParseIncremental(ctx context.Context, filePath continue } + parserState.Repository.ObserveClaude(entry) + if entry.Message == nil { commitLine() if readErr == io.EOF { @@ -214,6 +234,8 @@ func (p *claudeIncrementalParser) ParseIncremental(ctx context.Context, filePath } } + parserState.Repository.Enrich(result.Metrics, result.Logs) + // Update state workingState.ByteOffset += committedBytes workingState.MessageCount = parserState.MessageIndex diff --git a/backend/internal/watcher/codex_incremental.go b/backend/internal/watcher/codex_incremental.go index 6c03693..0b23851 100644 --- a/backend/internal/watcher/codex_incremental.go +++ b/backend/internal/watcher/codex_incremental.go @@ -18,10 +18,11 @@ import ( ) type codexParserState struct { - SessionID string `json:"sessionId"` - CurrentModel string `json:"currentModel"` - MessageIndex int `json:"messageIndex"` - LastTokenCount *importer.CodexTokenCount `json:"lastTokenCount"` + Repository *importer.RepositoryMetadata `json:"repository,omitempty"` + SessionID string `json:"sessionId"` + CurrentModel string `json:"currentModel"` + MessageIndex int `json:"messageIndex"` + LastTokenCount *importer.CodexTokenCount `json:"lastTokenCount"` } type codexIncrementalParser struct{} @@ -53,6 +54,47 @@ func (p *codexIncrementalParser) ParseIncremental(ctx context.Context, filePath parserState.MessageIndex = workingState.MessageCount } + if parserState.Repository == nil { + parserState.Repository = &importer.RepositoryMetadata{} + recoverTokens := parserState.LastTokenCount == nil + recoverModel := parserState.CurrentModel == "" + recoverSession := parserState.SessionID == "" + if err := replayRepositoryPrefix(ctx, file, workingState.ByteOffset, func(line []byte) { + var entry importer.CodexJSONLEntry + if json.Unmarshal(line, &entry) != nil { + return + } + if _, err := time.Parse(time.RFC3339Nano, entry.Timestamp); err != nil { + return + } + parserState.Repository.ObserveCodex(entry) + switch entry.Type { + case "session_meta": + var meta importer.CodexSessionMeta + if json.Unmarshal(entry.Payload, &meta) == nil { + if recoverSession && meta.ID != "" { + parserState.SessionID = meta.ID + } + if recoverModel && meta.Model != "" { + parserState.CurrentModel = meta.Model + } + } + case "turn_context": + var turn struct{ Model string } + if recoverModel && json.Unmarshal(entry.Payload, &turn) == nil && turn.Model != "" { + parserState.CurrentModel = turn.Model + } + case "event_msg": + var event importer.CodexEventMsg + if recoverTokens && json.Unmarshal(entry.Payload, &event) == nil && event.Type == "token_count" && event.Info != nil && event.Info.TotalTokenUsage != nil { + parserState.LastTokenCount = event.Info.TotalTokenUsage + } + } + }); err != nil { + return nil, err + } + } + // Use filename as fallback session ID if parserState.SessionID == "" { parserState.SessionID = strings.TrimSuffix(filepath.Base(filePath), ".jsonl") @@ -115,6 +157,8 @@ readLoop: } } + parserState.Repository.ObserveCodex(entry) + switch entry.Type { case "session_meta": var meta importer.CodexSessionMeta @@ -417,6 +461,8 @@ readLoop: } } + parserState.Repository.Enrich(result.Metrics, result.Logs) + // Update state workingState.ByteOffset += committedBytes workingState.MessageCount = parserState.MessageIndex diff --git a/backend/internal/watcher/repository.go b/backend/internal/watcher/repository.go new file mode 100644 index 0000000..b7cfae2 --- /dev/null +++ b/backend/internal/watcher/repository.go @@ -0,0 +1,31 @@ +package watcher + +import ( + "bufio" + "context" + "fmt" + "io" + "os" +) + +// replayRepositoryPrefix recovers context from old watcher state or a file +// skipped without backfill or after full import. It reads only the committed prefix and +// emits no records. SectionReader leaves the incremental file position intact. +func replayRepositoryPrefix(ctx context.Context, file *os.File, offset int64, observe func([]byte)) error { + reader := bufio.NewReader(io.NewSectionReader(file, 0, offset)) + for { + if err := ctx.Err(); err != nil { + return err + } + line, err := reader.ReadBytes('\n') + if err != nil && err != io.EOF { + return fmt.Errorf("recovering repository context: %w", err) + } + if len(line) > 0 { + observe(line) + } + if err == io.EOF { + return nil + } + } +} diff --git a/backend/internal/watcher/repository_test.go b/backend/internal/watcher/repository_test.go new file mode 100644 index 0000000..b7da263 --- /dev/null +++ b/backend/internal/watcher/repository_test.go @@ -0,0 +1,253 @@ +package watcher + +import ( + "context" + "encoding/json" + "os" + "reflect" + "strings" + "testing" + + "github.com/tobilg/ai-observer/internal/api" + "github.com/tobilg/ai-observer/internal/importer" + "github.com/tobilg/ai-observer/internal/storage" +) + +const repositoryCodexMeta = `{"timestamp":"2026-09-01T12:00:00Z","type":"session_meta","payload":{"id":"s1","cwd":"/missing/project","model":"gpt-4o","git":{"repository_url":"https://credential@github.com/acme/project.git","branch":"main"}}}` +const repositoryCodexTokens = `{"timestamp":"2026-09-01T12:00:01Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":100,"output_tokens":50}}}}` +const repositoryCodexNextTokens = `{"timestamp":"2026-09-01T12:00:02Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":150,"output_tokens":60}}}}` +const repositoryClaudeMeta = `{"type":"user","timestamp":"2026-09-01T12:00:00Z","sessionId":"s1","cwd":"/missing/project","gitBranch":"main","message":{"role":"user","content":[{"type":"text","text":"https://github.com/acme/project/pull/7"}]}}` +const repositoryClaudeTokens = `{"type":"assistant","timestamp":"2026-09-01T12:00:01Z","sessionId":"s1","requestId":"r1","message":{"id":"m1","role":"assistant","model":"claude-sonnet-4-20250514","content":[{"type":"text","text":"hello"}],"usage":{"input_tokens":100,"output_tokens":50}}}` +const repositoryClaudeNextTokens = `{"type":"assistant","timestamp":"2026-09-01T12:00:02Z","sessionId":"s1","requestId":"r2","message":{"id":"m2","role":"assistant","model":"claude-sonnet-4-20250514","content":[{"type":"text","text":"next"}],"usage":{"input_tokens":50,"output_tokens":10}}}` + +type repositoryParserCase struct { + name, prefix, suffix, tokenName string + full func() importer.SessionParser + watch func() IncrementalParser +} + +func repositoryParsers() []repositoryParserCase { + return []repositoryParserCase{ + {"Claude", repositoryClaudeMeta + "\n" + repositoryClaudeTokens + "\n", repositoryClaudeNextTokens + "\n", importer.ClaudeTokenUsageMetric, func() importer.SessionParser { return importer.NewClaudeParser() }, func() IncrementalParser { return &claudeIncrementalParser{} }}, + {"Codex", repositoryCodexMeta + "\n" + repositoryCodexTokens + "\n", repositoryCodexNextTokens + "\n", "codex_cli_rs.token.usage", func() importer.SessionParser { return importer.NewCodexParser() }, func() IncrementalParser { return &codexIncrementalParser{} }}, + } +} + +func assertRepositoryBatch(t *testing.T, metrics []api.MetricDataPoint, logs []api.LogRecord, repo string) { + t.Helper() + for _, metric := range metrics { + if got := metric.Attributes["repository"]; got != repo { + t.Errorf("%s repository = %q, want %q", metric.MetricName, got, repo) + } + if got := metric.Attributes["git_branch"]; got != "main" { + t.Errorf("%s branch = %q", metric.MetricName, got) + } + } + for _, log := range logs { + if got := log.LogAttributes["repository"]; got != repo { + t.Errorf("%s repository = %q, want %q", log.Body, got, repo) + } + if got := log.LogAttributes["git_branch"]; got != "main" { + t.Errorf("%s branch = %q", log.Body, got) + } + } +} + +func assertRepositoryTokenValue(t *testing.T, metrics []api.MetricDataPoint, name string, want float64) { + t.Helper() + var total float64 + for _, metric := range metrics { + if metric.MetricName == name && metric.Attributes["type"] == "input" && metric.Value != nil { + total += *metric.Value + } + } + if total != want { + t.Fatalf("input token total = %v, want %v", total, want) + } +} + +func appendRepositoryFile(t *testing.T, path, content string) { + t.Helper() + f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0600) + if err != nil { + t.Fatal(err) + } + defer f.Close() + if _, err := f.WriteString(content); err != nil { + t.Fatal(err) + } +} + +func TestRepositoryFullAndWatchParity(t *testing.T) { + for _, tc := range repositoryParsers() { + t.Run(tc.name, func(t *testing.T) { + path := writeWatcherTestFile(t, tc.prefix) + full, err := tc.full().ParseFile(context.Background(), path) + if err != nil { + t.Fatal(err) + } + state := &storage.ImportState{FilePath: path} + watched, err := tc.watch().ParseIncremental(context.Background(), path, state) + if err != nil { + t.Fatal(err) + } + assertRepositoryBatch(t, full.Metrics, full.Logs, "acme/project") + assertRepositoryBatch(t, watched.Metrics, watched.Logs, "acme/project") + assertRepositoryTokenValue(t, full.Metrics, tc.tokenName, 100) + assertRepositoryTokenValue(t, watched.Metrics, tc.tokenName, 100) + if len(full.Logs) != len(watched.Logs) { + t.Fatalf("full/watch log counts differ: %d/%d", len(full.Logs), len(watched.Logs)) + } + for _, metric := range full.Metrics { + if metric.MetricName == "claude_code.pull_request.count" || metric.MetricName == "claude_code.commit.count" { + t.Errorf("text reference created activity: %s", metric.MetricName) + } + } + if strings.Contains(state.ParserState, "credential") { + t.Fatal("repository state persisted URL credentials") + } + }) + } +} + +func TestRepositoryPersistsAcrossRestart(t *testing.T) { + for _, tc := range repositoryParsers() { + t.Run(tc.name, func(t *testing.T) { + path := writeWatcherTestFile(t, tc.prefix) + state := &storage.ImportState{FilePath: path} + if _, err := tc.watch().ParseIncremental(context.Background(), path, state); err != nil { + t.Fatal(err) + } + // Round-trip the persisted state and instantiate a new parser. + data, err := json.Marshal(state) + if err != nil { + t.Fatal(err) + } + var restored storage.ImportState + if err := json.Unmarshal(data, &restored); err != nil { + t.Fatal(err) + } + appendRepositoryFile(t, path, tc.suffix) + result, err := tc.watch().ParseIncremental(context.Background(), path, &restored) + if err != nil { + t.Fatal(err) + } + assertRepositoryBatch(t, result.Metrics, result.Logs, "acme/project") + assertRepositoryTokenValue(t, result.Metrics, tc.tokenName, 50) + if restored.ByteOffset != int64(len(tc.prefix+tc.suffix)) { + t.Fatalf("offset = %d", restored.ByteOffset) + } + empty, err := tc.watch().ParseIncremental(context.Background(), path, &restored) + if err != nil { + t.Fatal(err) + } + if len(empty.Logs)+len(empty.Metrics) != 0 { + t.Fatal("replayed data with no append") + } + }) + } +} + +func TestRepositoryRecoversLegacyState(t *testing.T) { + for _, tc := range repositoryParsers() { + for _, oldState := range []string{"", `{"messageIndex":9,"seenRequests":{"m1:r1":true},"sessionId":"s1","currentModel":"gpt-4o","lastTokenCount":{"input_tokens":100,"output_tokens":50}}`} { + t.Run(tc.name+"/"+oldState, func(t *testing.T) { + path := writeWatcherTestFile(t, tc.prefix+tc.suffix) + state := &storage.ImportState{FilePath: path, ByteOffset: int64(len(tc.prefix)), MessageCount: 9, ParserState: oldState} + result, err := tc.watch().ParseIncremental(context.Background(), path, state) + if err != nil { + t.Fatal(err) + } + assertRepositoryBatch(t, result.Metrics, result.Logs, "acme/project") + assertRepositoryTokenValue(t, result.Metrics, tc.tokenName, 50) + for _, log := range result.Logs { + if log.Body == "hello" || log.Body == "conversation_starts" { + t.Errorf("re-emitted prefix log: %s", log.Body) + } + if log.LogAttributes["event.name"] == "transcript.message" && log.LogAttributes["message.index"] != "9" { + t.Errorf("message index was reset: %v", log.LogAttributes) + } + } + if !strings.Contains(state.ParserState, `"repository"`) { + t.Fatal("recovered context not persisted") + } + }) + } + } +} + +func TestRepositoryRecoveryCancellationPreservesState(t *testing.T) { + for _, tc := range repositoryParsers() { + t.Run(tc.name, func(t *testing.T) { + path := writeWatcherTestFile(t, tc.prefix+tc.suffix) + state := storage.ImportState{FilePath: path, ByteOffset: int64(len(tc.prefix)), MessageCount: 9} + before := state + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := tc.watch().ParseIncremental(ctx, path, &state); err != context.Canceled { + t.Fatalf("error = %v", err) + } + if !reflect.DeepEqual(before, state) { + t.Fatal("cancellation changed persisted state") + } + }) + } +} + +func TestCodexAllLogsReceiveRepository(t *testing.T) { + lines := []string{ + repositoryCodexMeta, + `{"timestamp":"2026-09-01T12:00:01Z","type":"event_msg","payload":{"type":"user_message"}}`, + `{"timestamp":"2026-09-01T12:00:02Z","type":"event_msg","payload":{"type":"agent_message"}}`, + `{"timestamp":"2026-09-01T12:00:03Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"message"}]}}`, + `{"timestamp":"2026-09-01T12:00:04Z","type":"response_item","payload":{"type":"function_call","name":"exec_command","arguments":"{\"cmd\":\"echo hello\"}","call_id":"c1"}}`, + `{"timestamp":"2026-09-01T12:00:05Z","type":"response_item","payload":{"type":"function_call_output","output":"hello","call_id":"c1"}}`, + `{"timestamp":"2026-09-01T12:00:06Z","type":"response_item","payload":{"type":"reasoning","summary":[{"type":"summary_text","text":"reasoning"}]}}`, + repositoryCodexTokens, + } + path := writeWatcherTestFile(t, strings.Join(lines, "\n")+"\n") + full, err := importer.NewCodexParser().ParseFile(context.Background(), path) + if err != nil { + t.Fatal(err) + } + watched, err := (&codexIncrementalParser{}).ParseIncremental(context.Background(), path, &storage.ImportState{}) + if err != nil { + t.Fatal(err) + } + if len(full.Logs) != 7 || len(watched.Logs) != 7 { + t.Fatalf("log counts = %d/%d, want 7/7", len(full.Logs), len(watched.Logs)) + } + assertRepositoryBatch(t, full.Metrics, full.Logs, "acme/project") + assertRepositoryBatch(t, watched.Metrics, watched.Logs, "acme/project") +} + +func TestRepositoryPartialMetadataAndLateReference(t *testing.T) { + // A late reference enriches the current batch, without changing records + // returned by an earlier call. Incomplete JSON contributes no evidence. + meta := strings.Replace(repositoryCodexMeta, `"git":{"repository_url":"https://credential@github.com/acme/project.git","branch":"main"}`, `"git":{"branch":"main"}`, 1) + path := writeWatcherTestFile(t, meta+"\n"+repositoryCodexTokens+"\n") + state := &storage.ImportState{} + first, err := (&codexIncrementalParser{}).ParseIncremental(context.Background(), path, state) + if err != nil { + t.Fatal(err) + } + assertRepositoryBatch(t, first.Metrics, first.Logs, "project") + reference := `{"timestamp":"2026-09-01T12:00:03Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"https://github.com/acme/project"}]}}` + appendRepositoryFile(t, path, reference[:len(reference)-4]) + offset := state.ByteOffset + partial, err := (&codexIncrementalParser{}).ParseIncremental(context.Background(), path, state) + if err != nil { + t.Fatal(err) + } + if state.ByteOffset != offset || len(partial.Logs)+len(partial.Metrics) != 0 { + t.Fatal("incomplete metadata was committed") + } + appendRepositoryFile(t, path, reference[len(reference)-4:]+"\n"+repositoryCodexNextTokens+"\n") + last, err := (&codexIncrementalParser{}).ParseIncremental(context.Background(), path, state) + if err != nil { + t.Fatal(err) + } + assertRepositoryBatch(t, last.Metrics, last.Logs, "acme/project") + assertRepositoryBatch(t, first.Metrics, first.Logs, "project") + assertRepositoryTokenValue(t, last.Metrics, "codex_cli_rs.token.usage", 50) +} diff --git a/docs/import.md b/docs/import.md index ecef8cd..e9c0a99 100644 --- a/docs/import.md +++ b/docs/import.md @@ -43,10 +43,27 @@ Full `import claude-code` reconstructs selected activity metrics from JSONL reco Tool activity without a matching successful result is omitted. LOC is also omitted for overwrites without original contents or a patch, `replace_all` without a patch, and unusually large fallback diffs (over four million line comparisons after shared edges are removed). Commit aliases, wrappers, dynamically constructed commands, quiet commits, and unrecognized/localized completion output may be omitted. These omissions favor reporting supported activity over inventing counts from incomplete records. -Repository attribution uses the working tree's Git remote when available, then matching PR repository candidates, then the directory name. A linked PR in another repository keeps that repository on its own metric and does not override session usage attribution. Codex imports prefer the Git URL recorded in session metadata. On another machine or without Git installed, attribution may fall back to a directory name. - Imported metrics carry `import_source=local_jsonl`. Avoid summing overlapping imports and live telemetry as if they were independent activity. The normal import state skips unchanged files; forcing an import does not deduplicate against existing metrics. Use the existing `--purge` workflow when replacing previously imported data. +## Repository attribution in import and watch + +Both full imports and incremental `watch` attach `repository` to Claude Code and Codex token/cost metrics and logs when the session provides enough evidence. This includes Codex session-start, user/agent event, transcript, tool-call, tool-result, and reasoning logs. The first recorded Git branch is attached as `git_branch` when available. + +Repository selection uses this order: + +1. Codex's recorded `session_meta.git.repository_url`. +2. The local working tree's `origin` remote, including Git worktrees and Claude's recorded original working directory. +3. An unambiguous repository candidate whose name matches the working directory. An exact directory-name match takes precedence over a matching ancestor path segment. +4. The working directory's name. If no working directory is known, a single distinct candidate can identify the repository; multiple candidates leave it unset. + +Candidates come from structured Claude `pr-link` entries and references in user/assistant message text, text tool results, and supported shell tool inputs (`Bash`, `exec_command`, `shell_command`, and `shell`). Mining recognizes GitHub repository/PR URLs and literal `gh pr ... --repo OWNER/REPO` or `-R` arguments, including quoted values and `--repo=OWNER/REPO`. Codex string tool outputs and user/agent event text are included. System/developer messages, reasoning summaries, arbitrary tool arguments, and structured Codex output objects are excluded. Markdown/prose can supply URLs; command flags are extracted only when the text parses as a shell command. + +Mined examples such as `owner/repo`, lookalike GitHub hosts, and dynamically constructed repository names are ignored. References are fallback evidence: they cannot replace a recorded Git URL or local remote. Conflicting owners for a matching repository name fall back to the directory name. Text references never create PR or commit activity metrics. A structured PR in another repository keeps that repository on its own activity metric without overriding session attribution. + +Full import can use evidence anywhere in a file. Watch mode uses evidence available through the end of each parsed batch and saves that context, including a discovered local remote, across restarts. A reference arriving later can improve subsequent batches; previously stored records are not rewritten. When upgrading older watch state or starting after a full import/without backfill, the watcher reads the already-skipped prefix once to recover context and missing token baselines without emitting its old records. Existing database records are not automatically backfilled with new attributes. + +On another machine or without Git installed, attribution may fall back to matching references or the directory name. Repository identifiers contain the normalized owner/name, not credentials from Git URLs. The activity-metric differences between full import and watch described above still apply. + ## Options | Option | Description |