Skip to content

Propagate cancellation context through workflow job detail/status GitHub API calls #46963

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
pelikhan merged 6 commits into from
Jul 21, 2026
Merged

Propagate cancellation context through workflow job detail/status GitHub API calls #46963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d4ade77
Initial plan
Copilot Jul 21, 2026
3ba4f9c
chore: outline plan for cancellable job details fetch
Copilot Jul 21, 2026
82708a5
fix: pass context through fetchJobDetailsWithCounts
Copilot Jul 21, 2026
c0d9544
fix: propagate context through job detail/status fetch wrappers
Copilot Jul 21, 2026
d2561d5
chore: initial plan - resolve merge conflicts
Copilot Jul 21, 2026
12247f5
fix: resolve merge conflict in audit.go - preserve doc comments and c…
Copilot Jul 21, 2026
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
14 changes: 7 additions & 7 deletions pkg/cli/audit.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -377,7 +377,7 @@ func AuditWorkflowRun(ctx context.Context, runID int64, opts AuditOptions) error
if err != nil {
return err
}
results := collectAuditAnalysisResults(run, cfg.outputDir, cfg.verbose, artifactMatchesFilter(constants.AgentArtifactName, cfg.artifactFilter))
results := collectAuditAnalysisResults(ctx, run, cfg.outputDir, cfg.verbose, artifactMatchesFilter(constants.AgentArtifactName, cfg.artifactFilter))
run = applyAuditMetrics(run, results)
processedRun := buildProcessedAuditRun(run, results)
saveAuditRunSummary(cfg.outputDir, run, processedRun, results, cfg.verbose)
Expand Down Expand Up @@ -660,10 +660,10 @@ func prepareRunForAnalysis(run WorkflowRun, cfg auditRunConfig, useLocalCache bo
return run
}

func collectAuditAnalysisResults(run WorkflowRun, runOutputDir string, verbose bool, hasFirewallArtifact bool) auditAnalysisResults {
func collectAuditAnalysisResults(ctx context.Context, run WorkflowRun, runOutputDir string, verbose bool, hasFirewallArtifact bool) auditAnalysisResults {
results := auditAnalysisResults{}
var wg sync.WaitGroup
launchCoreAuditAnalyses(&wg, &results, run, runOutputDir, verbose)
launchCoreAuditAnalyses(ctx, &wg, &results, run, runOutputDir, verbose)
if hasFirewallArtifact {
launchFirewallAuditAnalyses(&wg, &results, runOutputDir, verbose)
}
Expand All @@ -673,13 +673,13 @@ func collectAuditAnalysisResults(run WorkflowRun, runOutputDir string, verbose b
}

// launchCoreAuditAnalyses exclusively writes missingTools, missingData, noops, mcpFailures, and accessAnalysis.
func launchCoreAuditAnalyses(wg *sync.WaitGroup, results *auditAnalysisResults, run WorkflowRun, runOutputDir string, verbose bool) {
func launchCoreAuditAnalyses(ctx context.Context, wg *sync.WaitGroup, results *auditAnalysisResults, run WorkflowRun, runOutputDir string, verbose bool) {
// Resolve experiment assignment once so all goroutines reuse the same values
// rather than each reading state.json independently.
expName, expVariant, _ := firstExperimentAssignment(extractExperimentData(runOutputDir))

launchMetricsAnalysis(wg, results, runOutputDir, verbose, run.WorkflowPath)
launchJobDetailsAnalysis(wg, results, run.DatabaseID, verbose)
launchJobDetailsAnalysis(ctx, wg, results, run.DatabaseID, verbose)
runAuditAnalysis(wg, verbose, "extractMissingToolsFromRun", "Failed to extract missing tools", func(v []MissingToolReport) {
results.missingTools = v
}, func() ([]MissingToolReport, error) {
Expand Down Expand Up @@ -723,9 +723,9 @@ func launchMetricsAnalysis(wg *sync.WaitGroup, results *auditAnalysisResults, ru
}

// launchJobDetailsAnalysis exclusively writes results.jobDetails and results.failedJobCount.
func launchJobDetailsAnalysis(wg *sync.WaitGroup, results *auditAnalysisResults, runID int64, verbose bool) {
func launchJobDetailsAnalysis(ctx context.Context, wg *sync.WaitGroup, results *auditAnalysisResults, runID int64, verbose bool) {
wg.Go(func() {
jobDetails, failedJobCount, err := fetchJobDetailsWithCounts(runID, verbose)
jobDetails, failedJobCount, err := fetchJobDetailsWithCounts(ctx, runID, verbose)
if err != nil {
auditLog.Printf("fetchJobDetailsWithCounts failed: %v", err)
if verbose {
Expand Down
12 changes: 6 additions & 6 deletions pkg/cli/logs_github_api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,13 +76,13 @@ func buildCreatedFilter(startDate, endDate, beforeDate string) string {
// call and returns the full detail slice together with the count of failed jobs.
// It is the single source of truth for the jobs endpoint; fetchJobDetails and
// fetchJobStatuses are thin wrappers that each return only the value they need.
func fetchJobDetailsWithCounts(runID int64, verbose bool) ([]JobInfoWithDuration, int, error) {
func fetchJobDetailsWithCounts(ctx context.Context, runID int64, verbose bool) ([]JobInfoWithDuration, int, error) {
logsGitHubAPILog.Printf("Fetching job details: runID=%d", runID)
if verbose {
fmt.Fprintln(os.Stderr, console.FormatVerboseMessage(fmt.Sprintf("Fetching job details for run %d", runID)))
}

output, err := workflow.RunGHCombined("Fetching job details...", "api",
output, err := workflow.RunGHCombinedContext(ctx, "Fetching job details...", "api",
fmt.Sprintf("repos/{owner}/{repo}/actions/runs/%d/jobs", runID),
"--jq", ".jobs[] | {name: .name, status: .status, conclusion: (.conclusion // \"\"), started_at: .started_at, completed_at: .completed_at, steps: ((.steps // []) | map({name: .name, status: .status, conclusion: (.conclusion // \"\")}))}")
if err != nil {
Expand Down Expand Up @@ -130,8 +130,8 @@ func fetchJobDetailsWithCounts(runID int64, verbose bool) ([]JobInfoWithDuration
// fetchJobDetails gets detailed job information including durations for a workflow run.
// Errors from the underlying API call are suppressed so that callers can continue
// processing even when job data is unavailable (e.g. missing permissions).
func fetchJobDetails(runID int64, verbose bool) ([]JobInfoWithDuration, error) {
jobs, _, err := fetchJobDetailsWithCounts(runID, verbose)
func fetchJobDetails(ctx context.Context, runID int64, verbose bool) ([]JobInfoWithDuration, error) {
jobs, _, err := fetchJobDetailsWithCounts(ctx, runID, verbose)
if err != nil {
// Don't fail the entire operation if we can't get job info
return nil, nil
Expand All @@ -142,8 +142,8 @@ func fetchJobDetails(runID int64, verbose bool) ([]JobInfoWithDuration, error) {
// fetchJobStatuses gets the count of failed jobs for a workflow run.
// Errors from the underlying API call are suppressed so that callers can continue
// processing even when job data is unavailable (e.g. missing permissions).
func fetchJobStatuses(runID int64, verbose bool) (int, error) {
_, failedJobs, err := fetchJobDetailsWithCounts(runID, verbose)
func fetchJobStatuses(ctx context.Context, runID int64, verbose bool) (int, error) {
_, failedJobs, err := fetchJobDetailsWithCounts(ctx, runID, verbose)
if err != nil {
// Don't fail the entire operation if we can't get job info
return 0, nil
Expand Down
5 changes: 3 additions & 2 deletions pkg/cli/logs_github_api_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
package cli

import (
"context"
"encoding/json"
"os"
"path/filepath"
Expand Down Expand Up @@ -239,7 +240,7 @@ func TestFetchJobDetailsWithCountsIncludesSteps(t *testing.T) {

t.Setenv("PATH", fakeBinDir+string(os.PathListSeparator)+os.Getenv("PATH"))

jobs, failedJobs, err := fetchJobDetailsWithCounts(28307653871, false)
jobs, failedJobs, err := fetchJobDetailsWithCounts(context.Background(), 28307653871, false)
require.NoError(t, err)
require.Len(t, jobs, 1)
assert.Equal(t, 1, failedJobs, "failed job count should include failed jobs")
Expand Down Expand Up @@ -270,7 +271,7 @@ func TestFetchJobDetailsWithCountsNullConclusion(t *testing.T) {

t.Setenv("PATH", fakeBinDir+string(os.PathListSeparator)+os.Getenv("PATH"))

jobs, failedJobs, err := fetchJobDetailsWithCounts(28307653871, false)
jobs, failedJobs, err := fetchJobDetailsWithCounts(context.Background(), 28307653871, false)
require.NoError(t, err)
require.Len(t, jobs, 1, "in-progress jobs with null conclusion should not be dropped")
assert.Equal(t, 0, failedJobs, "in-progress job should not count as failed")
Expand Down
2 changes: 1 addition & 1 deletion pkg/cli/logs_orchestrator_filters.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ func buildProcessedRun(result DownloadResult, verbose, logFailedJobs bool) Proce
}

// Add failed jobs to error count.
if failedJobCount, err := fetchJobStatusesForProcessedRun(run.DatabaseID, verbose); err == nil {
if failedJobCount, err := fetchJobStatusesForProcessedRun(context.Background(), run.DatabaseID, verbose); err == nil {
run.ErrorCount += failedJobCount
if verbose && logFailedJobs && failedJobCount > 0 {
fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Added %d failed jobs to error count for run %d", failedJobCount, run.DatabaseID)))
Expand Down
14 changes: 7 additions & 7 deletions pkg/cli/logs_orchestrator_filters_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import (
"github.com/stretchr/testify/require"
)

func stubFetchJobStatusesForProcessedRun(t *testing.T, fn func(int64, bool) (int, error)) {
func stubFetchJobStatusesForProcessedRun(t *testing.T, fn func(context.Context, int64, bool) (int, error)) {
t.Helper()
previous := fetchJobStatusesForProcessedRun
fetchJobStatusesForProcessedRun = fn
Expand Down Expand Up @@ -211,7 +211,7 @@ func TestApplyRunFilters_FilteredIntegrity(t *testing.T) {
// the ProcessedRun fields from a DownloadResult.
func TestBuildProcessedRun(t *testing.T) {
t.Run("basic fields are propagated", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil })
stubFetchJobStatusesForProcessedRun(t, func(context.Context, int64, bool) (int, error) { return 0, nil })
now := time.Now()
tmpDir := t.TempDir()
awCtx := &AwContext{Repo: "owner/repo"}
Expand All @@ -235,7 +235,7 @@ func TestBuildProcessedRun(t *testing.T) {
})

t.Run("duration and action minutes are computed", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil })
stubFetchJobStatusesForProcessedRun(t, func(context.Context, int64, bool) (int, error) { return 0, nil })
base := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
result := DownloadResult{
Run: WorkflowRun{
Expand All @@ -253,7 +253,7 @@ func TestBuildProcessedRun(t *testing.T) {
})

t.Run("zero timestamps leave duration unset", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil })
stubFetchJobStatusesForProcessedRun(t, func(context.Context, int64, bool) (int, error) { return 0, nil })
result := DownloadResult{
Run: WorkflowRun{DatabaseID: 7},
LogsPath: t.TempDir(),
Expand All @@ -264,7 +264,7 @@ func TestBuildProcessedRun(t *testing.T) {
})

t.Run("effective tokens are propagated", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil })
stubFetchJobStatusesForProcessedRun(t, func(context.Context, int64, bool) (int, error) { return 0, nil })
usage := &TokenUsageSummary{TotalEffectiveTokens: 5000}
result := DownloadResult{
Run: WorkflowRun{DatabaseID: 3},
Expand All @@ -276,7 +276,7 @@ func TestBuildProcessedRun(t *testing.T) {
})

t.Run("zero effective tokens not propagated", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil })
stubFetchJobStatusesForProcessedRun(t, func(context.Context, int64, bool) (int, error) { return 0, nil })
usage := &TokenUsageSummary{TotalEffectiveTokens: 0}
result := DownloadResult{
Run: WorkflowRun{DatabaseID: 4},
Expand All @@ -288,7 +288,7 @@ func TestBuildProcessedRun(t *testing.T) {
})

t.Run("failed job count is added via test seam", func(t *testing.T) {
stubFetchJobStatusesForProcessedRun(t, func(runID int64, verbose bool) (int, error) {
stubFetchJobStatusesForProcessedRun(t, func(_ context.Context, runID int64, verbose bool) (int, error) {
assert.Equal(t, int64(88), runID)
assert.False(t, verbose)
return 2, nil
Expand Down
10 changes: 5 additions & 5 deletions pkg/cli/logs_run_processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ func processSingleRunDownload(
if params.evalsArtifactRequested && !runHasEvals(runOutputDir, params.verbose) {
tryDownloadEvalsArtifactFallback(ctx, run.DatabaseID, runOutputDir, perRunParams)
}
analyzeRunArtifacts(result, runOutputDir, params.verbose, params.artifactFilter)
analyzeRunArtifacts(ctx, result, runOutputDir, params.verbose, params.artifactFilter)
}
} else {
logsOrchestratorLog.Printf("Cache hit for run %d, using cached summary", run.DatabaseID)
Expand Down Expand Up @@ -329,7 +329,7 @@ func tryLoadCachedRunResult(
// analyzeRunArtifacts populates a DownloadResult with all analysis data derived from
// freshly-downloaded artifacts in runOutputDir. Called only when the download succeeded
// and no valid cached summary was found.
func analyzeRunArtifacts(result *DownloadResult, runOutputDir string, verbose bool, artifactFilter []string) {
func analyzeRunArtifacts(ctx context.Context, result *DownloadResult, runOutputDir string, verbose bool, artifactFilter []string) {
metrics := extractRunMetricsAndMetadata(result, runOutputDir, verbose)

usageActivitySummary, usageActivityErr := loadUsageActivitySummary(runOutputDir)
Expand All @@ -350,7 +350,7 @@ func analyzeRunArtifacts(result *DownloadResult, runOutputDir string, verbose bo

applyRunUsageMetrics(result, &metrics, runOutputDir, verbose, usageActivitySummary, hasFirewallArtifact)

finalizeAndSaveRunSummary(result, runOutputDir, metrics, verbose)
finalizeAndSaveRunSummary(ctx, result, runOutputDir, metrics, verbose)
}

// extractRunMetricsAndMetadata extracts log metrics, infers missing workflow path, and
Expand Down Expand Up @@ -488,8 +488,8 @@ func applyRunUsageMetrics(result *DownloadResult, metrics *LogMetrics, runOutput
// finalizeAndSaveRunSummary fetches job details, derives the agentic analysis, builds the
// RunSummary struct, and writes it to disk. It also sets the agentic-analysis fields on
// result directly so they are available to the caller.
func finalizeAndSaveRunSummary(result *DownloadResult, runOutputDir string, metrics LogMetrics, verbose bool) {
jobDetails, jobErr := fetchJobDetails(result.Run.DatabaseID, verbose)
func finalizeAndSaveRunSummary(ctx context.Context, result *DownloadResult, runOutputDir string, metrics LogMetrics, verbose bool) {
jobDetails, jobErr := fetchJobDetails(ctx, result.Run.DatabaseID, verbose)
if jobErr != nil && verbose {
fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to fetch job details for run %d: %v", result.Run.DatabaseID, jobErr)))
} else {
Expand Down
Loading