diff --git a/docs/adr/44051-split-logs-orchestrator-focused-modules.md b/docs/adr/44051-split-logs-orchestrator-focused-modules.md new file mode 100644 index 00000000000..bc50b9e2d84 --- /dev/null +++ b/docs/adr/44051-split-logs-orchestrator-focused-modules.md @@ -0,0 +1,45 @@ +# ADR-44051: Split logs_orchestrator.go into Focused Modules + +**Date**: 2026-07-07 +**Status**: Draft +**Deciders**: Unknown (generated from PR #44051 diff) + +--- + +### Context + +`pkg/cli/logs_orchestrator.go` had grown to 1284 lines, mixing orchestration, filter logic, type definitions, output rendering, and stdin processing in a single file. Approximately 150 lines of run-filter logic and `ProcessedRun` construction were duplicated verbatim between `DownloadWorkflowLogs` and `DownloadWorkflowLogsFromStdin`. Any bug fix or behavioral change to the filter pipeline had to be applied in two places, creating an active maintenance risk. The codebase uses the `pkg/cli` package for all command-line surface area; Go does not require splitting a package across multiple files, but readability and testability suffer when a single file grows past ~300–400 lines. + +### Decision + +We will split `logs_orchestrator.go` into five focused files within the same `cli` package: `logs_orchestrator_types.go` (option structs and internal types), `logs_orchestrator_filters.go` (filter helpers `applyRunFilters` and `buildProcessedRun`), `logs_orchestrator_render.go` (output rendering), `logs_orchestrator_stdin.go` (`DownloadWorkflowLogsFromStdin`), and a trimmed `logs_orchestrator.go` retaining only the pagination loop and `DownloadWorkflowLogs`. The public API (`DownloadWorkflowLogs`, `DownloadWorkflowLogsFromStdin`, `LogsDownloadOptions`, `StdinLogsOptions`) is unchanged. Unit tests for the newly extracted helpers are added in `logs_orchestrator_filters_test.go`. + +### Alternatives Considered + +#### Alternative 1: Deduplicate helpers without file splitting + +Introduce `applyRunFilters` and `buildProcessedRun` as private helpers inside `logs_orchestrator.go` without moving any code to new files. This eliminates the duplication and is the minimal change, but the file would remain ~900 lines with type definitions, rendering, stdin processing, and the core pagination loop still entangled. Navigation and isolated testing of each concern would still be difficult. + +#### Alternative 2: Move each concern into a separate Go package + +Create sub-packages such as `pkg/cli/logsfilters` and `pkg/cli/logsrender`. This enforces hard encapsulation boundaries at the package level and prevents accidental coupling. However, it requires exporting all shared types (e.g., `DownloadResult`, `ProcessedRun`) that are currently package-private, widens the public API surface, and increases the cost of future changes that cross the new package boundaries. Given the single-package nature of the CLI layer, the encapsulation benefit does not justify the added complexity. + +### Consequences + +#### Positive +- Eliminates ~150 lines of duplicated filter and `ProcessedRun` construction code, so future filter changes are applied once and tested once. +- Smaller, single-concern files (87–252 lines each) are faster to navigate, review, and modify independently. +- The new `applyRunFilters` and `buildProcessedRun` helpers are directly unit-testable without exercising the full download pipeline. +- No callers outside the package need to change; the public API is stable. + +#### Negative +- Tracing the complete logic for a single download path (e.g., `DownloadWorkflowLogs`) now requires opening multiple files rather than scrolling within one. +- If the filter semantics for the stdin path and the standard path ever need to diverge significantly, the shared `applyRunFilters` function becomes a constraint that must be refactored. + +#### Neutral +- All five files remain in `package cli`; there is no package boundary or import-cycle risk introduced. +- The `logs_orchestrator_filters_test.go` file uses `package cli` (not `package cli_test`), so it retains access to package-private identifiers needed to construct test fixtures. + +--- + +*ADR created by [adr-writer agent]. Review and finalize before changing status from Draft to Accepted.* diff --git a/pkg/cli/audit_cross_run_render.go b/pkg/cli/audit_cross_run_render.go index f840ab8ce1b..4e698ffd068 100644 --- a/pkg/cli/audit_cross_run_render.go +++ b/pkg/cli/audit_cross_run_render.go @@ -3,6 +3,7 @@ package cli import ( "encoding/json" "fmt" + "io" "os" "strconv" "strings" @@ -25,157 +26,165 @@ func renderCrossRunReportJSON(report *CrossRunAuditReport) error { // renderCrossRunReportMarkdown outputs the cross-run report as Markdown to stdout. func renderCrossRunReportMarkdown(report *CrossRunAuditReport) { + renderCrossRunReportMarkdownToWriter(os.Stdout, report) +} + +func renderCrossRunReportMarkdownToWriter(w io.Writer, report *CrossRunAuditReport) { crossRunRenderLog.Printf("Rendering cross-run report as markdown: runs_analyzed=%d, domains=%d", report.RunsAnalyzed, len(report.DomainInventory)) - fmt.Fprintln(os.Stdout, "# Audit Report — Cross-Run Analysis") - fmt.Fprintln(os.Stdout) - - renderMarkdownExecutiveSummary(report) - renderMarkdownMetricsTrend(report.MetricsTrend) - renderMarkdownMCPHealth(report) - renderMarkdownErrorTrend(report) - renderMarkdownDomainInventory(report) - renderMarkdownDrain3Insights(report.Drain3Insights) - renderMarkdownPerRunBreakdown(report.PerRunBreakdown) + fmt.Fprintln(w, "# Audit Report — Cross-Run Analysis") + fmt.Fprintln(w) + + renderMarkdownExecutiveSummaryToWriter(w, report) + renderMarkdownMetricsTrendToWriter(w, report.MetricsTrend) + renderMarkdownMCPHealthToWriter(w, report) + renderMarkdownErrorTrendToWriter(w, report) + renderMarkdownDomainInventoryToWriter(w, report) + renderMarkdownDrain3InsightsToWriter(w, report.Drain3Insights) + renderMarkdownPerRunBreakdownToWriter(w, report.PerRunBreakdown) } -func renderMarkdownExecutiveSummary(report *CrossRunAuditReport) { - fmt.Fprintln(os.Stdout, "## Executive Summary") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Metric | Value |\n") - fmt.Fprintf(os.Stdout, "|--------|-------|\n") - fmt.Fprintf(os.Stdout, "| Runs analyzed | %d |\n", report.RunsAnalyzed) - fmt.Fprintf(os.Stdout, "| Runs with firewall data | %d |\n", report.RunsWithData) - fmt.Fprintf(os.Stdout, "| Runs without firewall data | %d |\n", report.RunsWithoutData) - fmt.Fprintf(os.Stdout, "| Total requests | %d |\n", report.Summary.TotalRequests) - fmt.Fprintf(os.Stdout, "| Allowed requests | %d |\n", report.Summary.TotalAllowed) - fmt.Fprintf(os.Stdout, "| Blocked requests | %d |\n", report.Summary.TotalBlocked) - fmt.Fprintf(os.Stdout, "| Overall denial rate | %.1f%% |\n", report.Summary.OverallDenyRate*100) - fmt.Fprintf(os.Stdout, "| Unique domains | %d |\n", report.Summary.UniqueDomains) - fmt.Fprintln(os.Stdout) +func renderMarkdownExecutiveSummaryToWriter(w io.Writer, report *CrossRunAuditReport) { + fmt.Fprintln(w, "## Executive Summary") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Metric | Value |\n") + fmt.Fprintf(w, "|--------|-------|\n") + fmt.Fprintf(w, "| Runs analyzed | %d |\n", report.RunsAnalyzed) + fmt.Fprintf(w, "| Runs with firewall data | %d |\n", report.RunsWithData) + fmt.Fprintf(w, "| Runs without firewall data | %d |\n", report.RunsWithoutData) + fmt.Fprintf(w, "| Total requests | %d |\n", report.Summary.TotalRequests) + fmt.Fprintf(w, "| Allowed requests | %d |\n", report.Summary.TotalAllowed) + fmt.Fprintf(w, "| Blocked requests | %d |\n", report.Summary.TotalBlocked) + fmt.Fprintf(w, "| Overall denial rate | %.1f%% |\n", report.Summary.OverallDenyRate*100) + fmt.Fprintf(w, "| Unique domains | %d |\n", report.Summary.UniqueDomains) + fmt.Fprintln(w) } func renderMarkdownMetricsTrend(mt MetricsTrendData) { + renderMarkdownMetricsTrendToWriter(os.Stdout, mt) +} + +func renderMarkdownMetricsTrendToWriter(w io.Writer, mt MetricsTrendData) { if mt.TotalTokens == 0 && mt.TotalTurns == 0 && mt.AvgDurationNs == 0 { return } - fmt.Fprintln(os.Stdout, "## Metrics Trends") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Metric | Total | Avg/run | Min | Max | Spikes |\n") - fmt.Fprintf(os.Stdout, "|--------|-------|---------|-----|-----|--------|\n") + fmt.Fprintln(w, "## Metrics Trends") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Metric | Total | Avg/run | Min | Max | Spikes |\n") + fmt.Fprintf(w, "|--------|-------|---------|-----|-----|--------|\n") if mt.TotalTokens > 0 { spikes := "—" if len(mt.TokenSpikes) > 0 { spikes = "⚠ " + formatRunIDs(mt.TokenSpikes) } - fmt.Fprintf(os.Stdout, "| Token Trend | %d | %d | %d | %d | %s |\n", + fmt.Fprintf(w, "| Token Trend | %d | %d | %d | %d | %s |\n", mt.TotalTokens, mt.AvgTokens, mt.MinTokens, mt.MaxTokens, spikes) } if mt.TotalTurns > 0 { - fmt.Fprintf(os.Stdout, "| Turns | %d | %.1f | — | %d | — |\n", + fmt.Fprintf(w, "| Turns | %d | %.1f | — | %d | — |\n", mt.TotalTurns, mt.AvgTurns, mt.MaxTurns) } if mt.AvgDurationNs > 0 { - fmt.Fprintf(os.Stdout, "| Duration | — | %s | %s | %s | — |\n", + fmt.Fprintf(w, "| Duration | — | %s | %s | %s | — |\n", timeutil.FormatDurationNs(mt.AvgDurationNs), timeutil.FormatDurationNs(mt.MinDurationNs), timeutil.FormatDurationNs(mt.MaxDurationNs)) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } -func renderMarkdownMCPHealth(report *CrossRunAuditReport) { +func renderMarkdownMCPHealthToWriter(w io.Writer, report *CrossRunAuditReport) { if len(report.MCPHealth) == 0 { return } - fmt.Fprintf(os.Stdout, "## MCP Server Health (%d runs)\n\n", report.RunsAnalyzed) - fmt.Fprintf(os.Stdout, "| Server | Connected | Error Rate | Total Calls | Errors | Status |\n") - fmt.Fprintf(os.Stdout, "|--------|-----------|------------|-------------|--------|--------|\n") + fmt.Fprintf(w, "## MCP Server Health (%d runs)\n\n", report.RunsAnalyzed) + fmt.Fprintf(w, "| Server | Connected | Error Rate | Total Calls | Errors | Status |\n") + fmt.Fprintf(w, "|--------|-----------|------------|-------------|--------|--------|\n") for _, h := range report.MCPHealth { status := "✅ ok" if h.Unreliable { status = "⚠ unreliable" } - fmt.Fprintf(os.Stdout, "| `%s` | %d/%d | %.1f%% | %d | %d | %s |\n", + fmt.Fprintf(w, "| `%s` | %d/%d | %.1f%% | %d | %d | %s |\n", h.ServerName, h.RunsConnected, h.TotalRuns, h.ErrorRate*100, h.TotalCalls, h.TotalErrors, status) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } -func renderMarkdownErrorTrend(report *CrossRunAuditReport) { +func renderMarkdownErrorTrendToWriter(w io.Writer, report *CrossRunAuditReport) { et := report.ErrorTrend if et.TotalErrors == 0 && et.TotalWarnings == 0 { return } - fmt.Fprintln(os.Stdout, "## Error Trend") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Metric | Value |\n") - fmt.Fprintf(os.Stdout, "|--------|-------|\n") - fmt.Fprintf(os.Stdout, "| Runs with errors | %d/%d (%.0f%%) |\n", + fmt.Fprintln(w, "## Error Trend") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Metric | Value |\n") + fmt.Fprintf(w, "|--------|-------|\n") + fmt.Fprintf(w, "| Runs with errors | %d/%d (%.0f%%) |\n", et.RunsWithErrors, report.RunsAnalyzed, safePercent(et.RunsWithErrors, report.RunsAnalyzed)) - fmt.Fprintf(os.Stdout, "| Total errors | %d |\n", et.TotalErrors) - fmt.Fprintf(os.Stdout, "| Avg errors/run | %.2f |\n", et.AvgErrorsPerRun) + fmt.Fprintf(w, "| Total errors | %d |\n", et.TotalErrors) + fmt.Fprintf(w, "| Avg errors/run | %.2f |\n", et.AvgErrorsPerRun) if et.TotalWarnings > 0 { - fmt.Fprintf(os.Stdout, "| Runs with warnings | %d/%d |\n", et.RunsWithWarnings, report.RunsAnalyzed) - fmt.Fprintf(os.Stdout, "| Total warnings | %d |\n", et.TotalWarnings) + fmt.Fprintf(w, "| Runs with warnings | %d/%d |\n", et.RunsWithWarnings, report.RunsAnalyzed) + fmt.Fprintf(w, "| Total warnings | %d |\n", et.TotalWarnings) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } -func renderMarkdownDomainInventory(report *CrossRunAuditReport) { +func renderMarkdownDomainInventoryToWriter(w io.Writer, report *CrossRunAuditReport) { if len(report.DomainInventory) == 0 { return } - fmt.Fprintln(os.Stdout, "## Domain Inventory") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Domain | Status | Seen In | Allowed | Blocked |\n") - fmt.Fprintf(os.Stdout, "|--------|--------|---------|---------|--------|\n") + fmt.Fprintln(w, "## Domain Inventory") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Domain | Status | Seen In | Allowed | Blocked |\n") + fmt.Fprintf(w, "|--------|--------|---------|---------|--------|\n") for _, entry := range report.DomainInventory { - fmt.Fprintf(os.Stdout, "| `%s` | %s %s | %d/%d runs | %d | %d |\n", + fmt.Fprintf(w, "| `%s` | %s %s | %d/%d runs | %d | %d |\n", entry.Domain, firewallStatusEmoji(entry.OverallStatus), entry.OverallStatus, entry.SeenInRuns, report.RunsAnalyzed, entry.TotalAllowed, entry.TotalBlocked) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } -func renderMarkdownDrain3Insights(insights []ObservabilityInsight) { +func renderMarkdownDrain3InsightsToWriter(w io.Writer, insights []ObservabilityInsight) { if len(insights) == 0 { return } crossRunRenderLog.Printf("Rendering markdown drain3 insights: count=%d", len(insights)) - fmt.Fprintln(os.Stdout, "## Agent Event Pattern Analysis") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Severity | Category | Title | Summary |\n") - fmt.Fprintf(os.Stdout, "|----------|----------|-------|--------|\n") + fmt.Fprintln(w, "## Agent Event Pattern Analysis") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Severity | Category | Title | Summary |\n") + fmt.Fprintf(w, "|----------|----------|-------|--------|\n") for _, insight := range insights { summary := insight.Summary if insight.Evidence != "" { summary += " (" + insight.Evidence + ")" } - fmt.Fprintf(os.Stdout, "| %s %s | %s | %s | %s |\n", + fmt.Fprintf(w, "| %s %s | %s | %s | %s |\n", renderSeverityIcon(insight.Severity), insight.Severity, insight.Category, insight.Title, summary) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } -func renderMarkdownPerRunBreakdown(runs []PerRunFirewallBreakdown) { +func renderMarkdownPerRunBreakdownToWriter(w io.Writer, runs []PerRunFirewallBreakdown) { if len(runs) == 0 { return } - fmt.Fprintln(os.Stdout, "## Per-Run Breakdown") - fmt.Fprintln(os.Stdout) - fmt.Fprintf(os.Stdout, "| Run ID | Workflow | Conclusion | Duration | Firewall | Tokens | Turns | MCP Err | Errors |\n") - fmt.Fprintf(os.Stdout, "|--------|----------|------------|----------|----------|--------|-------|---------|--------|\n") + fmt.Fprintln(w, "## Per-Run Breakdown") + fmt.Fprintln(w) + fmt.Fprintf(w, "| Run ID | Workflow | Conclusion | Duration | Firewall | Tokens | Turns | MCP Err | Errors |\n") + fmt.Fprintf(w, "|--------|----------|------------|----------|----------|--------|-------|---------|--------|\n") for _, run := range runs { firewallCol, tokenStr, turnsStr, durStr := markdownPerRunFields(run) - fmt.Fprintf(os.Stdout, "| %d | %s | %s | %s | %s | %s | %s | %d | %d |\n", + fmt.Fprintf(w, "| %d | %s | %s | %s | %s | %s | %s | %d | %d |\n", run.RunID, run.WorkflowName, run.Conclusion, durStr, firewallCol, tokenStr, turnsStr, run.MCPErrors, run.ErrorCount) } - fmt.Fprintln(os.Stdout) + fmt.Fprintln(w) } func markdownPerRunFields(run PerRunFirewallBreakdown) (string, string, string, string) { diff --git a/pkg/cli/logs_orchestrator.go b/pkg/cli/logs_orchestrator.go index f8a651bf7df..f6cafba3d5e 100644 --- a/pkg/cli/logs_orchestrator.go +++ b/pkg/cli/logs_orchestrator.go @@ -14,18 +14,15 @@ import ( "context" "errors" "fmt" - "math" "os" "path/filepath" "strings" "time" "github.com/github/gh-aw/pkg/console" - "github.com/github/gh-aw/pkg/constants" "github.com/github/gh-aw/pkg/envutil" "github.com/github/gh-aw/pkg/fileutil" "github.com/github/gh-aw/pkg/logger" - "github.com/github/gh-aw/pkg/parser" ) var logsOrchestratorLog = logger.New("cli:logs_orchestrator") @@ -96,45 +93,6 @@ func getMaxConcurrentDownloads() int { return envutil.GetIntFromEnv("GH_AW_MAX_CONCURRENT_DOWNLOADS", MaxConcurrentDownloads, 1, 100, logsOrchestratorLog) } -// matchEngineFilter checks whether the run recorded in awInfo matches the -// requested engine filter string. It returns (matches, detectedEngineID). -// detectedEngineID is "" when awInfo is unavailable or carries no engine_id. -func matchEngineFilter(awInfo *AwInfo, awInfoErr error, filterEngine string) (bool, string) { - if awInfoErr != nil || awInfo == nil || awInfo.EngineID == "" { - return false, "" - } - return awInfo.EngineID == filterEngine, awInfo.EngineID -} - -type LogsDownloadOptions struct { - WorkflowName string - Count int - StartDate string - EndDate string - OutputDir string - Engine string - Ref string - BeforeRunID int64 - AfterRunID int64 - RepoOverride string - Verbose bool - ToolGraph bool - NoStaged bool - FirewallOnly bool - NoFirewall bool - Parse bool - JSONOutput bool - TimeoutMinutes int - SummaryFile string - SafeOutputType string - FilteredIntegrity bool - Train bool - Format string - ArtifactSets []string - After string - ReportFile string -} - func shouldStopPagination(totalFetched, batchSize int) bool { return totalFetched < batchSize } @@ -158,17 +116,6 @@ func selectPaginationCursorDate(filteredRuns []WorkflowRun, oldestFetchedCreated // - countLimitReached: in fetchAllInRange mode the count cap was hit before the // date window was exhausted; the next page starts just before the oldest run // returned in this batch. -type continuationOptions struct { - workflowName string - startDate string - endDate string - engine string - branch string - afterRunID int64 - count int - timeoutMinutes int -} - func buildContinuationIfNeeded( processedRuns []ProcessedRun, timeoutReached, countLimitReached bool, @@ -318,6 +265,15 @@ func DownloadWorkflowLogs(ctx context.Context, opts LogsDownloadOptions) error { // and drives continuation-data generation so callers can page through the full range. var countLimitReached bool + filters := runFilterOpts{ + engine: engine, + noStaged: noStaged, + firewallOnly: firewallOnly, + noFirewall: noFirewall, + safeOutputType: safeOutputType, + filteredIntegrity: filteredIntegrity, + } + // Iterative algorithm: keep fetching runs until we have enough or exhaust available runs outerLoop: for iteration < MaxIterations { @@ -503,154 +459,11 @@ outerLoop: continue } - // Parse aw_info.json once for all filters that need it (optimization) - var awInfo *AwInfo - var awInfoErr error - awInfoPath := filepath.Join(result.LogsPath, "aw_info.json") - - // Only parse if we need it for any filter - if engine != "" || noStaged || firewallOnly || noFirewall { - awInfo, awInfoErr = parseAwInfo(awInfoPath, verbose) - } - - // Apply engine filtering if specified - if engine != "" { - engineMatches, detectedEngineID := matchEngineFilter(awInfo, awInfoErr, engine) - if !engineMatches { - if detectedEngineID == "" { - detectedEngineID = "unknown" - } - logsOrchestratorLog.Printf("Skipping run %d: engine filter=%s, detected=%s", result.Run.DatabaseID, engine, detectedEngineID) - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: engine '%s' does not match filter '%s'", result.Run.DatabaseID, detectedEngineID, engine))) - } - continue - } - } - - // Apply staged filtering if --no-staged flag is specified - if noStaged { - var isStaged bool - if awInfoErr == nil && awInfo != nil { - isStaged = awInfo.Staged - } - - if isStaged { - logsOrchestratorLog.Printf("Skipping run %d: staged workflow filtered by --no-staged", result.Run.DatabaseID) - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow is staged (filtered out by --no-staged)", result.Run.DatabaseID))) - } - continue - } - } - - // Apply firewall filtering if --firewall or --no-firewall flag is specified - if firewallOnly || noFirewall { - var hasFirewall bool - if awInfoErr == nil && awInfo != nil { - // Firewall is enabled if steps.firewall is non-empty (e.g., "squid") - hasFirewall = awInfo.Steps.Firewall != "" - } - - // Check if the run matches the filter - if firewallOnly && !hasFirewall { - logsOrchestratorLog.Printf("Skipping run %d: no firewall detected, filtered by --firewall", result.Run.DatabaseID) - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow does not use firewall (filtered by --firewall)", result.Run.DatabaseID))) - } - continue - } - if noFirewall && hasFirewall { - logsOrchestratorLog.Printf("Skipping run %d: firewall detected, filtered by --no-firewall", result.Run.DatabaseID) - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow uses firewall (filtered by --no-firewall)", result.Run.DatabaseID))) - } - continue - } - } - - // Apply safe output type filtering if --safe-output flag is specified - if safeOutputType != "" { - hasSafeOutputType, checkErr := runContainsSafeOutputType(result.LogsPath, safeOutputType, verbose) - if checkErr != nil && verbose { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check safe output type for run %d: %v", result.Run.DatabaseID, checkErr))) - } - - if !hasSafeOutputType { - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no '%s' safe output messages found", result.Run.DatabaseID, safeOutputType))) - } - continue - } - } - - // Apply filtered-integrity filtering if --filtered-integrity flag is specified - if filteredIntegrity { - hasFiltered, checkErr := runHasDifcFilteredItems(result.LogsPath, verbose) - if checkErr != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check DIFC filtered items for run %d: %v", result.Run.DatabaseID, checkErr))) - continue - } - - if !hasFiltered { - logsOrchestratorLog.Printf("Skipping run %d: no DIFC filtered items found", result.Run.DatabaseID) - if verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no DIFC integrity-filtered items found in gateway logs", result.Run.DatabaseID))) - } - continue - } - } - - // Update run with metrics and path - run := result.Run - run.TokenUsage = result.Metrics.TokenUsage - applyMetricsTurnsToRun(&run, result.Metrics) - run.AvgTimeBetweenTurns = result.Metrics.AvgTimeBetweenTurns - run.ErrorCount = 0 - run.WarningCount = 0 - run.LogsPath = result.LogsPath - - // Propagate effective tokens from cached firewall proxy summary when available - if result.TokenUsage != nil && result.TokenUsage.TotalEffectiveTokens > 0 { - run.EffectiveTokens = result.TokenUsage.TotalEffectiveTokens - } - - // Add failed jobs to error count - if failedJobCount, err := fetchJobStatuses(run.DatabaseID, verbose); err == nil { - run.ErrorCount += failedJobCount - if verbose && failedJobCount > 0 { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Added %d failed jobs to error count for run %d", failedJobCount, run.DatabaseID))) - } - } - - // Always use GitHub API timestamps for duration calculation - if !run.StartedAt.IsZero() && !run.UpdatedAt.IsZero() { - run.Duration = run.UpdatedAt.Sub(run.StartedAt) - // Estimate billable Actions minutes from wall-clock time. - // GitHub Actions bills per minute, rounded up per job. - run.ActionMinutes = math.Ceil(run.Duration.Minutes()) + if applyRunFilters(result, filters, verbose) { + continue } - processedRun := ProcessedRun{ - Run: run, - AwContext: result.AwContext, - TaskDomain: result.TaskDomain, - BehaviorFingerprint: result.BehaviorFingerprint, - AgenticAssessments: result.AgenticAssessments, - AccessAnalysis: result.AccessAnalysis, - FirewallAnalysis: result.FirewallAnalysis, - RedactedDomainsAnalysis: result.RedactedDomainsAnalysis, - MissingTools: result.MissingTools, - MissingData: result.MissingData, - Noops: result.Noops, - MCPFailures: result.MCPFailures, - MCPToolUsage: result.MCPToolUsage, - TokenUsage: result.TokenUsage, - GitHubRateLimitUsage: result.GitHubRateLimitUsage, - JobDetails: result.JobDetails, - } - processedRuns = append(processedRuns, processedRun) - batchProcessed++ + processedRun := buildProcessedRun(result, verbose, true) // If --parse flag is set, parse the agent log and write to log.md if parse { @@ -659,27 +472,30 @@ outerLoop: detectedEngine := extractEngineFromAwInfo(awInfoPath, verbose) if err := parseAgentLog(result.LogsPath, detectedEngine, verbose); err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse log for run %d: %v", run.DatabaseID, err))) + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse log for run %d: %v", processedRun.Run.DatabaseID, err))) } else { // Always show success message for parsing, not just in verbose mode logMdPath := filepath.Join(result.LogsPath, "log.md") if fileutil.FileExists(logMdPath) { - fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed log for run %d → %s", run.DatabaseID, logMdPath))) + fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed log for run %d → %s", processedRun.Run.DatabaseID, logMdPath))) } } // Also parse firewall logs if they exist if err := parseFirewallLogs(result.LogsPath, verbose); err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse firewall logs for run %d: %v", run.DatabaseID, err))) + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse firewall logs for run %d: %v", processedRun.Run.DatabaseID, err))) } else { // Show success message if firewall.md was created firewallMdPath := filepath.Join(result.LogsPath, "firewall.md") if fileutil.FileExists(firewallMdPath) { - fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed firewall logs for run %d → %s", run.DatabaseID, firewallMdPath))) + fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed firewall logs for run %d → %s", processedRun.Run.DatabaseID, firewallMdPath))) } } } + processedRuns = append(processedRuns, processedRun) + batchProcessed++ + // Stop processing this batch once we've collected enough runs. if len(processedRuns) >= count { break @@ -785,500 +601,3 @@ outerLoop: artifactFilter: artifactFilter, }) } - -// renderLogsOutputOptions holds configuration for renderLogsOutput. -type renderLogsOutputOptions struct { - outputDir string - summaryFile string - format string - reportFile string - jsonOutput bool - toolGraph bool - train bool - continuation *ContinuationData - verbose bool - artifactFilter []string -} - -// renderLogsOutput finalizes processedRuns and renders them in the appropriate output -// format: JSON, console metrics table, or cross-run audit report (pretty/markdown). -// continuation is optional and only set when a timeout was reached during a paginated download. -func renderLogsOutput(processedRuns []ProcessedRun, opts renderLogsOutputOptions) error { - // Update MissingToolCount, MissingDataCount, and NoopCount in runs - for i := range processedRuns { - processedRuns[i].Run.MissingToolCount = len(processedRuns[i].MissingTools) - processedRuns[i].Run.MissingDataCount = len(processedRuns[i].MissingData) - processedRuns[i].Run.NoopCount = len(processedRuns[i].Noops) - } - - // Build structured logs data - logsOrchestratorLog.Printf("Building logs data from %d processed runs (continuation=%t)", len(processedRuns), opts.continuation != nil) - logsData := buildLogsData(processedRuns, opts.outputDir, opts.continuation) - - // When only the usage artifact was downloaded, add a hint so consumers know how - // to fetch additional artifact sets (agent logs, firewall data, etc.). - if isUsageOnlyArtifactFilter(opts.artifactFilter) { - logsData.Message = usageOnlyArtifactHintMessage() - } - - // Write summary file if requested (default behavior unless disabled with empty string) - if opts.summaryFile != "" { - summaryPath := filepath.Join(opts.outputDir, opts.summaryFile) - if err := writeSummaryFile(summaryPath, logsData, opts.verbose); err != nil { - return fmt.Errorf("failed to write summary file: %w", err) - } - } - - // Train drain3 weights if requested. - if opts.train { - if err := TrainDrain3Weights(processedRuns, opts.outputDir, opts.verbose); err != nil { - return fmt.Errorf("log pattern training: %w", err) - } - } - - // Render output based on format preference. - switch opts.format { - case "tsv": - if opts.verbose { - renderLogsTSVVerbose(logsData) - } else { - renderLogsTSV(logsData) - } - renderLogsArtifactHint(os.Stderr, logsData.Message) - return nil - - case "markdown", "pretty": - inputs := make([]crossRunInput, 0, len(processedRuns)) - for _, pr := range processedRuns { - inputs = append(inputs, crossRunInput{ - RunID: pr.Run.DatabaseID, - WorkflowName: pr.Run.WorkflowName, - Conclusion: pr.Run.Conclusion, - Duration: pr.Run.Duration, - FirewallAnalysis: pr.FirewallAnalysis, - Metrics: LogMetrics{ - TokenUsage: pr.Run.TokenUsage, - Turns: pr.Run.Turns, - }, - MCPToolUsage: pr.MCPToolUsage, - MCPFailures: pr.MCPFailures, - ErrorCount: pr.Run.ErrorCount, - }) - } - report := buildCrossRunAuditReport(inputs) - if opts.jsonOutput { - return renderCrossRunReportJSON(report) - } - if opts.format == "pretty" { - renderCrossRunReportPretty(report) - renderLogsArtifactHint(os.Stderr, logsData.Message) - return nil - } - if opts.reportFile != "" { - if err := os.MkdirAll(filepath.Dir(opts.reportFile), constants.DirPermPublic); err != nil { - return fmt.Errorf("failed to create report file directory: %w", err) - } - f, err := os.Create(opts.reportFile) - if err != nil { - return fmt.Errorf("failed to create report file: %w", err) - } - if err := func() (retErr error) { - defer func() { - if cerr := f.Close(); cerr != nil && retErr == nil { - retErr = cerr - } - }() - oldStdout := os.Stdout - defer func() { os.Stdout = oldStdout }() - os.Stdout = f - renderCrossRunReportMarkdown(report) - return nil - }(); err != nil { - return fmt.Errorf("failed to write report file: %w", err) - } - } else { - renderCrossRunReportMarkdown(report) - } - renderLogsArtifactHint(os.Stderr, logsData.Message) - return nil - - case "console": - // Explicit console format: decorated tables for human reading - if opts.jsonOutput { - if err := renderLogsJSON(logsData, opts.verbose); err != nil { - return fmt.Errorf("failed to render JSON output: %w", err) - } - } else { - renderLogsConsole(logsData) - displayAggregatedGatewayMetrics(processedRuns, opts.outputDir, opts.verbose) - displayUnifiedTimeline(processedRuns, opts.verbose) - if opts.toolGraph { - generateToolGraph(processedRuns, opts.verbose) - } - renderLogsArtifactHint(os.Stderr, logsData.Message) - } - return nil - } - - // Default: compact format optimized for agentic consumption - if opts.jsonOutput { - if err := renderLogsJSON(logsData, opts.verbose); err != nil { - return fmt.Errorf("failed to render JSON output: %w", err) - } - } else { - if opts.verbose { - renderLogsCompactVerbose(logsData) - } else { - renderLogsCompact(logsData) - } - } - - return nil -} - -func renderLogsArtifactHint(w *os.File, message string) { - if message == "" { - return - } - fmt.Fprintf(w, "[hint] %s\n", message) -} - -// StdinLogsOptions holds parameters for DownloadWorkflowLogsFromStdin. -type StdinLogsOptions struct { - RunURLs []string - OutputDir string - Engine string - RepoOverride string - Verbose bool - ToolGraph bool - NoStaged bool - FirewallOnly bool - NoFirewall bool - Parse bool - JSONOutput bool - Timeout int - SummaryFile string - SafeOutputType string - FilteredIntegrity bool - Train bool - Format string - ReportFile string - // ArtifactSets defaults to nil (download all artifacts) when this API is used - // programmatically. The CLI passes ["usage"] to match the logs command default. - ArtifactSets []string -} - -// DownloadWorkflowLogsFromStdin fetches and processes workflow run logs for runs -// provided as IDs or URLs, bypassing the GitHub API run-discovery step. -// This is used when the --stdin flag is passed to the logs command. -func DownloadWorkflowLogsFromStdin(ctx context.Context, opts StdinLogsOptions) error { - logsOrchestratorLog.Printf("Starting stdin log download: runs=%d, outputDir=%s", len(opts.RunURLs), opts.OutputDir) - - if err := ValidateArtifactSets(opts.ArtifactSets); err != nil { - return err - } - artifactFilter := ResolveArtifactFilter(opts.ArtifactSets) - if len(artifactFilter) > 0 { - logsOrchestratorLog.Printf("Artifact filter active: %v", artifactFilter) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage("Artifact filter: downloading only "+strings.Join(artifactFilter, ", "))) - } - } - - if err := ensureLogsGitignore(); err != nil { - logsOrchestratorLog.Printf("Failed to ensure logs .gitignore: %v", err) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to ensure .github/aw/logs/.gitignore: %v", err))) - } - } - - select { - case <-ctx.Done(): - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Operation cancelled")) - return ctx.Err() - default: - } - - if len(opts.RunURLs) == 0 { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No run IDs or URLs provided on stdin")) - return nil - } - - // Parse owner/repo (and optional GHES host) from --repo override if provided. - // Accepted formats: "owner/repo" or "HOST/owner/repo". - var hostOverride, ownerOverride, repoNameOverride string - if opts.RepoOverride != "" { - parts := strings.SplitN(opts.RepoOverride, "/", 3) - switch len(parts) { - case 3: // HOST/owner/repo - if parts[0] == "" || parts[1] == "" || parts[2] == "" { - return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) - } - hostOverride, ownerOverride, repoNameOverride = parts[0], parts[1], parts[2] - case 2: // owner/repo - if parts[0] == "" || parts[1] == "" { - return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) - } - ownerOverride, repoNameOverride = parts[0], parts[1] - default: - return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) - } - } - - // Start timeout timer if specified - var startTime time.Time - if opts.Timeout > 0 { - startTime = time.Now() - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Timeout set to %d minutes", opts.Timeout))) - } - } - - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Fetching metadata for %d runs from stdin...", len(opts.RunURLs)))) - } - - // Build WorkflowRun objects by fetching metadata for each provided URL - var runs []WorkflowRun - for _, rawURL := range opts.RunURLs { - select { - case <-ctx.Done(): - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Operation cancelled")) - return ctx.Err() - default: - } - - if opts.Timeout > 0 && time.Since(startTime).Seconds() >= float64(opts.Timeout)*60 { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Timeout reached before all run metadata could be fetched")) - break - } - - components, err := parser.ParseRunURLExtended(rawURL) - if err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping invalid run %q: %v", rawURL, err))) - continue - } - - // Prefer owner/repo embedded in the URL; fall back to --repo override. - // If neither source provides owner, the run cannot be fetched — return an - // actionable error rather than silently continuing with a broken API call. - owner := components.Owner - repo := components.Repo - host := components.Host - if owner == "" { - owner = ownerOverride - repo = repoNameOverride - if host == "" { - host = hostOverride - } - } - if owner == "" { - return fmt.Errorf("run %q does not include repository information; pass --repo owner/repo or provide full run URLs", rawURL) - } - - run, err := fetchWorkflowRunMetadata(ctx, components.Number, owner, repo, host, opts.Verbose) - if err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping run %d: failed to fetch metadata: %v", components.Number, err))) - continue - } - runs = append(runs, run) - } - - if len(runs) == 0 { - if opts.JSONOutput { - logsData := buildLogsData([]ProcessedRun{}, opts.OutputDir, nil) - logsData.Message = "No runs found. No valid runs could be loaded from the provided input." - if err := renderLogsJSON(logsData, opts.Verbose); err != nil { - return fmt.Errorf("failed to render JSON output: %w", err) - } - } - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No valid runs could be loaded from stdin")) - return nil - } - - // Download artifacts for all runs concurrently - downloadResults := downloadRunArtifactsConcurrent(ctx, runs, opts.OutputDir, opts.Verbose, len(runs), opts.RepoOverride, artifactFilter) - - // Process download results applying the same filters as DownloadWorkflowLogs - var processedRuns []ProcessedRun - for _, result := range downloadResults { - if result.Skipped { - if opts.Verbose && result.Error != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping run %d: %v", result.Run.DatabaseID, result.Error))) - } - continue - } - - if result.Error != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to download artifacts for run %d: %v", result.Run.DatabaseID, result.Error))) - continue - } - - awInfoPath := filepath.Join(result.LogsPath, "aw_info.json") - var awInfo *AwInfo - var awInfoErr error - if opts.Engine != "" || opts.NoStaged || opts.FirewallOnly || opts.NoFirewall { - awInfo, awInfoErr = parseAwInfo(awInfoPath, opts.Verbose) - } - - if opts.Engine != "" { - engineMatches, detectedEngineID := matchEngineFilter(awInfo, awInfoErr, opts.Engine) - if !engineMatches { - if detectedEngineID == "" { - detectedEngineID = "unknown" - } - logsOrchestratorLog.Printf("Skipping run %d: engine filter=%s, detected=%s", result.Run.DatabaseID, opts.Engine, detectedEngineID) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: engine '%s' does not match filter '%s'", result.Run.DatabaseID, detectedEngineID, opts.Engine))) - } - continue - } - } - - if opts.NoStaged { - var isStaged bool - if awInfoErr == nil && awInfo != nil { - isStaged = awInfo.Staged - } - if isStaged { - logsOrchestratorLog.Printf("Skipping run %d: staged workflow filtered by --no-staged", result.Run.DatabaseID) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow is staged (filtered by --no-staged)", result.Run.DatabaseID))) - } - continue - } - } - - if opts.FirewallOnly || opts.NoFirewall { - var hasFirewall bool - if awInfoErr == nil && awInfo != nil { - hasFirewall = awInfo.Steps.Firewall != "" - } - if opts.FirewallOnly && !hasFirewall { - logsOrchestratorLog.Printf("Skipping run %d: no firewall detected, filtered by --firewall", result.Run.DatabaseID) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow does not use firewall (filtered by --firewall)", result.Run.DatabaseID))) - } - continue - } - if opts.NoFirewall && hasFirewall { - logsOrchestratorLog.Printf("Skipping run %d: firewall detected, filtered by --no-firewall", result.Run.DatabaseID) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow uses firewall (filtered by --no-firewall)", result.Run.DatabaseID))) - } - continue - } - } - - if opts.SafeOutputType != "" { - hasSafeOutputType, checkErr := runContainsSafeOutputType(result.LogsPath, opts.SafeOutputType, opts.Verbose) - if checkErr != nil && opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check safe output type for run %d: %v", result.Run.DatabaseID, checkErr))) - } - if !hasSafeOutputType { - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no '%s' safe output messages found", result.Run.DatabaseID, opts.SafeOutputType))) - } - continue - } - } - - if opts.FilteredIntegrity { - hasFiltered, checkErr := runHasDifcFilteredItems(result.LogsPath, opts.Verbose) - if checkErr != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check DIFC filtered items for run %d: %v", result.Run.DatabaseID, checkErr))) - continue - } - if !hasFiltered { - logsOrchestratorLog.Printf("Skipping run %d: no DIFC filtered items found", result.Run.DatabaseID) - if opts.Verbose { - fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no DIFC integrity-filtered items found in gateway logs", result.Run.DatabaseID))) - } - continue - } - } - - run := result.Run - run.TokenUsage = result.Metrics.TokenUsage - applyMetricsTurnsToRun(&run, result.Metrics) - run.AvgTimeBetweenTurns = result.Metrics.AvgTimeBetweenTurns - run.ErrorCount = 0 - run.WarningCount = 0 - run.LogsPath = result.LogsPath - - if result.TokenUsage != nil && result.TokenUsage.TotalEffectiveTokens > 0 { - run.EffectiveTokens = result.TokenUsage.TotalEffectiveTokens - } - if failedJobCount, err := fetchJobStatuses(run.DatabaseID, opts.Verbose); err == nil { - run.ErrorCount += failedJobCount - } - if !run.StartedAt.IsZero() && !run.UpdatedAt.IsZero() { - run.Duration = run.UpdatedAt.Sub(run.StartedAt) - run.ActionMinutes = math.Ceil(run.Duration.Minutes()) - } - - processedRun := ProcessedRun{ - Run: run, - AwContext: result.AwContext, - TaskDomain: result.TaskDomain, - BehaviorFingerprint: result.BehaviorFingerprint, - AgenticAssessments: result.AgenticAssessments, - AccessAnalysis: result.AccessAnalysis, - FirewallAnalysis: result.FirewallAnalysis, - RedactedDomainsAnalysis: result.RedactedDomainsAnalysis, - MissingTools: result.MissingTools, - MissingData: result.MissingData, - Noops: result.Noops, - MCPFailures: result.MCPFailures, - MCPToolUsage: result.MCPToolUsage, - TokenUsage: result.TokenUsage, - GitHubRateLimitUsage: result.GitHubRateLimitUsage, - JobDetails: result.JobDetails, - } - processedRuns = append(processedRuns, processedRun) - - if opts.Parse { - detectedEngine := extractEngineFromAwInfo(awInfoPath, opts.Verbose) - if err := parseAgentLog(result.LogsPath, detectedEngine, opts.Verbose); err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse log for run %d: %v", run.DatabaseID, err))) - } else { - logMdPath := filepath.Join(result.LogsPath, "log.md") - if fileutil.FileExists(logMdPath) { - fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed log for run %d → %s", run.DatabaseID, logMdPath))) - } - } - if err := parseFirewallLogs(result.LogsPath, opts.Verbose); err != nil { - fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse firewall logs for run %d: %v", run.DatabaseID, err))) - } else { - firewallMdPath := filepath.Join(result.LogsPath, "firewall.md") - if fileutil.FileExists(firewallMdPath) { - fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed firewall logs for run %d → %s", run.DatabaseID, firewallMdPath))) - } - } - } - } - - if len(processedRuns) == 0 { - if opts.JSONOutput { - logsData := buildLogsData([]ProcessedRun{}, opts.OutputDir, nil) - logsData.Message = "No runs found matching the specified criteria." - if err := renderLogsJSON(logsData, opts.Verbose); err != nil { - return fmt.Errorf("failed to render JSON output: %w", err) - } - } - fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No workflow runs with artifacts found matching the specified criteria")) - return nil - } - - return renderLogsOutput(processedRuns, renderLogsOutputOptions{ - outputDir: opts.OutputDir, - summaryFile: opts.SummaryFile, - format: opts.Format, - reportFile: opts.ReportFile, - jsonOutput: opts.JSONOutput, - toolGraph: opts.ToolGraph, - train: opts.Train, - verbose: opts.Verbose, - artifactFilter: artifactFilter, - }) -} diff --git a/pkg/cli/logs_orchestrator_filters.go b/pkg/cli/logs_orchestrator_filters.go new file mode 100644 index 00000000000..252c3f2e819 --- /dev/null +++ b/pkg/cli/logs_orchestrator_filters.go @@ -0,0 +1,186 @@ +// This file provides command-line interface functionality for gh-aw. +// This file (logs_orchestrator_filters.go) contains run-filter helpers for the +// logs orchestrator: deciding whether a downloaded run should be included in +// results and constructing a ProcessedRun from a DownloadResult. + +package cli + +import ( + "fmt" + "math" + "os" + "path/filepath" + + "github.com/github/gh-aw/pkg/console" +) + +// runFilterOpts bundles the filter flags passed to applyRunFilters. +type runFilterOpts struct { + engine string + noStaged bool + firewallOnly bool + noFirewall bool + safeOutputType string + filteredIntegrity bool +} + +var fetchJobStatusesForProcessedRun = fetchJobStatuses + +// matchEngineFilter checks whether the run recorded in awInfo matches the +// requested engine filter string. It returns (matches, detectedEngineID). +// detectedEngineID is "" when awInfo is unavailable or carries no engine_id. +func matchEngineFilter(awInfo *AwInfo, awInfoErr error, filterEngine string) (bool, string) { + if awInfoErr != nil || awInfo == nil || awInfo.EngineID == "" { + return false, "" + } + return awInfo.EngineID == filterEngine, awInfo.EngineID +} + +// applyRunFilters applies all configured run filters to a DownloadResult. +// It parses aw_info.json once (lazily) when any filter that needs it is active. +// Returns true when the run should be skipped / excluded from results. +func applyRunFilters(result DownloadResult, opts runFilterOpts, verbose bool) bool { + // Parse aw_info.json once for all filters that need it (optimization). + var awInfo *AwInfo + var awInfoErr error + if opts.engine != "" || opts.noStaged || opts.firewallOnly || opts.noFirewall { + awInfoPath := filepath.Join(result.LogsPath, "aw_info.json") + awInfo, awInfoErr = parseAwInfo(awInfoPath, verbose) + } + + // Apply engine filtering if specified. + if opts.engine != "" { + engineMatches, detectedEngineID := matchEngineFilter(awInfo, awInfoErr, opts.engine) + if !engineMatches { + if detectedEngineID == "" { + detectedEngineID = "unknown" + } + logsOrchestratorLog.Printf("Skipping run %d: engine filter=%s, detected=%s", result.Run.DatabaseID, opts.engine, detectedEngineID) + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: engine '%s' does not match filter '%s'", result.Run.DatabaseID, detectedEngineID, opts.engine))) + } + return true + } + } + + // Apply staged filtering if --no-staged flag is specified. + if opts.noStaged { + var isStaged bool + if awInfoErr == nil && awInfo != nil { + isStaged = awInfo.Staged + } + if isStaged { + logsOrchestratorLog.Printf("Skipping run %d: staged workflow filtered by --no-staged", result.Run.DatabaseID) + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow is staged (filtered out by --no-staged)", result.Run.DatabaseID))) + } + return true + } + } + + // Apply firewall filtering if --firewall or --no-firewall flag is specified. + if opts.firewallOnly || opts.noFirewall { + var hasFirewall bool + if awInfoErr == nil && awInfo != nil { + // Firewall is enabled if steps.firewall is non-empty (e.g. "squid"). + hasFirewall = awInfo.Steps.Firewall != "" + } + if opts.firewallOnly && !hasFirewall { + logsOrchestratorLog.Printf("Skipping run %d: no firewall detected, filtered by --firewall", result.Run.DatabaseID) + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow does not use firewall (filtered by --firewall)", result.Run.DatabaseID))) + } + return true + } + if opts.noFirewall && hasFirewall { + logsOrchestratorLog.Printf("Skipping run %d: firewall detected, filtered by --no-firewall", result.Run.DatabaseID) + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: workflow uses firewall (filtered by --no-firewall)", result.Run.DatabaseID))) + } + return true + } + } + + // Apply safe output type filtering if --safe-output flag is specified. + if opts.safeOutputType != "" { + hasSafeOutputType, checkErr := runContainsSafeOutputType(result.LogsPath, opts.safeOutputType, verbose) + if checkErr != nil && verbose { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check safe output type for run %d: %v", result.Run.DatabaseID, checkErr))) + } + if !hasSafeOutputType { + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no '%s' safe output messages found", result.Run.DatabaseID, opts.safeOutputType))) + } + return true + } + } + + // Apply filtered-integrity filtering if --filtered-integrity flag is specified. + if opts.filteredIntegrity { + hasFiltered, checkErr := runHasDifcFilteredItems(result.LogsPath, verbose) + if checkErr != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to check DIFC filtered items for run %d: %v", result.Run.DatabaseID, checkErr))) + return true + } + if !hasFiltered { + logsOrchestratorLog.Printf("Skipping run %d: no DIFC filtered items found", result.Run.DatabaseID) + if verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Skipping run %d: no DIFC integrity-filtered items found in gateway logs", result.Run.DatabaseID))) + } + return true + } + } + + return false +} + +// buildProcessedRun constructs a ProcessedRun from a DownloadResult, computing +// duration, action minutes, effective tokens, and job-failure counts. +func buildProcessedRun(result DownloadResult, verbose, logFailedJobs bool) ProcessedRun { + run := result.Run + run.TokenUsage = result.Metrics.TokenUsage + applyMetricsTurnsToRun(&run, result.Metrics) + run.AvgTimeBetweenTurns = result.Metrics.AvgTimeBetweenTurns + run.ErrorCount = 0 + run.WarningCount = 0 + run.LogsPath = result.LogsPath + + // Propagate effective tokens from cached firewall proxy summary when available. + if result.TokenUsage != nil && result.TokenUsage.TotalEffectiveTokens > 0 { + run.EffectiveTokens = result.TokenUsage.TotalEffectiveTokens + } + + // Add failed jobs to error count. + if failedJobCount, err := fetchJobStatusesForProcessedRun(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))) + } + } + + // Always use GitHub API timestamps for duration calculation. + // GitHub Actions bills per minute, rounded up per job. + if !run.StartedAt.IsZero() && !run.UpdatedAt.IsZero() { + run.Duration = run.UpdatedAt.Sub(run.StartedAt) + run.ActionMinutes = math.Ceil(run.Duration.Minutes()) + } + + return ProcessedRun{ + Run: run, + AwContext: result.AwContext, + TaskDomain: result.TaskDomain, + BehaviorFingerprint: result.BehaviorFingerprint, + AgenticAssessments: result.AgenticAssessments, + AccessAnalysis: result.AccessAnalysis, + FirewallAnalysis: result.FirewallAnalysis, + RedactedDomainsAnalysis: result.RedactedDomainsAnalysis, + MissingTools: result.MissingTools, + MissingData: result.MissingData, + Noops: result.Noops, + MCPFailures: result.MCPFailures, + MCPToolUsage: result.MCPToolUsage, + TokenUsage: result.TokenUsage, + GitHubRateLimitUsage: result.GitHubRateLimitUsage, + JobDetails: result.JobDetails, + } +} diff --git a/pkg/cli/logs_orchestrator_filters_test.go b/pkg/cli/logs_orchestrator_filters_test.go new file mode 100644 index 00000000000..4aa1dd42650 --- /dev/null +++ b/pkg/cli/logs_orchestrator_filters_test.go @@ -0,0 +1,304 @@ +//go:build !integration + +package cli + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func stubFetchJobStatusesForProcessedRun(t *testing.T, fn func(int64, bool) (int, error)) { + t.Helper() + previous := fetchJobStatusesForProcessedRun + fetchJobStatusesForProcessedRun = fn + t.Cleanup(func() { + fetchJobStatusesForProcessedRun = previous + }) +} + +// makeDownloadResult creates a DownloadResult pointing at a temporary directory +// that optionally contains an aw_info.json file. +func makeDownloadResult(t *testing.T, awInfoJSON string) DownloadResult { + t.Helper() + tmpDir := t.TempDir() + if awInfoJSON != "" { + require.NoError(t, os.WriteFile(filepath.Join(tmpDir, "aw_info.json"), []byte(awInfoJSON), 0644)) + } + return DownloadResult{ + Run: WorkflowRun{DatabaseID: 42}, + LogsPath: tmpDir, + } +} + +// TestApplyRunFilters_NoFilters verifies that an empty runFilterOpts always +// passes every run through (returns false = do not skip). +func TestApplyRunFilters_NoFilters(t *testing.T) { + result := makeDownloadResult(t, `{"engine_id":"claude"}`) + skip := applyRunFilters(result, runFilterOpts{}, false) + assert.False(t, skip, "no filters should never skip a run") +} + +// TestApplyRunFilters_Engine exercises both the matching and non-matching engine cases. +func TestApplyRunFilters_Engine(t *testing.T) { + tests := []struct { + name string + awInfo string + filterEngine string + wantSkip bool + }{ + { + name: "matching engine passes", + awInfo: `{"engine_id":"claude"}`, + filterEngine: "claude", + wantSkip: false, + }, + { + name: "non-matching engine skipped", + awInfo: `{"engine_id":"copilot"}`, + filterEngine: "claude", + wantSkip: true, + }, + { + name: "missing aw_info skipped", + awInfo: "", // no file + filterEngine: "claude", + wantSkip: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := makeDownloadResult(t, tt.awInfo) + skip := applyRunFilters(result, runFilterOpts{engine: tt.filterEngine}, false) + assert.Equal(t, tt.wantSkip, skip) + }) + } +} + +// TestApplyRunFilters_NoStaged verifies that --no-staged skips staged runs. +func TestApplyRunFilters_NoStaged(t *testing.T) { + tests := []struct { + name string + awInfo string + wantSkip bool + }{ + { + name: "staged run is skipped", + awInfo: `{"staged":true}`, + wantSkip: true, + }, + { + name: "non-staged run passes", + awInfo: `{"staged":false}`, + wantSkip: false, + }, + { + name: "missing staged field treated as false", + awInfo: `{}`, + wantSkip: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := makeDownloadResult(t, tt.awInfo) + skip := applyRunFilters(result, runFilterOpts{noStaged: true}, false) + assert.Equal(t, tt.wantSkip, skip) + }) + } +} + +// TestApplyRunFilters_Firewall verifies both --firewall and --no-firewall. +func TestApplyRunFilters_Firewall(t *testing.T) { + withFirewall := `{"steps":{"firewall":"squid"}}` + noFirewall := `{"steps":{}}` + + tests := []struct { + name string + awInfo string + firewallOnly bool + noFirewall bool + wantSkip bool + }{ + { + name: "firewallOnly: run with firewall passes", + awInfo: withFirewall, + firewallOnly: true, + wantSkip: false, + }, + { + name: "firewallOnly: run without firewall skipped", + awInfo: noFirewall, + firewallOnly: true, + wantSkip: true, + }, + { + name: "noFirewall: run without firewall passes", + awInfo: noFirewall, + noFirewall: true, + wantSkip: false, + }, + { + name: "noFirewall: run with firewall skipped", + awInfo: withFirewall, + noFirewall: true, + wantSkip: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := makeDownloadResult(t, tt.awInfo) + skip := applyRunFilters(result, runFilterOpts{firewallOnly: tt.firewallOnly, noFirewall: tt.noFirewall}, false) + assert.Equal(t, tt.wantSkip, skip) + }) + } +} + +// TestApplyRunFilters_SafeOutputType verifies safe-output-type filtering. +// When no agent_output.json is present, the run is skipped. +func TestApplyRunFilters_SafeOutputType(t *testing.T) { + t.Run("no agent_output.json skips run", func(t *testing.T) { + result := makeDownloadResult(t, "") // no aw_info.json either + skip := applyRunFilters(result, runFilterOpts{safeOutputType: "create-issue"}, false) + assert.True(t, skip) + }) + + t.Run("matching safe output type passes", func(t *testing.T) { + tmpDir := t.TempDir() + agentOutput := `{"items":[{"type":"create-issue"}]}` + require.NoError(t, os.WriteFile(filepath.Join(tmpDir, "agent_output.json"), []byte(agentOutput), 0644)) + result := DownloadResult{Run: WorkflowRun{DatabaseID: 1}, LogsPath: tmpDir} + skip := applyRunFilters(result, runFilterOpts{safeOutputType: "create-issue"}, false) + assert.False(t, skip) + }) + + t.Run("non-matching safe output type skips run", func(t *testing.T) { + tmpDir := t.TempDir() + agentOutput := `{"items":[{"type":"add-comment"}]}` + require.NoError(t, os.WriteFile(filepath.Join(tmpDir, "agent_output.json"), []byte(agentOutput), 0644)) + result := DownloadResult{Run: WorkflowRun{DatabaseID: 2}, LogsPath: tmpDir} + skip := applyRunFilters(result, runFilterOpts{safeOutputType: "create-issue"}, false) + assert.True(t, skip) + }) +} + +func TestApplyRunFilters_FilteredIntegrity(t *testing.T) { + t.Run("gateway log with DIFC filtered event passes", func(t *testing.T) { + tmpDir := t.TempDir() + gatewayLog := `{"timestamp":"2025-01-01T00:00:00Z","type":"DIFC_FILTERED","server_id":"github","tool_name":"create_issue","reason":"integrity"}` + "\n" + require.NoError(t, os.WriteFile(filepath.Join(tmpDir, "gateway.jsonl"), []byte(gatewayLog), 0644)) + result := DownloadResult{Run: WorkflowRun{DatabaseID: 5}, LogsPath: tmpDir} + + skip := applyRunFilters(result, runFilterOpts{filteredIntegrity: true}, false) + + assert.False(t, skip) + }) + + t.Run("missing gateway logs are skipped", func(t *testing.T) { + result := makeDownloadResult(t, "") + + skip := applyRunFilters(result, runFilterOpts{filteredIntegrity: true}, false) + + assert.True(t, skip) + }) +} + +// TestBuildProcessedRun verifies that buildProcessedRun correctly populates +// 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 }) + now := time.Now() + tmpDir := t.TempDir() + awCtx := &AwContext{Repo: "owner/repo"} + result := DownloadResult{ + Run: WorkflowRun{ + DatabaseID: 1234, + StartedAt: now.Add(-5 * time.Minute), + UpdatedAt: now, + }, + LogsPath: tmpDir, + AwContext: awCtx, + } + + pr := buildProcessedRun(result, false, false) + + assert.Equal(t, int64(1234), pr.Run.DatabaseID) + assert.Equal(t, tmpDir, pr.Run.LogsPath) + assert.Equal(t, awCtx, pr.AwContext) + assert.Equal(t, 0, pr.Run.ErrorCount) + assert.Equal(t, 0, pr.Run.WarningCount) + }) + + t.Run("duration and action minutes are computed", func(t *testing.T) { + stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil }) + base := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + result := DownloadResult{ + Run: WorkflowRun{ + DatabaseID: 999, + StartedAt: base, + UpdatedAt: base.Add(90 * time.Second), // 1.5 minutes + }, + LogsPath: t.TempDir(), + } + + pr := buildProcessedRun(result, false, false) + + assert.Equal(t, 90*time.Second, pr.Run.Duration) + assert.InDelta(t, 2.0, pr.Run.ActionMinutes, 0.001) // ceil(1.5) = 2 + }) + + t.Run("zero timestamps leave duration unset", func(t *testing.T) { + stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil }) + result := DownloadResult{ + Run: WorkflowRun{DatabaseID: 7}, + LogsPath: t.TempDir(), + } + pr := buildProcessedRun(result, false, false) + assert.Equal(t, time.Duration(0), pr.Run.Duration) + assert.InDelta(t, 0.0, pr.Run.ActionMinutes, 0.001) + }) + + t.Run("effective tokens are propagated", func(t *testing.T) { + stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil }) + usage := &TokenUsageSummary{TotalEffectiveTokens: 5000} + result := DownloadResult{ + Run: WorkflowRun{DatabaseID: 3}, + LogsPath: t.TempDir(), + TokenUsage: usage, + } + pr := buildProcessedRun(result, false, false) + assert.Equal(t, 5000, pr.Run.EffectiveTokens) + }) + + t.Run("zero effective tokens not propagated", func(t *testing.T) { + stubFetchJobStatusesForProcessedRun(t, func(int64, bool) (int, error) { return 0, nil }) + usage := &TokenUsageSummary{TotalEffectiveTokens: 0} + result := DownloadResult{ + Run: WorkflowRun{DatabaseID: 4}, + LogsPath: t.TempDir(), + TokenUsage: usage, + } + pr := buildProcessedRun(result, false, false) + assert.Equal(t, 0, pr.Run.EffectiveTokens) + }) + + t.Run("failed job count is added via test seam", func(t *testing.T) { + stubFetchJobStatusesForProcessedRun(t, func(runID int64, verbose bool) (int, error) { + assert.Equal(t, int64(88), runID) + assert.False(t, verbose) + return 2, nil + }) + result := DownloadResult{ + Run: WorkflowRun{DatabaseID: 88}, + LogsPath: t.TempDir(), + } + + pr := buildProcessedRun(result, false, false) + + assert.Equal(t, 2, pr.Run.ErrorCount) + }) +} diff --git a/pkg/cli/logs_orchestrator_render.go b/pkg/cli/logs_orchestrator_render.go new file mode 100644 index 00000000000..5bd5a7e37ba --- /dev/null +++ b/pkg/cli/logs_orchestrator_render.go @@ -0,0 +1,155 @@ +// This file provides command-line interface functionality for gh-aw. +// This file (logs_orchestrator_render.go) contains output-rendering helpers for the +// logs orchestrator: transforming []ProcessedRun into console, JSON, markdown, +// TSV, or "pretty" cross-run audit output. + +package cli + +import ( + "fmt" + "os" + "path/filepath" + + "github.com/github/gh-aw/pkg/constants" +) + +// renderLogsOutput finalizes processedRuns and renders them in the appropriate output +// format: JSON, console metrics table, or cross-run audit report (pretty/markdown). +// continuation is optional and only set when a timeout was reached during a paginated download. +func renderLogsOutput(processedRuns []ProcessedRun, opts renderLogsOutputOptions) error { + // Update MissingToolCount, MissingDataCount, and NoopCount in runs + for i := range processedRuns { + processedRuns[i].Run.MissingToolCount = len(processedRuns[i].MissingTools) + processedRuns[i].Run.MissingDataCount = len(processedRuns[i].MissingData) + processedRuns[i].Run.NoopCount = len(processedRuns[i].Noops) + } + + // Build structured logs data + logsOrchestratorLog.Printf("Building logs data from %d processed runs (continuation=%t)", len(processedRuns), opts.continuation != nil) + logsData := buildLogsData(processedRuns, opts.outputDir, opts.continuation) + + // When only the usage artifact was downloaded, add a hint so consumers know how + // to fetch additional artifact sets (agent logs, firewall data, etc.). + if isUsageOnlyArtifactFilter(opts.artifactFilter) { + logsData.Message = usageOnlyArtifactHintMessage() + } + + // Write summary file if requested (default behavior unless disabled with empty string) + if opts.summaryFile != "" { + summaryPath := filepath.Join(opts.outputDir, opts.summaryFile) + if err := writeSummaryFile(summaryPath, logsData, opts.verbose); err != nil { + return fmt.Errorf("failed to write summary file: %w", err) + } + } + + // Train drain3 weights if requested. + if opts.train { + if err := TrainDrain3Weights(processedRuns, opts.outputDir, opts.verbose); err != nil { + return fmt.Errorf("log pattern training: %w", err) + } + } + + // Render output based on format preference. + switch opts.format { + case "tsv": + if opts.verbose { + renderLogsTSVVerbose(logsData) + } else { + renderLogsTSV(logsData) + } + renderLogsArtifactHint(os.Stderr, logsData.Message) + return nil + + case "markdown", "pretty": + inputs := make([]crossRunInput, 0, len(processedRuns)) + for _, pr := range processedRuns { + inputs = append(inputs, crossRunInput{ + RunID: pr.Run.DatabaseID, + WorkflowName: pr.Run.WorkflowName, + Conclusion: pr.Run.Conclusion, + Duration: pr.Run.Duration, + FirewallAnalysis: pr.FirewallAnalysis, + Metrics: LogMetrics{ + TokenUsage: pr.Run.TokenUsage, + Turns: pr.Run.Turns, + }, + MCPToolUsage: pr.MCPToolUsage, + MCPFailures: pr.MCPFailures, + ErrorCount: pr.Run.ErrorCount, + }) + } + report := buildCrossRunAuditReport(inputs) + if opts.jsonOutput { + return renderCrossRunReportJSON(report) + } + if opts.format == "pretty" { + renderCrossRunReportPretty(report) + renderLogsArtifactHint(os.Stderr, logsData.Message) + return nil + } + if opts.reportFile != "" { + if err := os.MkdirAll(filepath.Dir(opts.reportFile), constants.DirPermPublic); err != nil { + return fmt.Errorf("failed to create report file directory: %w", err) + } + f, err := os.Create(opts.reportFile) + if err != nil { + return fmt.Errorf("failed to create report file: %w", err) + } + if err := func() (retErr error) { + defer func() { + if cerr := f.Close(); cerr != nil && retErr == nil { + retErr = cerr + } + }() + renderCrossRunReportMarkdownToWriter(f, report) + return nil + }(); err != nil { + return fmt.Errorf("failed to write report file: %w", err) + } + } else { + renderCrossRunReportMarkdown(report) + } + renderLogsArtifactHint(os.Stderr, logsData.Message) + return nil + + case "console": + // Explicit console format: decorated tables for human reading + if opts.jsonOutput { + if err := renderLogsJSON(logsData, opts.verbose); err != nil { + return fmt.Errorf("failed to render JSON output: %w", err) + } + } else { + renderLogsConsole(logsData) + displayAggregatedGatewayMetrics(processedRuns, opts.outputDir, opts.verbose) + displayUnifiedTimeline(processedRuns, opts.verbose) + if opts.toolGraph { + generateToolGraph(processedRuns, opts.verbose) + } + renderLogsArtifactHint(os.Stderr, logsData.Message) + } + return nil + } + + // Default: compact format optimised for agentic consumption + if opts.jsonOutput { + if err := renderLogsJSON(logsData, opts.verbose); err != nil { + return fmt.Errorf("failed to render JSON output: %w", err) + } + } else { + if opts.verbose { + renderLogsCompactVerbose(logsData) + } else { + renderLogsCompact(logsData) + } + } + + return nil +} + +// renderLogsArtifactHint writes a [hint] line to w when message is non-empty. +func renderLogsArtifactHint(w *os.File, message string) { + if message == "" { + return + } + fmt.Fprintf(w, "[hint] %s\n", message) +} diff --git a/pkg/cli/logs_orchestrator_stdin.go b/pkg/cli/logs_orchestrator_stdin.go new file mode 100644 index 00000000000..8eb5b030b61 --- /dev/null +++ b/pkg/cli/logs_orchestrator_stdin.go @@ -0,0 +1,229 @@ +// This file provides command-line interface functionality for gh-aw. +// This file (logs_orchestrator_stdin.go) contains DownloadWorkflowLogsFromStdin, +// the stdin-driven run-discovery and processing path used when --stdin is passed +// to the logs command. + +package cli + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "github.com/github/gh-aw/pkg/console" + "github.com/github/gh-aw/pkg/fileutil" + "github.com/github/gh-aw/pkg/parser" +) + +// DownloadWorkflowLogsFromStdin fetches and processes workflow run logs for runs +// provided as IDs or URLs, bypassing the GitHub API run-discovery step. +// This is used when the --stdin flag is passed to the logs command. +func DownloadWorkflowLogsFromStdin(ctx context.Context, opts StdinLogsOptions) error { + logsOrchestratorLog.Printf("Starting stdin log download: runs=%d, outputDir=%s", len(opts.RunURLs), opts.OutputDir) + + if err := ValidateArtifactSets(opts.ArtifactSets); err != nil { + return err + } + artifactFilter := ResolveArtifactFilter(opts.ArtifactSets) + if len(artifactFilter) > 0 { + logsOrchestratorLog.Printf("Artifact filter active: %v", artifactFilter) + if opts.Verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage("Artifact filter: downloading only "+strings.Join(artifactFilter, ", "))) + } + } + + if err := ensureLogsGitignore(); err != nil { + logsOrchestratorLog.Printf("Failed to ensure logs .gitignore: %v", err) + if opts.Verbose { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to ensure .github/aw/logs/.gitignore: %v", err))) + } + } + + select { + case <-ctx.Done(): + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Operation cancelled")) + return ctx.Err() + default: + } + + if len(opts.RunURLs) == 0 { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No run IDs or URLs provided on stdin")) + return nil + } + + // Parse owner/repo (and optional GHES host) from --repo override if provided. + // Accepted formats: "owner/repo" or "HOST/owner/repo". + var hostOverride, ownerOverride, repoNameOverride string + if opts.RepoOverride != "" { + parts := strings.SplitN(opts.RepoOverride, "/", 3) + switch len(parts) { + case 3: // HOST/owner/repo + if parts[0] == "" || parts[1] == "" || parts[2] == "" { + return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) + } + hostOverride, ownerOverride, repoNameOverride = parts[0], parts[1], parts[2] + case 2: // owner/repo + if parts[0] == "" || parts[1] == "" { + return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) + } + ownerOverride, repoNameOverride = parts[0], parts[1] + default: + return fmt.Errorf("invalid repository format '%s': expected '[HOST/]owner/repo'", opts.RepoOverride) + } + } + + // Start timeout timer if specified. + var startTime time.Time + if opts.Timeout > 0 { + startTime = time.Now() + if opts.Verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Timeout set to %d minutes", opts.Timeout))) + } + } + + if opts.Verbose { + fmt.Fprintln(os.Stderr, console.FormatInfoMessage(fmt.Sprintf("Fetching metadata for %d runs from stdin...", len(opts.RunURLs)))) + } + + // Build WorkflowRun objects by fetching metadata for each provided URL. + var runs []WorkflowRun + for _, rawURL := range opts.RunURLs { + select { + case <-ctx.Done(): + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Operation cancelled")) + return ctx.Err() + default: + } + + if opts.Timeout > 0 && time.Since(startTime).Seconds() >= float64(opts.Timeout)*60 { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("Timeout reached before all run metadata could be fetched")) + break + } + + components, err := parser.ParseRunURLExtended(rawURL) + if err != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping invalid run %q: %v", rawURL, err))) + continue + } + + // Prefer owner/repo embedded in the URL; fall back to --repo override. + // If neither source provides owner, the run cannot be fetched — return an + // actionable error rather than silently continuing with a broken API call. + owner := components.Owner + repo := components.Repo + host := components.Host + if owner == "" { + owner = ownerOverride + repo = repoNameOverride + if host == "" { + host = hostOverride + } + } + if owner == "" { + return fmt.Errorf("run %q does not include repository information; pass --repo owner/repo or provide full run URLs", rawURL) + } + + run, err := fetchWorkflowRunMetadata(ctx, components.Number, owner, repo, host, opts.Verbose) + if err != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping run %d: failed to fetch metadata: %v", components.Number, err))) + continue + } + runs = append(runs, run) + } + + if len(runs) == 0 { + if opts.JSONOutput { + logsData := buildLogsData([]ProcessedRun{}, opts.OutputDir, nil) + logsData.Message = "No runs found. No valid runs could be loaded from the provided input." + if err := renderLogsJSON(logsData, opts.Verbose); err != nil { + return fmt.Errorf("failed to render JSON output: %w", err) + } + } + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No valid runs could be loaded from stdin")) + return nil + } + + // Download artifacts for all runs concurrently. + downloadResults := downloadRunArtifactsConcurrent(ctx, runs, opts.OutputDir, opts.Verbose, len(runs), opts.RepoOverride, artifactFilter) + + filters := runFilterOpts{ + engine: opts.Engine, + noStaged: opts.NoStaged, + firewallOnly: opts.FirewallOnly, + noFirewall: opts.NoFirewall, + safeOutputType: opts.SafeOutputType, + filteredIntegrity: opts.FilteredIntegrity, + } + + // Process download results applying the same filters as DownloadWorkflowLogs. + var processedRuns []ProcessedRun + for _, result := range downloadResults { + if result.Skipped { + if opts.Verbose && result.Error != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Skipping run %d: %v", result.Run.DatabaseID, result.Error))) + } + continue + } + + if result.Error != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to download artifacts for run %d: %v", result.Run.DatabaseID, result.Error))) + continue + } + + if applyRunFilters(result, filters, opts.Verbose) { + continue + } + + processedRun := buildProcessedRun(result, opts.Verbose, false) + + if opts.Parse { + awInfoPath := filepath.Join(result.LogsPath, "aw_info.json") + detectedEngine := extractEngineFromAwInfo(awInfoPath, opts.Verbose) + if err := parseAgentLog(result.LogsPath, detectedEngine, opts.Verbose); err != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse log for run %d: %v", processedRun.Run.DatabaseID, err))) + } else { + logMdPath := filepath.Join(result.LogsPath, "log.md") + if fileutil.FileExists(logMdPath) { + fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed log for run %d → %s", processedRun.Run.DatabaseID, logMdPath))) + } + } + if err := parseFirewallLogs(result.LogsPath, opts.Verbose); err != nil { + fmt.Fprintln(os.Stderr, console.FormatWarningMessage(fmt.Sprintf("Failed to parse firewall logs for run %d: %v", processedRun.Run.DatabaseID, err))) + } else { + firewallMdPath := filepath.Join(result.LogsPath, "firewall.md") + if fileutil.FileExists(firewallMdPath) { + fmt.Fprintln(os.Stderr, console.FormatSuccessMessage(fmt.Sprintf("✓ Parsed firewall logs for run %d → %s", processedRun.Run.DatabaseID, firewallMdPath))) + } + } + } + + processedRuns = append(processedRuns, processedRun) + } + + if len(processedRuns) == 0 { + if opts.JSONOutput { + logsData := buildLogsData([]ProcessedRun{}, opts.OutputDir, nil) + logsData.Message = "No runs found matching the specified criteria." + if err := renderLogsJSON(logsData, opts.Verbose); err != nil { + return fmt.Errorf("failed to render JSON output: %w", err) + } + } + fmt.Fprintln(os.Stderr, console.FormatWarningMessage("No workflow runs with artifacts found matching the specified criteria")) + return nil + } + + return renderLogsOutput(processedRuns, renderLogsOutputOptions{ + outputDir: opts.OutputDir, + summaryFile: opts.SummaryFile, + format: opts.Format, + reportFile: opts.ReportFile, + jsonOutput: opts.JSONOutput, + toolGraph: opts.ToolGraph, + train: opts.Train, + verbose: opts.Verbose, + artifactFilter: artifactFilter, + }) +} diff --git a/pkg/cli/logs_orchestrator_types.go b/pkg/cli/logs_orchestrator_types.go new file mode 100644 index 00000000000..2bcc8dcf60d --- /dev/null +++ b/pkg/cli/logs_orchestrator_types.go @@ -0,0 +1,87 @@ +// This file provides command-line interface functionality for gh-aw. +// This file (logs_orchestrator_types.go) contains type definitions used by the +// logs orchestrator: option structs for public API functions and internal helper types. + +package cli + +// LogsDownloadOptions holds parameters for DownloadWorkflowLogs. +type LogsDownloadOptions struct { + WorkflowName string + Count int + StartDate string + EndDate string + OutputDir string + Engine string + Ref string + BeforeRunID int64 + AfterRunID int64 + RepoOverride string + Verbose bool + ToolGraph bool + NoStaged bool + FirewallOnly bool + NoFirewall bool + Parse bool + JSONOutput bool + TimeoutMinutes int + SummaryFile string + SafeOutputType string + FilteredIntegrity bool + Train bool + Format string + ArtifactSets []string + After string + ReportFile string +} + +// StdinLogsOptions holds parameters for DownloadWorkflowLogsFromStdin. +type StdinLogsOptions struct { + RunURLs []string + OutputDir string + Engine string + RepoOverride string + Verbose bool + ToolGraph bool + NoStaged bool + FirewallOnly bool + NoFirewall bool + Parse bool + JSONOutput bool + Timeout int + SummaryFile string + SafeOutputType string + FilteredIntegrity bool + Train bool + Format string + ReportFile string + // ArtifactSets defaults to nil (download all artifacts) when this API is used + // programmatically. The CLI passes ["usage"] to match the logs command default. + ArtifactSets []string +} + +// continuationOptions carries the parameters needed by buildContinuationIfNeeded +// to produce a ContinuationData cursor for paginated log fetches. +type continuationOptions struct { + workflowName string + startDate string + endDate string + engine string + branch string + afterRunID int64 + count int + timeoutMinutes int +} + +// renderLogsOutputOptions holds configuration for renderLogsOutput. +type renderLogsOutputOptions struct { + outputDir string + summaryFile string + format string + reportFile string + jsonOutput bool + toolGraph bool + train bool + continuation *ContinuationData + verbose bool + artifactFilter []string +}