Repository: cloudflare/daphne
Pull request #695 (2024-10-15 10:12:51)

Reorganize dapf
- Separate out human friendly parsers of DAP types to seperate module
- Cleanup TestAction::CreateAddTaskJson
- create dap functions module
- Move to functions::helper the code needed to make a request to the helper
    which will allow us to possibly reuse this code in a individual dapf function



Patch:
diff --git a/crates/dapf/Cargo.toml b/crates/dapf/Cargo.toml
index 6d39d2f..7319824 100644
--- a/crates/dapf/Cargo.toml
+++ b/crates/dapf/Cargo.toml
@@ -39,6 +39,3 @@ tracing.workspace = true
 url.workspace = true
 webpki.workspace = true
 x509-parser.workspace = true
-
-[features]
-experimental = ["daphne/experimental"]
diff --git a/crates/dapf/src/acceptance/mod.rs b/crates/dapf/src/acceptance/mod.rs
index f54d339..a12f973 100644
--- a/crates/dapf/src/acceptance/mod.rs
+++ b/crates/dapf/src/acceptance/mod.rs
@@ -18,18 +18,14 @@
 
 pub mod load_testing;
 
-use crate::{
-    deduce_dap_version_from_url, response_to_anyhow, test_durations::TestDurations, HttpClient,
-};
+use crate::{deduce_dap_version_from_url, functions, test_durations::TestDurations, HttpClient};
 use anyhow::{anyhow, bail, Context, Result};
 use async_trait::async_trait;
 use daphne::{
-    constants::DapMediaType,
-    error::aborts::ProblemDetails,
     hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        self, AggregateShareReq, AggregationJobId, AggregationJobResp, Base64Encode, BatchId,
-        BatchSelector, PartialBatchSelector, TaskId,
+        self, AggregateShareReq, AggregationJobId, Base64Encode, BatchId, BatchSelector,
+        PartialBatchSelector, TaskId,
     },
     metrics::DaphneMetrics,
     roles::DapReportInitializer,
@@ -39,9 +35,8 @@ use daphne::{
     DapQueryConfig, DapTaskConfig, DapTaskParameters, DapVersion, EarlyReportStateConsumed,
     EarlyReportStateInitialized, ReplayProtection,
 };
-use daphne_service_utils::{bearer_token::BearerToken, http_headers};
+use daphne_service_utils::bearer_token::BearerToken;
 use futures::{future::OptionFuture, StreamExt, TryStreamExt};
-use prio::codec::{Decode, ParameterizedEncode};
 use prometheus::{Encoder, HistogramVec, IntCounterVec, IntGaugeVec, TextEncoder};
 use rand::{rngs, Rng};
 use rayon::prelude::{IntoParallelIterator, ParallelIterator};
@@ -509,62 +504,30 @@ impl Test {
             .context("producing agg job init request")?;
 
         // Send AggregationJobInitReq.
-        let headers = construct_request_headers(
-            DapMediaType::AggregationJobInitReq.as_str_for_version(task_config.version),
-            taskprov_advertisement.as_deref(),
-            &self.bearer_token,
-        )
-        .context("constructing request headers for AggregationJobInitReq")?;
-        let url = self.helper_url.join(&format!(
-            "tasks/{}/aggregation_jobs/{}",
-            task_id.to_base64url(),
-            agg_job_id.to_base64url()
-        ))?;
-
         // wait for all agg jobs to be ready to fire.
         info!("Reports generated, waiting for other tasks...");
         let _guard = load_control.wait().await;
         info!("Starting AggregationJobInitReq");
         let start = Instant::now();
-        let resp = send(
-            self.http_client
-                .put(url)
-                .body(
-                    agg_job_init_req
-                        .get_encoded_with_param(&task_config.version)
-                        .unwrap(),
-                )
-                .headers(headers),
-        )
-        .await?;
+        let agg_job_resp = self
+            .http_client
+            .submit_aggregation_job_init_req(
+                self.helper_url.join(&format!(
+                    "tasks/{}/aggregation_jobs/{}",
+                    task_id.to_base64url(),
+                    agg_job_id.to_base64url()
+                ))?,
+                agg_job_init_req,
+                task_config.version,
+                functions::helper::Options {
+                    taskprov_advertisement: taskprov_advertisement.as_deref(),
+                    bearer_token: self.bearer_token.as_ref(),
+                },
+            )
+            .await?;
         let duration = start.elapsed();
         info!("Finished AggregationJobInitReq in {duration:#?}");
 
-        if resp.status() == 400 {
-            let text = resp.text().await?;
-            let problem_details: ProblemDetails =
-                serde_json::from_str(&text).with_context(|| {
-                    format!("400 Bad Request: failed to parse problem details document: {text:?}")
-                })?;
-            return Err(anyhow!("400 Bad Request: {problem_details:?}"));
-        } else if resp.status() == 500 {
-            return Err(anyhow::anyhow!(
-                "500 Internal Server Error: {}",
-                resp.text().await?
-            ));
-        } else if !resp.status().is_success() {
-            return Err(response_to_anyhow(resp).await)
-                .context("while running an AggregateInitReq");
-        }
-
-        // Handle AggregationJobResp..
-        let agg_job_resp = AggregationJobResp::get_decoded(
-            &resp
-                .bytes()
-                .await
-                .context("transfering bytes from the AggregateInitReq")?,
-        )
-        .with_context(|| "failed to parse response to AggregateInitReq from Helper")?;
         let agg_share_span = task_config.consume_agg_job_resp(
             task_id,
             agg_job_state,
@@ -603,43 +566,21 @@ impl Test {
         // Send AggregateShareReq.
         info!("Starting AggregationJobInitReq");
         let start = Instant::now();
-        let headers = construct_request_headers(
-            DapMediaType::AggregateShareReq.as_str_for_version(version),
-            taskprov_advertisement,
-            &self.bearer_token,
-        )?;
-        let url = self.helper_url.join(&format!(
-            "tasks/{}/aggregate_shares",
-            task_id.to_base64url()
-        ))?;
-        let resp = send(
-            self.http_client
-                .post(url)
-                .body(agg_share_req.get_encoded_with_param(&version).unwrap())
-                .headers(headers),
-        )
-        .await?;
-        let duration = start.elapsed();
-        info!("Finished AggregateShareReq in {duration:#?}");
-        if resp.status() == 400 {
-            let problem_details: ProblemDetails = serde_json::from_slice(
-                &resp
-                    .bytes()
-                    .await
-                    .context("transfering bytes for AggregateShareReq")?,
+        self.http_client
+            .get_aggregate_share(
+                self.helper_url.join(&format!(
+                    "tasks/{}/aggregate_shares",
+                    task_id.to_base64url()
+                ))?,
+                agg_share_req,
+                version,
+                functions::helper::Options {
+                    taskprov_advertisement,
+                    bearer_token: self.bearer_token.as_ref(),
+                },
             )
-            .with_context(|| "400 Bad Request: failed to parse problem details document")?;
-            return Err(anyhow!("400 Bad Request: {problem_details:?}"));
-        } else if resp.status() == 500 {
-            return Err(anyhow::anyhow!(
-                "500 Internal Server Error: {}",
-                resp.text().await?
-            ));
-        } else if !resp.status().is_success() {
-            return Err(response_to_anyhow(resp).await)
-                .context("while running an AggregateInitReq");
-        }
-        Ok(duration)
+            .await?;
+        Ok(start.elapsed())
     }
 
     pub async fn test_helper(&self, opt: &TestOptions) -> Result<TestDurations> {
@@ -794,60 +735,6 @@ impl DapReportInitializer for Test {
     }
 }
 
-fn construct_request_headers<'a, M, T, B>(
-    media_type: M,
-    taskprov: T,
-    bearer_token: B,
-) -> Result<reqwest::header::HeaderMap>
-where
-    M: Into<Option<&'a str>>,
-    T: Into<Option<&'a str>>,
-    B: Into<Option<&'a BearerToken>>,
-{
-    let mut headers = reqwest::header::HeaderMap::new();
-    if let Some(media_type) = media_type.into() {
-        headers.insert(
-            reqwest::header::CONTENT_TYPE,
-            reqwest::header::HeaderValue::from_str(media_type)?,
-        );
-    }
-    if let Some(taskprov) = taskprov.into() {
-        headers.insert(
-            reqwest::header::HeaderName::from_static(http_headers::DAP_TASKPROV),
-            reqwest::header::HeaderValue::from_str(taskprov)?,
-        );
-    }
-    if let Some(token) = bearer_token.into() {
-        headers.insert(
-            reqwest::header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
-            reqwest::header::HeaderValue::from_str(token.as_ref())?,
-        );
-    }
-    Ok(headers)
-}
-
-async fn send(req: reqwest::RequestBuilder) -> reqwest::Result<reqwest::Response> {
-    for i in 0..4 {
-        let resp = req.try_clone().unwrap().send().await;
-        match &resp {
-            Ok(r) if r.status() != reqwest::StatusCode::BAD_GATEWAY => {
-                return resp;
-            }
-            Ok(r) if r.status().is_client_error() => {
-                return resp;
-            }
-            Ok(_) => {}
-            Err(e) => {
-                tracing::error!("request failed: {e:?}");
-            }
-        }
-        if i == 3 {
-            return resp;
-        }
-    }
-    unreachable!()
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct Now(u64);
 pub fn now() -> Now {
diff --git a/crates/dapf/src/cli_parsers.rs b/crates/dapf/src/cli_parsers.rs
new file mode 100644
index 0000000..ed03a9f
--- /dev/null
+++ b/crates/dapf/src/cli_parsers.rs
@@ -0,0 +1,328 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+//! Human friendly parsers for common types of parameters to DAP functions.
+
+use std::{
+    fmt,
+    io::{self, IsTerminal as _},
+    ops::ControlFlow,
+    str::FromStr,
+};
+
+use anyhow::{anyhow, Context};
+use clap::{builder::PossibleValue, ValueEnum};
+use daphne::{
+    hpke::HpkeKemId,
+    messages::{Base64Encode, TaskId},
+    vdaf::{Prio3Config, VdafConfig},
+    DapQueryConfig,
+};
+
+/// Some defaults for ease of use from the CLI. Instead of specifying the entire vdaf config json
+/// these names can be used. Check the [`ValueEnum::to_possible_value`] implementation for the
+/// names of these in the commandline.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum DefaultVdafConfigs {
+    Prio2Dimension99k,
+    #[default]
+    Prio3NumProofs2,
+    Prio3NumProofs3,
+}
+
+impl fmt::Display for DefaultVdafConfigs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.to_possible_value().unwrap().get_name())
+    }
+}
+
+impl ValueEnum for DefaultVdafConfigs {
+    fn value_variants<'a>() -> &'a [Self] {
+        &[
+            Self::Prio2Dimension99k,
+            Self::Prio3NumProofs2,
+            Self::Prio3NumProofs3,
+        ]
+    }
+
+    fn to_possible_value(&self) -> Option<PossibleValue> {
+        match self {
+            Self::Prio2Dimension99k => Some(PossibleValue::new("prio2-dimension-99k")),
+            Self::Prio3NumProofs2 => Some(PossibleValue::new("prio3-num-proofs-2")),
+            Self::Prio3NumProofs3 => Some(PossibleValue::new("prio3-num-proofs-3")),
+        }
+    }
+}
+
+impl DefaultVdafConfigs {
+    fn into_vdaf_config(self) -> VdafConfig {
+        match self {
+            Self::Prio2Dimension99k => VdafConfig::Prio2 { dimension: 99_992 },
+            Self::Prio3NumProofs2 => {
+                VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                    bits: 1,
+                    length: 100_000,
+                    chunk_length: 320,
+                    num_proofs: 2,
+                })
+            }
+            Self::Prio3NumProofs3 => {
+                VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                    bits: 1,
+                    length: 100_000,
+                    chunk_length: 320,
+                    num_proofs: 3,
+                })
+            }
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub enum CliVdafConfig {
+    Default(DefaultVdafConfigs),
+    Custom(VdafConfig),
+}
+
+impl Default for CliVdafConfig {
+    fn default() -> Self {
+        Self::Default(Default::default())
+    }
+}
+
+impl fmt::Display for CliVdafConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Default(default) => write!(f, "{default}"),
+            Self::Custom(custom) => write!(f, "{custom}"),
+        }
+    }
+}
+
+impl FromStr for CliVdafConfig {
+    type Err = String;
+    fn from_str(s: &str) -> std::prelude::v1::Result<Self, Self::Err> {
+        DefaultVdafConfigs::from_str(s, false)
+            .map(Self::Default)
+            .or_else(|default_err| {
+                serde_json::from_str(s)
+                    .map(Self::Custom)
+                    .map_err(|json_err| {
+                        if s.contains('{') {
+                            json_err.to_string()
+                        } else {
+                            default_err
+                        }
+                    })
+            })
+    }
+}
+
+impl CliVdafConfig {
+    pub fn into_vdaf_config(self) -> VdafConfig {
+        match self {
+            Self::Default(d) => d.into_vdaf_config(),
+            Self::Custom(v) => v,
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct CliDapQueryConfig(pub DapQueryConfig);
+
+impl fmt::Display for CliDapQueryConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        <DapQueryConfig as fmt::Display>::fmt(&self.0, f)
+    }
+}
+
+impl From<CliDapQueryConfig> for DapQueryConfig {
+    fn from(CliDapQueryConfig(val): CliDapQueryConfig) -> Self {
+        val
+    }
+}
+
+impl From<DapQueryConfig> for CliDapQueryConfig {
+    fn from(value: DapQueryConfig) -> Self {
+        Self(value)
+    }
+}
+
+impl FromStr for CliDapQueryConfig {
+    type Err = String;
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        if s == "time-interval" {
+            Ok(Self(DapQueryConfig::TimeInterval))
+        } else if let Some(size) = s.strip_prefix("fixed-size") {
+            Ok(Self(DapQueryConfig::FixedSize {
+                max_batch_size: if let Some(size) = size.strip_prefix("-") {
+                    Some(
+                        size.parse()
+                            .map_err(|e| format!("{s} is an invalid query config: {e:?}"))?,
+                    )
+                } else if size.is_empty() {
+                    None
+                } else {
+                    return Err(format!("{size} is an invalid fixed size max batch size"));
+                },
+            }))
+        } else {
+            Err(format!("{s} is an invalid query config"))
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct CliTaskId(pub TaskId);
+
+impl fmt::Display for CliTaskId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        <TaskId as fmt::Display>::fmt(&self.0, f)
+    }
+}
+
+impl From<TaskId> for CliTaskId {
+    fn from(id: TaskId) -> Self {
+        Self(id)
+    }
+}
+
+impl From<CliTaskId> for TaskId {
+    fn from(CliTaskId(id): CliTaskId) -> Self {
+        id
+    }
+}
+
+impl FromStr for CliTaskId {
+    type Err = anyhow::Error;
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        TaskId::try_from_base64url(s)
+            .ok_or_else(|| anyhow!("failed to decode ID"))
+            .context("expected URL-safe, base64 string")
+            .map(Self)
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct CliHpkeKemId(pub HpkeKemId);
+
+impl Default for CliHpkeKemId {
+    fn default() -> Self {
+        Self(HpkeKemId::X25519HkdfSha256)
+    }
+}
+
+impl fmt::Display for CliHpkeKemId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.to_possible_value().unwrap().get_name())
+    }
+}
+
+impl ValueEnum for CliHpkeKemId {
+    fn value_variants<'a>() -> &'a [Self] {
+        &[
+            Self(HpkeKemId::X25519HkdfSha256),
+            Self(HpkeKemId::P256HkdfSha256),
+        ]
+    }
+
+    fn to_possible_value(&self) -> Option<PossibleValue> {
+        Some(match self.0 {
+            HpkeKemId::X25519HkdfSha256 => PossibleValue::new("x25519_hkdf_sha256"),
+            HpkeKemId::P256HkdfSha256 => PossibleValue::new("p256_hkdf_sha256"),
+            HpkeKemId::NotImplemented(id) => unreachable!("unhandled HPKE KEM ID {id}"),
+        })
+    }
+}
+
+pub fn use_or_request_from_user_or_default<T, U>(
+    value: Option<T>,
+    mut default: impl FnMut() -> U,
+    field: &'static str,
+) -> io::Result<T>
+where
+    U: Into<T>,
+    T: FromStr + fmt::Display,
+    T::Err: fmt::Debug,
+{
+    match value {
+        Some(v) => Ok(v),
+        None => Ok(request_from_user::<T>(
+            field,
+            format!("leave empty to default: {}", default().into()).as_str(),
+            || ControlFlow::Break(Some(default().into())),
+        )?
+        .unwrap()),
+    }
+}
+
+pub fn use_or_request_from_user<T>(value: Option<T>, field: &'static str) -> io::Result<T>
+where
+    T: FromStr + fmt::Display,
+    T::Err: fmt::Debug,
+{
+    match value {
+        Some(v) => Ok(v),
+        None => request_from_user::<T>(field, None, || ControlFlow::Continue(()))
+            .transpose()
+            .unwrap()
+            .map(Into::into),
+    }
+}
+
+pub fn request_from_user<'e, T>(
+    field: &'static str,
+    extra_info: impl Into<Option<&'e str>>,
+    mut on_empty: impl FnMut() -> ControlFlow<Option<T>>,
+) -> io::Result<Option<T>>
+where
+    T: FromStr + fmt::Display,
+    T::Err: fmt::Debug,
+{
+    if !std::io::stdin().is_terminal() {
+        return match on_empty() {
+            ControlFlow::Break(b) => {
+                eprintln!(
+                    "defaulting {field} to {}",
+                    b.as_ref()
+                        .map(|b| format!("{b}"))
+                        .unwrap_or_else(|| "None".into())
+                );
+                Ok(b)
+            }
+            ControlFlow::Continue(()) => Err(io::Error::other(format!(
+                "{field:?} is required. use command line options to specify it"
+            ))),
+        };
+    }
+    let extra_info = extra_info.into();
+    loop {
+        eprint!(
+            "{field}{}?: ",
+            extra_info.map(|e| format!(" ({e})")).unwrap_or_default()
+        );
+        let mut buf = String::new();
+        std::io::stdin().read_line(&mut buf)?;
+
+        if buf.is_empty() {
+            // <C-d>
+            break Err(io::ErrorKind::UnexpectedEof.into());
+        }
+
+        buf.pop();
+
+        if buf.is_empty() {
+            match on_empty() {
+                ControlFlow::Continue(()) => continue,
+                ControlFlow::Break(b) => break Ok(b),
+            }
+        }
+
+        match buf.parse::<T>() {
+            Ok(t) => break Ok(Some(t)),
+            Err(e) => {
+                eprintln!("{buf} is not a valid {field}: {e:?}");
+            }
+        }
+    }
+}
diff --git a/crates/dapf/src/functions/helper.rs b/crates/dapf/src/functions/helper.rs
new file mode 100644
index 0000000..d322857
--- /dev/null
+++ b/crates/dapf/src/functions/helper.rs
@@ -0,0 +1,137 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use anyhow::{anyhow, Context as _};
+use daphne::{
+    constants::DapMediaType,
+    error::aborts::ProblemDetails,
+    messages::{AggregateShareReq, AggregationJobInitReq, AggregationJobResp},
+    DapVersion,
+};
+use daphne_service_utils::{bearer_token::BearerToken, http_headers};
+use prio::codec::{Decode as _, ParameterizedEncode as _};
+use reqwest::header;
+use url::Url;
+
+use crate::{response_to_anyhow, HttpClient};
+
+impl HttpClient {
+    pub async fn submit_aggregation_job_init_req(
+        &self,
+        url: Url,
+        agg_job_init_req: AggregationJobInitReq,
+        version: DapVersion,
+        opts: Options<'_>,
+    ) -> anyhow::Result<AggregationJobResp> {
+        let resp = self
+            .put(url)
+            .body(agg_job_init_req.get_encoded_with_param(&version).unwrap())
+            .headers(construct_request_headers(
+                DapMediaType::AggregationJobInitReq
+                    .as_str_for_version(version)
+                    .with_context(|| {
+                        format!("AggregationJobInitReq media type is not defined for {version}")
+                    })?,
+                opts,
+            )?)
+            .send()
+            .await
+            .context("sending AggregationJobInitReq")?;
+        if resp.status() == 400 {
+            let text = resp.text().await?;
+            let problem_details: ProblemDetails =
+                serde_json::from_str(&text).with_context(|| {
+                    format!("400 Bad Request: failed to parse problem details document: {text:?}")
+                })?;
+            Err(anyhow!("400 Bad Request: {problem_details:?}"))
+        } else if resp.status() == 500 {
+            Err(anyhow::anyhow!(
+                "500 Internal Server Error: {}",
+                resp.text().await?
+            ))
+        } else if !resp.status().is_success() {
+            Err(response_to_anyhow(resp).await).context("while running an AggregationJobInitReq")
+        } else {
+            AggregationJobResp::get_decoded(
+                &resp
+                    .bytes()
+                    .await
+                    .context("transfering bytes from the AggregateInitReq")?,
+            )
+            .with_context(|| "failed to parse response to AggregateInitReq from Helper")
+        }
+    }
+
+    pub async fn get_aggregate_share(
+        &self,
+        url: Url,
+        agg_share_req: AggregateShareReq,
+        version: DapVersion,
+        opts: Options<'_>,
+    ) -> anyhow::Result<()> {
+        let resp = self
+            .post(url)
+            .body(agg_share_req.get_encoded_with_param(&version).unwrap())
+            .headers(construct_request_headers(
+                DapMediaType::AggregateShareReq
+                    .as_str_for_version(version)
+                    .with_context(|| {
+                        format!("AggregateShareReq media type is not defined for {version}")
+                    })?,
+                opts,
+            )?)
+            .send()
+            .await
+            .context("sending AggregateShareReq")?;
+        if resp.status() == 400 {
+            let problem_details: ProblemDetails = serde_json::from_slice(
+                &resp
+                    .bytes()
+                    .await
+                    .context("transfering bytes for AggregateShareReq")?,
+            )
+            .with_context(|| "400 Bad Request: failed to parse problem details document")?;
+            Err(anyhow!("400 Bad Request: {problem_details:?}"))
+        } else if resp.status() == 500 {
+            Err(anyhow!("500 Internal Server Error: {}", resp.text().await?))
+        } else if !resp.status().is_success() {
+            Err(response_to_anyhow(resp).await).context("while running an AggregateShareReq")
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[derive(Default, Debug)]
+pub struct Options<'s> {
+    pub taskprov_advertisement: Option<&'s str>,
+    pub bearer_token: Option<&'s BearerToken>,
+}
+
+fn construct_request_headers(
+    media_type: &str,
+    options: Options<'_>,
+) -> Result<header::HeaderMap, header::InvalidHeaderValue> {
+    let mut headers = header::HeaderMap::new();
+    headers.insert(
+        header::CONTENT_TYPE,
+        header::HeaderValue::from_str(media_type)?,
+    );
+    let Options {
+        taskprov_advertisement,
+        bearer_token,
+    } = options;
+    if let Some(taskprov) = taskprov_advertisement {
+        headers.insert(
+            const { header::HeaderName::from_static(http_headers::DAP_TASKPROV) },
+            header::HeaderValue::from_str(taskprov)?,
+        );
+    }
+    if let Some(token) = bearer_token {
+        headers.insert(
+            const { header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN) },
+            header::HeaderValue::from_str(token.as_str())?,
+        );
+    }
+    Ok(headers)
+}
diff --git a/crates/dapf/src/functions/hpke.rs b/crates/dapf/src/functions/hpke.rs
new file mode 100644
index 0000000..7937bb5
--- /dev/null
+++ b/crates/dapf/src/functions/hpke.rs
@@ -0,0 +1,55 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use std::{io::Cursor, path::Path};
+
+use anyhow::{anyhow, Context as _};
+use daphne::messages::{decode_base64url_vec, HpkeConfigList};
+use daphne_service_utils::http_headers;
+use prio::codec::Decode as _;
+use url::Url;
+use webpki::{EndEntityCert, ECDSA_P256_SHA256};
+use x509_parser::pem::Pem;
+
+use crate::{response_to_anyhow, HttpClient};
+
+impl HttpClient {
+    pub async fn get_hpke_config(
+        &self,
+        base_url: &Url,
+        certificate_file: Option<&Path>,
+    ) -> anyhow::Result<HpkeConfigList> {
+        let url = base_url.join("hpke_config")?;
+        let resp = self
+            .get(url.as_str())
+            .send()
+            .await
+            .with_context(|| "request failed")?;
+        if !resp.status().is_success() {
+            return Err(response_to_anyhow(resp).await);
+        }
+        let maybe_signature = resp.headers().get(http_headers::HPKE_SIGNATURE).cloned();
+        let hpke_config_bytes = resp.bytes().await.context("failed to read hpke config")?;
+        if let Some(cert_path) = certificate_file {
+            let cert = std::fs::read_to_string(cert_path).context("reading the certificate")?;
+            let Some(signature) = maybe_signature else {
+                anyhow::bail!("Aggregator did not sign its response");
+            };
+            let signature_bytes =
+                decode_base64url_vec(signature.as_bytes()).context("decoding the signature")?;
+            let (cert_pem, _bytes_read) =
+                Pem::read(Cursor::new(cert.as_bytes())).context("reading PEM certificate")?;
+            let cert = EndEntityCert::try_from(cert_pem.contents.as_ref())
+                .map_err(|e| anyhow!("{e:?}")) // webpki::Error does not implement std::error::Error
+                .context("parsing PEM certificate")?;
+
+            cert.verify_signature(
+                &ECDSA_P256_SHA256,
+                &hpke_config_bytes,
+                signature_bytes.as_ref(),
+            )
+            .map_err(|e| anyhow!("signature not verified: {}", e.to_string()))?;
+        }
+        Ok(HpkeConfigList::get_decoded(&hpke_config_bytes)?)
+    }
+}
diff --git a/crates/dapf/src/functions/mod.rs b/crates/dapf/src/functions/mod.rs
new file mode 100644
index 0000000..a8caa12
--- /dev/null
+++ b/crates/dapf/src/functions/mod.rs
@@ -0,0 +1,8 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+//! The various DAP functions dapf can perform.
+
+pub mod helper;
+pub mod hpke;
+pub mod test_routes;
diff --git a/crates/dapf/src/http_client.rs b/crates/dapf/src/http_client.rs
index 28bfbf7..8d3a972 100644
--- a/crates/dapf/src/http_client.rs
+++ b/crates/dapf/src/http_client.rs
@@ -1,20 +1,12 @@
 // Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use std::{borrow::Cow, env, io::Cursor, path::Path};
+use std::{borrow::Cow, env};
 
-use anyhow::{anyhow, bail, Context};
-use daphne::messages::{decode_base64url_vec, HpkeConfigList};
-use daphne_service_utils::http_headers;
-use prio::codec::Decode;
+use anyhow::{anyhow, bail};
 use reqwest::{Client, IntoUrl, RequestBuilder};
 use rustls::pki_types::{CertificateDer, PrivateKeyDer};
 use rustls_pemfile::Item;
-use url::Url;
-use webpki::{EndEntityCert, ECDSA_P256_SHA256};
-use x509_parser::pem::Pem;
-
-use crate::response_to_anyhow;
 
 pub struct HttpClient {
     using_mtls: bool,
@@ -151,43 +143,4 @@ impl HttpClient {
     pub fn put<U: IntoUrl>(&self, url: U) -> RequestBuilder {
         self.client().put(url)
     }
-
-    pub async fn get_hpke_config(
-        &self,
-        base_url: &Url,
-        certificate_file: Option<&Path>,
-    ) -> anyhow::Result<HpkeConfigList> {
-        let url = base_url.join("hpke_config")?;
-        let resp = self
-            .get(url.as_str())
-            .send()
-            .await
-            .with_context(|| "request failed")?;
-        if !resp.status().is_success() {
-            return Err(response_to_anyhow(resp).await);
-        }
-        let maybe_signature = resp.headers().get(http_headers::HPKE_SIGNATURE).cloned();
-        let hpke_config_bytes = resp.bytes().await.context("failed to read hpke config")?;
-        if let Some(cert_path) = certificate_file {
-            let cert = std::fs::read_to_string(cert_path).context("reading the certificate")?;
-            let Some(signature) = maybe_signature else {
-                anyhow::bail!("Aggregator did not sign its response");
-            };
-            let signature_bytes =
-                decode_base64url_vec(signature.as_bytes()).context("decoding the signature")?;
-            let (cert_pem, _bytes_read) =
-                Pem::read(Cursor::new(cert.as_bytes())).context("reading PEM certificate")?;
-            let cert = EndEntityCert::try_from(cert_pem.contents.as_ref())
-                .map_err(|e| anyhow!("{e:?}")) // webpki::Error does not implement std::error::Error
-                .context("parsing PEM certificate")?;
-
-            cert.verify_signature(
-                &ECDSA_P256_SHA256,
-                &hpke_config_bytes,
-                signature_bytes.as_ref(),
-            )
-            .map_err(|e| anyhow!("signature not verified: {}", e.to_string()))?;
-        }
-        Ok(HpkeConfigList::get_decoded(&hpke_config_bytes)?)
-    }
 }
diff --git a/crates/dapf/src/lib.rs b/crates/dapf/src/lib.rs
index 56c4847..81af060 100644
--- a/crates/dapf/src/lib.rs
+++ b/crates/dapf/src/lib.rs
@@ -2,9 +2,10 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 pub mod acceptance;
+pub mod cli_parsers;
+pub mod functions;
 pub mod http_client;
 mod test_durations;
-pub mod test_routes;
 
 use anyhow::{anyhow, Context};
 use daphne::DapVersion;
diff --git a/crates/dapf/src/main.rs b/crates/dapf/src/main.rs
index a0d44ba..a81c22e 100644
--- a/crates/dapf/src/main.rs
+++ b/crates/dapf/src/main.rs
@@ -2,20 +2,20 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 use anyhow::{anyhow, bail, Context, Result};
-use clap::{builder::PossibleValue, Args, Parser, Subcommand, ValueEnum};
+use clap::{Args, Parser, Subcommand};
 use dapf::{
     acceptance::{load_testing, LoadControlParams, LoadControlStride, TestOptions},
+    cli_parsers::{
+        use_or_request_from_user, use_or_request_from_user_or_default, CliDapQueryConfig,
+        CliHpkeKemId, CliTaskId, CliVdafConfig,
+    },
     deduce_dap_version_from_url, response_to_anyhow, HttpClient,
 };
 use daphne::{
     constants::DapMediaType,
     error::aborts::ProblemDetails,
-    hpke::{HpkeKemId, HpkeReceiverConfig},
-    messages::{
-        self, encode_base64url, Base64Encode, BatchSelector, Collection, CollectionReq, Query,
-        TaskId,
-    },
-    vdaf::{Prio3Config, VdafConfig},
+    hpke::HpkeReceiverConfig,
+    messages::{self, encode_base64url, BatchSelector, Collection, CollectionReq, Query, TaskId},
     DapAggregationParam, DapMeasurement, DapQueryConfig, DapVersion,
 };
 use daphne_service_utils::{
@@ -25,193 +25,16 @@ use daphne_service_utils::{
 };
 use prio::codec::{ParameterizedDecode, ParameterizedEncode};
 use rand::{thread_rng, Rng};
-use std::fmt;
 use std::{
-    io::{self, stdin, IsTerminal, Read},
-    ops::ControlFlow,
+    io::{stdin, IsTerminal, Read},
     path::PathBuf,
     process::Command,
-    str::FromStr,
     time::{Duration, SystemTime},
 };
 use tracing::level_filters::LevelFilter;
 use tracing_subscriber::EnvFilter;
 use url::Url;
 
-#[derive(Debug, Clone, Copy)]
-enum DefaultVdafConfigs {
-    Prio2Dimension99k,
-    Prio3NumProofs2,
-    Prio3NumProofs3,
-}
-
-impl fmt::Display for DefaultVdafConfigs {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(self.to_possible_value().unwrap().get_name())
-    }
-}
-
-impl ValueEnum for DefaultVdafConfigs {
-    fn value_variants<'a>() -> &'a [Self] {
-        &[
-            Self::Prio2Dimension99k,
-            Self::Prio3NumProofs2,
-            Self::Prio3NumProofs3,
-        ]
-    }
-
-    fn to_possible_value(&self) -> Option<PossibleValue> {
-        match self {
-            Self::Prio2Dimension99k => Some(PossibleValue::new("prio2-dimension-99k")),
-            Self::Prio3NumProofs2 => Some(PossibleValue::new("prio3-num-proofs-2")),
-            Self::Prio3NumProofs3 => Some(PossibleValue::new("prio3-num-proofs-3")),
-        }
-    }
-}
-
-impl DefaultVdafConfigs {
-    fn into_vdaf(self) -> VdafConfig {
-        match self {
-            Self::Prio2Dimension99k => VdafConfig::Prio2 { dimension: 99_992 },
-            Self::Prio3NumProofs2 => {
-                VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
-                    bits: 1,
-                    length: 100_000,
-                    chunk_length: 320,
-                    num_proofs: 2,
-                })
-            }
-            Self::Prio3NumProofs3 => {
-                VdafConfig::Prio3(Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
-                    bits: 1,
-                    length: 100_000,
-                    chunk_length: 320,
-                    num_proofs: 3,
-                })
-            }
-        }
-    }
-}
-
-#[derive(Clone, Debug)]
-enum CliVdafConfig {
-    Default(DefaultVdafConfigs),
-    Custom(VdafConfig),
-}
-
-impl fmt::Display for CliVdafConfig {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            Self::Default(default) => write!(f, "{default}"),
-            Self::Custom(custom) => write!(f, "{custom}"),
-        }
-    }
-}
-
-impl FromStr for CliVdafConfig {
-    type Err = String;
-    fn from_str(s: &str) -> std::prelude::v1::Result<Self, Self::Err> {
-        DefaultVdafConfigs::from_str(s, false)
-            .map(Self::Default)
-            .or_else(|default_err| {
-                serde_json::from_str(s)
-                    .map(Self::Custom)
-                    .map_err(|json_err| {
-                        if s.contains('{') {
-                            json_err.to_string()
-                        } else {
-                            default_err
-                        }
-                    })
-            })
-    }
-}
-
-impl CliVdafConfig {
-    fn into_vdaf(self) -> VdafConfig {
-        match self {
-            Self::Default(d) => d.into_vdaf(),
-            Self::Custom(v) => v,
-        }
-    }
-}
-
-#[derive(Clone, Debug)]
-struct CliDapQueryConfig(DapQueryConfig);
-
-impl fmt::Display for CliDapQueryConfig {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        <DapQueryConfig as fmt::Display>::fmt(&self.0, f)
-    }
-}
-
-impl From<CliDapQueryConfig> for DapQueryConfig {
-    fn from(CliDapQueryConfig(val): CliDapQueryConfig) -> Self {
-        val
-    }
-}
-
-impl From<DapQueryConfig> for CliDapQueryConfig {
-    fn from(value: DapQueryConfig) -> Self {
-        Self(value)
-    }
-}
-
-impl FromStr for CliDapQueryConfig {
-    type Err = String;
-    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-        if s == "time-interval" {
-            Ok(Self(DapQueryConfig::TimeInterval))
-        } else if let Some(size) = s.strip_prefix("fixed-size") {
-            Ok(Self(DapQueryConfig::FixedSize {
-                max_batch_size: if let Some(size) = size.strip_prefix("-") {
-                    Some(
-                        size.parse()
-                            .map_err(|e| format!("{s} is an invalid query config: {e:?}"))?,
-                    )
-                } else if size.is_empty() {
-                    None
-                } else {
-                    return Err(format!("{size} is an invalid fixed size max batch size"));
-                },
-            }))
-        } else {
-            Err(format!("{s} is an invalid query config"))
-        }
-    }
-}
-
-#[derive(Clone, Debug)]
-struct CliTaskId(TaskId);
-
-impl fmt::Display for CliTaskId {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        <TaskId as fmt::Display>::fmt(&self.0, f)
-    }
-}
-
-impl From<TaskId> for CliTaskId {
-    fn from(id: TaskId) -> Self {
-        Self(id)
-    }
-}
-
-impl From<CliTaskId> for TaskId {
-    fn from(CliTaskId(id): CliTaskId) -> Self {
-        id
-    }
-}
-
-impl FromStr for CliTaskId {
-    type Err = anyhow::Error;
-    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-        TaskId::try_from_base64url(s)
-            .ok_or_else(|| anyhow!("failed to decode ID"))
-            .context("expected URL-safe, base64 string")
-            .map(Self)
-    }
-}
-
 /// DAP Functions, a utility for interacting with DAP deployments.
 #[derive(Parser, Debug)]
 #[clap(version, about, long_about = None)]
@@ -353,7 +176,7 @@ enum HpkeAction {
     /// first by prio and then in base64. This version can later be used when TODO
     Generate {
         #[arg(default_value_t)]
-        kem_alg: KemAlg,
+        kem_alg: CliHpkeKemId,
     },
     /// Rotate the HPKE config advertised by the Aggregator.
     RotateReceiverConfig {
@@ -364,7 +187,7 @@ enum HpkeAction {
         #[arg(default_value_t, long)]
         dap_version: DapVersion,
         #[arg(default_value_t, long)]
-        kem_alg: KemAlg,
+        kem_alg: CliHpkeKemId,
     },
 }
 
@@ -375,7 +198,7 @@ enum TestAction {
     AddHpkeConfig {
         aggregator_url: Url,
         #[arg(short, long, default_value_t)]
-        kem_alg: KemAlg,
+        kem_alg: CliHpkeKemId,
     },
     CreateAddTaskJson {
         #[arg(long)]
@@ -409,7 +232,7 @@ enum TestAction {
         /// If a new hpke config should be inserted after clearing all storage, use this flag to
         /// specify the key algorithm to use for the new hpke config
         #[arg(short, long)]
-        set_hpke_key: Option<KemAlg>,
+        set_hpke_key: Option<CliHpkeKemId>,
     },
 }
 
@@ -516,38 +339,6 @@ impl LoadControlParamsCli {
     }
 }
 
-#[derive(Clone, Debug)]
-struct KemAlg(HpkeKemId);
-
-impl Default for KemAlg {
-    fn default() -> Self {
-        Self(HpkeKemId::X25519HkdfSha256)
-    }
-}
-
-impl fmt::Display for KemAlg {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "{}", self.to_possible_value().unwrap().get_name())
-    }
-}
-
-impl ValueEnum for KemAlg {
-    fn value_variants<'a>() -> &'a [Self] {
-        &[
-            Self(HpkeKemId::X25519HkdfSha256),
-            Self(HpkeKemId::P256HkdfSha256),
-        ]
-    }
-
-    fn to_possible_value(&self) -> Option<PossibleValue> {
-        Some(match self.0 {
-            HpkeKemId::X25519HkdfSha256 => PossibleValue::new("x25519_hkdf_sha256"),
-            HpkeKemId::P256HkdfSha256 => PossibleValue::new("p256_hkdf_sha256"),
-            HpkeKemId::NotImplemented(id) => unreachable!("unhandled HPKE KEM ID {id}"),
-        })
-    }
-}
-
 #[tokio::main]
 async fn main() -> Result<()> {
     let mut cli = Cli::parse();
@@ -627,7 +418,7 @@ async fn handle_leader_actions(
             let version = deduce_dap_version_from_url(&leader_url)?;
             // Generate a report for the measurement.
             let report = vdaf_config
-                .into_vdaf()
+                .into_vdaf_config()
                 .produce_report(
                     &[leader_hpke_config, helper_hpke_config],
                     now,
@@ -752,15 +543,17 @@ async fn handle_leader_actions(
             })?;
             let version = deduce_dap_version_from_url(&uri)?;
             let collect_resp = Collection::get_decoded_with_param(&version, &resp.bytes().await?)?;
-            let agg_res = vdaf_config.into_vdaf().consume_encrypted_agg_shares(
-                receiver,
-                &task_id.into(),
-                &batch_selector,
-                collect_resp.report_count,
-                &DapAggregationParam::Empty,
-                collect_resp.encrypted_agg_shares.to_vec(),
-                version,
-            )?;
+            let agg_res = vdaf_config
+                .into_vdaf_config()
+                .consume_encrypted_agg_shares(
+                    receiver,
+                    &task_id.into(),
+                    &batch_selector,
+                    collect_resp.report_count,
+                    &DapAggregationParam::Empty,
+                    collect_resp.encrypted_agg_shares.to_vec(),
+                    version,
+                )?;
 
             print!("{}", serde_json::to_string(&agg_res)?);
             Ok(())
@@ -787,7 +580,7 @@ async fn handle_helper_actions(
 
             let t = dapf::acceptance::Test::from_env(
                 helper_url,
-                vdaf_config.into_vdaf(),
+                vdaf_config.into_vdaf_config(),
                 hpke_signing_certificate_path,
                 http_client,
                 load_control,
@@ -829,7 +622,7 @@ async fn handle_helper_actions(
         } => {
             load_testing::execute_single_combination_from_env(
                 helper_url,
-                vdaf_config.into_vdaf(),
+                vdaf_config.into_vdaf_config(),
                 reports_per_batch,
                 reports_per_agg_job,
                 http_client,
@@ -1041,7 +834,7 @@ async fn handle_decode_actions(action: DecodeAction) -> anyhow::Result<()> {
                         }
                     };
                     let agg_shares = vdaf_config
-                        .into_vdaf()
+                        .into_vdaf_config()
                         .consume_encrypted_agg_shares(
                             &hpke_config,
                             &task_id.into(),
@@ -1075,17 +868,24 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
         TestAction::AddHpkeConfig {
             aggregator_url,
             kem_alg,
-        } => dapf::test_routes::add_hpke_config(&http_client, &aggregator_url, kem_alg.0).await,
+        } => {
+            http_client
+                .add_hpke_config(&aggregator_url, kem_alg.0)
+                .await?;
+            println!("config added");
+            Ok(())
+        }
         TestAction::ClearStorage {
             aggregator_url,
             set_hpke_key,
         } => {
-            dapf::test_routes::delete_all_storage(
-                &http_client,
-                &aggregator_url,
-                set_hpke_key.map(|k| k.0),
-            )
-            .await
+            http_client.delete_all_storage(&aggregator_url).await?;
+            if let Some(CliHpkeKemId(kem_alg)) = set_hpke_key {
+                http_client
+                    .add_hpke_config(&aggregator_url, kem_alg)
+                    .await?;
+            }
+            Ok(())
         }
         TestAction::CreateAddTaskJson {
             task_id,
@@ -1101,54 +901,9 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
             time_precision,
             expires_in_seconds: task_expiration,
         } => {
-            let (vdaf_verify_key, vdaf) = use_or_request_from_user_or_default(
-                vdaf,
-                || CliVdafConfig::Default(DefaultVdafConfigs::Prio3NumProofs3),
-                "vdaf",
-            )
-            .map(|vdaf| {
-                let vdaf = vdaf.into_vdaf();
-                let vdaf_verify_key = vdaf.gen_verify_key();
-                let (typ, bits, length, chunk_length) = match vdaf {
-                    VdafConfig::Prio3(prio3) => match prio3 {
-                        Prio3Config::Count => ("Prio3Count", None, None, None),
-                        Prio3Config::Sum { bits } => ("Prio3Sum", Some(bits), None, None),
-                        Prio3Config::Histogram {
-                            length,
-                            chunk_length,
-                        } => ("Prio3Histogram", None, Some(length), Some(chunk_length)),
-                        Prio3Config::SumVec {
-                            bits,
-                            length,
-                            chunk_length,
-                        } => ("Prio3SumVec", Some(bits), Some(length), Some(chunk_length)),
-                        Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
-                            bits,
-                            length,
-                            chunk_length,
-                            num_proofs: _,
-                        } => (
-                            "Prio3SumVecField64MultiproofHmacSha256Aes128",
-                            Some(bits),
-                            Some(length),
-                            Some(chunk_length),
-                        ),
-                    },
-                    VdafConfig::Prio2 { .. } => ("Prio2", None, None, None),
-                    VdafConfig::Pine(_) => ("Pine", None, None, None),
-                    #[cfg(feature = "experimental")]
-                    VdafConfig::Mastic { .. } => todo!(),
-                };
-                (
-                    encode_base64url(vdaf_verify_key),
-                    InternalTestVdaf {
-                        typ: typ.into(),
-                        bits: bits.map(|a| a.to_string()),
-                        length: length.map(|a| a.to_string()),
-                        chunk_length: chunk_length.map(|a| a.to_string()),
-                    },
-                )
-            })?;
+            let vdaf = use_or_request_from_user_or_default(vdaf, CliVdafConfig::default, "vdaf")?
+                .into_vdaf_config();
+            let vdaf_verify_key = encode_base64url(vdaf.gen_verify_key());
             let CliDapQueryConfig(query) = use_or_request_from_user_or_default(
                 query,
                 || DapQueryConfig::FixedSize {
@@ -1157,19 +912,17 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                 "query",
             )?;
             let role = use_or_request_from_user(role, "role")?;
+            let default_task_id = TaskId(thread_rng().gen());
             let internal_task = InternalTestAddTask {
                 task_id: use_or_request_from_user_or_default(
                     task_id,
-                    {
-                        let t = TaskId(thread_rng().gen());
-                        move || t
-                    },
+                    || default_task_id,
                     "task id",
                 )?
                 .into(),
                 leader: use_or_request_from_user(leader_url, "leader url")?,
                 helper: use_or_request_from_user(helper_url, "helper url")?,
-                vdaf,
+                vdaf: InternalTestVdaf::from(vdaf),
                 leader_authentication_token: use_or_request_from_user(
                     leader_auth_token,
                     "leader auth token",
@@ -1226,95 +979,3 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
         }
     }
 }
-
-fn use_or_request_from_user_or_default<T, U>(
-    value: Option<T>,
-    mut default: impl FnMut() -> U,
-    field: &'static str,
-) -> io::Result<T>
-where
-    U: Into<T>,
-    T: FromStr + fmt::Display,
-    T::Err: fmt::Debug,
-{
-    match value {
-        Some(v) => Ok(v),
-        None => Ok(request_from_user::<T>(
-            field,
-            format!("leave empty to default: {}", default().into()).as_str(),
-            || ControlFlow::Break(Some(default().into())),
-        )?
-        .unwrap()),
-    }
-}
-
-fn use_or_request_from_user<T>(value: Option<T>, field: &'static str) -> io::Result<T>
-where
-    T: FromStr + fmt::Display,
-    T::Err: fmt::Debug,
-{
-    match value {
-        Some(v) => Ok(v),
-        None => request_from_user::<T>(field, None, || ControlFlow::Continue(()))
-            .transpose()
-            .unwrap()
-            .map(Into::into),
-    }
-}
-
-fn request_from_user<'e, T>(
-    field: &'static str,
-    extra_info: impl Into<Option<&'e str>>,
-    mut on_empty: impl FnMut() -> ControlFlow<Option<T>>,
-) -> io::Result<Option<T>>
-where
-    T: FromStr + fmt::Display,
-    T::Err: fmt::Debug,
-{
-    if !std::io::stdin().is_terminal() {
-        return match on_empty() {
-            ControlFlow::Break(b) => {
-                eprintln!(
-                    "defaulting {field} to {}",
-                    b.as_ref()
-                        .map(|b| format!("{b}"))
-                        .unwrap_or_else(|| "None".into())
-                );
-                Ok(b)
-            }
-            ControlFlow::Continue(()) => Err(io::Error::other(format!(
-                "{field:?} is required. use command line options to specify it"
-            ))),
-        };
-    }
-    let extra_info = extra_info.into();
-    loop {
-        eprint!(
-            "{field}{}?: ",
-            extra_info.map(|e| format!(" ({e})")).unwrap_or_default()
-        );
-        let mut buf = String::new();
-        std::io::stdin().read_line(&mut buf)?;
-
-        if buf.is_empty() {
-            // <C-d>
-            break Err(io::ErrorKind::UnexpectedEof.into());
-        }
-
-        buf.pop();
-
-        if buf.is_empty() {
-            match on_empty() {
-                ControlFlow::Continue(()) => continue,
-                ControlFlow::Break(b) => break Ok(b),
-            }
-        }
-
-        match buf.parse::<T>() {
-            Ok(t) => break Ok(Some(t)),
-            Err(e) => {
-                eprintln!("{buf} is not a valid {field}: {e:?}");
-            }
-        }
-    }
-}
diff --git a/crates/daphne-server/src/roles/aggregator.rs b/crates/daphne-server/src/roles/aggregator.rs
index 1d4168c..a66e33b 100644
--- a/crates/daphne-server/src/roles/aggregator.rs
+++ b/crates/daphne-server/src/roles/aggregator.rs
@@ -19,7 +19,7 @@ use daphne::{
 use daphne_service_utils::durable_requests::bindings::{
     self, AggregateStoreMergeOptions, AggregateStoreMergeReq, AggregateStoreMergeResp,
 };
-use futures::{future::try_join_all, StreamExt, TryStreamExt};
+use futures::{future::try_join_all, StreamExt, TryFutureExt, TryStreamExt};
 use mappable_rc::Marc;
 use rayon::prelude::{IntoParallelIterator, ParallelIterator};
 
@@ -316,15 +316,34 @@ impl DapAggregator for crate::App {
 
         futures::stream::iter(agg_span)
             .map(|bucket| async move {
-                Ok::<bool, DapError>(
-                    !self
-                        .durable()
-                        .request(bindings::AggregateStore::Get, (version, task_id, &bucket))
+                let durable = self.durable();
+                let params = (version, task_id, &bucket);
+
+                let get_report_count = || {
+                    durable
+                        .request(bindings::AggregateStore::ReportCount, params)
+                        .send::<u64>()
+                };
+
+                // TODO: remove this after the worker has this feature deployed.
+                let backwards_compat_get_report_count = || {
+                    durable
+                        .request(bindings::AggregateStore::Get, params)
                         .send::<DapAggregateShare>()
-                        .await
-                        .map_err(|e| fatal_error!(err = ?e, "failed to get an agg share"))?
-                        .empty(),
-                )
+                        .map_ok(|r| r.report_count)
+                };
+
+                let count = get_report_count()
+                    .or_else(|_| backwards_compat_get_report_count())
+                    .await
+                    .map_err(|e| {
+                        fatal_error!(
+                            err = ?e,
+                            params = ?params,
+                            "failed fetching report count of an agg share"
+                        )
+                    })?;
+                Ok(count > 0)
             })
             .buffer_unordered(usize::MAX)
             .collect::<Vec<_>>()
diff --git a/crates/daphne-service-utils/Cargo.toml b/crates/daphne-service-utils/Cargo.toml
index e9f74ad..ef95cd6 100644
--- a/crates/daphne-service-utils/Cargo.toml
+++ b/crates/daphne-service-utils/Cargo.toml
@@ -28,6 +28,7 @@ capnpc = { workspace = true, optional = true }
 [features]
 test-utils = ["dep:url", "daphne/prometheus", "daphne/test-utils"]
 durable_requests = ["dep:capnp", "dep:capnpc", "dep:prio"]
+experimental = ["daphne/experimental"]
 
 [lints]
 workspace = true
diff --git a/crates/daphne-service-utils/src/durable_requests/bindings/aggregate_store.rs b/crates/daphne-service-utils/src/durable_requests/bindings/aggregate_store.rs
index 2a27b97..4cfbdb0 100644
--- a/crates/daphne-service-utils/src/durable_requests/bindings/aggregate_store.rs
+++ b/crates/daphne-service-utils/src/durable_requests/bindings/aggregate_store.rs
@@ -25,6 +25,7 @@ super::define_do_binding! {
         Merge = "/internal/do/aggregate_store/merge",
         MarkCollected = "/internal/do/aggregate_store/mark_collected",
         CheckCollected = "/internal/do/aggregate_store/check_collected",
+        ReportCount = "/internal/do/aggregate_store/report_count"
     }
 
     fn name((version, task_id, bucket): (DapVersion, &'n TaskId, &'n DapBatchBucket)) -> ObjectIdFrom {
diff --git a/crates/daphne-worker/src/durable/aggregate_store.rs b/crates/daphne-worker/src/durable/aggregate_store.rs
index ff71f5b..b6f3fbf 100644
--- a/crates/daphne-worker/src/durable/aggregate_store.rs
+++ b/crates/daphne-worker/src/durable/aggregate_store.rs
@@ -234,6 +234,16 @@ impl AggregateStore {
         Ok(())
     }
 
+    async fn agg_share_report_count(&self) -> Result<u64> {
+        if let Some(agg_share) = &self.agg_share {
+            return Ok(agg_share.report_count);
+        };
+        Ok(self
+            .get::<DapAggregateShareMetadata>(METADATA_KEY)
+            .await?
+            .map_or(0, |m| m.report_count))
+    }
+
     async fn get_agg_share(&mut self) -> Result<&mut DapAggregateShare> {
         let keys = Self::agg_share_shard_keys();
         let agg_share = if let Some(agg_share) = self.agg_share.take() {
@@ -477,6 +487,15 @@ impl GcDurableObject for AggregateStore {
                 Response::from_json(&agg_share)
             }
 
+            // Get the number of reports in this aggregate share.
+            //
+            // Idempotent
+            // Output: `u64`
+            Some(bindings::AggregateStore::ReportCount) => {
+                let count = self.agg_share_report_count().await?;
+                Response::from_json(&count)
+            }
+
             // Mark this bucket as collected.
             //
             // Non-idempotent (do not retry)


Test patch:
diff --git a/crates/dapf/src/functions/test_routes.rs b/crates/dapf/src/functions/test_routes.rs
new file mode 100644
index 0000000..8dc6145
--- /dev/null
+++ b/crates/dapf/src/functions/test_routes.rs
@@ -0,0 +1,64 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+//! Functions used to interact with `/internal/*` routes that implement the [interop][interop] draft.
+//!
+//! [interop]: https://divergentdave.github.io/draft-dcook-ppm-dap-interop-test-design/draft-dcook-ppm-dap-interop-test-design.html
+
+use anyhow::Context;
+use daphne::{
+    hpke::{HpkeKemId, HpkeReceiverConfig},
+    messages::HpkeConfigList,
+};
+use rand::{thread_rng, Rng};
+use url::Url;
+
+use crate::HttpClient;
+
+impl HttpClient {
+    pub async fn add_hpke_config(
+        &self,
+        aggregator_url: &Url,
+        kem_alg: HpkeKemId,
+    ) -> anyhow::Result<()> {
+        let mut rng = thread_rng();
+        let HpkeConfigList { hpke_configs } = match self.get_hpke_config(aggregator_url, None).await
+        {
+            Ok(configs) => configs,
+            Err(_) => HpkeConfigList {
+                hpke_configs: vec![],
+            },
+        };
+
+        let receiver_config = loop {
+            let config = HpkeReceiverConfig::gen(rng.gen(), kem_alg)
+                .context("failed to generate HPKE receiver config")?;
+            if hpke_configs.iter().all(|c| c.id != config.config.id) {
+                break config;
+            }
+        };
+
+        self.post(
+            aggregator_url
+                .join("internal/test/add_hpke_config")
+                .unwrap(),
+        )
+        .json(&receiver_config)
+        .send()
+        .await
+        .context("adding the hpke config")?
+        .error_for_status()
+        .context("adding the hpke config")?;
+        Ok(())
+    }
+
+    pub async fn delete_all_storage(&self, aggregator_url: &Url) -> anyhow::Result<()> {
+        self.post(aggregator_url.join("/internal/delete_all").unwrap())
+            .send()
+            .await
+            .context("deleting storage")?
+            .error_for_status()
+            .context("deleting storage")?;
+        Ok(())
+    }
+}
diff --git a/crates/dapf/src/test_routes.rs b/crates/dapf/src/test_routes.rs
deleted file mode 100644
index 9c2b5a8..0000000
--- a/crates/dapf/src/test_routes.rs
+++ /dev/null
@@ -1,71 +0,0 @@
-// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
-// SPDX-License-Identifier: BSD-3-Clause
-
-use anyhow::Context;
-use daphne::{
-    hpke::{HpkeKemId, HpkeReceiverConfig},
-    messages::HpkeConfigList,
-};
-use rand::{thread_rng, Rng};
-use url::Url;
-
-use crate::HttpClient;
-
-pub async fn add_hpke_config(
-    http_client: &HttpClient,
-    aggregator_url: &Url,
-    kem_alg: HpkeKemId,
-) -> anyhow::Result<()> {
-    let mut rng = thread_rng();
-    let HpkeConfigList { hpke_configs } =
-        match http_client.get_hpke_config(aggregator_url, None).await {
-            Ok(configs) => configs,
-            Err(_) => HpkeConfigList {
-                hpke_configs: vec![],
-            },
-        };
-
-    let receiver_config = loop {
-        let config = HpkeReceiverConfig::gen(rng.gen(), kem_alg)
-            .context("failed to generate HPKE receiver config")?;
-        if hpke_configs.iter().all(|c| c.id != config.config.id) {
-            break config;
-        }
-    };
-
-    http_client
-        .post(
-            aggregator_url
-                .join("internal/test/add_hpke_config")
-                .unwrap(),
-        )
-        .json(&receiver_config)
-        .send()
-        .await
-        .context("adding the hpke config")?
-        .error_for_status()
-        .context("adding the hpke config")?;
-
-    println!("config added!");
-
-    Ok(())
-}
-
-pub async fn delete_all_storage(
-    http_client: &HttpClient,
-    aggregator_url: &Url,
-    kem_alg: Option<HpkeKemId>,
-) -> anyhow::Result<()> {
-    http_client
-        .post(aggregator_url.join("/internal/delete_all").unwrap())
-        .send()
-        .await
-        .context("deleting storage")?
-        .error_for_status()
-        .context("deleting storage")?;
-
-    if let Some(kem_alg) = kem_alg {
-        add_hpke_config(http_client, aggregator_url, kem_alg).await?;
-    }
-    Ok(())
-}
diff --git a/crates/daphne-service-utils/src/test_route_types.rs b/crates/daphne-service-utils/src/test_route_types.rs
index 2e89d72..6591188 100644
--- a/crates/daphne-service-utils/src/test_route_types.rs
+++ b/crates/daphne-service-utils/src/test_route_types.rs
@@ -1,7 +1,14 @@
 // Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use daphne::messages::{Duration, TaskId, Time};
+//! Types used in `/internal/*` routes to implement the [interop][interop] draft
+//!
+//! [interop]: https://divergentdave.github.io/draft-dcook-ppm-dap-interop-test-design/draft-dcook-ppm-dap-interop-test-design.html
+
+use daphne::{
+    messages::{Duration, TaskId, Time},
+    vdaf::{Prio3Config, VdafConfig},
+};
 use serde::{Deserialize, Serialize};
 use url::Url;
 
@@ -23,6 +30,47 @@ pub struct InternalTestVdaf {
     pub chunk_length: Option<String>,
 }
 
+impl From<VdafConfig> for InternalTestVdaf {
+    fn from(vdaf: VdafConfig) -> Self {
+        let (typ, bits, length, chunk_length) = match vdaf {
+            VdafConfig::Prio3(prio3) => match prio3 {
+                Prio3Config::Count => ("Prio3Count", None, None, None),
+                Prio3Config::Sum { bits } => ("Prio3Sum", Some(bits), None, None),
+                Prio3Config::Histogram {
+                    length,
+                    chunk_length,
+                } => ("Prio3Histogram", None, Some(length), Some(chunk_length)),
+                Prio3Config::SumVec {
+                    bits,
+                    length,
+                    chunk_length,
+                } => ("Prio3SumVec", Some(bits), Some(length), Some(chunk_length)),
+                Prio3Config::SumVecField64MultiproofHmacSha256Aes128 {
+                    bits,
+                    length,
+                    chunk_length,
+                    num_proofs: _unimplemented,
+                } => (
+                    "Prio3SumVecField64MultiproofHmacSha256Aes128",
+                    Some(bits),
+                    Some(length),
+                    Some(chunk_length),
+                ),
+            },
+            VdafConfig::Prio2 { .. } => ("Prio2", None, None, None),
+            VdafConfig::Pine(_) => ("Pine", None, None, None),
+            #[cfg(feature = "experimental")]
+            VdafConfig::Mastic { .. } => todo!(),
+        };
+        Self {
+            typ: typ.into(),
+            bits: bits.map(|a| a.to_string()),
+            length: length.map(|a| a.to_string()),
+            chunk_length: chunk_length.map(|a| a.to_string()),
+        }
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub struct InternalTestAddTask {