diff --git a/aggregator/src/aggregator.rs b/aggregator/src/aggregator.rs
index 901221e71..e60cf5902 100644
--- a/aggregator/src/aggregator.rs
+++ b/aggregator/src/aggregator.rs
@@ -724,7 +724,8 @@ impl<C: Clock> Aggregator<C> {
                 task_config.query_config().query().try_into()?,
                 vdaf_instance,
                 vdaf_verify_key,
-                Some(*task_config.task_expiration()),
+                None, // TODO(#3636): update taskprov implementation to specify task start
+                Some(*task_config.task_end()),
                 peer_aggregator.report_expiry_age().cloned(),
                 task_config.query_config().min_batch_size() as u64,
                 *task_config.query_config().time_precision(),
@@ -763,7 +764,7 @@ impl<C: Clock> Aggregator<C> {
 
     /// Validate and authorize a taskprov request. Returns values necessary for determining whether
     /// we can opt into the task. This function might return an opt-out error for conditions that
-    /// are relevant for all DAP workflows (e.g. task expiration).
+    /// are relevant for all DAP workflows (e.g. task end).
     #[tracing::instrument(skip(self, aggregator_auth_token), err(level = Level::DEBUG))]
     async fn taskprov_authorize_request(
         &self,
@@ -794,8 +795,8 @@ impl<C: Clock> Aggregator<C> {
             return Err(Error::UnauthorizedRequest(*task_id));
         }
 
-        if self.clock.now() > *task_config.task_expiration() {
-            return Err(Error::InvalidTask(*task_id, OptOutReason::TaskExpired));
+        if self.clock.now() > *task_config.task_end() {
+            return Err(Error::InvalidTask(*task_id, OptOutReason::TaskEnded));
         }
 
         debug!(
@@ -1608,11 +1609,18 @@ impl VdafOps {
             return Err(reject_report(ReportRejectionReason::TooEarly).await?);
         }
 
-        // Reject reports after a task has expired.
+        // Reject reports before a task has started.
+        if let Some(task_start) = task.task_start() {
+            if report.metadata().time().is_before(task_start) {
+                return Err(reject_report(ReportRejectionReason::TaskNotStarted).await?);
+            }
+        }
+
+        // Reject reports after a task has ended.
         // https://www.ietf.org/archive/id/draft-ietf-ppm-dap-07.html#section-4.4.2-20
-        if let Some(task_expiration) = task.task_expiration() {
-            if report.metadata().time().is_after(task_expiration) {
-                return Err(reject_report(ReportRejectionReason::TaskExpired).await?);
+        if let Some(task_end) = task.task_end() {
+            if report.metadata().time().is_after(task_end) {
+                return Err(reject_report(ReportRejectionReason::TaskEnded).await?);
             }
         }
 
diff --git a/aggregator/src/aggregator/error.rs b/aggregator/src/aggregator/error.rs
index c763aef5e..bb73a7fe0 100644
--- a/aggregator/src/aggregator/error.rs
+++ b/aggregator/src/aggregator/error.rs
@@ -225,10 +225,11 @@ pub enum ReportRejectionReason {
     IntervalCollected,
     DecryptFailure,
     DecodeFailure,
-    TaskExpired,
+    TaskEnded,
     Expired,
     TooEarly,
     OutdatedHpkeConfig(HpkeConfigId),
+    TaskNotStarted,
 }
 
 impl ReportRejectionReason {
@@ -239,12 +240,13 @@ impl ReportRejectionReason {
             }
             ReportRejectionReason::DecryptFailure => "Report share could not be decrypted.",
             ReportRejectionReason::DecodeFailure => "Report could not be decoded.",
-            ReportRejectionReason::TaskExpired => "Task has expired.",
+            ReportRejectionReason::TaskEnded => "Task has ended.",
             ReportRejectionReason::Expired => "Report timestamp is too old.",
             ReportRejectionReason::TooEarly => "Report timestamp is too far in the future.",
             ReportRejectionReason::OutdatedHpkeConfig(_) => {
                 "Report is using an outdated HPKE configuration."
             }
+            ReportRejectionReason::TaskNotStarted => "Task has not started.",
         }
     }
 }
@@ -260,8 +262,8 @@ impl Display for ReportRejectionReason {
 pub enum OptOutReason {
     #[error("this aggregator is not peered with the given {0} aggregator")]
     NoSuchPeer(Role),
-    #[error("task has expired")]
-    TaskExpired,
+    #[error("task has ended")]
+    TaskEnded,
     #[error("invalid task: {0}")]
     TaskParameters(#[from] task::Error),
     #[error("URL parse error: {0}")]
diff --git a/aggregator/src/aggregator/http_handlers/tests/report.rs b/aggregator/src/aggregator/http_handlers/tests/report.rs
index c9c29c74b..664132b0d 100644
--- a/aggregator/src/aggregator/http_handlers/tests/report.rs
+++ b/aggregator/src/aggregator/http_handlers/tests/report.rs
@@ -202,21 +202,21 @@ async fn upload_handler() {
     )
     .await;
 
-    // Reports with timestamps past the task's expiration should be rejected.
-    let task_expire_soon = TaskBuilder::new(BatchMode::TimeInterval, VdafInstance::Prio3Count)
-        .with_task_expiration(Some(clock.now().add(&Duration::from_seconds(60)).unwrap()))
+    // Reports with timestamps past the task's end time should be rejected.
+    let task_end_soon = TaskBuilder::new(BatchMode::TimeInterval, VdafInstance::Prio3Count)
+        .with_task_end(Some(clock.now().add(&Duration::from_seconds(60)).unwrap()))
         .build();
-    let leader_task_expire_soon = task_expire_soon.leader_view().unwrap();
+    let leader_task_end_soon = task_end_soon.leader_view().unwrap();
     datastore
-        .put_aggregator_task(&leader_task_expire_soon)
+        .put_aggregator_task(&leader_task_end_soon)
         .await
         .unwrap();
     let report_2 = create_report(
-        &leader_task_expire_soon,
+        &leader_task_end_soon,
         &hpke_keypair,
         clock.now().add(&Duration::from_seconds(120)).unwrap(),
     );
-    let mut test_conn = post(task_expire_soon.report_upload_uri().unwrap().path())
+    let mut test_conn = post(task_end_soon.report_upload_uri().unwrap().path())
         .with_request_header(KnownHeaderName::ContentType, Report::MEDIA_TYPE)
         .with_request_body(report_2.get_encoded().unwrap())
         .run_async(&handler)
@@ -226,8 +226,8 @@ async fn upload_handler() {
         Status::BadRequest,
         "reportRejected",
         "Report could not be processed.",
-        task_expire_soon.id(),
-        Some(ReportRejectionReason::TaskExpired.detail()),
+        task_end_soon.id(),
+        Some(ReportRejectionReason::TaskEnded.detail()),
     )
     .await;
 
diff --git a/aggregator/src/aggregator/problem_details.rs b/aggregator/src/aggregator/problem_details.rs
index f5d7f5e36..5f380e4d5 100644
--- a/aggregator/src/aggregator/problem_details.rs
+++ b/aggregator/src/aggregator/problem_details.rs
@@ -207,7 +207,7 @@ mod tests {
                             random(),
                             random(),
                             RealClock::default().now(),
-                            ReportRejectionReason::TaskExpired
+                            ReportRejectionReason::TaskEnded
                         ))
                     }),
                     Some(DapProblemType::ReportRejected),
diff --git a/aggregator/src/aggregator/report_writer.rs b/aggregator/src/aggregator/report_writer.rs
index b5adf5058..eee774a02 100644
--- a/aggregator/src/aggregator/report_writer.rs
+++ b/aggregator/src/aggregator/report_writer.rs
@@ -344,10 +344,11 @@ impl TaskUploadCounters {
             ReportRejectionReason::IntervalCollected => entry.increment_interval_collected(),
             ReportRejectionReason::DecryptFailure => entry.increment_report_decrypt_failure(),
             ReportRejectionReason::DecodeFailure => entry.increment_report_decode_failure(),
-            ReportRejectionReason::TaskExpired => entry.increment_task_expired(),
+            ReportRejectionReason::TaskEnded => entry.increment_task_ended(),
             ReportRejectionReason::Expired => entry.increment_report_expired(),
             ReportRejectionReason::TooEarly => entry.increment_report_too_early(),
             ReportRejectionReason::OutdatedHpkeConfig(_) => entry.increment_report_outdated_key(),
+            ReportRejectionReason::TaskNotStarted => entry.increment_task_not_started(),
         }
     }
 
diff --git a/aggregator/src/aggregator/taskprov_tests.rs b/aggregator/src/aggregator/taskprov_tests.rs
index c4ab23ae3..9fcb770f4 100644
--- a/aggregator/src/aggregator/taskprov_tests.rs
+++ b/aggregator/src/aggregator/taskprov_tests.rs
@@ -154,7 +154,7 @@ where
 
         let time_precision = Duration::from_seconds(1);
         let min_batch_size = 1;
-        let task_expiration = clock.now().add(&Duration::from_hours(24).unwrap()).unwrap();
+        let task_end = clock.now().add(&Duration::from_hours(24).unwrap()).unwrap();
         let task_config = TaskConfig::new(
             Vec::from("foobar".as_bytes()),
             "https://leader.example.com/".as_bytes().try_into().unwrap(),
@@ -164,7 +164,7 @@ where
                 min_batch_size,
                 TaskprovQuery::LeaderSelected,
             ),
-            task_expiration,
+            task_end,
             vdaf_config,
         )
         .unwrap();
@@ -185,7 +185,7 @@ where
         .with_leader_aggregator_endpoint(Url::parse("https://leader.example.com/").unwrap())
         .with_helper_aggregator_endpoint(Url::parse("https://helper.example.com/").unwrap())
         .with_vdaf_verify_key(vdaf_verify_key)
-        .with_task_expiration(Some(task_expiration))
+        .with_task_end(Some(task_end))
         .with_report_expiry_age(peer_aggregator.report_expiry_age().copied())
         .with_min_batch_size(min_batch_size as u64)
         .with_time_precision(Duration::from_seconds(1))
@@ -560,7 +560,7 @@ async fn taskprov_aggregate_init_malformed_extension() {
 }
 
 #[tokio::test]
-async fn taskprov_opt_out_task_expired() {
+async fn taskprov_opt_out_task_ended() {
     let test = TaskprovTestCase::new().await;
 
     let (transcript, report_share, _) = test.next_report_share();
@@ -582,7 +582,7 @@ async fn taskprov_opt_out_task_expired() {
         .primary_aggregator_auth_token()
         .request_authentication();
 
-    // Advance clock past task expiry.
+    // Advance clock past task end time.
     test.clock.advance(&Duration::from_hours(48).unwrap());
 
     let mut test_conn = put(test
@@ -631,7 +631,7 @@ async fn taskprov_opt_out_mismatched_task_id() {
 
     let aggregation_job_id: AggregationJobId = random();
 
-    let task_expiration = test
+    let task_end = test
         .clock
         .now()
         .add(&Duration::from_hours(24).unwrap())
@@ -646,7 +646,7 @@ async fn taskprov_opt_out_mismatched_task_id() {
             100,
             TaskprovQuery::LeaderSelected,
         ),
-        task_expiration,
+        task_end,
         VdafConfig::new(
             DpConfig::new(DpMechanism::None),
             VdafType::Fake { rounds: 2 },
@@ -708,7 +708,7 @@ async fn taskprov_opt_out_peer_aggregator_wrong_role() {
 
     let aggregation_job_id: AggregationJobId = random();
 
-    let task_expiration = test
+    let task_end = test
         .clock
         .now()
         .add(&Duration::from_hours(24).unwrap())
@@ -723,7 +723,7 @@ async fn taskprov_opt_out_peer_aggregator_wrong_role() {
             100,
             TaskprovQuery::LeaderSelected,
         ),
-        task_expiration,
+        task_end,
         VdafConfig::new(
             DpConfig::new(DpMechanism::None),
             VdafType::Fake { rounds: 2 },
@@ -786,7 +786,7 @@ async fn taskprov_opt_out_peer_aggregator_does_not_exist() {
 
     let aggregation_job_id: AggregationJobId = random();
 
-    let task_expiration = test
+    let task_end = test
         .clock
         .now()
         .add(&Duration::from_hours(24).unwrap())
@@ -801,7 +801,7 @@ async fn taskprov_opt_out_peer_aggregator_does_not_exist() {
             100,
             TaskprovQuery::LeaderSelected,
         ),
-        task_expiration,
+        task_end,
         VdafConfig::new(
             DpConfig::new(DpMechanism::None),
             VdafType::Fake { rounds: 2 },
diff --git a/aggregator/src/aggregator/upload_tests.rs b/aggregator/src/aggregator/upload_tests.rs
index f8a0d6a23..f432f711d 100644
--- a/aggregator/src/aggregator/upload_tests.rs
+++ b/aggregator/src/aggregator/upload_tests.rs
@@ -167,7 +167,9 @@ async fn upload() {
 
     assert_eq!(
         got_counter,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 1, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 1, 0, 0, 0
+        ))
     )
 }
 
@@ -230,7 +232,9 @@ async fn upload_batch() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 100, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 100, 0, 0, 0
+        ))
     );
 }
 
@@ -297,7 +301,9 @@ async fn upload_wrong_hpke_config_id() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 1, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 1, 0, 0, 0, 0
+        ))
     )
 }
 
@@ -344,7 +350,9 @@ async fn upload_report_in_the_future_boundary_condition() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 1, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 1, 0, 0, 0
+        ))
     )
 }
 
@@ -400,7 +408,9 @@ async fn upload_report_in_the_future_past_clock_skew() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 0, 1, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 0, 1, 0, 0
+        ))
     )
 }
 
@@ -481,12 +491,14 @@ async fn upload_report_for_collected_batch() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(1, 0, 0, 0, 0, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            1, 0, 0, 0, 0, 0, 0, 0, 0
+        ))
     )
 }
 
 #[tokio::test]
-async fn upload_report_task_expired() {
+async fn upload_report_task_not_started() {
     let mut runtime_manager = TestRuntimeManager::new();
     let UploadTest {
         aggregator,
@@ -501,14 +513,77 @@ async fn upload_report_task_expired() {
     )
     .await;
 
+    // Set the task start time to the future, and generate & upload a report from before that time.
     let task = TaskBuilder::new(BatchMode::TimeInterval, VdafInstance::Prio3Count)
-        .with_task_expiration(Some(clock.now()))
+        .with_task_start(Some(
+            clock.now().add(&Duration::from_seconds(3600)).unwrap(),
+        ))
         .build()
         .leader_view()
         .unwrap();
     datastore.put_aggregator_task(&task).await.unwrap();
 
-    // Advance the clock to expire the task.
+    let report = create_report(&task, &hpke_keypair, clock.now());
+
+    // Try to upload the report, verify that we get the expected error.
+    let error = aggregator
+        .handle_upload(task.id(), &report.get_encoded().unwrap())
+        .await
+        .unwrap_err();
+    assert_matches!(
+        error.as_ref(),
+        Error::ReportRejected(rejection) => {
+            assert_eq!(task.id(), rejection.task_id());
+            assert_eq!(report.metadata().id(), rejection.report_id());
+            assert_eq!(report.metadata().time(), rejection.time());
+            assert_matches!(rejection.reason(), ReportRejectionReason::TaskNotStarted);
+        }
+    );
+
+    // Wait for the report writer to have completed one write task.
+    runtime_manager
+        .wait_for_completed_tasks("aggregator", 1)
+        .await;
+
+    let got_counters = datastore
+        .run_unnamed_tx(|tx| {
+            let task_id = *task.id();
+            Box::pin(async move { tx.get_task_upload_counter(&task_id).await })
+        })
+        .await
+        .unwrap();
+    assert_eq!(
+        got_counters,
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 0, 0, 1, 0
+        ))
+    )
+}
+
+#[tokio::test]
+async fn upload_report_task_ended() {
+    let mut runtime_manager = TestRuntimeManager::new();
+    let UploadTest {
+        aggregator,
+        clock,
+        datastore,
+        ephemeral_datastore: _ephemeral_datastore,
+        hpke_keypair,
+        ..
+    } = UploadTest::new_with_runtime(
+        default_aggregator_config(),
+        runtime_manager.with_label("aggregator"),
+    )
+    .await;
+
+    let task = TaskBuilder::new(BatchMode::TimeInterval, VdafInstance::Prio3Count)
+        .with_task_end(Some(clock.now()))
+        .build()
+        .leader_view()
+        .unwrap();
+    datastore.put_aggregator_task(&task).await.unwrap();
+
+    // Advance the clock to end the task.
     clock.advance(&Duration::from_seconds(1));
     let report = create_report(&task, &hpke_keypair, clock.now());
 
@@ -523,7 +598,7 @@ async fn upload_report_task_expired() {
             assert_eq!(task.id(), rejection.task_id());
             assert_eq!(report.metadata().id(), rejection.report_id());
             assert_eq!(report.metadata().time(), rejection.time());
-            assert_matches!(rejection.reason(), ReportRejectionReason::TaskExpired);
+            assert_matches!(rejection.reason(), ReportRejectionReason::TaskEnded);
         }
     );
 
@@ -541,7 +616,9 @@ async fn upload_report_task_expired() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 0, 0, 1))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 0, 0, 0, 0, 0, 1
+        ))
     )
 }
 
@@ -602,7 +679,9 @@ async fn upload_report_report_expired() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 0, 1, 0, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 0, 1, 0, 0, 0, 0, 0
+        ))
     )
 }
 
@@ -662,7 +741,9 @@ async fn upload_report_faulty_encryption() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 0, 1, 0, 0, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 0, 1, 0, 0, 0, 0, 0, 0
+        ))
     )
 }
 
@@ -723,7 +804,9 @@ async fn upload_report_public_share_decode_failure() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 1, 0, 0, 0, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 1, 0, 0, 0, 0, 0, 0, 0
+        ))
     )
 }
 
@@ -798,6 +881,8 @@ async fn upload_report_leader_input_share_decode_failure() {
         .unwrap();
     assert_eq!(
         got_counters,
-        Some(TaskUploadCounter::new_with_values(0, 1, 0, 0, 0, 0, 0, 0))
+        Some(TaskUploadCounter::new_with_values(
+            0, 1, 0, 0, 0, 0, 0, 0, 0
+        ))
     )
 }
diff --git a/aggregator/src/binaries/janus_cli.rs b/aggregator/src/binaries/janus_cli.rs
index b969341e3..02c1b3bc2 100644
--- a/aggregator/src/binaries/janus_cli.rs
+++ b/aggregator/src/binaries/janus_cli.rs
@@ -1328,7 +1328,7 @@ mod tests {
     bits: 2
   role: Leader
   vdaf_verify_key:
-  task_expiration: 9000000000
+  task_end: 9000000000
   min_batch_size: 10
   time_precision: 300
   tolerable_clock_skew: 600
@@ -1351,7 +1351,7 @@ mod tests {
     bits: 2
   role: Helper
   vdaf_verify_key:
-  task_expiration: 9000000000
+  task_end: 9000000000
   min_batch_size: 10
   time_precision: 300
   tolerable_clock_skew: 600
diff --git a/aggregator/src/cache.rs b/aggregator/src/cache.rs
index 9ae36ce3b..fbf8d757e 100644
--- a/aggregator/src/cache.rs
+++ b/aggregator/src/cache.rs
@@ -367,14 +367,12 @@ mod tests {
         assert_eq!(task_aggregator.task.id(), task.id());
 
         // Modify the task.
-        let new_expiration = Time::from_seconds_since_epoch(100);
+        let new_end = Time::from_seconds_since_epoch(100);
         datastore
             .run_unnamed_tx(|tx| {
                 let task_id = *task.id();
                 Box::pin(async move {
-                    tx.update_task_expiration(&task_id, Some(&new_expiration))
-                        .await
-                        .unwrap();
+                    tx.update_task_end(&task_id, Some(&new_end)).await.unwrap();
                     Ok(())
                 })
             })
@@ -385,8 +383,8 @@ mod tests {
         // previous task.
         let task_aggregator = task_aggregators.get(task.id()).await.unwrap().unwrap();
         assert!(
-            (task_aggregator.task.task_expiration() == task.task_expiration())
-                || (task_aggregator.task.task_expiration() == Some(&new_expiration))
+            (task_aggregator.task.task_end() == task.task_end())
+                || (task_aggregator.task.task_end() == Some(&new_end))
         );
 
         // Unfortunately, because moka doesn't provide any facility for a fake clock, we have to resort
@@ -394,10 +392,7 @@ mod tests {
         sleep(Duration::from_secs(1)).await;
 
         let task_aggregator = task_aggregators.get(task.id()).await.unwrap().unwrap();
-        assert_eq!(
-            task_aggregator.task.task_expiration(),
-            Some(&new_expiration)
-        );
+        assert_eq!(task_aggregator.task.task_end(), Some(&new_end));
     }
 
     #[tokio::test]
@@ -441,14 +436,12 @@ mod tests {
         assert_eq!(task_aggregator.task.id(), task.id());
 
         // Modify the task.
-        let new_expiration = Time::from_seconds_since_epoch(100);
+        let new_end = Time::from_seconds_since_epoch(100);
         datastore
             .run_unnamed_tx(|tx| {
                 let task_id = *task.id();
                 Box::pin(async move {
-                    tx.update_task_expiration(&task_id, Some(&new_expiration))
-                        .await
-                        .unwrap();
+                    tx.update_task_end(&task_id, Some(&new_end)).await.unwrap();
                     Ok(())
                 })
             })
@@ -459,16 +452,13 @@ mod tests {
         // previous value.
         let task_aggregator = task_aggregators.get(task.id()).await.unwrap().unwrap();
         assert!(
-            (task_aggregator.task.task_expiration() == task.task_expiration())
-                || (task_aggregator.task.task_expiration() == Some(&new_expiration))
+            (task_aggregator.task.task_end() == task.task_end())
+                || (task_aggregator.task.task_end() == Some(&new_end))
         );
 
         sleep(Duration::from_secs(1)).await;
 
         let task_aggregator = task_aggregators.get(task.id()).await.unwrap().unwrap();
-        assert_eq!(
-            task_aggregator.task.task_expiration(),
-            Some(&new_expiration)
-        );
+        assert_eq!(task_aggregator.task.task_end(), Some(&new_end));
     }
 }
diff --git a/aggregator_api/src/models.rs b/aggregator_api/src/models.rs
index 655bead05..0b6e82006 100644
--- a/aggregator_api/src/models.rs
+++ b/aggregator_api/src/models.rs
@@ -72,8 +72,10 @@ pub(crate) struct PostTaskReq {
     /// The VDAF verification key used for this DAP task, as Base64 encoded bytes. Task ID is
     /// derived from the verify key.
     pub(crate) vdaf_verify_key: String,
+    /// The time before which the task is considered invalid.
+    pub(crate) task_start: Option<Time>,
     /// The time after which the task is considered invalid.
-    pub(crate) task_expiration: Option<Time>,
+    pub(crate) task_end: Option<Time>,
     /// The minimum number of reports in a batch to allow it to be collected.
     pub(crate) min_batch_size: u64,
     /// The duration to which clients should round their reported timestamps, as seconds since
@@ -93,7 +95,7 @@ pub(crate) struct PostTaskReq {
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub(crate) struct PatchTaskReq {
     #[serde(default, deserialize_with = "deserialize_some")]
-    pub(crate) task_expiration: Option<Option<Time>>,
+    pub(crate) task_end: Option<Option<Time>>,
 }
 
 #[derive(Clone, Derivative, PartialEq, Eq, Serialize, Deserialize)]
@@ -114,8 +116,10 @@ pub(crate) struct TaskResp {
     /// The VDAF verification key used for this DAP task, as Base64 encoded bytes. Task ID is
     /// derived from the verify key.
     pub(crate) vdaf_verify_key: String,
+    /// The time before which the task is considered invalid.
+    pub(crate) task_start: Option<Time>,
     /// The time after which the task is considered invalid.
-    pub(crate) task_expiration: Option<Time>,
+    pub(crate) task_end: Option<Time>,
     /// The age after which a report is considered to be "expired" and will be considered a
     /// candidate for garbage collection.
     pub(crate) report_expiry_age: Option<Duration>,
@@ -145,7 +149,8 @@ impl TryFrom<&AggregatorTask> for TaskResp {
             vdaf: task.vdaf().clone(),
             role: *task.role(),
             vdaf_verify_key: URL_SAFE_NO_PAD.encode(task.opaque_vdaf_verify_key().as_ref()),
-            task_expiration: task.task_expiration().copied(),
+            task_start: task.task_start().copied(),
+            task_end: task.task_end().copied(),
             report_expiry_age: task.report_expiry_age().cloned(),
             min_batch_size: task.min_batch_size(),
             time_precision: *task.time_precision(),
diff --git a/aggregator_api/src/routes.rs b/aggregator_api/src/routes.rs
index bf7fc5bcb..75391dbb9 100644
--- a/aggregator_api/src/routes.rs
+++ b/aggregator_api/src/routes.rs
@@ -172,7 +172,8 @@ pub(super) async fn post_task<C: Clock>(
             /* batch_mode */ req.batch_mode,
             /* vdaf */ req.vdaf,
             vdaf_verify_key,
-            /* task_expiration */ req.task_expiration,
+            /* task_start */ req.task_start,
+            /* task_end */ req.task_end,
             /* report_expiry_age */
             Some(Duration::from_seconds(3600 * 24 * 7 * 2)), // 2 weeks
             /* min_batch_size */ req.min_batch_size,
@@ -195,7 +196,8 @@ pub(super) async fn post_task<C: Clock>(
                 && existing_task.vdaf() == task.vdaf()
                 && existing_task.opaque_vdaf_verify_key() == task.opaque_vdaf_verify_key()
                 && existing_task.role() == task.role()
-                && existing_task.task_expiration() == task.task_expiration()
+                && existing_task.task_start() == task.task_start()
+                && existing_task.task_end() == task.task_end()
                 && existing_task.min_batch_size() == task.min_batch_size()
                 && existing_task.time_precision() == task.time_precision()
                 && existing_task.collector_hpke_config() == task.collector_hpke_config() {
@@ -266,9 +268,8 @@ pub(super) async fn patch_task<C: Clock>(
     let task = ds
         .run_tx("patch_task", |tx| {
             Box::pin(async move {
-                if let Some(task_expiration) = req.task_expiration {
-                    tx.update_task_expiration(&task_id, task_expiration.as_ref())
-                        .await?;
+                if let Some(task_end) = req.task_end {
+                    tx.update_task_end(&task_id, task_end.as_ref()).await?;
                 }
                 tx.get_aggregator_task(&task_id).await
             })
diff --git a/aggregator_api/src/tests.rs b/aggregator_api/src/tests.rs
index a07214e37..101782f83 100644
--- a/aggregator_api/src/tests.rs
+++ b/aggregator_api/src/tests.rs
@@ -192,7 +192,8 @@ async fn post_task_bad_role() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Collector,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: Some(Time::from_seconds_since_epoch(12340)),
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -225,7 +226,8 @@ async fn post_task_unauthorized() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Helper,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: None,
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -259,7 +261,8 @@ async fn post_task_helper_no_optional_fields() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Helper,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: None,
+        task_end: None,
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -308,7 +311,7 @@ async fn post_task_helper_no_optional_fields() {
     assert_eq!(&req.batch_mode, got_task.batch_mode());
     assert_eq!(&req.vdaf, got_task.vdaf());
     assert_eq!(&req.role, got_task.role());
-    assert_eq!(req.task_expiration.as_ref(), got_task.task_expiration());
+    assert_eq!(req.task_end.as_ref(), got_task.task_end());
     assert_eq!(req.min_batch_size, got_task.min_batch_size());
     assert_eq!(&req.time_precision, got_task.time_precision());
     assert!(got_task.aggregator_auth_token().is_none());
@@ -339,7 +342,8 @@ async fn post_task_helper_with_aggregator_auth_token() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Helper,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: None,
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -374,12 +378,12 @@ async fn post_task_idempotence() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Leader,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: Some(Time::from_seconds_since_epoch(12340)),
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
         aggregator_auth_token: Some(aggregator_auth_token.clone()),
-
         collector_auth_token_hash: Some(AuthenticationTokenHash::from(&random())),
     };
 
@@ -448,7 +452,8 @@ async fn post_task_leader_all_optional_fields() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Leader,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: None,
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -491,7 +496,7 @@ async fn post_task_leader_all_optional_fields() {
     assert_eq!(&req.vdaf, got_task.vdaf());
     assert_eq!(&req.role, got_task.role());
     assert_eq!(&vdaf_verify_key, got_task.opaque_vdaf_verify_key());
-    assert_eq!(req.task_expiration.as_ref(), got_task.task_expiration());
+    assert_eq!(req.task_end.as_ref(), got_task.task_end());
     assert_eq!(req.min_batch_size, got_task.min_batch_size());
     assert_eq!(&req.time_precision, got_task.time_precision());
     assert_eq!(
@@ -526,7 +531,8 @@ async fn post_task_leader_no_aggregator_auth_token() {
         vdaf: VdafInstance::Prio3Count,
         role: Role::Leader,
         vdaf_verify_key: URL_SAFE_NO_PAD.encode(&vdaf_verify_key),
-        task_expiration: Some(Time::from_seconds_since_epoch(12345)),
+        task_start: Some(Time::from_seconds_since_epoch(12340)),
+        task_end: Some(Time::from_seconds_since_epoch(12345)),
         min_batch_size: 223,
         time_precision: Duration::from_seconds(62),
         collector_hpke_config: HpkeKeypair::test().config().clone(),
@@ -686,7 +692,7 @@ async fn patch_task(#[case] role: Role) {
     let (handler, _ephemeral_datastore, ds) = setup_api_test().await;
 
     let task = TaskBuilder::new(BatchMode::TimeInterval, VdafInstance::Fake { rounds: 1 })
-        .with_task_expiration(Some(Time::from_seconds_since_epoch(1000)))
+        .with_task_end(Some(Time::from_seconds_since_epoch(1000)))
         .build()
         .view_for_role(role)
         .unwrap();
@@ -718,15 +724,15 @@ async fn patch_task(#[case] role: Role) {
         .await
         .unwrap();
     assert_eq!(
-        task.unwrap().task_expiration(),
+        task.unwrap().task_end(),
         Some(&Time::from_seconds_since_epoch(1000))
     );
 
-    // Verify: patching the task with a null task expiration returns the expected result.
+    // Verify: patching the task with a null task end time returns the expected result.
     let mut conn = patch(format!("/tasks/{}", task_id))
         .with_request_header("Authorization", format!("Bearer {AUTH_TOKEN}"))
         .with_request_header("Accept", CONTENT_TYPE)
-        .with_request_body(r#"{"task_expiration": null}"#)
+        .with_request_body(r#"{"task_end": null}"#)
         .run_async(&handler)
         .await;
     assert_status!(conn, Status::Ok);
@@ -739,19 +745,19 @@ async fn patch_task(#[case] role: Role) {
             .unwrap(),
     )
     .unwrap();
-    assert_eq!(got_task_resp.task_expiration, None);
+    assert_eq!(got_task_resp.task_end, None);
     let task = ds
         .run_unnamed_tx(|tx| Box::pin(async move { tx.get_aggregator_task(&task_id).await }))
         .await
         .unwrap();
-    assert_eq!(task.unwrap().task_expiration(), None);
+    assert_eq!(task.unwrap().task_end(), None);
 
-    // Verify: patching the task with a task expiration returns the expected result.
+    // Verify: patching the task with a task end time returns the expected result.
     let expected_time = Some(Time::from_seconds_since_epoch(2000));
     let mut conn = patch(format!("/tasks/{}", task_id))
         .with_request_header("Authorization", format!("Bearer {AUTH_TOKEN}"))
         .with_request_header("Accept", CONTENT_TYPE)
-        .with_request_body(r#"{"task_expiration": 2000}"#)
+        .with_request_body(r#"{"task_end": 2000}"#)
         .run_async(&handler)
         .await;
     assert_status!(conn, Status::Ok);
@@ -764,12 +770,12 @@ async fn patch_task(#[case] role: Role) {
             .unwrap(),
     )
     .unwrap();
-    assert_eq!(got_task_resp.task_expiration, expected_time);
+    assert_eq!(got_task_resp.task_end, expected_time);
     let task = ds
         .run_unnamed_tx(|tx| Box::pin(async move { tx.get_aggregator_task(&task_id).await }))
         .await
         .unwrap();
-    assert_eq!(task.unwrap().task_expiration(), expected_time.as_ref());
+    assert_eq!(task.unwrap().task_end(), expected_time.as_ref());
 
     // Verify: patching a nonexistent task returns NotFound.
     assert_response!(
@@ -832,7 +838,7 @@ async fn get_task_upload_metrics() {
             tx.increment_task_upload_counter(
                 &task_id,
                 1,
-                &TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 12),
+                &TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 22, 12),
             )
             .await
         })
@@ -847,7 +853,7 @@ async fn get_task_upload_metrics() {
             .await,
         Status::Ok,
         serde_json::to_string(&GetTaskUploadMetricsResp(
-            TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 12)
+            TaskUploadCounter::new_with_values(0, 0, 2, 4, 6, 100, 25, 22, 12)
         ))
         .unwrap(),
     );
@@ -1656,7 +1662,8 @@ fn post_task_req_serialization() {
             },
             role: Role::Helper,
             vdaf_verify_key: "encoded".to_owned(),
-            task_expiration: None,
+            task_start: None,
+            task_end: None,
             min_batch_size: 100,
             time_precision: Duration::from_seconds(3600),
             collector_hpke_config: HpkeConfig::new(
@@ -1672,7 +1679,7 @@ fn post_task_req_serialization() {
         &[
             Token::Struct {
                 name: "PostTaskReq",
-                len: 11,
+                len: 12,
             },
             Token::Str("peer_aggregator_endpoint"),
             Token::Str("https://example.com/"),
@@ -1713,7 +1720,9 @@ fn post_task_req_serialization() {
             },
             Token::Str("vdaf_verify_key"),
             Token::Str("encoded"),
-            Token::Str("task_expiration"),
+            Token::Str("task_start"),
+            Token::None,
+            Token::Str("task_end"),
             Token::None,
             Token::Str("min_batch_size"),
             Token::U64(100),
@@ -1771,7 +1780,8 @@ fn post_task_req_serialization() {
             },
             role: Role::Leader,
             vdaf_verify_key: "encoded".to_owned(),
-            task_expiration: Some(Time::from_seconds_since_epoch(1000)),
+            task_start: Some(Time::from_seconds_since_epoch(500)),
+            task_end: Some(Time::from_seconds_since_epoch(1000)),
             min_batch_size: 100,
             time_precision: Duration::from_seconds(3600),
             collector_hpke_config: HpkeConfig::new(
@@ -1791,7 +1801,7 @@ fn post_task_req_serialization() {
         &[
             Token::Struct {
                 name: "PostTaskReq",
-                len: 11,
+                len: 12,
             },
             Token::Str("peer_aggregator_endpoint"),
             Token::Str("https://example.com/"),
@@ -1832,7 +1842,11 @@ fn post_task_req_serialization() {
             },
             Token::Str("vdaf_verify_key"),
             Token::Str("encoded"),
-            Token::Str("task_expiration"),
+            Token::Str("task_start"),
+            Token::Some,
+            Token::NewtypeStruct { name: "Time" },
+            Token::U64(500),
+            Token::Str("task_end"),
             Token::Some,
             Token::NewtypeStruct { name: "Time" },
             Token::U64(1000),
@@ -1919,6 +1933,7 @@ fn task_resp_serialization() {
         SecretBytes::new(b"vdaf verify key!".to_vec()),
         None,
         None,
+        None,
         100,
         Duration::from_seconds(3600),
         Duration::from_seconds(60),
@@ -1946,7 +1961,7 @@ fn task_resp_serialization() {
         &[
             Token::Struct {
                 name: "TaskResp",
-                len: 13,
+                len: 14,
             },
             Token::Str("task_id"),
             Token::Str("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"),
@@ -1989,7 +2004,9 @@ fn task_resp_serialization() {
             },
             Token::Str("vdaf_verify_key"),
             Token::Str("dmRhZiB2ZXJpZnkga2V5IQ"),
-            Token::Str("task_expiration"),
+            Token::Str("task_start"),
+            Token::None,
+            Token::Str("task_end"),
             Token::None,
             Token::Str("report_expiry_age"),
             Token::None,
@@ -2039,14 +2056,16 @@ fn task_resp_serialization() {
 #[test]
 fn get_task_upload_metrics_serialization() {
     assert_ser_tokens(
-        &GetTaskUploadMetricsResp(TaskUploadCounter::new_with_values(0, 1, 2, 3, 4, 5, 6, 7)),
+        &GetTaskUploadMetricsResp(TaskUploadCounter::new_with_values(
+            0, 1, 2, 3, 4, 5, 6, 7, 8,
+        )),
         &[
             Token::NewtypeStruct {
                 name: "GetTaskUploadMetricsResp",
             },
             Token::Struct {
                 name: "TaskUploadCounter",
-                len: 8,
+                len: 9,
             },
             Token::Str("interval_collected"),
             Token::U64(0),
@@ -2062,8 +2081,10 @@ fn get_task_upload_metrics_serialization() {
             Token::U64(5),
             Token::Str("report_too_early"),
             Token::U64(6),
-            Token::Str("task_expired"),
+            Token::Str("task_not_started"),
             Token::U64(7),
+            Token::Str("task_ended"),
+            Token::U64(8),
             Token::StructEnd,
         ],
     )
diff --git a/aggregator_core/src/datastore.rs b/aggregator_core/src/datastore.rs
index fcfd8b0a5..fe0251e53 100644
--- a/aggregator_core/src/datastore.rs
+++ b/aggregator_core/src/datastore.rs
@@ -647,14 +647,14 @@ WHERE success = TRUE ORDER BY version DESC LIMIT(1)",
                 "-- put_aggregator_task()
 INSERT INTO tasks (
     task_id, aggregator_role, peer_aggregator_endpoint, batch_mode, vdaf,
-    task_expiration, report_expiry_age, min_batch_size, time_precision,
+    task_start, task_end, report_expiry_age, min_batch_size, time_precision,
     tolerable_clock_skew, collector_hpke_config, vdaf_verify_key,
     taskprov_task_info, aggregator_auth_token_type, aggregator_auth_token,
     aggregator_auth_token_hash, collector_auth_token_type,
     collector_auth_token_hash, created_at, updated_at, updated_by)
 VALUES (
     $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
-    $19, $20, $21
+    $19, $20, $21, $22
 )
 ON CONFLICT DO NOTHING",
             )
@@ -669,11 +669,13 @@ ON CONFLICT DO NOTHING",
                     &task.peer_aggregator_endpoint().as_str(),
                     /* batch_mode */ &Json(task.batch_mode()),
                     /* vdaf */ &Json(task.vdaf()),
-                    /* task_expiration */
+                    /* task_start */
                     &task
-                        .task_expiration()
+                        .task_start()
                         .map(Time::as_naive_date_time)
                         .transpose()?,
+                    /* task_end */
+                    &task.task_end().map(Time::as_naive_date_time).transpose()?,
                     /* report_expiry_age */
                     &task
                         .report_expiry_age()
@@ -760,17 +762,17 @@ DELETE FROM tasks WHERE task_id = $1",
         Ok(())
     }
 
-    /// Sets or unsets the expiration date of a task.
+    /// Sets or unsets the end date of a task.
     #[tracing::instrument(skip(self), err(level = Level::DEBUG))]
-    pub async fn update_task_expiration(
+    pub async fn update_task_end(
         &self,
         task_id: &TaskId,
-        task_expiration: Option<&Time>,
+        task_end: Option<&Time>,
     ) -> Result<(), Error> {
         let stmt = self
             .prepare_cached(
-                "-- update_task_expiration()
-UPDATE tasks SET task_expiration = $1, updated_at = $2, updated_by = $3
+                "-- update_task_end()
+UPDATE tasks SET task_end = $1, updated_at = $2, updated_by = $3
    WHERE task_id = $4",
             )
             .await?;
@@ -778,8 +780,8 @@ UPDATE tasks SET task_expiration = $1, updated_at = $2, updated_by = $3
             self.execute(
                 &stmt,
                 &[
-                    /* task_expiration */
-                    &task_expiration.map(Time::as_naive_date_time).transpose()?,
+                    /* task_end */
+                    &task_end.map(Time::as_naive_date_time).transpose()?,
                     /* updated_at */ &self.clock.now().as_naive_date_time()?,
                     /* updated_by */ &self.name,
                     /* task_id */ &task_id.as_ref(),
@@ -799,8 +801,9 @@ UPDATE tasks SET task_expiration = $1, updated_at = $2, updated_by = $3
         let stmt = self
             .prepare_cached(
                 "-- get_aggregator_task()
-SELECT aggregator_role, peer_aggregator_endpoint, batch_mode, vdaf,
-    task_expiration, report_expiry_age, min_batch_size, time_precision,
+SELECT
+    aggregator_role, peer_aggregator_endpoint, batch_mode, vdaf, task_start,
+    task_end, report_expiry_age, min_batch_size, time_precision,
     tolerable_clock_skew, collector_hpke_config, vdaf_verify_key,
     taskprov_task_info, aggregator_auth_token_type, aggregator_auth_token,
     aggregator_auth_token_hash, collector_auth_token_type, collector_auth_token_hash
@@ -820,8 +823,9 @@ FROM tasks WHERE task_id = $1",
         let stmt = self
             .prepare_cached(
                 "-- get_aggregator_tasks()
-SELECT task_id, aggregator_role, peer_aggregator_endpoint, batch_mode, vdaf,
-    task_expiration, report_expiry_age, min_batch_size, time_precision,
+SELECT
+    task_id, aggregator_role, peer_aggregator_endpoint, batch_mode, vdaf, task_start,
+    task_end, report_expiry_age, min_batch_size, time_precision,
     tolerable_clock_skew, collector_hpke_config, vdaf_verify_key,
     taskprov_task_info, aggregator_auth_token_type, aggregator_auth_token,
     aggregator_auth_token_hash, collector_auth_token_type, collector_auth_token_hash
@@ -843,8 +847,12 @@ FROM tasks",
         let peer_aggregator_endpoint = row.get::<_, String>("peer_aggregator_endpoint").parse()?;
         let batch_mode = row.try_get::<_, Json<task::BatchMode>>("batch_mode")?.0;
         let vdaf = row.try_get::<_, Json<VdafInstance>>("vdaf")?.0;
-        let task_expiration = row
-            .get::<_, Option<NaiveDateTime>>("task_expiration")
+        let task_start = row
+            .get::<_, Option<NaiveDateTime>>("task_start")
+            .as_ref()
+            .map(Time::from_naive_date_time);
+        let task_end = row
+            .get::<_, Option<NaiveDateTime>>("task_end")
             .as_ref()
             .map(Time::from_naive_date_time);
         let report_expiry_age = row
@@ -942,7 +950,8 @@ FROM tasks",
             batch_mode,
             vdaf,
             vdaf_verify_key,
-            task_expiration,
+            task_start,
+            task_end,
             report_expiry_age,
             min_batch_size,
             time_precision,
@@ -5312,7 +5321,8 @@ SELECT
     COALESCE(SUM(report_outdated_key)::BIGINT, 0) AS report_outdated_key,
     COALESCE(SUM(report_success)::BIGINT, 0) AS report_success,
     COALESCE(SUM(report_too_early)::BIGINT, 0) AS report_too_early,
-    COALESCE(SUM(task_expired)::BIGINT, 0) AS task_expired
+    COALESCE(SUM(task_not_started)::BIGINT, 0) AS task_not_started,
+    COALESCE(SUM(task_ended)::BIGINT, 0) AS task_ended
 FROM task_upload_counters
 RIGHT JOIN tasks on tasks.id = task_upload_counters.task_id
 WHERE tasks.task_id = $1
@@ -5331,7 +5341,8 @@ GROUP BY tasks.id",
                     report_outdated_key: row.get_bigint_and_convert("report_outdated_key")?,
                     report_success: row.get_bigint_and_convert("report_success")?,
                     report_too_early: row.get_bigint_and_convert("report_too_early")?,
-                    task_expired: row.get_bigint_and_convert("task_expired")?,
+                    task_not_started: row.get_bigint_and_convert("task_not_started")?,
+                    task_ended: row.get_bigint_and_convert("task_ended")?,
                 })
             })
             .transpose()
@@ -5347,12 +5358,13 @@ GROUP BY tasks.id",
         ord: u64,
         counter: &TaskUploadCounter,
     ) -> Result<(), Error> {
-        let stmt =
-            "-- increment_task_upload_counter()
-INSERT INTO task_upload_counters (task_id, ord, interval_collected, report_decode_failure,
-        report_decrypt_failure, report_expired, report_outdated_key, report_success, report_too_early,
-        task_expired)
-VALUES ((SELECT id FROM tasks WHERE task_id = $1), $2, $3, $4, $5, $6, $7, $8, $9, $10)
+        let stmt = "-- increment_task_upload_counter()
+INSERT INTO task_upload_counters (
+    task_id, ord, interval_collected, report_decode_failure,
+    report_decrypt_failure, report_expired, report_outdated_key, report_success, report_too_early,
+    task_not_started, task_ended
+)
+VALUES ((SELECT id FROM tasks WHERE task_id = $1), $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
 ON CONFLICT (task_id, ord) DO UPDATE SET
     interval_collected = task_upload_counters.interval_collected + $3,
     report_decode_failure = task_upload_counters.report_decode_failure + $4,
@@ -5361,7 +5373,8 @@ ON CONFLICT (task_id, ord) DO UPDATE SET
     report_outdated_key = task_upload_counters.report_outdated_key + $7,
     report_success = task_upload_counters.report_success + $8,
     report_too_early = task_upload_counters.report_too_early + $9,
-    task_expired = task_upload_counters.task_expired + $10";
+    task_not_started = task_upload_counters.task_not_started + $10,
+    task_ended = task_upload_counters.task_ended + $11";
 
         let stmt = self.prepare_cached(stmt).await?;
         check_single_row_mutation(
@@ -5377,7 +5390,8 @@ ON CONFLICT (task_id, ord) DO UPDATE SET
                     &i64::try_from(counter.report_outdated_key)?,
                     &i64::try_from(counter.report_success)?,
                     &i64::try_from(counter.report_too_early)?,
-                    &i64::try_from(counter.task_expired)?,
+                    &i64::try_from(counter.task_not_started)?,
+                    &i64::try_from(counter.task_ended)?,
                 ],
             )
             .await?,
diff --git a/aggregator_core/src/datastore/models.rs b/aggregator_core/src/datastore/models.rs
index 97a02de6d..4b9f69fb8 100644
--- a/aggregator_core/src/datastore/models.rs
+++ b/aggregator_core/src/datastore/models.rs
@@ -2283,8 +2283,10 @@ pub struct TaskUploadCounter {
     pub(crate) report_success: u64,
     /// Reports that contain a timestamp too far in the future.
     pub(crate) report_too_early: u64,
-    /// Reports that were submitted to the task after the task's expiry.
-    pub(crate) task_expired: u64,
+    /// Reports that were submitted to the task before the task's start time.
+    pub(crate) task_not_started: u64,
+    /// Reports that were submitted to the task after the task's end time.
+    pub(crate) task_ended: u64,
 }
 
 impl TaskUploadCounter {
@@ -2298,7 +2300,8 @@ impl TaskUploadCounter {
         report_outdated_key: u64,
         report_success: u64,
         report_too_early: u64,
-        task_expired: u64,
+        task_not_started: u64,
+        task_ended: u64,
     ) -> Self {
         Self {
             interval_collected,
@@ -2308,7 +2311,8 @@ impl TaskUploadCounter {
             report_outdated_key,
             report_success,
             report_too_early,
-            task_expired,
+            task_not_started,
+            task_ended,
         }
     }
 
@@ -2340,8 +2344,12 @@ impl TaskUploadCounter {
         self.report_too_early += 1
     }
 
-    pub fn increment_task_expired(&mut self) {
-        self.task_expired += 1
+    pub fn increment_task_not_started(&mut self) {
+        self.task_not_started += 1
+    }
+
+    pub fn increment_task_ended(&mut self) {
+        self.task_ended += 1
     }
 
     pub fn interval_collected(&self) -> u64 {
@@ -2372,8 +2380,12 @@ impl TaskUploadCounter {
         self.report_too_early
     }
 
-    pub fn task_expired(&self) -> u64 {
-        self.task_expired
+    pub fn task_not_started(&self) -> u64 {
+        self.task_not_started
+    }
+
+    pub fn task_ended(&self) -> u64 {
+        self.task_ended
     }
 }
 
diff --git a/aggregator_core/src/datastore/tests.rs b/aggregator_core/src/datastore/tests.rs
index 7aa055872..9f2c6d60c 100644
--- a/aggregator_core/src/datastore/tests.rs
+++ b/aggregator_core/src/datastore/tests.rs
@@ -196,6 +196,8 @@ async fn roundtrip_task(ephemeral_datastore: EphemeralDatastore) {
         ),
     ] {
         let task = TaskBuilder::new(task::BatchMode::TimeInterval, vdaf)
+            .with_task_start(Some(Time::from_seconds_since_epoch(1000)))
+            .with_task_end(Some(Time::from_seconds_since_epoch(4000)))
             .with_report_expiry_age(Some(Duration::from_seconds(3600)))
             .build()
             .view_for_role(role)
@@ -289,12 +291,12 @@ async fn roundtrip_task(ephemeral_datastore: EphemeralDatastore) {
 
 #[rstest_reuse::apply(schema_versions_template)]
 #[tokio::test]
-async fn update_task_expiration(ephemeral_datastore: EphemeralDatastore) {
+async fn update_task_end(ephemeral_datastore: EphemeralDatastore) {
     install_test_trace_subscriber();
     let ds = ephemeral_datastore.datastore(MockClock::default()).await;
 
     let task = TaskBuilder::new(task::BatchMode::TimeInterval, VdafInstance::Prio3Count)
-        .with_task_expiration(Some(Time::from_seconds_since_epoch(1000)))
+        .with_task_end(Some(Time::from_seconds_since_epoch(1000)))
         .build()
         .leader_view()
         .unwrap();
@@ -305,27 +307,27 @@ async fn update_task_expiration(ephemeral_datastore: EphemeralDatastore) {
         Box::pin(async move {
             let task = tx.get_aggregator_task(&task_id).await.unwrap().unwrap();
             assert_eq!(
-                task.task_expiration().cloned(),
+                task.task_end().cloned(),
                 Some(Time::from_seconds_since_epoch(1000))
             );
 
-            tx.update_task_expiration(&task_id, Some(&Time::from_seconds_since_epoch(2000)))
+            tx.update_task_end(&task_id, Some(&Time::from_seconds_since_epoch(2000)))
                 .await
                 .unwrap();
 
             let task = tx.get_aggregator_task(&task_id).await.unwrap().unwrap();
             assert_eq!(
-                task.task_expiration().cloned(),
+                task.task_end().cloned(),
                 Some(Time::from_seconds_since_epoch(2000))
             );
 
-            tx.update_task_expiration(&task_id, None).await.unwrap();
+            tx.update_task_end(&task_id, None).await.unwrap();
 
             let task = tx.get_aggregator_task(&task_id).await.unwrap().unwrap();
-            assert_eq!(task.task_expiration().cloned(), None);
+            assert_eq!(task.task_end().cloned(), None);
 
             let result = tx
-                .update_task_expiration(&random(), Some(&Time::from_seconds_since_epoch(2000)))
+                .update_task_end(&random(), Some(&Time::from_seconds_since_epoch(2000)))
                 .await;
             assert_matches!(result, Err(Error::MutationTargetNotFound));
 
@@ -7472,7 +7474,7 @@ async fn roundtrip_task_upload_counter(ephemeral_datastore: EphemeralDatastore)
                 tx.increment_task_upload_counter(
                     &task_id,
                     ord,
-                    &TaskUploadCounter::new_with_values(2, 4, 6, 8, 10, 100, 25, 12),
+                    &TaskUploadCounter::new_with_values(2, 4, 6, 8, 10, 100, 25, 22, 12),
                 )
                 .await
                 .unwrap();
@@ -7481,7 +7483,7 @@ async fn roundtrip_task_upload_counter(ephemeral_datastore: EphemeralDatastore)
                 tx.increment_task_upload_counter(
                     &task_id,
                     ord,
-                    &TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 0, 0, 8),
+                    &TaskUploadCounter::new_with_values(0, 0, 0, 0, 0, 0, 0, 0, 8),
                 )
                 .await
                 .unwrap();
@@ -7502,7 +7504,8 @@ async fn roundtrip_task_upload_counter(ephemeral_datastore: EphemeralDatastore)
                         report_outdated_key: 10,
                         report_success: 100,
                         report_too_early: 25,
-                        task_expired: 20,
+                        task_not_started: 22,
+                        task_ended: 20,
                     })
                 );
 
diff --git a/aggregator_core/src/task.rs b/aggregator_core/src/task.rs
index f75d38918..ab81e773f 100644
--- a/aggregator_core/src/task.rs
+++ b/aggregator_core/src/task.rs
@@ -99,8 +99,10 @@ struct CommonTaskParameters {
     vdaf: VdafInstance,
     /// Secret verification key shared by the aggregators.
     vdaf_verify_key: SecretBytes,
+    /// The time before which the task is considered invalid.
+    task_start: Option<Time>,
     /// The time after which the task is considered invalid.
-    task_expiration: Option<Time>,
+    task_end: Option<Time>,
     /// The age after which a report is considered to be "expired" and will be considered a
     /// candidate for garbage collection. A value of `None` indicates that garbage collection is
     /// disabled.
@@ -128,7 +130,8 @@ impl CommonTaskParameters {
         batch_mode: BatchMode,
         vdaf: VdafInstance,
         vdaf_verify_key: SecretBytes,
-        task_expiration: Option<Time>,
+        task_start: Option<Time>,
+        task_end: Option<Time>,
         report_expiry_age: Option<Duration>,
         min_batch_size: u64,
         time_precision: Duration,
@@ -155,10 +158,20 @@ impl CommonTaskParameters {
                 return Err(Error::InvalidParameter("report_expiry_age too large"));
             }
         }
-        if let Some(task_expiration) = task_expiration {
-            task_expiration
+        if let Some(task_start) = task_start {
+            task_start
                 .as_naive_date_time()
-                .map_err(|_| Error::InvalidParameter("task_expiration out of range"))?;
+                .map_err(|_| Error::InvalidParameter("task_start out of range"))?;
+        }
+        if let Some(task_end) = task_end {
+            task_end
+                .as_naive_date_time()
+                .map_err(|_| Error::InvalidParameter("task_end out of range"))?;
+        }
+        if let (Some(task_start), Some(task_end)) = (task_start, task_end) {
+            if task_end < task_start {
+                return Err(Error::InvalidParameter("task_end before task_start"));
+            }
         }
 
         if time_precision.as_seconds() == 0 {
@@ -170,7 +183,8 @@ impl CommonTaskParameters {
             batch_mode,
             vdaf,
             vdaf_verify_key,
-            task_expiration,
+            task_start,
+            task_end,
             report_expiry_age,
             min_batch_size,
             time_precision,
@@ -213,7 +227,8 @@ impl AggregatorTask {
         batch_mode: BatchMode,
         vdaf: VdafInstance,
         vdaf_verify_key: SecretBytes,
-        task_expiration: Option<Time>,
+        task_start: Option<Time>,
+        task_end: Option<Time>,
         report_expiry_age: Option<Duration>,
         min_batch_size: u64,
         time_precision: Duration,
@@ -225,7 +240,8 @@ impl AggregatorTask {
             batch_mode,
             vdaf,
             vdaf_verify_key,
-            task_expiration,
+            task_start,
+            task_end,
             report_expiry_age,
             min_batch_size,
             time_precision,
@@ -300,9 +316,14 @@ impl AggregatorTask {
         &self.common_parameters.vdaf_verify_key
     }
 
-    /// Retrieves the task expiration associated with this task.
-    pub fn task_expiration(&self) -> Option<&Time> {
-        self.common_parameters.task_expiration.as_ref()
+    /// Retrieves the task start time associated with this task.
+    pub fn task_start(&self) -> Option<&Time> {
+        self.common_parameters.task_start.as_ref()
+    }
+
+    /// Retrieves the task end time associated with this task.
+    pub fn task_end(&self) -> Option<&Time> {
+        self.common_parameters.task_end.as_ref()
     }
 
     /// Retrieves the report expiry age associated with this task.
@@ -549,7 +570,8 @@ pub struct SerializedAggregatorTask {
     vdaf: VdafInstance,
     role: Role,
     vdaf_verify_key: Option<String>, // in unpadded base64url
-    task_expiration: Option<Time>,
+    task_start: Option<Time>,
+    task_end: Option<Time>,
     report_expiry_age: Option<Duration>,
     min_batch_size: u64,
     time_precision: Duration,
@@ -605,7 +627,8 @@ impl Serialize for AggregatorTask {
             vdaf: self.vdaf().clone(),
             role: *self.role(),
             vdaf_verify_key: Some(URL_SAFE_NO_PAD.encode(self.opaque_vdaf_verify_key())),
-            task_expiration: self.task_expiration().copied(),
+            task_start: self.task_start().copied(),
+            task_end: self.task_end().copied(),
             report_expiry_age: self.report_expiry_age().copied(),
             min_batch_size: self.min_batch_size(),
             time_precision: *self.time_precision(),
@@ -668,7 +691,8 @@ impl TryFrom<SerializedAggregatorTask> for AggregatorTask {
             serialized_task.batch_mode,
             serialized_task.vdaf,
             SecretBytes::new(URL_SAFE_NO_PAD.decode(vdaf_verify_key)?),
-            serialized_task.task_expiration,
+            serialized_task.task_start,
+            serialized_task.task_end,
             serialized_task.report_expiry_age,
             serialized_task.min_batch_size,
             serialized_task.time_precision,
@@ -747,7 +771,8 @@ pub mod test_util {
             batch_mode: BatchMode,
             vdaf: VdafInstance,
             vdaf_verify_key: SecretBytes,
-            task_expiration: Option<Time>,
+            task_start: Option<Time>,
+            task_end: Option<Time>,
             report_expiry_age: Option<Duration>,
             min_batch_size: u64,
             time_precision: Duration,
@@ -775,7 +800,8 @@ pub mod test_util {
                     batch_mode,
                     vdaf,
                     vdaf_verify_key,
-                    task_expiration,
+                    task_start,
+                    task_end,
                     report_expiry_age,
                     min_batch_size,
                     time_precision,
@@ -825,9 +851,14 @@ pub mod test_util {
             &self.common_parameters.vdaf_verify_key
         }
 
-        /// Retrieves the task expiration associated with this task.
-        pub fn task_expiration(&self) -> Option<&Time> {
-            self.common_parameters.task_expiration.as_ref()
+        /// Retrieves the task start time associated with this task.
+        pub fn task_start(&self) -> Option<&Time> {
+            self.common_parameters.task_start.as_ref()
+        }
+
+        /// Retrieves the task end time associated with this task.
+        pub fn task_end(&self) -> Option<&Time> {
+            self.common_parameters.task_end.as_ref()
         }
 
         /// Retrieves the report expiry age associated with this task.
@@ -1009,6 +1040,7 @@ pub mod test_util {
                 vdaf_verify_key,
                 None,
                 None,
+                None,
                 1,
                 Duration::from_hours(8).unwrap(),
                 Duration::from_minutes(10).unwrap(),
@@ -1168,11 +1200,22 @@ pub mod test_util {
             })
         }
 
-        /// Sets the task expiration time.
-        pub fn with_task_expiration(self, task_expiration: Option<Time>) -> Self {
+        /// Sets the task start time.
+        pub fn with_task_start(self, task_start: Option<Time>) -> Self {
             Self(Task {
                 common_parameters: CommonTaskParameters {
-                    task_expiration,
+                    task_start,
+                    ..self.0.common_parameters
+                },
+                ..self.0
+            })
+        }
+
+        /// Sets the task end time.
+        pub fn with_task_end(self, task_end: Option<Time>) -> Self {
+            Self(Task {
+                common_parameters: CommonTaskParameters {
+                    task_end,
                     ..self.0.common_parameters
                 },
                 ..self.0
@@ -1231,7 +1274,8 @@ mod tests {
         vdaf::vdaf_dp_strategies,
     };
     use janus_messages::{
-        Duration, HpkeAeadId, HpkeConfig, HpkeConfigId, HpkeKdfId, HpkeKemId, HpkePublicKey, TaskId,
+        Duration, HpkeAeadId, HpkeConfig, HpkeConfigId, HpkeKdfId, HpkeKemId, HpkePublicKey,
+        TaskId, Time,
     };
     use rand::random;
     use serde_json::json;
@@ -1353,6 +1397,7 @@ mod tests {
                 SecretBytes::new(b"1234567812345678".to_vec()),
                 None,
                 None,
+                None,
                 10,
                 Duration::from_seconds(3600),
                 Duration::from_seconds(60),
@@ -1378,7 +1423,7 @@ mod tests {
             &[
                 Token::Struct {
                     name: "SerializedAggregatorTask",
-                    len: 15,
+                    len: 16,
                 },
                 Token::Str("task_id"),
                 Token::Some,
@@ -1403,7 +1448,9 @@ mod tests {
                 Token::Str("vdaf_verify_key"),
                 Token::Some,
                 Token::Str("MTIzNDU2NzgxMjM0NTY3OA"),
-                Token::Str("task_expiration"),
+                Token::Str("task_start"),
+                Token::None,
+                Token::Str("task_end"),
                 Token::None,
                 Token::Str("report_expiry_age"),
                 Token::None,
@@ -1491,7 +1538,8 @@ mod tests {
                     dp_strategy: vdaf_dp_strategies::Prio3SumVec::NoDifferentialPrivacy,
                 },
                 SecretBytes::new(b"1234567812345678".to_vec()),
-                None,
+                Some(Time::from_seconds_since_epoch(1000)),
+                Some(Time::from_seconds_since_epoch(2000)),
                 Some(Duration::from_seconds(1800)),
                 10,
                 Duration::from_seconds(3600),
@@ -1516,7 +1564,7 @@ mod tests {
             &[
                 Token::Struct {
                     name: "SerializedAggregatorTask",
-                    len: 15,
+                    len: 16,
                 },
                 Token::Str("task_id"),
                 Token::Some,
@@ -1561,8 +1609,14 @@ mod tests {
                 Token::Str("vdaf_verify_key"),
                 Token::Some,
                 Token::Str("MTIzNDU2NzgxMjM0NTY3OA"),
-                Token::Str("task_expiration"),
-                Token::None,
+                Token::Str("task_start"),
+                Token::Some,
+                Token::NewtypeStruct { name: "Time" },
+                Token::U64(1000),
+                Token::Str("task_end"),
+                Token::Some,
+                Token::NewtypeStruct { name: "Time" },
+                Token::U64(2000),
                 Token::Str("report_expiry_age"),
                 Token::Some,
                 Token::NewtypeStruct { name: "Duration" },
diff --git a/core/src/time.rs b/core/src/time.rs
index ffb95bea7..6fdffbed0 100644
--- a/core/src/time.rs
+++ b/core/src/time.rs
@@ -203,7 +203,10 @@ pub trait TimeExt: Sized {
     /// `self` is before `other`, the result is zero.
     fn saturating_difference(&self, other: &Self) -> Duration;
 
-    /// Returns true if this [`Time`] occurs after `time`.
+    /// Returns true if and only if this [`Time`] occurs before `time`.
+    fn is_before(&self, time: &Time) -> bool;
+
+    /// Returns true if and only if this [`Time`] occurs after `time`.
     fn is_after(&self, time: &Time) -> bool;
 }
 
@@ -272,6 +275,10 @@ impl TimeExt for Time {
         )
     }
 
+    fn is_before(&self, time: &Time) -> bool {
+        self.as_seconds_since_epoch() < time.as_seconds_since_epoch()
+    }
+
     fn is_after(&self, time: &Time) -> bool {
         self.as_seconds_since_epoch() > time.as_seconds_since_epoch()
     }
diff --git a/db/00000000000001_initial_schema.up.sql b/db/00000000000001_initial_schema.up.sql
index 5c9b81b38..7e89725a9 100644
--- a/db/00000000000001_initial_schema.up.sql
+++ b/db/00000000000001_initial_schema.up.sql
@@ -98,7 +98,8 @@ CREATE TABLE tasks(
     peer_aggregator_endpoint    TEXT NOT NULL,             -- peer aggregator's API endpoint
     batch_mode                  JSONB NOT NULL,            -- the batch mode in use for this task, along with its parameters
     vdaf                        JSON NOT NULL,             -- the VDAF instance in use for this task, along with its parameters
-    task_expiration             TIMESTAMP,                 -- the time after which client reports are no longer accepted
+    task_start                  TIMESTAMP,                 -- the time before which client reports are not accepted
+    task_end                    TIMESTAMP,                 -- the time after which client reports are no longer accepted
     report_expiry_age           BIGINT,                    -- the maximum age of a report before it is considered expired (and acceptable for garbage collection), in seconds. NULL means that GC is disabled.
     min_batch_size              BIGINT NOT NULL,           -- the minimum number of reports in a batch to allow it to be collected
     time_precision              BIGINT NOT NULL,           -- the duration to which clients are expected to round their report timestamps, in seconds
@@ -149,17 +150,17 @@ CREATE INDEX task_id_index ON tasks(task_id);
 CREATE TABLE task_upload_counters(
     id                     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,  -- artificial ID, internal-only
     task_id                BIGINT NOT NULL,
-
-    interval_collected     BIGINT NOT NULL DEFAULT 0, -- Reports submitted for an interval that was already collected.
-    report_decode_failure  BIGINT NOT NULL DEFAULT 0, -- Reports which failed to decode.
-    report_decrypt_failure BIGINT NOT NULL DEFAULT 0, -- Reports which failed to decrypt.
-    report_expired         BIGINT NOT NULL DEFAULT 0, -- Reports that were older than the task's report_expiry_age.
-    report_outdated_key    BIGINT NOT NULL DEFAULT 0, -- Reports that were encrypted with an unknown or outdated HPKE key.
-    report_success         BIGINT NOT NULL DEFAULT 0, -- Reports that were successfully uploaded.
-    report_too_early       BIGINT NOT NULL DEFAULT 0, -- Reports whose timestamp is too far in the future.
-    task_expired           BIGINT NOT NULL DEFAULT 0, -- Reports sent to the task while it is expired.
-
-    ord                    BIGINT NOT NULL,           -- Index of this task_upload_counters shard.
+    ord                    BIGINT NOT NULL,           -- Index of this task_upload_counters shard.s
+
+    interval_collected     BIGINT NOT NULL DEFAULT 0, -- reports submitted for an interval that was already collected.
+    report_decode_failure  BIGINT NOT NULL DEFAULT 0, -- reports which failed to decode.
+    report_decrypt_failure BIGINT NOT NULL DEFAULT 0, -- reports which failed to decrypt.
+    report_expired         BIGINT NOT NULL DEFAULT 0, -- reports that were older than the task's report_expiry_age.
+    report_outdated_key    BIGINT NOT NULL DEFAULT 0, -- reports that were encrypted with an unknown or outdated HPKE key.
+    report_success         BIGINT NOT NULL DEFAULT 0, -- reports that were successfully uploaded.
+    report_too_early       BIGINT NOT NULL DEFAULT 0, -- reports whose timestamp is too far in the future.
+    task_not_started       BIGINT NOT NULL DEFAULT 0, -- reports sent to the task before it started.
+    task_ended             BIGINT NOT NULL DEFAULT 0, -- reports sent to the task after it ended.
 
     CONSTRAINT fk_task_id FOREIGN KEY(task_id) REFERENCES tasks(id) ON DELETE CASCADE,
     CONSTRAINT task_upload_counters_unique UNIQUE(task_id, ord)
diff --git a/docs/samples/tasks.yaml b/docs/samples/tasks.yaml
index 994b4e295..d88462f98 100644
--- a/docs/samples/tasks.yaml
+++ b/docs/samples/tasks.yaml
@@ -24,8 +24,8 @@
   # determined by the task's VDAF.
   vdaf_verify_key: "1CmuYNtBLYIoXN8bU0T_XA"
 
-  # The task's expiration time, as a number of seconds after the Unix epoch.
-  task_expiration: 1704088800
+  # The task's end time, as a number of seconds after the Unix epoch.
+  task_end: 1704088800
 
   # Time in seconds after which reports expire and may be garbage collected.
   # This is a Janus-specific parameter. Garbage collection for a task may
@@ -99,7 +99,7 @@
   vdaf: Prio3Count
   role: Helper
   vdaf_verify_key: "ZXtE4kLqtsCOr8h_pNUeoQ"
-  task_expiration: 1704088800
+  task_end: 1704088800
   report_expiry_age: null
   min_batch_size: 100
   time_precision: 300
diff --git a/integration_tests/src/daphne.rs b/integration_tests/src/daphne.rs
index 3a5bf86c9..8d051ce49 100644
--- a/integration_tests/src/daphne.rs
+++ b/integration_tests/src/daphne.rs
@@ -113,11 +113,11 @@ impl Daphne {
             );
         }
 
-        // Daphne does not support unset task expiration values. Work around this by specifying an
-        // arbitrary, far-future task expiration time, instead.
-        let task = if task.task_expiration().is_none() {
+        // Daphne does not support unset task end time values. Work around this by specifying an
+        // arbitrary, far-future task end time, instead.
+        let task = if task.task_end().is_none() {
             TaskBuilder::from(task.clone())
-                .with_task_expiration(Some(Time::from_seconds_since_epoch(2000000000)))
+                .with_task_end(Some(Time::from_seconds_since_epoch(2000000000)))
                 .build()
         } else {
             task.clone()
diff --git a/interop_binaries/src/commands/janus_interop_aggregator.rs b/interop_binaries/src/commands/janus_interop_aggregator.rs
index 2358744f2..9e794ac0d 100644
--- a/interop_binaries/src/commands/janus_interop_aggregator.rs
+++ b/interop_binaries/src/commands/janus_interop_aggregator.rs
@@ -102,7 +102,8 @@ async fn handle_add_task(
         batch_mode,
         vdaf,
         vdaf_verify_key,
-        request.task_expiration.map(Time::from_seconds_since_epoch),
+        request.task_start.map(Time::from_seconds_since_epoch),
+        request.task_end.map(Time::from_seconds_since_epoch),
         None,
         request.min_batch_size,
         time_precision,
diff --git a/interop_binaries/src/lib.rs b/interop_binaries/src/lib.rs
index 4258120f6..34a60083a 100644
--- a/interop_binaries/src/lib.rs
+++ b/interop_binaries/src/lib.rs
@@ -299,7 +299,8 @@ pub struct AggregatorAddTaskRequest {
     pub min_batch_size: u64,
     pub time_precision: u64,           // in seconds
     pub collector_hpke_config: String, // in unpadded base64url
-    pub task_expiration: Option<u64>,  // in seconds since the epoch
+    pub task_start: Option<u64>,       // in seconds since the epoch
+    pub task_end: Option<u64>,         // in seconds since the epoch
 }
 
 impl AggregatorAddTaskRequest {
@@ -333,7 +334,8 @@ impl AggregatorAddTaskRequest {
                     .get_encoded()
                     .unwrap(),
             ),
-            task_expiration: task.task_expiration().map(Time::as_seconds_since_epoch),
+            task_start: task.task_start().map(Time::as_seconds_since_epoch),
+            task_end: task.task_end().map(Time::as_seconds_since_epoch),
         }
     }
 }
diff --git a/messages/src/lib.rs b/messages/src/lib.rs
index 3c085e285..33cfbf2e2 100644
--- a/messages/src/lib.rs
+++ b/messages/src/lib.rs
@@ -2145,6 +2145,7 @@ pub enum ReportError {
     TaskExpired = 7,
     InvalidMessage = 8,
     ReportTooEarly = 9,
+    TaskNotStarted = 10,
 }
 
 impl Encode for ReportError {
diff --git a/messages/src/taskprov.rs b/messages/src/taskprov.rs
index a56e5c856..cec6ac801 100644
--- a/messages/src/taskprov.rs
+++ b/messages/src/taskprov.rs
@@ -24,7 +24,7 @@ pub struct TaskConfig {
     /// Determines the properties that all batches for this task must have.
     query_config: QueryConfig,
     /// Time up to which Clients are expected to upload to this task.
-    task_expiration: Time,
+    task_end: Time,
     /// Determines VDAF type and all properties.
     vdaf_config: VdafConfig,
 }
@@ -35,7 +35,7 @@ impl TaskConfig {
         leader_aggregator_endpoint: Url,
         helper_aggregator_endpoint: Url,
         query_config: QueryConfig,
-        task_expiration: Time,
+        task_end: Time,
         vdaf_config: VdafConfig,
     ) -> Result<Self, Error> {
         if task_info.is_empty() {
@@ -47,7 +47,7 @@ impl TaskConfig {
             leader_aggregator_endpoint,
             helper_aggregator_endpoint,
             query_config,
-            task_expiration,
+            task_end,
             vdaf_config,
         })
     }
@@ -68,8 +68,8 @@ impl TaskConfig {
         &self.query_config
     }
 
-    pub fn task_expiration(&self) -> &Time {
-        &self.task_expiration
+    pub fn task_end(&self) -> &Time {
+        &self.task_end
     }
 
     pub fn vdaf_config(&self) -> &VdafConfig {
@@ -83,7 +83,7 @@ impl Encode for TaskConfig {
         self.leader_aggregator_endpoint.encode(bytes)?;
         self.helper_aggregator_endpoint.encode(bytes)?;
         encode_u16_items(bytes, &(), &self.query_config.get_encoded()?)?;
-        self.task_expiration.encode(bytes)?;
+        self.task_end.encode(bytes)?;
         encode_u16_items(bytes, &(), &self.vdaf_config.get_encoded()?)?;
         Ok(())
     }
@@ -94,7 +94,7 @@ impl Encode for TaskConfig {
                 + self.leader_aggregator_endpoint.encoded_len()?
                 + self.helper_aggregator_endpoint.encoded_len()?
                 + (2 + self.query_config.encoded_len()?)
-                + self.task_expiration.encoded_len()?
+                + self.task_end.encoded_len()?
                 + (2 + self.vdaf_config.encoded_len()?),
         )
     }
@@ -111,7 +111,7 @@ impl Decode for TaskConfig {
         let leader_aggregator_endpoint = Url::decode(bytes)?;
         let helper_aggregator_endpoint = Url::decode(bytes)?;
         let query_config = QueryConfig::get_decoded(&decode_u16_items(&(), bytes)?)?;
-        let task_expiration = Time::decode(bytes)?;
+        let task_end = Time::decode(bytes)?;
         let vdaf_config = VdafConfig::get_decoded(&decode_u16_items(&(), bytes)?)?;
 
         Ok(Self {
@@ -119,7 +119,7 @@ impl Decode for TaskConfig {
             leader_aggregator_endpoint,
             helper_aggregator_endpoint,
             query_config,
-            task_expiration,
+            task_end,
             vdaf_config,
         })
     }
@@ -904,7 +904,7 @@ mod tests {
                         "0000CCCC",         // min_batch_size
                         "02",               // batch_mode
                     ),
-                    "000000000000EEEE", // task_expiration
+                    "000000000000EEEE", // task_end
                     concat!(
                         // vdaf_config
                         "0007", // vdaf_config length
@@ -960,7 +960,7 @@ mod tests {
                         "0000CCCC",         // min_batch_size
                         "01",               // batch_mode
                     ),
-                    "000000000000EEEE", // task_expiration
+                    "000000000000EEEE", // task_end
                     concat!(
                         // vdaf_config
                         "000F", // vdaf_config length
@@ -1005,7 +1005,7 @@ mod tests {
                         "0000CCCC",         // min_batch_size
                         "01",               // batch_mode
                     ),
-                    "000000000000EEEE", // task_expiration
+                    "000000000000EEEE", // task_end
                     concat!(
                         // vdaf_config
                         "0007", // vdaf_config length
diff --git a/messages/src/tests/aggregation.rs b/messages/src/tests/aggregation.rs
index 5f898e058..942341495 100644
--- a/messages/src/tests/aggregation.rs
+++ b/messages/src/tests/aggregation.rs
@@ -280,6 +280,7 @@ fn roundtrip_report_share_error() {
         (ReportError::TaskExpired, "07"),
         (ReportError::InvalidMessage, "08"),
         (ReportError::ReportTooEarly, "09"),
+        (ReportError::TaskNotStarted, "0A"),
     ])
 }