diff --git a/atst/jobs.py b/atst/jobs.py
index ceea19db..8977cd7e 100644
--- a/atst/jobs.py
+++ b/atst/jobs.py
@@ -6,6 +6,7 @@ from atst.models import EnvironmentJobFailure, EnvironmentRoleJobFailure
 from atst.domain.csp.cloud import CloudProviderInterface, GeneralCSPException
 from atst.domain.environments import Environments
 from atst.domain.users import Users
+from atst.models import Application, Portfolio, TaskOrder, CLIN, Environment
 
 
 class RecordEnvironmentFailure(celery.Task):
@@ -117,3 +118,23 @@ def create_atat_admin_user(self, environment_id=None):
 @celery.task(bind=True)
 def create_environment_baseline(self, environment_id=None):
     do_work(do_create_environment_baseline, self, app.csp.cloud, **kwargs)
+
+
+def environments_to_create(now):
+    query = (
+        db.session.query(Environment.id)
+        .join(Application)
+        .join(Portfolio)
+        .join(TaskOrder)
+        .join(CLIN)
+        .filter(CLIN.start_date <= now)
+        .filter(CLIN.end_date > now)
+        .filter(Environment.cloud_id == None)
+    )
+    return [environment_id for (environment_id,) in query.all()]
+
+
+@celery.task(bind=True)
+def dispatch_create_environment(self):
+    for environment_id in environments_to_create(pendulum.now()):
+        create_environment.delay(environment_id=environment_id, atat_user_id="TODO")
diff --git a/tests/test_jobs.py b/tests/test_jobs.py
index e1ef988f..8a2bf19d 100644
--- a/tests/test_jobs.py
+++ b/tests/test_jobs.py
@@ -1,18 +1,26 @@
+import pendulum
 import pytest
 
 from atst.jobs import RecordEnvironmentFailure, RecordEnvironmentRoleFailure
 from tests.factories import EnvironmentFactory, EnvironmentRoleFactory, UserFactory
 from uuid import uuid4
 from unittest.mock import Mock
-
+from tests.factories import (
+    UserFactory,
+    PortfolioFactory,
+    CLINFactory,
+    TaskOrderFactory,
+    EnvironmentFactory
+)
+from atst.domain.csp.cloud import MockCloudProvider
 from atst.jobs import (
     do_create_environment,
     do_create_atat_admin_user,
     do_create_environment_baseline,
+    environments_to_create,
 )
 from atst.models import Environment
 
-
 def test_environment_job_failure(celery_app, celery_worker):
     @celery_app.task(bind=True, base=RecordEnvironmentFailure)
     def _fail_hard(self, environment_id=None):
@@ -47,8 +55,10 @@ def test_environment_role_job_failure(celery_app, celery_worker):
     assert role.job_failures
     job_failure = role.job_failures[0]
     assert job_failure.task == task
-from tests.factories import EnvironmentFactory, UserFactory
-from atst.domain.csp.cloud import MockCloudProvider
+
+now = pendulum.now()
+yesterday = now.subtract(days=1)
+tomorrow = now.add(days=1)
 
 
 @pytest.fixture(autouse=True, scope="function")
@@ -99,3 +109,40 @@ def test_create_environment_baseline(csp, session):
     updated_environment = session.query(Environment).get(environment_id)
 
     assert updated_environment.baseline_info
+
+
+def create_portfolio_with_clins(start_and_end_dates):
+    return PortfolioFactory.create(
+        applications=[
+            {
+                "name": "Mos Eisley",
+                "description": "Where Han shot first",
+                "environments": [{"name": "thebar"}],
+            }
+        ],
+        task_orders=[
+            TaskOrderFactory.create(
+                clins=[
+                    CLINFactory.create(start_date=start_date, end_date=end_date)
+                    for (start_date, end_date) in start_and_end_dates
+                ]
+            )
+        ],
+    )
+
+
+def test_dispatch_query_with_expired_clins(session):
+    create_portfolio_with_clins([(yesterday, yesterday)])
+    assert len(environments_to_create(pendulum.now())) == 0
+
+
+def test_dispatch_query_with_active_clins(session):
+    portfolio = create_portfolio_with_clins([(yesterday, tomorrow)])
+    environments_to_create(pendulum.now()) == [
+        portfolio.applications[0].environments[0].id
+    ]
+
+
+def test_dispatch_query_with_future_clins(session):
+    create_portfolio_with_clins([(tomorrow, tomorrow)])
+    assert len(environments_to_create(pendulum.now())) == 0