feat(ci): use job selector for historical logs

pkg/api/ci.go

@@ -114,12 +114,34 @@ func CIGetJobSummary(ctx context.Context, token, orgID string, req *civ1.GetJobS
 
 // CIGetJobAttemptLogs returns all log lines for a job attempt, paginating through all pages.
 func CIGetJobAttemptLogs(ctx context.Context, token, orgID, attemptID string) ([]*civ1.LogLine, error) {
+	return CIGetJobAttemptLogsForTarget(ctx, token, orgID, CILogStreamTarget{AttemptID: attemptID})
+}
+
+type CILogStreamTarget struct {
+	AttemptID string
+	JobID     string
+}
+
+// CIGetJobAttemptLogsForTarget returns all log lines for a job attempt or the
+// latest attempt of a job, paginating through all pages.
+func CIGetJobAttemptLogsForTarget(ctx context.Context, token, orgID string, target CILogStreamTarget) ([]*civ1.LogLine, error) {
+	if target.AttemptID == "" && target.JobID == "" {
+		return nil, fmt.Errorf("exactly one of attempt ID or job ID is required")
+	}
+	if target.AttemptID != "" && target.JobID != "" {
+		return nil, fmt.Errorf("exactly one of attempt ID or job ID is required")
+	}
+
 	client := newCIServiceClient()
 	var allLines []*civ1.LogLine
 	var pageToken string
 
 	for {
-		req := &civ1.GetJobAttemptLogsRequest{AttemptId: attemptID, PageToken: pageToken}
+		req := &civ1.GetJobAttemptLogsRequest{
+			AttemptId: target.AttemptID,
+			JobId:     optionalString(target.JobID),
+			PageToken: pageToken,
+		}
 		resp, err := client.GetJobAttemptLogs(ctx, WithAuthenticationAndOrg(connect.NewRequest(req), token, orgID))
 		if err != nil {
 			return nil, err
@@ -134,11 +156,6 @@ func CIGetJobAttemptLogs(ctx context.Context, token, orgID, attemptID string) ([
 	return allLines, nil
 }
 
-type CILogStreamTarget struct {
-	AttemptID string
-	JobID     string
-}
-
 // CIExportJobAttemptLogs streams a finite exported log snapshot for a job
 // attempt or the latest attempt of a job into w. The returned metadata is
 // advisory; callers choose their own destination path.
@@ -551,9 +568,9 @@ func CIListArtifacts(ctx context.Context, token, orgID, runID string, options CI
 	for {
 		req := &civ1.ListArtifactsRequest{
 			RunId:      runID,
-			WorkflowId: optionalArtifactFilter(options.WorkflowID),
-			JobId:      optionalArtifactFilter(options.JobID),
-			AttemptId:  optionalArtifactFilter(options.AttemptID),
+			WorkflowId: optionalString(options.WorkflowID),
+			JobId:      optionalString(options.JobID),
+			AttemptId:  optionalString(options.AttemptID),
 			PageSize:   500,
 			PageToken:  pageToken,
 		}
@@ -572,7 +589,7 @@ func CIListArtifacts(ctx context.Context, token, orgID, runID string, options CI
 	return artifacts, nil
 }
 
-func optionalArtifactFilter(value string) *string {
+func optionalString(value string) *string {
 	if value == "" {
 		return nil
 	}

pkg/api/ci_test.go

@@ -363,6 +363,94 @@ func TestCIArtifactsWrappers(t *testing.T) {
 	})
 }
 
+type getLogsTargetRecorder struct {
+	civ1connect.UnimplementedCIServiceHandler
+	t        *testing.T
+	want     CILogStreamTarget
+	requests []*civ1.GetJobAttemptLogsRequest
+}
+
+func (r *getLogsTargetRecorder) GetJobAttemptLogs(_ context.Context, req *connect.Request[civ1.GetJobAttemptLogsRequest]) (*connect.Response[civ1.GetJobAttemptLogsResponse], error) {
+	assertAuthAndOrg(r.t, req.Header())
+	r.requests = append(r.requests, proto.Clone(req.Msg).(*civ1.GetJobAttemptLogsRequest))
+
+	if got := req.Msg.GetAttemptId(); got != r.want.AttemptID {
+		r.t.Fatalf("AttemptId = %q, want %q", got, r.want.AttemptID)
+	}
+	if got := req.Msg.GetJobId(); got != r.want.JobID {
+		r.t.Fatalf("JobId = %q, want %q", got, r.want.JobID)
+	}
+	if r.want.JobID == "" && req.Msg.JobId != nil {
+		r.t.Fatalf("JobId pointer = %q, want nil", req.Msg.GetJobId())
+	}
+	if r.want.JobID != "" && req.Msg.JobId == nil {
+		r.t.Fatalf("JobId pointer = nil, want %q", r.want.JobID)
+	}
+
+	switch req.Msg.GetPageToken() {
+	case "":
+		return connect.NewResponse(&civ1.GetJobAttemptLogsResponse{
+			Lines:         []*civ1.LogLine{testLogLine("step-1", 1, "first")},
+			NextPageToken: "next",
+		}), nil
+	case "next":
+		return connect.NewResponse(&civ1.GetJobAttemptLogsResponse{
+			Lines: []*civ1.LogLine{testLogLine("step-1", 2, "second")},
+		}), nil
+	default:
+		r.t.Fatalf("PageToken = %q, want empty or next", req.Msg.GetPageToken())
+		return nil, nil
+	}
+}
+
+func TestCIGetJobAttemptLogsForTargetSendsSingleSelector(t *testing.T) {
+	tests := []struct {
+		name   string
+		target CILogStreamTarget
+	}{
+		{name: "attempt", target: CILogStreamTarget{AttemptID: "attempt-123"}},
+		{name: "job", target: CILogStreamTarget{JobID: "job-123"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			recorder := &getLogsTargetRecorder{t: t, want: tt.target}
+			_, handler := civ1connect.NewCIServiceHandler(recorder)
+			server := httptest.NewServer(h2c.NewHandler(handler, &http2.Server{}))
+			t.Cleanup(server.Close)
+
+			originalBaseURLFunc := baseURLFunc
+			baseURLFunc = func() string { return server.URL }
+			t.Cleanup(func() { baseURLFunc = originalBaseURLFunc })
+
+			lines, err := CIGetJobAttemptLogsForTarget(context.Background(), "token-123", "org-123", tt.target)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got, want := len(lines), 2; got != want {
+				t.Fatalf("lines = %d, want %d", got, want)
+			}
+			if got := lines[0].GetBody(); got != "first" {
+				t.Fatalf("first body = %q, want first", got)
+			}
+			if got := lines[1].GetBody(); got != "second" {
+				t.Fatalf("second body = %q, want second", got)
+			}
+			if got, want := len(recorder.requests), 2; got != want {
+				t.Fatalf("requests = %d, want %d", got, want)
+			}
+			for i, req := range recorder.requests {
+				if got := req.GetAttemptId(); got != tt.target.AttemptID {
+					t.Fatalf("request %d AttemptId = %q, want %q", i, got, tt.target.AttemptID)
+				}
+				if got := req.GetJobId(); got != tt.target.JobID {
+					t.Fatalf("request %d JobId = %q, want %q", i, got, tt.target.JobID)
+				}
+			}
+		})
+	}
+}
+
 func withTestCIService(t *testing.T, fn func()) {
 	t.Helper()
 

pkg/cmd/ci/logs.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	"connectrpc.com/connect"
 	"github.com/briandowns/spinner"
 	"github.com/depot/cli/pkg/api"
 	"github.com/depot/cli/pkg/config"
@@ -29,11 +30,12 @@ const (
 )
 
 var (
-	ciGetRunStatus             = api.CIGetRunStatus
-	ciGetJobAttemptLogs        = api.CIGetJobAttemptLogs
-	ciExportJobAttemptLogs     = api.CIExportJobAttemptLogs
-	ciStreamJobAttemptLogs     = api.CIStreamJobAttemptLogs
-	ciStreamJobAttemptLogLines = api.CIStreamJobAttemptLogLines
+	ciGetRunStatus               = api.CIGetRunStatus
+	ciGetJobAttemptLogs          = api.CIGetJobAttemptLogs
+	ciGetJobAttemptLogsForTarget = api.CIGetJobAttemptLogsForTarget
+	ciExportJobAttemptLogs       = api.CIExportJobAttemptLogs
+	ciStreamJobAttemptLogs       = api.CIStreamJobAttemptLogs
+	ciStreamJobAttemptLogLines   = api.CIStreamJobAttemptLogLines
 )
 
 func NewCmdLogs() *cobra.Command {
@@ -201,10 +203,18 @@ ID, use --job and --workflow to disambiguate by workflow job key.`,
 					return fmt.Errorf("failed to stream logs: %w", err)
 				}
 			} else {
-				lines, err := ciGetJobAttemptLogs(ctx, tokenVal, orgID, id)
+				lines, err := getUnresolvedHistoricalLogs(ctx, tokenVal, orgID, id)
 				if err != nil {
-					// Both paths failed — show both errors so the user can
-					// distinguish "bad ID" from "auth/network failure".
+					if unresolvedErr, ok := err.(*unresolvedHistoricalLogsError); ok {
+						return fmt.Errorf(
+							"could not resolve %q as a run, job, or attempt ID:\n  as run/job: %v\n  as attempt: %v\n  as job: %v",
+							id,
+							runErr,
+							unresolvedErr.attemptErr,
+							unresolvedErr.jobErr,
+						)
+					}
+
 					return fmt.Errorf(
 						"could not resolve %q as a run, job, or attempt ID:\n  as run/job: %v\n  as attempt: %v",
 						id,
@@ -453,6 +463,32 @@ func exportUnresolvedLogsToFile(ctx context.Context, tokenVal string, orgID stri
 	return fmt.Errorf("as job: %v; as attempt: %v", jobErr, attemptErr)
 }
 
+type unresolvedHistoricalLogsError struct {
+	attemptErr error
+	jobErr     error
+}
+
+func (e *unresolvedHistoricalLogsError) Error() string {
+	return fmt.Sprintf("as attempt: %v; as job: %v", e.attemptErr, e.jobErr)
+}
+
+func getUnresolvedHistoricalLogs(ctx context.Context, tokenVal string, orgID string, id string) ([]*civ1.LogLine, error) {
+	lines, attemptErr := ciGetJobAttemptLogs(ctx, tokenVal, orgID, id)
+	if attemptErr == nil || isContextDoneError(attemptErr) {
+		return lines, attemptErr
+	}
+	if connect.CodeOf(attemptErr) != connect.CodeNotFound {
+		return nil, attemptErr
+	}
+
+	lines, jobErr := ciGetJobAttemptLogsForTarget(ctx, tokenVal, orgID, api.CILogStreamTarget{JobID: id})
+	if jobErr == nil || isContextDoneError(jobErr) {
+		return lines, jobErr
+	}
+
+	return nil, &unresolvedHistoricalLogsError{attemptErr: attemptErr, jobErr: jobErr}
+}
+
 type logTarget struct {
 	attemptID      string
 	attemptNumber  int32