feat(cli): dissolve the real-data blueprint sweep into the campaign path (#210 T4)

aura sweep <bp.json> --real ... --axis ... now runs as thin sugar over
its generated campaign document through the one campaign executor: the
dispatch registers the canonical blueprint (topology_hash IS
content_id_of over the same canonical bytes, so the strategy ref
resolves against the existing single store write), strips the CLI's
wrapped axis names to the raw campaign namespace (#203 bind
convention), converts the probed archive window back through the
ingest seam's own epoch_ns_to_unix_ms (made pub; C3 — the seam owns
the ms/ns convention, never an inline division), and hands off to
verb_sugar::run_sweep_sugar. The inline real-data arm is gone; the
synthetic arm and --list-axes are untouched.

campaign_run gains RunPresentation {Full, MemberLinesOnly}:
run_campaign keeps Full (record line unchanged, project gate
unchanged); the sugar path runs MemberLinesOnly — the generated doc's
emit already limits stdout to family_table member lines, the mode
suppresses only the final record line. The record append is identical
in both modes; no project gate on the by-id path (the dissolved verb
works project-less, exactly as before).

The characterization pin survives the re-cut with exactly the one
sanctioned assertion flip (instrument-absence -> instrument==GER40)
and now also pins the generated artifacts: one process + one campaign
document auto-registered (dedup on a second identical invocation),
raw axis keys in the stored document, the stored window a ms sub-range
of the request, C1 at the member-report level with the registry's
run-suffix increment isolated.

Suite 1055/0, clippy clean, gated real-data paths executed locally.

refs #210

crates/aura-cli/src/campaign_run.rs

@@ -262,6 +262,19 @@ impl MemberRunner for CliMemberRunner<'_> {
     }
 }
 
+/// Stdout shape of a campaign run. `Full` is `aura campaign run` (emit-gated
+/// member/selection lines + the always-on final `campaign_run` record line).
+/// `MemberLinesOnly` is dissolved-verb sugar: the generated document's
+/// `emit: ["family_table"]` already limits emission to member lines; the mode
+/// additionally suppresses the final record line so the verb's stdout
+/// contract is reproduced byte-for-byte. The record append is identical in
+/// both modes — presentation changes, the record does not.
+#[derive(Clone, Copy, PartialEq)]
+pub(crate) enum RunPresentation {
+    Full,
+    MemberLinesOnly,
+}
+
 /// `aura campaign run <target>`: resolve, gate, execute, emit. Every `Err`
 /// is a refusal `campaign_cmd` renders as `aura: {msg}` + exit 1.
 pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> {
@@ -297,11 +310,26 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> {
         ));
     };
 
-    // One uniform path from here: fetch the stored canonical bytes by id (so
-    // file addressing and id addressing produce the same realization by
-    // construction) and re-run the intrinsic tier on them.
+    run_campaign_by_id(&campaign_id, env, RunPresentation::Full)
+}
+
+/// The one campaign executor path from a resolved content id onward: fetch
+/// the stored canonical bytes by id (so file addressing and id addressing
+/// produce the same realization by construction) and re-run the intrinsic
+/// tier on them, execute, and emit per `presentation`. Shared by
+/// `run_campaign` (`RunPresentation::Full`) and the dissolved-verb sugar path
+/// (`verb_sugar::run_sweep_sugar`, `RunPresentation::MemberLinesOnly`) — no
+/// project gate here: the sugar path must work project-less exactly as the
+/// inline verb it replaces did (store mechanics run against `env.registry()`
+/// in both cases).
+pub(crate) fn run_campaign_by_id(
+    campaign_id: &str,
+    env: &Env,
+    presentation: RunPresentation,
+) -> Result<(), String> {
+    let registry = env.registry();
     let campaign_text = registry
-        .get_campaign(&campaign_id)
+        .get_campaign(campaign_id)
         .map_err(|e| e.to_string())?
         .ok_or_else(|| format!("no campaign {campaign_id} in the project store"))?;
     let campaign = parse_campaign(&campaign_text)
@@ -369,7 +397,7 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> {
         server: Arc::new(data_server::DataServer::new(env.data_path())),
     };
     let outcome = aura_campaign::execute(
-        &campaign_id,
+        campaign_id,
         &campaign,
         &process,
         &strategies,
@@ -437,11 +465,13 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> {
             }
         }
     }
-    println!(
-        "{}",
-        serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
-            .expect("campaign run record serializes")
-    );
+    if presentation == RunPresentation::Full {
+        println!(
+            "{}",
+            serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
+                .expect("campaign run record serializes")
+        );
+    }
 
     // Trace persistence runs LAST, after the always-on record line (0109):
     // stdout stays data-pure (the record line is the wire contract) and the

crates/aura-cli/src/main.rs

@@ -3201,6 +3201,10 @@ fn blueprint_mc_family(
 /// valid sweep on a name collision). `persist` is therefore not yet load-bearing here;
 /// it is retained for the deferred per-member trace path. The family record itself is
 /// written unconditionally, so lineage (C18/C21) holds whether or not `--trace` is given.
+///
+/// Synthetic-only: real-data invocations route through
+/// `verb_sugar::run_sweep_sugar` at the dispatch boundary and never reach
+/// this fn.
 fn run_blueprint_sweep(
     doc: &str, axes: &[(String, Vec<Scalar>)], name: &str, persist: bool, data: DataSource,
     env: &project::Env,
@@ -4473,7 +4477,79 @@ fn dispatch_sweep(a: SweepCmd, env: &project::Env) {
                 eprintln!("aura: {m}");
                 std::process::exit(2);
             });
-            run_blueprint_sweep(&doc, &axes, &name, persist, DataSource::from_choice(data, env), env);
+            match &data {
+                DataChoice::Real { symbol, .. } => {
+                    // The dissolved branch: real-data blueprint sweeps run as
+                    // sugar over a generated campaign document through the
+                    // one campaign executor. The blueprint store write (by
+                    // topology hash) is kept — the
+                    // canonical bytes are also the strategy ref's content, and
+                    // `topology_hash` IS `content_id_of` applied to the same
+                    // canonical bytes (verified by reading: `topology_hash` ==
+                    // `content_id` == `aura_research::content_id_of`), so the
+                    // strategy ref resolves against this one write; a second
+                    // put under `content_id_of(&canonical)` would target the
+                    // identical key and is not needed.
+                    let symbol = symbol.clone();
+                    let source = DataSource::from_choice(data, env);
+                    let (from_ts, to_ts) = source.full_window(env);
+                    // `full_window` probes the ARCHIVE's actual first/last bar
+                    // and returns aura's native epoch-ns `Timestamp` (the
+                    // ms->ns crossing happens once, at the ingest seam — C3);
+                    // the campaign document's `Window` field is Unix-ms (same
+                    // currency as `--from`/`--to` and every existing campaign
+                    // fixture, e.g. `campaign_doc_json` in research_docs.rs).
+                    // Convert back through the seam's own
+                    // `aura_ingest::epoch_ns_to_unix_ms` at this one
+                    // seam-crossing (never reimplement the division inline) so
+                    // the executor's `unix_ms_to_epoch_ns` re-normalizes
+                    // exactly once downstream, not twice.
+                    let (from_ms, to_ms) = (
+                        aura_ingest::epoch_ns_to_unix_ms(from_ts),
+                        aura_ingest::epoch_ns_to_unix_ms(to_ts),
+                    );
+                    let blueprint = blueprint_from_json(&doc, &|t| env.resolve(t))
+                        .expect("doc parse-validated at the dispatch boundary");
+                    let canonical =
+                        blueprint_to_json(&blueprint).expect("a loaded blueprint re-serializes");
+                    let reg = env.registry();
+                    let topo = topology_hash(&blueprint);
+                    reg.put_blueprint(&topo, &canonical).unwrap_or_else(|e| {
+                        eprintln!("aura: {e}");
+                        std::process::exit(1);
+                    });
+                    // Campaign documents speak the RAW
+                    // campaign-axis namespace (`bind_axes`'s convention, #203) —
+                    // everything after the wrapper's one node segment — while
+                    // `--axis` on a blueprint file is typed against the WRAPPED
+                    // probe (`blueprint_axis_probe`/`list_blueprint_axes`, e.g.
+                    // `sma_signal.fast.length`). Strip that one segment before
+                    // generating the campaign doc, or `validate_campaign_refs`
+                    // (which checks axes against the UNWRAPPED blueprint's own
+                    // `param_space`) refuses every axis as unknown.
+                    let raw_axes: Vec<(String, Vec<Scalar>)> = axes
+                        .iter()
+                        .map(|(n, v)| {
+                            let raw =
+                                n.split_once('.').map(|(_, rest)| rest.to_string()).unwrap_or_else(|| n.clone());
+                            (raw, v.clone())
+                        })
+                        .collect();
+                    verb_sugar::run_sweep_sugar(
+                        &raw_axes, &name, &symbol, from_ms, to_ms, &canonical, env,
+                    )
+                    .unwrap_or_else(|m| {
+                        eprintln!("aura: {m}");
+                        std::process::exit(1);
+                    });
+                }
+                DataChoice::Synthetic => {
+                    run_blueprint_sweep(
+                        &doc, &axes, &name, persist,
+                        DataSource::from_choice(data, env), env,
+                    );
+                }
+            }
         }
         None => {
             let usage = || "Usage: aura sweep [--strategy <sma|momentum|r-sma|r-breakout|r-meanrev>] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--name <n> | --trace <n>] [--fast <csv>] [--slow <csv>] [--stop-length <csv>] [--stop-k <csv>] [--channel <csv>] [--window <csv>] [--band-k <csv>]".to_string();

crates/aura-cli/src/verb_sugar.rs

@@ -7,10 +7,6 @@
 //! becomes durable, diffable, reproducible intent), then run through the one
 //! campaign executor in sugar presentation mode (member lines only).
 
-// Temporary until the dispatch rewire wires this module (plan Task 4, which
-// removes this line): the translator lands one commit ahead of its caller.
-#![allow(dead_code)]
-
 use std::collections::BTreeMap;
 
 use aura_core::{Scalar, ScalarKind};
@@ -116,6 +112,23 @@ pub(crate) fn register_generated(
     Ok((process_id, campaign_id))
 }
 
+/// Run one dissolved sweep invocation end-to-end: register the generated
+/// documents, then execute through the one campaign path in sugar mode.
+pub(crate) fn run_sweep_sugar(
+    axes: &[(String, Vec<Scalar>)],
+    name: &str,
+    symbol: &str,
+    from_ms: i64,
+    to_ms: i64,
+    blueprint_canonical: &str,
+    env: &crate::project::Env,
+) -> Result<(), String> {
+    let generated = translate_sweep(axes, name, symbol, from_ms, to_ms, blueprint_canonical)?;
+    let reg = env.registry();
+    let (_process_id, campaign_id) = register_generated(&reg, &generated)?;
+    crate::campaign_run::run_campaign_by_id(&campaign_id, env, crate::campaign_run::RunPresentation::MemberLinesOnly)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/aura-cli/tests/cli_run.rs

@@ -1403,22 +1403,22 @@ fn sweep_real_is_byte_deterministic_across_runs() {
 const GER40_SEPT2024_FROM_MS: &str = "1725148800000";
 const GER40_SEPT2024_TO_MS: &str = "1727740799999";
 
-/// Property (docs/specs/sweep-dissolution.md, Testing strategy 1 — the
-/// characterization pin ahead of the sweep-dissolution re-cut): today's inline
-/// `aura sweep <blueprint.json> --real SYM --axis …` path (`run_blueprint_sweep`)
+/// Property: the real-data blueprint sweep path (`aura sweep <blueprint.json>
+/// --real SYM --axis …`, dispatched as sugar over the one campaign executor)
 /// prints one member line per grid point, IN AXIS ORDER (odometer: the first
 /// `--axis` varies slowest), each line JSON with a `family_id` and a `report`
 /// whose `manifest.params` carries every swept param binding — and, the
-/// load-bearing pin the dissolution cycle must preserve unchanged, **`report.manifest`
-/// carries NO `instrument` key at all** (the inline path leaves `RunManifest::instrument`
-/// `None`, and `#[serde(skip_serializing_if = "Option::is_none")]` omits the key
-/// entirely from the wire form — contrast the campaign substrate's stamped sibling,
-/// `campaign_run_real_e2e_sweep_gate_walkforward` in `research_docs.rs`). Also pins
-/// that the run persists exactly one new `FamilyKind::Sweep` family with the grid's
-/// member count. Gated on the local GER40 archive (the project's skip-on-no-data
-/// convention); skips cleanly when absent.
+/// sanctioned additive delta over the synthetic path, **`report.manifest`
+/// carries a stamped `instrument` key** (the campaign substrate stamps every
+/// member's manifest with the instrument under test, exactly like the sibling
+/// `campaign_run_real_e2e_sweep_gate_walkforward` in `research_docs.rs`). Also
+/// pins that the run persists exactly one new `FamilyKind::Sweep` family with
+/// the grid's member count, and that the underlying generated process/campaign
+/// documents and campaign-run record are durably auto-registered. Gated on the
+/// local GER40 archive (the project's skip-on-no-data convention); skips
+/// cleanly when absent.
 #[test]
-fn sweep_real_blueprint_member_lines_pin_the_inline_contract() {
+fn sweep_real_blueprint_member_lines_pin_the_dissolved_contract() {
     if !local_data_present() {
         eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
         return;
@@ -1466,10 +1466,12 @@ fn sweep_real_blueprint_member_lines_pin_the_inline_contract() {
                 .and_then(|p| p[1]["I64"].as_i64());
             assert_eq!(bound, Some(value), "manifest carries the swept binding {name}={value}: {line}");
         }
-        // the load-bearing pin: the inline real-data path never stamps `instrument`.
-        assert!(
-            manifest.get("instrument").is_none(),
-            "the inline blueprint sweep must NOT stamp an instrument key: {line}"
+        // the sanctioned additive delta: the campaign substrate stamps every
+        // member's manifest with the instrument.
+        assert_eq!(
+            manifest["instrument"].as_str(),
+            Some("GER40"),
+            "the dissolved sweep stamps the instrument: {line}"
         );
     }
 
@@ -1485,6 +1487,131 @@ fn sweep_real_blueprint_member_lines_pin_the_inline_contract() {
     assert!(fams_out.contains("\"kind\":\"Sweep\""), "families: {fams_out}");
     assert!(fams_out.contains("\"members\":4"), "2x2 grid -> four members: {fams_out}");
 
+    // Auto-registration: generated documents are durable intent — exactly
+    // one process doc, one campaign doc, one campaign run record.
+    let count = |sub: &str| {
+        std::fs::read_dir(dir.join("runs").join(sub))
+            .map(|d| d.count())
+            .unwrap_or(0)
+    };
+    assert_eq!(count("processes"), 1, "one generated process document registered");
+    assert_eq!(count("campaigns"), 1, "one generated campaign document registered");
+    let runs_log = std::fs::read_to_string(dir.join("runs").join("campaign_runs.jsonl"))
+        .expect("campaign_runs.jsonl exists after a sugar run");
+    assert_eq!(runs_log.lines().count(), 1, "one campaign run recorded");
+    // `CampaignRunRecord` (aura-registry's lineage.rs)
+    // carries no name field at all (campaign id/process id/run/seed/cells/
+    // generalizations/trace_name only) — the `--name` handle is NOT part of
+    // the run-log line. It IS part of the generated CAMPAIGN DOCUMENT
+    // (`translate_sweep` sets `campaign.name = name`), so the handle is
+    // checked on that persisted document instead.
+    let campaigns_dir = dir.join("runs").join("campaigns");
+    let campaign_doc_path = std::fs::read_dir(&campaigns_dir)
+        .expect("campaigns dir exists")
+        .next()
+        .expect("exactly one campaign document")
+        .expect("readable dir entry")
+        .path();
+    let campaign_doc_json = std::fs::read_to_string(&campaign_doc_path).expect("read campaign doc");
+    assert!(
+        campaign_doc_json.contains("\"name\":\"brp\""),
+        "the --name handle lands on the generated campaign document: {campaign_doc_json}"
+    );
+
+    // Property (dispatch-rewire seam, docs/specs/sweep-dissolution.md Task 4
+    // Step 3): the CLI's `--axis` names are typed against the WRAPPED probe
+    // namespace (`sma_signal.fast.length`), but a campaign document's
+    // `strategies[].axes` speak the RAW `param_space` namespace
+    // (`fast.length`) — `bind_axes`'s convention (#203). `dispatch_sweep`
+    // strips the wrapper's one leading node segment before generating the
+    // document; a regression that stopped stripping (or stripped one segment
+    // too many/few) would persist the wrong axis keys and
+    // `validate_campaign_refs` would refuse every axis as unknown before any
+    // member ran — this pin catches that at the artifact level, directly.
+    let campaign_doc: serde_json::Value =
+        serde_json::from_str(&campaign_doc_json).expect("generated campaign doc parses as JSON");
+    let axes = campaign_doc["strategies"][0]["axes"]
+        .as_object()
+        .expect("strategies[0].axes is an object");
+    assert!(
+        axes.contains_key("fast.length"),
+        "the raw (unwrapped) axis name is the document key: {campaign_doc_json}"
+    );
+    assert!(
+        axes.contains_key("slow.length"),
+        "the raw (unwrapped) axis name is the document key: {campaign_doc_json}"
+    );
+    assert!(
+        !axes.contains_key("sma_signal.fast.length") && !axes.contains_key("sma_signal.slow.length"),
+        "the wrapped CLI probe name must never leak into the stored document: {campaign_doc_json}"
+    );
+
+    // Property (dispatch-rewire seam: the ms<->ns crossing, C3): the probed
+    // archive window is carried through `DataSource::full_window` (aura's
+    // native epoch-ns `Timestamp`, clamped to the archive's actual first/last
+    // bar inside the requested range) and converted back to Unix-ms via
+    // `aura_ingest::epoch_ns_to_unix_ms` exactly once before landing in the
+    // generated document's `data.windows[0]`. A regression that skipped the
+    // conversion (or applied it twice, or swapped ms<->ns) would persist a
+    // window off by a factor of ~1e6 from the requested `--from`/`--to` —
+    // silently, since the executor would still run (over the wrong or an
+    // empty range) rather than refuse outright. Pin the sane invariant that
+    // survives the archive's exact clamp: the stored window is a
+    // (from <= to) sub-range of the requested Unix-ms bounds, not a value
+    // orders of magnitude away.
+    let from_req = GER40_SEPT2024_FROM_MS.parse::<i64>().unwrap();
+    let to_req = GER40_SEPT2024_TO_MS.parse::<i64>().unwrap();
+    let window = &campaign_doc["data"]["windows"][0];
+    let from_stored = window["from_ms"].as_i64().expect("from_ms is an integer");
+    let to_stored = window["to_ms"].as_i64().expect("to_ms is an integer");
+    assert!(
+        from_req <= from_stored && from_stored <= to_stored && to_stored <= to_req,
+        "the stored window is a ms sub-range of the requested [{from_req}, {to_req}]: {campaign_doc_json}"
+    );
+
+    // Dedup: a second identical invocation reuses both documents (content-
+    // addressed) and appends a second run record.
+    let out2 = std::process::Command::new(BIN)
+        .args([
+            "sweep", &fixture,
+            "--real", "GER40",
+            "--from", GER40_SEPT2024_FROM_MS,
+            "--to", GER40_SEPT2024_TO_MS,
+            "--axis", "sma_signal.fast.length=2,4",
+            "--axis", "sma_signal.slow.length=8,16",
+            "--name", "brp",
+        ])
+        .current_dir(&dir)
+        .output()
+        .unwrap();
+    assert!(out2.status.success(), "second run stderr: {}", String::from_utf8_lossy(&out2.stderr));
+    // `family_id` carries a per-execution uniqueness
+    // suffix ("the '-{run}' suffix is appended by the registry",
+    // aura-campaign/src/exec.rs `run_cell`) — a fresh family record is
+    // written on every execution even when the campaign/process documents
+    // themselves dedupe, so two identical invocations are NOT literally
+    // byte-identical on stdout: only the trailing run-suffix (0 -> 1)
+    // differs. C1 determinism is checked at the member-report level instead
+    // (every report field byte-identical), plus the run-suffix increment.
+    let out2_stdout = String::from_utf8_lossy(&out2.stdout).into_owned();
+    let lines1: Vec<&str> = stdout.lines().collect();
+    let lines2: Vec<&str> = out2_stdout.lines().collect();
+    assert_eq!(lines2.len(), lines1.len(), "same member count on the dedup run: {out2_stdout}");
+    for (l1, l2) in lines1.iter().zip(&lines2) {
+        let v1: serde_json::Value = serde_json::from_str(l1).expect("first run member line parses");
+        let v2: serde_json::Value = serde_json::from_str(l2).expect("second run member line parses");
+        assert_eq!(v1["report"], v2["report"], "identical invocations reproduce identical member reports (C1): {l1} vs {l2}");
+        let (fam1, fam2) = (v1["family_id"].as_str().unwrap(), v2["family_id"].as_str().unwrap());
+        assert_eq!(
+            fam1.trim_end_matches(|c: char| c.is_ascii_digit()),
+            fam2.trim_end_matches(|c: char| c.is_ascii_digit()),
+            "same family name up to the run-suffix: {fam1} vs {fam2}"
+        );
+        assert!(fam1.ends_with("-0") && fam2.ends_with("-1"), "run-suffix increments 0 -> 1: {fam1} vs {fam2}");
+    }
+    assert_eq!(count("processes"), 1, "second identical run dedupes the process doc");
+    assert_eq!(count("campaigns"), 1, "second identical run dedupes the campaign doc");
+
     let _ = std::fs::remove_dir_all(&dir);
 }
 

crates/aura-ingest/src/lib.rs

@@ -42,10 +42,13 @@ pub fn unix_ms_to_epoch_ns(time_ms: i64) -> Timestamp {
 }
 
 /// Inverse of [`unix_ms_to_epoch_ns`]: project aura's canonical epoch-ns
-/// [`Timestamp`] back to data-server's Unix-millisecond time. **Private** — the
-/// seam owns the ms↔ns convention (C3); a consumer threads `Timestamp`s and never
-/// converts. Floor division by 1e6 ns/ms (archived M1 data has no sub-ms instant).
-fn epoch_ns_to_unix_ms(ts: Timestamp) -> i64 {
+/// [`Timestamp`] back to data-server's Unix-millisecond time. The seam owns the
+/// ms↔ns convention (C3): an in-engine consumer threads `Timestamp`s and never
+/// converts, but a consumer that must persist or report in ms currency (a
+/// generated document, a CLI window field) calls through this one named
+/// function rather than reimplementing the division inline. Floor division by
+/// 1e6 ns/ms (archived M1 data has no sub-ms instant).
+pub fn epoch_ns_to_unix_ms(ts: Timestamp) -> i64 {
     ts.0 / 1_000_000
 }