diff --git a/src/aind_data_transfer_service/__init__.py b/src/aind_data_transfer_service/__init__.py
index dd99e9c..bacf75a 100644
--- a/src/aind_data_transfer_service/__init__.py
+++ b/src/aind_data_transfer_service/__init__.py
@@ -1,7 +1,7 @@
 """Init package"""
 import os
 
-__version__ = "1.1.0"
+__version__ = "1.2.0"
 
 # Global constants
 OPEN_DATA_BUCKET_NAME = os.getenv("OPEN_DATA_BUCKET_NAME", "open")
diff --git a/src/aind_data_transfer_service/models.py b/src/aind_data_transfer_service/models.py
index 5e3a656..ebf9d04 100644
--- a/src/aind_data_transfer_service/models.py
+++ b/src/aind_data_transfer_service/models.py
@@ -1,9 +1,11 @@
 """Module for data models used in application"""
 
-from datetime import datetime
-from typing import List, Optional
+import ast
+from datetime import datetime, timedelta, timezone
+from typing import List, Optional, Union
 
-from pydantic import AwareDatetime, BaseModel, Field
+from pydantic import AwareDatetime, BaseModel, Field, field_validator
+from starlette.datastructures import QueryParams
 
 
 class AirflowDagRun(BaseModel):
@@ -37,12 +39,113 @@ class AirflowDagRunsRequestParameters(BaseModel):
 
     limit: int = 25
     offset: int = 0
-    order_by: str = "-start_date"
+    state: Optional[list[str]] = []
+    execution_date_gte: Optional[str] = (
+        datetime.now(timezone.utc) - timedelta(weeks=2)
+    ).isoformat()
+    execution_date_lte: Optional[str] = None
+    order_by: str = "-execution_date"
+
+    @field_validator("execution_date_gte", mode="after")
+    def validate_min_execution_date(cls, execution_date_gte: str):
+        """Validate the earliest submit date filter is within 2 weeks"""
+        min_execution_date = datetime.now(timezone.utc) - timedelta(weeks=2)
+        # datetime.fromisoformat does not support Z in python < 3.11
+        date_to_check = execution_date_gte.replace("Z", "+00:00")
+        if datetime.fromisoformat(date_to_check) < min_execution_date:
+            raise ValueError(
+                "execution_date_gte must be within the last 2 weeks"
+            )
+        return execution_date_gte
 
     @classmethod
-    def from_query_params(cls, query_params: dict):
+    def from_query_params(cls, query_params: QueryParams):
         """Maps the query parameters to the model"""
-        return cls(**query_params)
+        params = dict(query_params)
+        if "state" in params:
+            params["state"] = ast.literal_eval(params["state"])
+        return cls.model_validate(params)
+
+
+class AirflowDagRunRequestParameters(BaseModel):
+    """Model for parameters when requesting info from dag_run endpoint"""
+
+    dag_run_id: str = Field(..., min_length=1)
+
+    @classmethod
+    def from_query_params(cls, query_params: QueryParams):
+        """Maps the query parameters to the model"""
+        params = dict(query_params)
+        return cls.model_validate(params)
+
+
+class AirflowTaskInstancesRequestParameters(BaseModel):
+    """Model for parameters when requesting info from task_instances
+    endpoint"""
+
+    dag_run_id: str = Field(..., min_length=1)
+
+    @classmethod
+    def from_query_params(cls, query_params: QueryParams):
+        """Maps the query parameters to the model"""
+        params = dict(query_params)
+        return cls.model_validate(params)
+
+
+class AirflowTaskInstance(BaseModel):
+    """Data model for task_instance entry when requesting info from airflow"""
+
+    dag_id: Optional[str]
+    dag_run_id: Optional[str]
+    duration: Optional[Union[int, float]]
+    end_date: Optional[AwareDatetime]
+    execution_date: Optional[AwareDatetime]
+    executor_config: Optional[str]
+    hostname: Optional[str]
+    map_index: Optional[int]
+    max_tries: Optional[int]
+    note: Optional[str]
+    operator: Optional[str]
+    pid: Optional[int]
+    pool: Optional[str]
+    pool_slots: Optional[int]
+    priority_weight: Optional[int]
+    queue: Optional[str]
+    queued_when: Optional[AwareDatetime]
+    rendered_fields: Optional[dict]
+    sla_miss: Optional[dict]
+    start_date: Optional[AwareDatetime]
+    state: Optional[str]
+    task_id: Optional[str]
+    trigger: Optional[dict]
+    triggerer_job: Optional[dict]
+    try_number: Optional[int]
+    unixname: Optional[str]
+
+
+class AirflowTaskInstancesResponse(BaseModel):
+    """Data model for response when requesting info from task_instances
+    endpoint"""
+
+    task_instances: List[AirflowTaskInstance]
+    total_entries: int
+
+
+class AirflowTaskInstanceLogsRequestParameters(BaseModel):
+    """Model for parameters when requesting info from task_instance_logs
+    endpoint"""
+
+    # excluded fields are used to build the url
+    dag_run_id: str = Field(..., min_length=1, exclude=True)
+    task_id: str = Field(..., min_length=1, exclude=True)
+    try_number: int = Field(..., ge=0, exclude=True)
+    full_content: bool = True
+
+    @classmethod
+    def from_query_params(cls, query_params: QueryParams):
+        """Maps the query parameters to the model"""
+        params = dict(query_params)
+        return cls.model_validate(params)
 
 
 class JobStatus(BaseModel):
@@ -74,3 +177,38 @@ def from_airflow_dag_run(cls, airflow_dag_run: AirflowDagRun):
     def jinja_dict(self):
         """Map model to a dictionary that jinja can render"""
         return self.model_dump(exclude_none=True)
+
+
+class JobTasks(BaseModel):
+    """Model for what is rendered to the user for each task."""
+
+    job_id: Optional[str] = Field(None)
+    task_id: Optional[str] = Field(None)
+    try_number: Optional[int] = Field(None)
+    task_state: Optional[str] = Field(None)
+    priority_weight: Optional[int] = Field(None)
+    map_index: Optional[int] = Field(None)
+    submit_time: Optional[datetime] = Field(None)
+    start_time: Optional[datetime] = Field(None)
+    end_time: Optional[datetime] = Field(None)
+    duration: Optional[Union[int, float]] = Field(None)
+    comment: Optional[str] = Field(None)
+
+    @classmethod
+    def from_airflow_task_instance(
+        cls, airflow_task_instance: AirflowTaskInstance
+    ):
+        """Maps the fields from the HpcJobStatusResponse to this model"""
+        return cls(
+            job_id=airflow_task_instance.dag_run_id,
+            task_id=airflow_task_instance.task_id,
+            try_number=airflow_task_instance.try_number,
+            task_state=airflow_task_instance.state,
+            priority_weight=airflow_task_instance.priority_weight,
+            map_index=airflow_task_instance.map_index,
+            submit_time=airflow_task_instance.execution_date,
+            start_time=airflow_task_instance.start_date,
+            end_time=airflow_task_instance.end_date,
+            duration=airflow_task_instance.duration,
+            comment=airflow_task_instance.note,
+        )
diff --git a/src/aind_data_transfer_service/server.py b/src/aind_data_transfer_service/server.py
index ff8f815..3a9c890 100644
--- a/src/aind_data_transfer_service/server.py
+++ b/src/aind_data_transfer_service/server.py
@@ -30,9 +30,15 @@
 from aind_data_transfer_service.hpc.client import HpcClient, HpcClientConfigs
 from aind_data_transfer_service.hpc.models import HpcJobSubmitSettings
 from aind_data_transfer_service.models import (
+    AirflowDagRun,
+    AirflowDagRunRequestParameters,
     AirflowDagRunsRequestParameters,
     AirflowDagRunsResponse,
+    AirflowTaskInstanceLogsRequestParameters,
+    AirflowTaskInstancesRequestParameters,
+    AirflowTaskInstancesResponse,
     JobStatus,
+    JobTasks,
 )
 
 template_directory = os.path.abspath(
@@ -390,23 +396,40 @@ async def get_job_status_list(request: Request):
     """Get status of jobs with default pagination of limit=25 and offset=0."""
     # TODO: Use httpx async client
     try:
-        params = AirflowDagRunsRequestParameters.from_query_params(
-            request.query_params
-        )
+        url = os.getenv("AIND_AIRFLOW_SERVICE_JOBS_URL", "").strip("/")
+        get_one_job = request.query_params.get("dag_run_id") is not None
+        if get_one_job:
+            params = AirflowDagRunRequestParameters.from_query_params(
+                request.query_params
+            )
+            url = f"{url}/{params.dag_run_id}"
+        else:
+            params = AirflowDagRunsRequestParameters.from_query_params(
+                request.query_params
+            )
         params_dict = json.loads(params.model_dump_json())
+        # Send request to Airflow to ListDagRuns or GetDagRun
         response_jobs = requests.get(
-            url=os.getenv("AIND_AIRFLOW_SERVICE_JOBS_URL"),
+            url=url,
             auth=(
                 os.getenv("AIND_AIRFLOW_SERVICE_USER"),
                 os.getenv("AIND_AIRFLOW_SERVICE_PASSWORD"),
             ),
-            params=params_dict,
+            params=None if get_one_job else params_dict,
         )
         status_code = response_jobs.status_code
         if response_jobs.status_code == 200:
-            dag_runs = AirflowDagRunsResponse.model_validate_json(
-                json.dumps(response_jobs.json())
-            )
+            if get_one_job:
+                dag_run = AirflowDagRun.model_validate_json(
+                    json.dumps(response_jobs.json())
+                )
+                dag_runs = AirflowDagRunsResponse(
+                    dag_runs=[dag_run], total_entries=1
+                )
+            else:
+                dag_runs = AirflowDagRunsResponse.model_validate_json(
+                    json.dumps(response_jobs.json())
+                )
             job_status_list = [
                 JobStatus.from_airflow_dag_run(d) for d in dag_runs.dag_runs
             ]
@@ -420,7 +443,10 @@ async def get_job_status_list(request: Request):
             }
         else:
             message = "Error retrieving job status list from airflow"
-            data = {"params": params_dict, "errors": [response_jobs.json()]}
+            data = {
+                "params": params_dict,
+                "errors": [response_jobs.json()],
+            }
     except ValidationError as e:
         logging.error(e)
         status_code = 406
@@ -440,6 +466,115 @@ async def get_job_status_list(request: Request):
     )
 
 
+async def get_tasks_list(request: Request):
+    """Get list of tasks instances given a job id."""
+    try:
+        url = os.getenv("AIND_AIRFLOW_SERVICE_JOBS_URL", "").strip("/")
+        params = AirflowTaskInstancesRequestParameters.from_query_params(
+            request.query_params
+        )
+        params_dict = json.loads(params.model_dump_json())
+        response_tasks = requests.get(
+            url=f"{url}/{params.dag_run_id}/taskInstances",
+            auth=(
+                os.getenv("AIND_AIRFLOW_SERVICE_USER"),
+                os.getenv("AIND_AIRFLOW_SERVICE_PASSWORD"),
+            ),
+        )
+        status_code = response_tasks.status_code
+        if response_tasks.status_code == 200:
+            task_instances = AirflowTaskInstancesResponse.model_validate_json(
+                json.dumps(response_tasks.json())
+            )
+            job_tasks_list = sorted(
+                [
+                    JobTasks.from_airflow_task_instance(t)
+                    for t in task_instances.task_instances
+                ],
+                key=lambda t: (-t.priority_weight, t.map_index),
+            )
+            message = "Retrieved job tasks list from airflow"
+            data = {
+                "params": params_dict,
+                "total_entries": task_instances.total_entries,
+                "job_tasks_list": [
+                    json.loads(t.model_dump_json()) for t in job_tasks_list
+                ],
+            }
+        else:
+            message = "Error retrieving job tasks list from airflow"
+            data = {
+                "params": params_dict,
+                "errors": [response_tasks.json()],
+            }
+    except ValidationError as e:
+        logging.error(e)
+        status_code = 406
+        message = "Error validating request parameters"
+        data = {"errors": json.loads(e.json())}
+    except Exception as e:
+        logging.error(e)
+        status_code = 500
+        message = "Unable to retrieve job tasks list from airflow"
+        data = {"errors": [f"{e.__class__.__name__}{e.args}"]}
+    return JSONResponse(
+        status_code=status_code,
+        content={
+            "message": message,
+            "data": data,
+        },
+    )
+
+
+async def get_task_logs(request: Request):
+    """Get task logs given a job id, task id, and task try number."""
+    try:
+        url = os.getenv("AIND_AIRFLOW_SERVICE_JOBS_URL", "").strip("/")
+        params = AirflowTaskInstanceLogsRequestParameters.from_query_params(
+            request.query_params
+        )
+        params_dict = json.loads(params.model_dump_json())
+        params_full = dict(params)
+        response_logs = requests.get(
+            url=(
+                f"{url}/{params.dag_run_id}/taskInstances/{params.task_id}"
+                f"/logs/{params.try_number}"
+            ),
+            auth=(
+                os.getenv("AIND_AIRFLOW_SERVICE_USER"),
+                os.getenv("AIND_AIRFLOW_SERVICE_PASSWORD"),
+            ),
+            params=params_dict,
+        )
+        status_code = response_logs.status_code
+        if response_logs.status_code == 200:
+            message = "Retrieved task logs from airflow"
+            data = {"params": params_full, "logs": response_logs.text}
+        else:
+            message = "Error retrieving task logs from airflow"
+            data = {
+                "params": params_full,
+                "errors": [response_logs.json()],
+            }
+    except ValidationError as e:
+        logging.error(e)
+        status_code = 406
+        message = "Error validating request parameters"
+        data = {"errors": json.loads(e.json())}
+    except Exception as e:
+        logging.error(e)
+        status_code = 500
+        message = "Unable to retrieve task logs from airflow"
+        data = {"errors": [f"{e.__class__.__name__}{e.args}"]}
+    return JSONResponse(
+        status_code=status_code,
+        content={
+            "message": message,
+            "data": data,
+        },
+    )
+
+
 async def index(request: Request):
     """GET|POST /: form handler"""
     return templates.TemplateResponse(
@@ -478,6 +613,45 @@ async def job_status_table(request: Request):
     )
 
 
+async def job_tasks_table(request: Request):
+    """Get Job Tasks table given a job id"""
+    response_tasks = await get_tasks_list(request)
+    response_tasks_json = json.loads(response_tasks.body)
+    data = response_tasks_json.get("data")
+    return templates.TemplateResponse(
+        name="job_tasks_table.html",
+        context=(
+            {
+                "request": request,
+                "status_code": response_tasks.status_code,
+                "message": response_tasks_json.get("message"),
+                "errors": data.get("errors", []),
+                "total_entries": data.get("total_entries", 0),
+                "job_tasks_list": data.get("job_tasks_list", []),
+            }
+        ),
+    )
+
+
+async def task_logs(request: Request):
+    """Get task logs given a job id, task id, and task try number."""
+    response_tasks = await get_task_logs(request)
+    response_tasks_json = json.loads(response_tasks.body)
+    data = response_tasks_json.get("data")
+    return templates.TemplateResponse(
+        name="task_logs.html",
+        context=(
+            {
+                "request": request,
+                "status_code": response_tasks.status_code,
+                "message": response_tasks_json.get("message"),
+                "errors": data.get("errors", []),
+                "logs": data.get("logs"),
+            }
+        ),
+    )
+
+
 async def jobs(request: Request):
     """Get Job Status page with pagination"""
     default_limit = AirflowDagRunsRequestParameters.model_fields[
@@ -486,6 +660,9 @@ async def jobs(request: Request):
     default_offset = AirflowDagRunsRequestParameters.model_fields[
         "offset"
     ].default
+    default_state = AirflowDagRunsRequestParameters.model_fields[
+        "state"
+    ].default
     return templates.TemplateResponse(
         name="job_status.html",
         context=(
@@ -493,6 +670,7 @@ async def jobs(request: Request):
                 "request": request,
                 "default_limit": default_limit,
                 "default_offset": default_offset,
+                "default_state": default_state,
                 "project_names_url": os.getenv(
                     "AIND_METADATA_SERVICE_PROJECT_NAMES_URL"
                 ),
@@ -545,8 +723,12 @@ async def download_job_template(_: Request):
         endpoint=get_job_status_list,
         methods=["GET"],
     ),
+    Route("/api/v1/get_tasks_list", endpoint=get_tasks_list, methods=["GET"]),
+    Route("/api/v1/get_task_logs", endpoint=get_task_logs, methods=["GET"]),
     Route("/jobs", endpoint=jobs, methods=["GET"]),
     Route("/job_status_table", endpoint=job_status_table, methods=["GET"]),
+    Route("/job_tasks_table", endpoint=job_tasks_table, methods=["GET"]),
+    Route("/task_logs", endpoint=task_logs, methods=["GET"]),
     Route(
         "/api/job_upload_template",
         endpoint=download_job_template,
diff --git a/src/aind_data_transfer_service/templates/job_status.html b/src/aind_data_transfer_service/templates/job_status.html
index 75890f7..ebdac0f 100644
--- a/src/aind_data_transfer_service/templates/job_status.html
+++ b/src/aind_data_transfer_service/templates/job_status.html
@@ -3,8 +3,12 @@
 <head>
     <meta charset="UTF-8">
     <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+    <link href="https://cdnjs.cloudflare.com/ajax/libs/bootstrap-icons/1.8.1/font/bootstrap-icons.min.css" rel="stylesheet">
     <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
     <script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.29.3/moment.min.js"></script>
+    <script type="text/javascript" src="https://cdn.jsdelivr.net/jquery/latest/jquery.min.js"></script>
+    <script type="text/javascript" src="https://cdn.jsdelivr.net/npm/daterangepicker/daterangepicker.min.js"></script>
+    <link rel="stylesheet" type="text/css" href="https://cdn.jsdelivr.net/npm/daterangepicker/daterangepicker.css" />
     <title>{% block title %} {% endblock %} AIND Data Transfer Service Jobs</title>
     <style>
         body {
@@ -35,10 +39,59 @@
     </nav>
     <div class="content">
         <!-- display total entries from child iframe -->
-        <h2 class="mb-3">Jobs Submitted: <span id="jobs-iframe-total-entries"></h2>
+        <h4 class="mb-2">Jobs Submitted: <span id="jobs-iframe-total-entries"></h4>
+        <!-- filters for job status results-->
+        <div class="card mb-4 small" style="width:400px">
+            <div class="card-header py-1" type="button" data-bs-toggle="collapse" data-bs-target="#collapse-filters" aria-expanded="false" aria-controls="collapse-filters">
+                <i class="bi bi-filter"></i><span class="ms-2">Filter by</span>
+                <i class="bi bi-chevron-expand float-end"></i>
+            </div>
+            <div id="collapse-filters" class="collapse card-body p-2">
+                <!-- filter by job status -->
+                <div class="input-group input-group-sm mb-1">
+                    <span class="input-group-text" style="width:35%">Status</span>
+                    <select class="form-select" onchange="filterJobsByStatus(this.value);this.blur();">
+                        {% for s in [
+                            {"label": "all", "value": [], "class": "text-dark"},
+                            {"label": "queued", "value": ["queued"], "class": "text-secondary"},
+                            {"label": "running", "value": ["running"], "class": "text-info"},
+                            {"label": "failed", "value": ["failed"], "class": "text-danger"},
+                            {"label": "success", "value": ["success"], "class": "text-success"},
+                        ] %}
+                        {% if s.value==default_state %}<option class="{{ s.class }}" value="{{ s.value }}" selected>{{ s.label }}</option>
+                        {% else %}<option class="{{ s.class }}" value="{{ s.value }}">{{ s.label }}</option>
+                        {% endif %}
+                        {% endfor %}
+                    </select>
+                </div>
+                <!-- filter by job submitted date range -->
+                <div class="input-group input-group-sm">
+                    <span class="input-group-text" style="width:35%">Submit Time</span>
+                    <input id="submit-date-range" class="form-select" type="text" />
+                </div>
+                <!-- search by job id (exact match only) -->
+                <div class="d-flex align-items-center">
+                    <hr class="flex-grow-1 border-secondary">
+                    <span class="mx-2"><small class="d-block">or</small></span>
+                    <hr class="flex-grow-1 border-secondary">
+                </div>
+                <form onsubmit="searchByJobId(event)">
+                    <div class="input-group input-group-sm">
+                        <span class="input-group-text" style="width:35%">Job ID</span>
+                        <input id="job-id-input" type="text" class="form-control" placeholder="exact match only">
+                        <button class="btn btn-outline-secondary" type="submit" title="Search">
+                            <i class="bi bi-search"></i>
+                        </button>
+                        <button class="btn btn-outline-secondary" type="button" onclick="clearJobIdResult(event)" title="Clear">
+                            <i class="bi bi-x-lg"></i>
+                        </button>
+                    </div>
+                </form>
+            </div>
+        </div>
         <!-- toolbar to change jobs per page and navigate pages -->
         <div id="jobs-toolbar" class="btn-toolbar justify-content-between mb-2" role="toolbar">
-            <div class="input-group">
+            <div class="input-group input-group-sm">
                 <select class="form-select" onchange="updateJobStatusTableLimit(this.value);this.blur();">
                     {% for l in [10, 25, 50, 100] %}
                     {% if l==default_limit %}<option value="{{ l }}" selected>{{ l }}</option>
@@ -48,11 +101,11 @@ <h2 class="mb-3">Jobs Submitted: <span id="jobs-iframe-total-entries"></h2>
                 </select>
                 <span class="input-group-text">jobs per page</span>
             </div>
-            <div class="btn-group" role="group">
+            <div class="btn-group btn-group-sm" role="group">
                 <!-- display current jobs range e.g, "1 to 10 of 100" -->
                 <span id="jobs-iframe-showing" class="btn" style="cursor:default;"></span>
             </div>
-            <div class="btn-group pagination" role="group">
+            <div class="btn-group pagination pagination-sm" role="group">
                 <button id="jobs-page-btn-first" type="button" class="btn page-link"
                     onclick="updateJobStatusTablePage(PaginateTo.FIRST);this.blur();">&laquo; First</button>
                 <button id="jobs-page-btn-prev" type="button" class="btn page-link"
@@ -65,7 +118,7 @@ <h2 class="mb-3">Jobs Submitted: <span id="jobs-iframe-total-entries"></h2>
             </div>
         </div>
         <!-- iframe to display paginated job status table -->
-        <iframe id="jobs-iframe" src="{{ url_for('job_status_table').include_query_params(limit=default_limit, offset=default_offset)}}"></iframe>
+        <iframe id="jobs-iframe" src=""></iframe>
     </div>
     <script>
         const PaginateTo = {
@@ -74,12 +127,80 @@ <h2 class="mb-3">Jobs Submitted: <span id="jobs-iframe-total-entries"></h2>
             FIRST: 'first',
             LAST: 'last'
         };
-        function updateJobStatusTableLimit(newLimit) {
+        $(document).ready(function() {
+            const today = moment();
+            const twoWeeksAgo = moment().subtract(13, 'days');
+            
+            // initialize daterangepicker for submit date filter
+            $('#submit-date-range').daterangepicker({
+                startDate: twoWeeksAgo,
+                endDate: today,
+                minDate: twoWeeksAgo,
+                maxDate: today,
+                ranges: {
+                    'Today': [today, today],
+                    'Last 3 Days': [moment().subtract(2, 'days'), today],
+                    'Last 7 Days': [moment().subtract(6, 'days'), today],
+                    'Last 14 Days': [twoWeeksAgo, today],
+                }
+            }, filterJobsBySubmitTimeRange);
+
+            // initialize job status table with default values
+            const initialParams = {
+                limit: "{{ default_limit }}",
+                offset: "{{ default_offset }}",
+                state: "{{ default_state }}",
+                // set default submit date range client-side for browser's local time
+                execution_date_gte: twoWeeksAgo.startOf('day').toISOString(),
+                execution_date_lte: today.endOf('day').toISOString(),
+            };
+            updateJobStatusTable(initialParams, false, true);
+        });
+        // EVENT HANDLERS ------------------------------------------------
+        function updateJobStatusTable(newParams, isSearchJobId=false, isInitial=false) {
             var iframe = document.getElementById('jobs-iframe');
-            var currentUrl = new URL(iframe.src);
-            currentUrl.searchParams.set('limit', newLimit);
+            var currentUrl = isInitial ?  new URL("{{ url_for('job_status_table') }}") : new URL(iframe.src);
+            Object.entries(newParams).forEach(([key, value]) => {
+                currentUrl.searchParams.set(key, value);
+            });
+            // reset job_id filter
+            if (!isSearchJobId) {
+                document.getElementById('job-id-input').value = '';
+                currentUrl.searchParams.delete('dag_run_id');
+            }
             iframe.src = currentUrl.toString();
         }
+        // Filters
+        function filterJobsByStatus(newStatus) {
+            updateJobStatusTable({
+                state: newStatus,
+                offset: 0 // reset to first page
+            });
+        }
+        function filterJobsBySubmitTimeRange(start, end) {
+            // NOTE: daterangepicker already has 00:00:00.000 and 23:59:59.999 for start and end
+            updateJobStatusTable({
+                execution_date_gte: start.toISOString(),
+                execution_date_lte: end.toISOString(),
+                offset: 0 // reset to first page
+            });
+        }
+        function searchByJobId(event) {
+            event.preventDefault();
+            // search by job id and reset to first page
+            const jobId = document.getElementById('job-id-input').value;
+            console.log('Searching for job ID:', jobId);
+            updateJobStatusTable({ dag_run_id: jobId, offset: 0}, true);
+        }
+        function clearJobIdResult(event) {
+            event.preventDefault();
+            // clear job id filter and reset to first page
+            updateJobStatusTable({ offset: 0 }, false);
+        }
+        // Pagination
+        function updateJobStatusTableLimit(newLimit) {
+            updateJobStatusTable({limit: newLimit});
+        }
         function updateJobStatusTablePage(paginateTo) {
             var iframe = document.getElementById('jobs-iframe');
             var currentUrl = new URL(iframe.src);
@@ -101,8 +222,7 @@ <h2 class="mb-3">Jobs Submitted: <span id="jobs-iframe-total-entries"></h2>
                     offset = Math.floor(totalEntries / limit) * limit;
                     break;
             }
-            currentUrl.searchParams.set('offset', offset);
-            iframe.src = currentUrl.toString();
+            updateJobStatusTable({offset: offset});
         }
     </script>
     </body>
diff --git a/src/aind_data_transfer_service/templates/job_status_table.html b/src/aind_data_transfer_service/templates/job_status_table.html
index 9d26b4d..3ecb89f 100644
--- a/src/aind_data_transfer_service/templates/job_status_table.html
+++ b/src/aind_data_transfer_service/templates/job_status_table.html
@@ -4,21 +4,28 @@
 <head>
   <meta charset="UTF-8">
   <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+  <link href="https://cdnjs.cloudflare.com/ajax/libs/bootstrap-icons/1.8.1/font/bootstrap-icons.min.css" rel="stylesheet">
   <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
   <script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.29.3/moment.min.js"></script>
+  <style>
+    .modal-body {
+      height: calc(100vh - 100px);
+    }
+</style>
 </head>
 
 <body>
   <!-- job status table -->
-  <table class="table table-bordered table-striped table-hover table-sm">
+  <table class="table table-bordered table-striped table-hover table-sm" style="font-size: small">
     <tr>
       <th>Asset Name</th>
       <th>Job ID</th>
       <th>Status</th>
       <th>Submit Time</th>
       <th>Start Time</th>
-      <th>End time</th>
+      <th>End Time</th>
       <th>Comment</th>
+      <th>Tasks</th>
     </tr>
     {% for job_status in job_status_list %}
     <tr>
@@ -27,14 +34,38 @@
       <td class="{% if job_status.job_state == 'success' %}table-success
                 {% elif job_status.job_state == 'failed' %}table-danger
                 {% elif job_status.job_state == 'running' %}table-info
+                {% elif job_status.job_state == 'queued' %}table-secondary
                 {% endif %}">{{job_status.job_state}}</td>
       <td class="datetime_to_be_adjusted">{{job_status.submit_time}}</td>
       <td class="datetime_to_be_adjusted">{{job_status.start_time}}</td>
       <td class="datetime_to_be_adjusted">{{job_status.end_time}}</td>
       <td>{{job_status.comment}}</td>
+      <td>
+        <button type="button" class="btn btn-outline-primary btn-sm" data-bs-toggle="modal" data-bs-target="#tasks-modal" data-job-id="{{ job_status.job_id }}" data-job-name="{{ job_status.name}}" data-job-state="{{ job_status.job_state }}">
+          <i class="bi bi-box-arrow-up-right" title="View tasks and logs"></i>
+        </button>
+      </td>
     </tr>
     {% endfor %}
   </table>
+  <!-- modal for displaying tasks per job -->
+  <div class="modal fade" id="tasks-modal" tabindex="-1" aria-labelledby="tasks-modal-label" aria-hidden="true">
+    <div class="modal-dialog modal-xl">
+      <div class="modal-content">
+        <div class="modal-header p-2">
+          <div class="modal-title fw-bold" id="tasks-modal-label" style="font-size: small">
+            <span id="modal-title-job-name" class="me-2"></span>
+            <span id="modal-title-job-id" class="me-2"></span>
+            <span id="modal-title-job-state" class="badge"></span>
+          </div>
+          <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
+        </div>
+        <div class="modal-body p-2">
+            <iframe id="tasks-iframe" class="w-100 h-100" src=""></iframe>
+        </div>
+      </div>
+    </div>
+  </div>
   <!-- display errors to user -->
   {% if status_code != 200 %}
   <div class="alert alert-danger" role="alert">
@@ -63,10 +94,12 @@ <h4 class="alert-heading">{{ message }}</h4>
         limit = parseInt('{{limit}}');
         total_entries = parseInt('{{total_entries}}');
         parent.document.getElementById('jobs-iframe-total-entries').innerText = '{{total_entries}}';
-        parent.document.getElementById('jobs-iframe-showing').innerText = `${offset + 1} to ${Math.min(offset + limit, total_entries)} of ${total_entries}`;
+        parent.document.getElementById('jobs-iframe-showing').innerText = (total_entries > 1)
+          ? `${offset + 1} to ${Math.min(offset + limit, total_entries)} of ${total_entries}`
+          : `${total_entries} to ${total_entries} of ${total_entries}`;
         // also update the pagination buttons
-        isFirst = (offset == 0);
-        isLast = (offset + limit >= total_entries);
+        isFirst = (total_entries > 1) ? (offset == 0) : true;
+        isLast = (total_entries > 1) ? (offset + limit >= total_entries) : true;
         parent.document.getElementById('jobs-page-btn-first').disabled = isFirst;
         parent.document.getElementById('jobs-page-btn-prev').disabled = isFirst;
         parent.document.getElementById('jobs-page-btn-next').disabled = isLast;
@@ -79,6 +112,40 @@ <h4 class="alert-heading">{{ message }}</h4>
           el.innerText = utcTime.format('YYYY-MM-DD h:mm:ss a');  // Write the local time back to the element
         };
       });
+      var tasksModal = document.getElementById('tasks-modal');
+      tasksModal.addEventListener('show.bs.modal', function (event) {
+        var sourceData = event.relatedTarget?.dataset;
+        updateJobTasksModal(sourceData?.jobId, sourceData?.jobName, sourceData?.jobState);
+      })
+      tasksModal.addEventListener('hidden.bs.modal', function (event) {
+        updateJobTasksModal(null, null, null);
+      })
+    }
+    function updateJobTasksModal(jobId, jobName, jobState) {
+      // Update the modal header with the job id and name
+      document.getElementById('modal-title-job-id').textContent = `(${jobId})`;
+      document.getElementById('modal-title-job-name').textContent = jobName;
+      var modalTitleJobState = document.getElementById('modal-title-job-state');
+      modalTitleJobState.textContent = jobState;
+      if (jobState) {
+        modalTitleJobState.classList.add(
+          jobState === 'success' ? 'bg-success'
+          : jobState === 'failed' ? 'bg-danger'
+          : jobState === 'running' ? 'bg-info'
+          : 'bg-secondary'
+        );
+      } else {
+        modalTitleJobState.classList.value = 'badge';
+      }
+      // Update the iframe src with the job id
+      var tasksIframe = document.getElementById('tasks-iframe');
+      if (jobId) {
+        var url = new URL("{{ url_for('job_tasks_table') }}");
+        url.searchParams.append('dag_run_id', jobId);
+        tasksIframe.src = url;
+      } else {
+        tasksIframe.src = "";
+      }
     }
   </script>
 </body>
diff --git a/src/aind_data_transfer_service/templates/job_tasks_table.html b/src/aind_data_transfer_service/templates/job_tasks_table.html
new file mode 100644
index 0000000..b230605
--- /dev/null
+++ b/src/aind_data_transfer_service/templates/job_tasks_table.html
@@ -0,0 +1,133 @@
+<!DOCTYPE html>
+<html>
+
+<head>
+  <meta charset="UTF-8">
+  <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+  <link href="https://cdnjs.cloudflare.com/ajax/libs/bootstrap-icons/1.8.1/font/bootstrap-icons.min.css" rel="stylesheet">
+  <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
+  <script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.29.3/moment.min.js"></script>
+</head>
+
+<body>
+  <!-- tasks table -->
+  <table class="table table-bordered table-striped table-hover table-sm" style="font-size: small">
+    <tr>
+      <th>Task ID</th>
+      <th>Try Number</th>
+      <th>Status</th>
+      <th>Submit Time</th>
+      <th>Start Time</th>
+      <th>End Time</th>
+      <th>Duration</th>
+      <th>Logs</th>
+    </tr>
+    {% for job_task in job_tasks_list %}
+    <tr>
+      <td>{{job_task.task_id}}</td>
+      <td>{{job_task.try_number}}</td>
+      <td class="{% if job_task.task_state == 'success' %}table-success
+                {% elif job_task.task_state == 'failed' %}table-danger
+                {% elif job_task.task_state == 'running' %}table-info
+                {% elif job_task.task_state == 'queued' %}table-secondary
+                {% endif %}">{{job_task.task_state}}</td>
+      <td class="datetime_to_be_adjusted">{{job_task.submit_time}}</td>
+      <td class="datetime_to_be_adjusted">{{job_task.start_time}}</td>
+      <td class="datetime_to_be_adjusted">{{job_task.end_time}}</td>
+      <td>{{job_task.duration}}</td>
+      <td>
+        {% if job_task.try_number > 0 %}
+        <button type="button" class="btn btn-outline-primary btn-sm" data-bs-toggle="modal" data-bs-target="#logs-modal"
+            data-job-id="{{ job_task.job_id }}"
+            data-task-id="{{ job_task.task_id }}"
+            data-task-state="{{ job_task.task_state }}"
+            data-task-try-number="{{ job_task.try_number }}"
+          ><i class="bi bi-box-arrow-up-right" title="View logs"></i>
+          </button>
+        {% endif %}
+      </td>
+    </tr>
+    {% endfor %}
+  </table>
+  <!-- modal for displaying logs per task -->
+  <div class="modal fade" id="logs-modal" tabindex="-1" aria-labelledby="logs-modal-label" aria-hidden="true">
+    <div class="modal-dialog modal-fullscreen">
+      <div class="modal-content">
+        <div class="modal-header p-2">
+          <div class="modal-title fw-bold" id="logs-modal-label" style="font-size: small">
+            <span id="modal-title-task" class="me-2"></span><span id="modal-title-task-state" class="badge"></span>
+          </div>
+          <button type="button" class="btn-close" data-bs-dismiss="modal" aria-label="Close"></button>
+        </div>
+        <div class="modal-body p-2">
+            <iframe id="logs-iframe" class="w-100 h-100" src=""></iframe>
+        </div>
+      </div>
+    </div>
+  </div>
+  <!-- display errors to user -->
+  {% if status_code != 200 %}
+  <div class="alert alert-danger" role="alert">
+    <h4 class="alert-heading">{{ message }}</h4>
+    <p>
+      {% for error in errors %}
+        {% if error is string %}{{ error }}
+        {% elif error is mapping %}
+          {% for key, value in error.items() %}
+            <strong>{{ key }}:</strong> {{ value }}<br>
+          {% endfor %}
+        {% endif %}
+      {% endfor %}
+    </p>
+  </div>
+  {% endif %}
+  <script>
+    window.onload = function () {
+      document.querySelectorAll(".datetime_to_be_adjusted").forEach(function (el) {
+        if (el.innerHTML !== "") {
+          var utcTime = moment.utc(el.innerText);  // This is the time in UTC
+          utcTime.local();  // Switch to using the browser's local timezone
+          el.innerText = utcTime.format('YYYY-MM-DD h:mm:ss a');  // Write the local time back to the element
+        };
+      });
+      var logsModal = document.getElementById('logs-modal');
+      logsModal.addEventListener('show.bs.modal', function (event) {
+        var sourceData = event.relatedTarget?.dataset;
+        updateTaskLogsModal(sourceData?.jobId, sourceData?.taskId, sourceData?.taskState, sourceData?.taskTryNumber);
+      })
+      logsModal.addEventListener('hidden.bs.modal', function (event) {
+        updateTaskLogsModal(null, null, null, null);
+      })
+    }
+    function updateTaskLogsModal(jobId, taskId, taskState, taskTryNumber) {
+      console.log('updateTaskLogsModal', jobId, taskId, taskState, taskTryNumber);
+      // Update the modal header with the task id, try number, and state
+      document.getElementById('modal-title-task').textContent = `${taskId} (try ${taskTryNumber})`;
+      var modalTitleTaskState = document.getElementById('modal-title-task-state');
+      modalTitleTaskState.textContent = taskState;
+      if (taskState) {
+        modalTitleTaskState.classList.add(
+          taskState === 'success' ? 'bg-success'
+          : taskState === 'failed' ? 'bg-danger'
+          : taskState === 'running' ? 'bg-info'
+          : 'bg-secondary'
+        );
+      } else {
+        modalTitleTaskState.classList.value = 'badge';
+      }
+      // Update the iframe src to get the logs for the task
+      var logsIframe = document.getElementById('logs-iframe');
+      if (jobId) {
+        var url = new URL("{{ url_for('task_logs') }}");
+        url.searchParams.append('dag_run_id', jobId);
+        url.searchParams.append('task_id', taskId);
+        url.searchParams.append('try_number', taskTryNumber);
+        logsIframe.src = url;
+      } else {
+        logsIframe.src = "";
+      }
+    }
+  </script>
+</body>
+
+</html>
\ No newline at end of file
diff --git a/src/aind_data_transfer_service/templates/task_logs.html b/src/aind_data_transfer_service/templates/task_logs.html
new file mode 100644
index 0000000..6c1f2f9
--- /dev/null
+++ b/src/aind_data_transfer_service/templates/task_logs.html
@@ -0,0 +1,31 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="UTF-8">
+  <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css" rel="stylesheet">
+  <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.bundle.min.js"></script>
+</head>
+<body>
+  {% if status_code == 200 %}
+  <!-- display task logs -->
+  <div>
+    <pre class="bg-light p-1 border rounded" style="font-size: x-small">{{logs}}</pre>
+  </div>
+  {% else %}
+  <!-- display errors to user -->
+  <div class="alert alert-danger" role="alert">
+    <h4 class="alert-heading">{{ message }}</h4>
+    <p>
+      {% for error in errors %}
+        {% if error is string %}{{ error }}
+        {% elif error is mapping %}
+          {% for key, value in error.items() %}
+            <strong>{{ key }}:</strong> {{ value }}<br>
+          {% endfor %}
+        {% endif %}
+      {% endfor %}
+    </p>
+  </div>
+  {% endif %}
+</body>
+</html>
\ No newline at end of file
diff --git a/tests/resources/airflow_dag_run_response.json b/tests/resources/airflow_dag_run_response.json
new file mode 100644
index 0000000..a18698e
--- /dev/null
+++ b/tests/resources/airflow_dag_run_response.json
@@ -0,0 +1,51 @@
+{
+  "conf": {
+     "acq_datetime": "2000-01-01T01:40:03",
+     "force_cloud_sync": false,
+     "metadata_dir": null,
+     "metadata_dir_force": false,
+     "modalities": [
+        {
+           "compress_raw_data": true,
+           "extra_configs": null,
+           "modality": {
+              "abbreviation": "ecephys",
+              "name": "Extracellular electrophysiology"
+           },
+           "slurm_settings": null,
+           "source": "/a_source/ecephys"
+        },
+        {
+           "compress_raw_data": false,
+           "extra_configs": null,
+           "modality": {
+              "abbreviation": "behavior-videos",
+              "name": "Behavior videos"
+           },
+           "slurm_settings": null,
+           "source": "/b_source/ecephys"
+        }
+     ],
+     "platform": {
+        "abbreviation": "ecephys",
+        "name": "Electrophysiology platform"
+     },
+     "process_capsule_id": null,
+     "project_name": "Ephys Platform",
+     "s3_bucket": "private",
+     "subject_id": "655019"
+  },
+  "dag_id": "transform_and_upload",
+  "dag_run_id": "manual__2024-05-18T22:08:52.286765+00:00",
+  "data_interval_end": "2024-05-18T22:08:52.286765+00:00",
+  "data_interval_start": "2024-05-18T22:08:52.286765+00:00",
+  "end_date": "2024-05-18T22:09:28.530534+00:00",
+  "execution_date": "2024-05-18T22:08:52.286765+00:00",
+  "external_trigger": true,
+  "last_scheduling_decision": "2024-05-18T22:09:28.527134+00:00",
+  "logical_date": "2024-05-18T22:08:52.286765+00:00",
+  "note": null,
+  "run_type": "manual",
+  "start_date": "2024-05-18T22:08:52.637098+00:00",
+  "state": "failed"
+}
\ No newline at end of file
diff --git a/tests/resources/airflow_task_instances_response.json b/tests/resources/airflow_task_instances_response.json
new file mode 100644
index 0000000..4cc57af
--- /dev/null
+++ b/tests/resources/airflow_task_instances_response.json
@@ -0,0 +1,501 @@
+{
+  "task_instances": [
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 1.042997,
+          "end_date": "2024-08-21T16:17:27.278459+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-create-default-settings-msnfo8r2",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 12,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:17:11.816045+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {},
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:17:26.235462+00:00",
+          "state": "success",
+          "task_id": "create_default_settings",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 0.899175,
+          "end_date": "2024-08-21T16:17:11.720301+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-send-job-start-email-s1wxty2a",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 13,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:16:55.335522+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {},
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:17:10.821126+00:00",
+          "state": "success",
+          "task_id": "send_job_start_email",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 0.805628,
+          "end_date": "2024-08-21T16:18:00.491290+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-create-default-slurm-environment-1xph7dgd",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 10,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:17:44.511503+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "default_transfer_settings": "default_transfer_settings"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:17:59.685662+00:00",
+          "state": "success",
+          "task_id": "create_default_slurm_environment",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 1.062627,
+          "end_date": "2024-08-21T16:17:44.463969+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-check-s3-folder-exist-jy7ufnu9",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 11,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:17:28.252817+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "default_transfer_settings": "default_transfer_settings"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:17:43.401342+00:00",
+          "state": "success",
+          "task_id": "check_s3_folder_exist",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 61.686775,
+          "end_date": "2024-08-21T16:20:04.200273+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-create-folder-hxbnqe9h",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 8,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:18:47.513924+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "default_transfer_settings": "default_transfer_settings",
+                  "slurm_job_props": "slurm_job_props"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:19:02.513498+00:00",
+          "state": "success",
+          "task_id": "create_folder",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 31.563301,
+          "end_date": "2024-08-21T16:18:47.027590+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-check-source-folders-exist-xfz0r3qi",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 9,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:18:00.998977+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "slurm_job_props": "slurm_job_props"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:18:15.464289+00:00",
+          "state": "success",
+          "task_id": "check_source_folders_exist",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 0.941469,
+          "end_date": "2024-08-21T16:31:46.502387+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-send-job-end-email-aby2oyt5",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 1,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:31:30.258599+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {},
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:31:45.560918+00:00",
+          "state": "success",
+          "task_id": "send_job_end_email",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 0.755632,
+          "end_date": "2024-08-21T16:17:13.886610+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-make-modality-list-3epc13xm",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 7,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:16:55.335522+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {},
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:17:13.130978+00:00",
+          "state": "success",
+          "task_id": "make_modality_list",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 121.900177,
+          "end_date": "2024-08-21T16:26:54.539535+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-gather-final-metadata-5cn2eudo",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 5,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:24:38.460800+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": "op_kwargs",
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:24:52.639358+00:00",
+          "state": "success",
+          "task_id": "gather_final_metadata",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 121.718611,
+          "end_date": "2024-08-21T16:22:20.807546+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-gather-preliminary-metadata-2jggqh8n",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 7,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:20:04.384178+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "slurm_job_props": "slurm_job_props",
+                  "staging_folder": "/allen/aind/stage/svc_aind_airflow/dev/ISI_721679_job_1724257080_fc7ee"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:20:19.088935+00:00",
+          "state": "success",
+          "task_id": "gather_preliminary_metadata",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 121.897442,
+          "end_date": "2024-08-21T16:24:38.400648+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-compress-data-giexmirv",
+          "map_index": 0,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 6,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:22:21.400173+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "default_transfer_settings": "default_transfer_settings",
+                  "modality_conf": {
+                      "compress_raw_data": true,
+                      "extra_configs": null,
+                      "modality": {
+                          "abbreviation": "ISI",
+                          "name": "Intrinsic signal imaging"
+                      },
+                      "output_folder_name": "ISI",
+                      "platform": "ISI",
+                      "s3_bucket": "private",
+                      "s3_prefix": "ISI_721679_2024-08-21_00-00-00",
+                      "slurm_settings": null,
+                      "source": "/allen/aind/scratch/svc_aind_upload/test_data_sets/tmp_data"
+                  },
+                  "prelim_metadata": "Success",
+                  "slurm_job_props": "slurm_job_props",
+                  "staging_folder": "/allen/aind/stage/svc_aind_airflow/dev/ISI_721679_job_1724257080_fc7ee"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:22:36.503206+00:00",
+          "state": "success",
+          "task_id": "compress_data",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 10.542992,
+          "end_date": "2024-08-21T16:29:39.612352+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-send-codeocean-request-vxltb79h",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonVirtualenvDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 2,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:29:12.187996+00:00",
+          "rendered_fields": {
+              "index_urls": null,
+              "op_args": [
+                  {
+                      "context_as_str": "{\"acq_datetime\": \"2024-08-21T00:00:00\", \"email_notification_types\": [\"fail\"], \"force_cloud_sync\": false, \"input_data_mount\": null, \"metadata_dir\": null, \"metadata_dir_force\": false, \"modalities\": [{\"compress_raw_data\": true, \"extra_configs\": null, \"modality\": {\"abbreviation\": \"ISI\", \"name\": \"Intrinsic signal imaging\"}, \"output_folder_name\": \"ISI\", \"slurm_settings\": null, \"source\": \"/allen/aind/scratch/svc_aind_upload/test_data_sets/tmp_data\"}], \"platform\": {\"abbreviation\": \"ISI\", \"name\": \"Intrinsic signal imaging platform\"}, \"process_capsule_id\": null, \"project_name\": \"MSMA Platform\", \"s3_bucket\": \"private\", \"s3_prefix\": \"ISI_721679_2024-08-21_00-00-00\", \"subject_id\": \"721679\", \"user_email\": null}",
+                      "default_transfer_settings": "default_transfer_settings"
+                  }
+              ],
+              "op_kwargs": {},
+              "requirements": [
+                  "aind-codeocean-api==0.4.2"
+              ],
+              "templates_dict": null,
+              "venv_cache_path": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:29:29.069360+00:00",
+          "state": "success",
+          "task_id": "send_codeocean_request",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 121.805605,
+          "end_date": "2024-08-21T16:31:29.653235+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-remove-folder-prhy7w2q",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 20,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 2,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:29:12.187996+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "slurm_job_props": "slurm_job_props",
+                  "staging_folder": "/allen/aind/stage/svc_aind_airflow/dev/ISI_721679_job_1724257080_fc7ee"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:29:27.847630+00:00",
+          "state": "success",
+          "task_id": "remove_folder",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      },
+      {
+          "dag_id": "transform_and_upload",
+          "dag_run_id": "manual__2024-08-21T16:16:54.302335+00:00",
+          "duration": 121.853576,
+          "end_date": "2024-08-21T16:29:11.984181+00:00",
+          "execution_date": "2024-08-21T16:16:54.302335+00:00",
+          "executor_config": "{}",
+          "hostname": "transform-and-upload-upload-data-to-s3-eg7g6zr2",
+          "map_index": -1,
+          "max_tries": 0,
+          "note": null,
+          "operator": "_PythonDecoratedOperator",
+          "pid": 19,
+          "pool": "default_pool",
+          "pool_slots": 1,
+          "priority_weight": 4,
+          "queue": "default",
+          "queued_when": "2024-08-21T16:26:55.339993+00:00",
+          "rendered_fields": {
+              "op_args": [],
+              "op_kwargs": {
+                  "default_transfer_settings": "default_transfer_settings",
+                  "slurm_job_props": "slurm_job_props",
+                  "staging_folder": "/allen/aind/stage/svc_aind_airflow/dev/ISI_721679_job_1724257080_fc7ee"
+              },
+              "templates_dict": null
+          },
+          "sla_miss": null,
+          "start_date": "2024-08-21T16:27:10.130605+00:00",
+          "state": "success",
+          "task_id": "upload_data_to_s3",
+          "trigger": null,
+          "triggerer_job": null,
+          "try_number": 1,
+          "unixname": "airflow"
+      }
+  ],
+  "total_entries": 14
+}
\ No newline at end of file
diff --git a/tests/test_server.py b/tests/test_server.py
index 01162cb..9f3ff45 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -4,7 +4,7 @@
 import os
 import unittest
 from copy import deepcopy
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from io import BytesIO
 from pathlib import Path, PurePosixPath
 from unittest.mock import MagicMock, patch
@@ -43,9 +43,15 @@
 NEW_SAMPLE_CSV = TEST_DIRECTORY / "resources" / "new_sample.csv"
 MALFORMED_SAMPLE2_CSV = TEST_DIRECTORY / "resources" / "sample_malformed_2.csv"
 
-DAG_RUN_RESPONSE = (
+LIST_DAG_RUNS_RESPONSE = (
     TEST_DIRECTORY / "resources" / "airflow_dag_runs_response.json"
 )
+GET_DAG_RUN_RESPONSE = (
+    TEST_DIRECTORY / "resources" / "airflow_dag_run_response.json"
+)
+LIST_TASK_INSTANCES_RESPONSE = (
+    TEST_DIRECTORY / "resources" / "airflow_task_instances_response.json"
+)
 
 
 class TestServer(unittest.TestCase):
@@ -69,6 +75,7 @@ class TestServer(unittest.TestCase):
         "HPC_AWS_PARAM_STORE_NAME": "/some/param/store",
         "OPEN_DATA_AWS_SECRET_ACCESS_KEY": "open_data_aws_key",
         "OPEN_DATA_AWS_ACCESS_KEY_ID": "open_data_aws_key_id",
+        "AIND_AIRFLOW_SERVICE_JOBS_URL": "airflow_jobs_url",
     }
 
     with open(SAMPLE_CSV, "r") as file:
@@ -77,8 +84,14 @@ class TestServer(unittest.TestCase):
     with open(MOCK_DB_FILE) as f:
         json_contents = json.load(f)
 
-    with open(DAG_RUN_RESPONSE) as f:
-        dag_run_response = json.load(f)
+    with open(LIST_DAG_RUNS_RESPONSE) as f:
+        list_dag_runs_response = json.load(f)
+
+    with open(GET_DAG_RUN_RESPONSE) as f:
+        get_dag_run_response = json.load(f)
+
+    with open(LIST_TASK_INSTANCES_RESPONSE) as f:
+        list_task_instances_response = json.load(f)
 
     expected_job_configs = deepcopy(TestJobConfigs.expected_job_configs)
     for config in expected_job_configs:
@@ -534,15 +547,17 @@ def test_get_job_status_list_default(
         mock_dag_runs_response = Response()
         mock_dag_runs_response.status_code = 200
         mock_dag_runs_response._content = json.dumps(
-            self.dag_run_response
+            self.list_dag_runs_response
         ).encode("utf-8")
         mock_get.return_value = mock_dag_runs_response
         expected_message = "Retrieved job status list from airflow"
         expected_default_params = {
             "limit": 25,
             "offset": 0,
-            "start_date_gte": "mock_start_date_gte",
-            "order_by": "-start_date",
+            "state": [],
+            "execution_date_gte": "mock_execution_date_gte",
+            "execution_date_lte": None,
+            "order_by": "-execution_date",
         }
         expected_job_status_list = [
             {
@@ -596,8 +611,8 @@ def test_get_job_status_list_default(
         response_content = response.json()
         # small hack to mock the date
         response_content["data"]["params"][
-            "start_date_gte"
-        ] = "mock_start_date_gte"
+            "execution_date_gte"
+        ] = "mock_execution_date_gte"
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
             response_content,
@@ -605,7 +620,9 @@ def test_get_job_status_list_default(
                 "message": expected_message,
                 "data": {
                     "params": expected_default_params,
-                    "total_entries": self.dag_run_response["total_entries"],
+                    "total_entries": self.list_dag_runs_response[
+                        "total_entries"
+                    ],
                     "job_status_list": expected_job_status_list,
                 },
             },
@@ -622,13 +639,20 @@ def test_get_job_status_list_query_params(
         mock_dag_runs_response = Response()
         mock_dag_runs_response.status_code = 200
         mock_dag_runs_response._content = json.dumps(
-            self.dag_run_response
+            self.list_dag_runs_response
         ).encode("utf-8")
         mock_get.return_value = mock_dag_runs_response
         expected_message = "Retrieved job status list from airflow"
         with TestClient(app) as client:
             response = client.get(
-                "/api/v1/get_job_status_list?limit=10&offset=5"
+                "/api/v1/get_job_status_list",
+                params={
+                    "limit": 10,
+                    "offset": 5,
+                    "execution_date_gte": (
+                        datetime.now(timezone.utc) - timedelta(days=2)
+                    ).strftime("%Y-%m-%dT%H:%M:%SZ"),
+                },
             )
         response_content = response.json()
         self.assertEqual(response.status_code, 200)
@@ -637,22 +661,86 @@ def test_get_job_status_list_query_params(
         self.assertEqual(response_content["data"]["params"]["offset"], 5)
 
     @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
     @patch("logging.error")
     def test_get_job_status_list_validation_error(
         self,
         mock_log_error: MagicMock,
+        mock_get,
     ):
         """Tests get_job_status_list when query_params are invalid."""
+        invalid_queries = [
+            {"limit": "invalid", "offset": 5},
+            {"limit": 5, "offset": "invalid"},
+            {
+                "execution_date_gte": (
+                    datetime.now(timezone.utc)
+                    - timedelta(weeks=2)
+                    - timedelta(minutes=1)
+                ).strftime("%Y-%m-%dT%H:%M:%SZ")
+            },
+        ]
+        for query in invalid_queries:
+            with TestClient(app) as client:
+                response = client.get(
+                    "/api/v1/get_job_status_list", params=query
+                )
+            response_content = response.json()
+            self.assertEqual(response.status_code, 406)
+            self.assertEqual(
+                response_content["message"],
+                "Error validating request parameters",
+            )
+        mock_log_error.assert_called()
+        mock_get.assert_not_called()
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    def test_get_job_status_list_dag_run_id(
+        self,
+        mock_get,
+    ):
+        """Tests get_job_status_list gets 1 dagRun from airflow when dag_run_id
+        is provided."""
+        dag_run_id = "manual__2024-05-18T22:08:52.286765+00:00"
+        mock_dag_run_response = Response()
+        mock_dag_run_response.status_code = 200
+        mock_dag_run_response._content = json.dumps(
+            self.get_dag_run_response
+        ).encode("utf-8")
+        mock_get.return_value = mock_dag_run_response
+        expected_message = "Retrieved job status list from airflow"
+        expected_params = {
+            "dag_run_id": dag_run_id,
+        }
+        expected_job_status_list = [
+            {
+                "end_time": "2024-05-18T22:09:28.530534Z",
+                "job_id": "manual__2024-05-18T22:08:52.286765+00:00",
+                "job_state": "failed",
+                "name": "",
+                "comment": None,
+                "start_time": "2024-05-18T22:08:52.637098Z",
+                "submit_time": "2024-05-18T22:08:52.286765Z",
+            },
+        ]
         with TestClient(app) as client:
             response = client.get(
-                "/api/v1/get_job_status_list?limit=invalid&offset=5"
+                "/api/v1/get_job_status_list", params=expected_params
             )
         response_content = response.json()
-        self.assertEqual(response.status_code, 406)
+        self.assertEqual(response.status_code, 200)
         self.assertEqual(
-            response_content["message"], "Error validating request parameters"
+            response_content,
+            {
+                "message": expected_message,
+                "data": {
+                    "params": expected_params,
+                    "total_entries": 1,
+                    "job_status_list": expected_job_status_list,
+                },
+            },
         )
-        mock_log_error.assert_called_once()
 
     @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
     @patch("logging.error")
@@ -674,6 +762,374 @@ def test_get_job_status_list_error(
         )
         mock_log_error.assert_called_once()
 
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    def test_get_tasks_list_query_params(
+        self,
+        mock_get,
+    ):
+        """Tests get_tasks_list gets tasks from airflow using query_params."""
+        mock_task_instances_response = Response()
+        mock_task_instances_response.status_code = 200
+        mock_task_instances_response._content = json.dumps(
+            self.list_task_instances_response
+        ).encode("utf-8")
+        mock_get.return_value = mock_task_instances_response
+        expected_message = "Retrieved job tasks list from airflow"
+        expected_params = {
+            "dag_run_id": "mock_dag_run_id",
+        }
+        expected_task_list = [
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "send_job_start_email",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 13,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:17:10.821126Z",
+                "end_time": "2024-08-21T16:17:11.720301Z",
+                "duration": 0.899175,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "create_default_settings",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 12,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:17:26.235462Z",
+                "end_time": "2024-08-21T16:17:27.278459Z",
+                "duration": 1.042997,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "check_s3_folder_exist",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 11,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:17:43.401342Z",
+                "end_time": "2024-08-21T16:17:44.463969Z",
+                "duration": 1.062627,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "create_default_slurm_environment",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 10,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:17:59.685662Z",
+                "end_time": "2024-08-21T16:18:00.491290Z",
+                "duration": 0.805628,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "check_source_folders_exist",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 9,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:18:15.464289Z",
+                "end_time": "2024-08-21T16:18:47.027590Z",
+                "duration": 31.563301,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "create_folder",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 8,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:19:02.513498Z",
+                "end_time": "2024-08-21T16:20:04.200273Z",
+                "duration": 61.686775,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "make_modality_list",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 7,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:17:13.130978Z",
+                "end_time": "2024-08-21T16:17:13.886610Z",
+                "duration": 0.755632,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "gather_preliminary_metadata",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 7,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:20:19.088935Z",
+                "end_time": "2024-08-21T16:22:20.807546Z",
+                "duration": 121.718611,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "compress_data",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 6,
+                "map_index": 0,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:22:36.503206Z",
+                "end_time": "2024-08-21T16:24:38.400648Z",
+                "duration": 121.897442,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "gather_final_metadata",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 5,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:24:52.639358Z",
+                "end_time": "2024-08-21T16:26:54.539535Z",
+                "duration": 121.900177,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "upload_data_to_s3",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 4,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:27:10.130605Z",
+                "end_time": "2024-08-21T16:29:11.984181Z",
+                "duration": 121.853576,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "send_codeocean_request",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 2,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:29:29.069360Z",
+                "end_time": "2024-08-21T16:29:39.612352Z",
+                "duration": 10.542992,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "remove_folder",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 2,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:29:27.847630Z",
+                "end_time": "2024-08-21T16:31:29.653235Z",
+                "duration": 121.805605,
+                "comment": None,
+            },
+            {
+                "job_id": "manual__2024-08-21T16:16:54.302335+00:00",
+                "task_id": "send_job_end_email",
+                "try_number": 1,
+                "task_state": "success",
+                "priority_weight": 1,
+                "map_index": -1,
+                "submit_time": "2024-08-21T16:16:54.302335Z",
+                "start_time": "2024-08-21T16:31:45.560918Z",
+                "end_time": "2024-08-21T16:31:46.502387Z",
+                "duration": 0.941469,
+                "comment": None,
+            },
+        ]
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_tasks_list",
+                params={
+                    "dag_run_id": "mock_dag_run_id",
+                },
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response_content,
+            {
+                "message": expected_message,
+                "data": {
+                    "params": expected_params,
+                    "total_entries": self.list_task_instances_response[
+                        "total_entries"
+                    ],
+                    "job_tasks_list": expected_task_list,
+                },
+            },
+        )
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    @patch("logging.error")
+    def test_get_tasks_list_validation_error(
+        self,
+        mock_log_error: MagicMock,
+        mock_get,
+    ):
+        """Tests get_tasks_list when query_params are invalid."""
+        invalid_params = {
+            "job_id": "mock_dag_run_id",
+        }
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_tasks_list", params=invalid_params
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 406)
+        self.assertEqual(
+            response_content["message"],
+            "Error validating request parameters",
+        )
+        mock_log_error.assert_called()
+        mock_get.assert_not_called()
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("logging.error")
+    @patch("requests.get")
+    def test_get_tasks_list_error(
+        self,
+        mock_get: MagicMock,
+        mock_log_error: MagicMock,
+    ):
+        """Tests get_tasks_list when there is an error sending request."""
+        mock_get.side_effect = Exception("mock error")
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_tasks_list",
+                params={
+                    "dag_run_id": "mock_dag_run_id",
+                },
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 500)
+        self.assertEqual(
+            response_content["message"],
+            "Unable to retrieve job tasks list from airflow",
+        )
+        mock_log_error.assert_called_once()
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    def test_get_task_logs_query_params(
+        self,
+        mock_get,
+    ):
+        """Tests get_task_logs gets logs from airflow using query_params."""
+        mock_logs_response = Response()
+        mock_logs_response.status_code = 200
+        mock_logs_response._content = b"mock logs"
+        mock_get.return_value = mock_logs_response
+        expected_message = "Retrieved task logs from airflow"
+        expected_default_params = {
+            "dag_run_id": "mock_dag_run_id",
+            "task_id": "mock_task_id",
+            "try_number": 1,
+            "full_content": True,
+        }
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_task_logs",
+                params={
+                    "dag_run_id": "mock_dag_run_id",
+                    "task_id": "mock_task_id",
+                    "try_number": 1,
+                },
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(
+            response_content,
+            {
+                "message": expected_message,
+                "data": {
+                    "params": expected_default_params,
+                    "logs": "mock logs",
+                },
+            },
+        )
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    @patch("logging.error")
+    def test_get_task_logs_validation_error(
+        self,
+        mock_log_error: MagicMock,
+        mock_get,
+    ):
+        """Tests get_task_logs when query_params are invalid."""
+        invalid_params = {
+            "dag_run_id": "mock_dag_run_id",
+            "task_id": "mock_task_id",
+            "try_number": "invalid",
+        }
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_task_logs", params=invalid_params
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 406)
+        self.assertEqual(
+            response_content["message"],
+            "Error validating request parameters",
+        )
+        mock_log_error.assert_called()
+        mock_get.assert_not_called()
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("logging.error")
+    @patch("requests.get")
+    def test_get_task_logs_error(
+        self,
+        mock_get: MagicMock,
+        mock_log_error: MagicMock,
+    ):
+        """Tests get_task_logs when there is an error sending request."""
+        mock_get.side_effect = Exception("mock error")
+        with TestClient(app) as client:
+            response = client.get(
+                "/api/v1/get_task_logs",
+                params={
+                    "dag_run_id": "mock_dag_run_id",
+                    "task_id": "mock_task_id",
+                    "try_number": 1,
+                },
+            )
+        response_content = response.json()
+        self.assertEqual(response.status_code, 500)
+        self.assertEqual(
+            response_content["message"],
+            "Unable to retrieve task logs from airflow",
+        )
+        mock_log_error.assert_called_once()
+
     @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
     def test_index(self):
         """Tests that form renders at startup as expected."""
@@ -695,9 +1151,9 @@ def test_jobs_table_success(self, mock_get: MagicMock):
         """Tests that job status table renders as expected."""
         mock_response = Response()
         mock_response.status_code = 200
-        mock_response._content = json.dumps(self.dag_run_response).encode(
-            "utf-8"
-        )
+        mock_response._content = json.dumps(
+            self.list_dag_runs_response
+        ).encode("utf-8")
         mock_get.return_value = mock_response
         with TestClient(app) as client:
             response = client.get("/job_status_table")
@@ -723,6 +1179,87 @@ def test_jobs_table_failure(self, mock_get: MagicMock):
         )
         self.assertIn("test airflow error", response.text)
 
+    @patch("requests.get")
+    def test_tasks_table_success(self, mock_get: MagicMock):
+        """Tests that job tasks table renders as expected."""
+        mock_response = Response()
+        mock_response.status_code = 200
+        mock_response._content = json.dumps(
+            self.list_task_instances_response
+        ).encode("utf-8")
+        mock_get.return_value = mock_response
+        with TestClient(app) as client:
+            response = client.get(
+                "/job_tasks_table", params={"dag_run_id": "dag_run_id"}
+            )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Task ID", response.text)
+        self.assertIn("Try Number", response.text)
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    def test_tasks_table_failure(self, mock_get: MagicMock):
+        """Tests that job status table renders error message from airflow."""
+        mock_response = Response()
+        mock_response.status_code = 500
+        mock_response._content = json.dumps(
+            {"message": "test airflow error"}
+        ).encode("utf-8")
+        mock_get.return_value = mock_response
+        with TestClient(app) as client:
+            response = client.get(
+                "/job_tasks_table", params={"dag_run_id": "dag_run_id"}
+            )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Task ID", response.text)
+        self.assertIn("Try Number", response.text)
+        self.assertIn(
+            "Error retrieving job tasks list from airflow", response.text
+        )
+        self.assertIn("test airflow error", response.text)
+
+    @patch("requests.get")
+    def test_logs_success(self, mock_get: MagicMock):
+        """Tests that task logs page renders as expected."""
+        mock_response = Response()
+        mock_response.status_code = 200
+        mock_response._content = b"mock log content"
+        mock_get.return_value = mock_response
+        with TestClient(app) as client:
+            response = client.get(
+                "/task_logs",
+                params={
+                    "dag_run_id": "dag_run_id",
+                    "task_id": "task_id",
+                    "try_number": 1,
+                },
+            )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("mock log content", response.text)
+
+    @patch.dict(os.environ, EXAMPLE_ENV_VAR1, clear=True)
+    @patch("requests.get")
+    def test_logs_failure(self, mock_get: MagicMock):
+        """Tests that task logs page renders error message from airflow."""
+        mock_response = Response()
+        mock_response.status_code = 500
+        mock_response._content = json.dumps(
+            {"message": "test airflow error"}
+        ).encode("utf-8")
+        mock_get.return_value = mock_response
+        with TestClient(app) as client:
+            response = client.get(
+                "/task_logs",
+                params={
+                    "dag_run_id": "dag_run_id",
+                    "task_id": "task_id",
+                    "try_number": 1,
+                },
+            )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Error retrieving task logs from airflow", response.text)
+        self.assertIn("test airflow error", response.text)
+
     def test_download_job_template(self):
         """Tests that job template downloads as xlsx file."""