datasynth_group/aggregate/

driver.rs

//! Aggregate phase driver — Task 9.1.
//!
//! Wires Chunks 5–8 into a single entrypoint: [`run_aggregate`] walks the
//! per-entity shard archives produced by
//! [`crate::shard::run_shard`], folds them through pre-elimination,
//! IC matching, eliminations, IAS 21 translation, NCI / equity-method
//! overlays, and consolidated FS assembly, and emits every
//! aggregate-phase artefact under `{out_dir}/consolidated/` and
//! `{out_dir}/ic_eliminations/`.
//!
//! # v5.0 ordering
//!
//! The driver runs the chunks **in this strict order** because each
//! step depends on the prior step's outputs:
//!
//! 1. **Chunk 5** — Pre-elimination aggregation, IC matching,
//!    elimination JEs, post-elimination consolidated TB.  The
//!    presentation currency check on the pre-elim aggregator means
//!    every contributing entity must already be denominated in the
//!    presentation currency at this point — the v5.0 fixture (Mini-
//!    Acme) is single-currency CHF so this trivially holds.  Multi-
//!    currency engagements will need to translate first; that path is
//!    documented in the spec but not exercised in v5.0 driver-level
//!    tests.
//! 2. **Chunk 6** — IAS 21 per-entity translation, CTA rollforward,
//!    translation worksheet emission.  Runs *after* the post-elim TB
//!    so the consolidated TB and the translation worksheet are both
//!    grounded in the same set of contributing entities.
//! 3. **Chunk 7** — NCI rollforward per Full-method subsidiary, and
//!    equity-method investment rollforward per EquityMethod investee.
//!    These overlays sit on top of the post-elim TB; the driver applies
//!    them via `apply_nci_and_equity_method` before assembling the FS.
//! 4. **Chunk 8** — Consolidated FS assembly (BS / IS / CF / Changes
//!    in Equity), consolidation schedule, notes, and JSON writer.
//!
//! # Missing-shard handling
//!
//! [`AggregateOptions::tolerate_missing_shards`] toggles between fail-
//! fast and best-effort recovery semantics:
//!
//! - `false` (default) — any missing entity directory or
//!   `period_close/trial_balances.json` produces a
//!   [`crate::errors::GroupError::Aggregate`] error naming the entity
//!   and the path the driver expected.  This is the production
//!   contract: a complete shard archive is required for a complete
//!   consolidation.
//! - `true` — missing entities are pushed to
//!   [`AggregateSummary::entities_missing`] and a `tracing::warn!` is
//!   logged.  The driver then continues with the remaining entities,
//!   which is useful for partial-archive recovery scenarios.
//!
//! Equity-method / fair-value / proportional entities are still
//! **expected** to have shards on disk (the runner generated them) but
//! their TBs are captured separately via
//! [`AggregateOptions::deferred_entity_tbs`-style sidecar] for Chunk 7
//! to consume — the driver does not fold them into the consolidated TB.
//!
//! # Prior-period plumbing
//!
//! When [`AggregateOptions::prior_period_aggregate`] is `Some(path)`,
//! the driver reads opening NCI and equity-method carrying values from
//! `path/consolidated/{nci_rollforward,equity_method_investments}.json`
//! via the existing ingestion helpers in
//! [`crate::aggregate::nci::opening`] and
//! [`crate::aggregate::equity_method`].  Opening CTA balances are
//! similarly loaded from the prior `cta_rollforward.json`.  When `None`,
//! every opening defaults to zero — the engagement's first period.
//!
//! # Determinism
//!
//! Every step the driver invokes is deterministic given identical
//! inputs.  The driver itself walks `manifest.ownership_graph.entities`
//! in declaration order; sub-steps that need lexicographic ordering
//! (pre-elim aggregator, IC matcher) sort internally.  Two runs of
//! `run_aggregate` over the same shard archive produce byte-identical
//! `consolidated/*.json` and `ic_eliminations/*.json` files.

use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};

use chrono::NaiveDate;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};

use datasynth_core::models::balance::TrialBalance;
use datasynth_core::models::business_combination::BusinessCombination;
use datasynth_core::models::JournalEntry;
use datasynth_standards::framework::AccountingFramework;

use crate::aggregate::coverage_report::{build_coverage_report, write_coverage_report};
use crate::aggregate::elimination::{eliminations_to_journal_entries, generate_eliminations};
use crate::aggregate::equity_method::{
    compute_equity_method_investment, ingest_opening_equity_method_carrying_values,
    ingest_opening_suppressed_losses, write_equity_method_investments, write_suppressed_losses,
    EquityMethodInputs, EquityMethodInvestment,
};
use crate::aggregate::fs::{
    build_consolidated_balance_sheet_with_names, build_consolidated_cash_flow,
    build_consolidated_income_statement_with_names,
    build_consolidation_schedule_with_contributions, build_notes_to_consolidated_fs,
    build_statement_of_changes_in_equity, write_consolidated_fs, AccountNameDictionary,
    CashFlowInputs, ConsolidatedFinancialStatements, EquityChangesInputs, NotesInputs,
};
#[allow(unused_imports)]
use crate::aggregate::ic_matcher::match_ic_pairs;
use crate::aggregate::nci::{
    compute_nci_rollforward, ingest_opening_nci_balances, write_nci_rollforward, NciInputs,
    NciRollforward,
};
use crate::aggregate::post_elim::{apply_eliminations_to_tb, apply_nci_and_equity_method};
use crate::aggregate::pre_elim::{
    accumulate_entity_into_aggregate, empty_aggregate, finalise_streaming_aggregate, AggregatedTb,
};
use crate::aggregate::tb_loader::load_entity_trial_balance;
use crate::aggregate::translation::cta::{
    cta_rollforward, write_cta_rollforward, CtaRollforward, CONSOLIDATED_SUBDIR,
    CTA_ROLLFORWARD_FILENAME,
};
use crate::aggregate::translation::restatement::{select_restatement_path, RestatementPath};
use crate::aggregate::translation::translate::{
    translate_entity_tb_with_hyperinflation, translate_entity_tb_with_indexed_restatement, DrCr,
    TranslatedTb,
};
use crate::aggregate::translation::worksheet::write_translation_worksheet;
use crate::config::ConsolidationMethod;
use crate::errors::{GroupError, GroupResult};
use crate::manifest::builder::{GroupManifest, ManifestEntity};

// ── Public types ──────────────────────────────────────────────────────────────

/// Knobs the caller can supply to tune [`run_aggregate`] behaviour.
///
/// All fields are optional; the [`Default`] impl gives the production
/// fail-fast contract (no prior period, missing shards are errors).
#[derive(Debug, Clone, Default)]
pub struct AggregateOptions {
    /// Optional path to a prior period's `consolidated/` directory's
    /// **parent** — i.e. the prior period's `out_dir`.  When supplied,
    /// opening NCI, equity-method carrying values, and CTA balances are
    /// loaded from
    /// `prior_period_aggregate/consolidated/{nci_rollforward,equity_method_investments,cta_rollforward}.json`.
    /// When `None`, every opening defaults to zero per entity.
    pub prior_period_aggregate: Option<PathBuf>,
    /// When `true`, missing per-entity shard archives produce a warning
    /// rather than an error — useful for partial-archive recovery
    /// scenarios.  Defaults to `false`: missing shards fail fast.
    pub tolerate_missing_shards: bool,
    /// **v5.2 IAS 36 § 10** — per-period CGU goodwill impairment test
    /// inputs.  Each entry references a CGU defined in
    /// [`crate::manifest::GroupManifest::cgu_plan`] and supplies the
    /// CGU's other-asset carrying amount, fair-value-less-costs, and
    /// value-in-use estimates for this period.  When non-empty,
    /// [`run_aggregate`] joins these against the manifest plan, runs
    /// [`datasynth_core::models::CguImpairmentTest::run`] per CGU, and
    /// emits `consolidated/cgu_impairment_tests.json`.  When empty
    /// (default), no impairment tests run and no artefact is emitted —
    /// preserves backwards compatibility byte-for-byte for engagements
    /// without CGU configuration.
    pub cgu_test_inputs: Vec<crate::aggregate::cgu_impairment::CguTestInputs>,
    /// **v5.5.2 IAS 29 § 12** — per-currency general price index (CPI)
    /// series, keyed by ISO 4217 currency code.  When supplied, the
    /// aggregate driver matches each entity in
    /// [`datasynth_core::models::HyperinflationStatus::Hyperinflationary`]
    /// against this map by its functional currency.  Matched entities
    /// are translated via [`translate_entity_tb_with_indexed_restatement`]
    /// (composing IAS 29 § 12 indexed restatement with IAS 21 § 42(b)
    /// closing-rate translation).  Hyperinflationary entities without a
    /// matching currency entry log a warning and fall back to the
    /// closing-rate-only path (the v5.5.0 behaviour).  Non-hyperinflationary
    /// entities ignore this map.  Empty (default) preserves
    /// backwards-compatible behaviour byte-for-byte.
    pub cpi_series_by_currency:
        BTreeMap<String, datasynth_core::models::hyperinflation::GeneralPriceIndex>,
}

/// Top-level result returned by [`run_aggregate`].
///
/// All numeric fields mirror the post-overlay consolidated TB; the
/// `artifacts_written` list is the absolute paths of every file the
/// driver emitted, in the order they were written.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct AggregateSummary {
    /// Group identifier, mirrors [`GroupManifest::group_id`].
    pub group_id: String,
    /// Group presentation currency.
    pub presentation_currency: String,
    /// Period end date.
    pub as_of_date: NaiveDate,
    /// Codes of every entity whose shard archive contributed to the
    /// pre-elimination aggregation (Parent + Full).  Includes
    /// equity-method / fair-value / proportional entities whose TBs were
    /// loaded for Chunk 7 use even though they were not folded into the
    /// pre-elim totals.  Sorted lexicographically.
    pub entities_processed: Vec<String>,
    /// Codes of entities the driver expected but could not find on
    /// disk.  Always empty when [`AggregateOptions::tolerate_missing_shards`]
    /// is `false`.  Sorted lexicographically.
    pub entities_missing: Vec<String>,
    /// Codes of entities held back from the pre-elim aggregation per
    /// their consolidation method (equity-method / fair-value /
    /// proportional).  Sorted lexicographically.
    pub deferred_entities: Vec<String>,
    /// Number of IC pairs successfully matched.
    pub matched_pairs: usize,
    /// IC matching coverage = matched / planned.
    pub coverage: f64,
    /// Sum of `total_assets` from the consolidated balance sheet.
    pub total_assets: Decimal,
    /// Sum of `total_liabilities` from the consolidated balance sheet.
    pub total_liabilities: Decimal,
    /// Sum of equity attributable to owners of the parent.
    pub total_equity: Decimal,
    /// Sum of non-controlling interest equity.
    pub total_nci: Decimal,
    /// Absolute paths of every artefact the driver wrote, in
    /// emission order:
    ///
    /// 1. `ic_eliminations/ic_matching_coverage.json`
    /// 2. `consolidated/cta_rollforward.json`
    /// 3. `consolidated/translation_worksheet.json`
    /// 4. `consolidated/nci_rollforward.json`
    /// 5. `consolidated/equity_method_investments.json`
    /// 6. `consolidated/consolidated_financial_statements.json`
    /// 7. `consolidated/consolidation_schedule.json`
    /// 8. `consolidated/notes_to_consolidated_fs.json`
    pub artifacts_written: Vec<PathBuf>,
}

// ── Public API ────────────────────────────────────────────────────────────────

/// Drive the aggregate phase end-to-end.
///
/// Walks the per-entity shard archives under `shards_dir/entities/`,
/// folds them through Chunks 5–8, and writes every artefact under
/// `out_dir`.  Returns an [`AggregateSummary`] describing the
/// consolidation outcome and the absolute paths of every emitted file.
///
/// `shards_dir == out_dir` is supported (and used by
/// [`crate::standalone::generate_standalone`] — the runner writes
/// shards directly into the same root the aggregate driver consumes).
///
/// # Errors
///
/// - [`GroupError::Aggregate`] if a missing shard archive is fatal
///   (`tolerate_missing_shards == false`).
/// - [`GroupError::Aggregate`] propagated from the sub-modules
///   (currency mismatch, balance regressions, FX rate gaps, etc.).
/// - [`GroupError::Io`] if the per-entity TB / JE files cannot be read
///   or the consolidated artefacts cannot be written.
/// - [`GroupError::Serde`] if any file fails to deserialise / serialise.
pub fn run_aggregate(
    manifest: &GroupManifest,
    shards_dir: &Path,
    out_dir: &Path,
    opts: &AggregateOptions,
) -> GroupResult<AggregateSummary> {
    // Resolve framework once for downstream consumers (translation +
    // notes) so we don't re-parse it per-entity.
    let framework = resolve_primary_framework(manifest);

    // ── 1+2+9. v5.31 C1 streaming walk ──────────────────────────────────
    //
    // The previous v5.0–v5.30 driver loaded all 1800–2000 contributing
    // TBs into a `Vec<(String, TrialBalance)>` (`walk_entity_archives`),
    // ran pre-elim accumulation, then translated *all* TBs into a
    // second 1800–2000-element vector (`translate_all_contributing`).
    // Combined hold peaked at ~400 GB on the 2026-05-26 2000-entity
    // run, OOM-killing the aggregate phase at 226 GB anon-rss.
    //
    // The streaming variant fuses the walk, pre-elim accumulation,
    // translation, and consolidation-contribution-map build into ONE
    // pass per entity directory. Each source TB drops at the end of
    // its loop iteration; the only structures that survive past the
    // walk are the (small) derived aggregates.
    let StreamingWalkOutcome {
        pre_elim,
        translated_tbs,
        entity_contributions,
        deferred_tbs,
        ic_journal_entries,
        mut je_network_writer,
        entities_missing,
    } = walk_aggregate_streaming(
        manifest,
        shards_dir,
        out_dir,
        framework,
        &opts.cpi_series_by_currency,
        opts.tolerate_missing_shards,
    )?;

    let entities_processed: Vec<String> = pre_elim
        .contributing_entities
        .iter()
        .cloned()
        .chain(deferred_tbs.iter().map(|(c, _)| c.clone()))
        .collect();
    let mut entities_processed_sorted = entities_processed.clone();
    entities_processed_sorted.sort();
    entities_processed_sorted.dedup();

    // ── 3. (already covered above by walk_aggregate_streaming) ──────────

    // ── 4. Match IC pairs (Task 5.3) — v5.31 C1 Phase 2: input is the
    //      IC-filtered JE subset (~5 % of total) instead of all JEs.
    //      v5.31 C1 Phase 3: input consumed by value to eliminate the
    //      double-clone hold in the IC matcher (~45 GB saving at 2k).
    let match_result =
        crate::aggregate::ic_matcher::match_ic_pairs_consuming(manifest, ic_journal_entries)?;

    // ── 5. Build + write coverage report (Task 5.7) ─────────────────────
    let coverage_report = build_coverage_report(&match_result);
    let coverage_path = write_coverage_report(&coverage_report, out_dir)?;

    // ── 6. Generate eliminations (Task 5.4) ─────────────────────────────
    let elim_result = generate_eliminations(&match_result.matched, manifest)?;

    // ── 7. Convert to elimination JEs (Task 5.5) ────────────────────────
    let elim_jes = eliminations_to_journal_entries(&elim_result);

    // ── 7b. v5.31 C1 Phase 2: write elimination edges to the streaming
    //      JE-network writer, then finalise. Per-entity edges already
    //      flowed through the writer during walk_aggregate_streaming;
    //      the elim edges only land in the consolidated CSV (per the
    //      v5.0 contract — no per-entity emit for eliminations).
    je_network_writer.write_elim_edges(&elim_jes)?;
    let je_network_summary = je_network_writer.finalize()?;
    tracing::info!(
        "v5.10 je_network: {} per-entity files, {} elim edges, {} consolidated edges -> {:?}",
        je_network_summary.per_entity_edge_count.len(),
        je_network_summary.elim_edge_count,
        je_network_summary.consolidated_edge_count,
        je_network_summary.consolidated_csv_path,
    );

    // ── 8. Apply eliminations to pre-elim TB (Task 5.6) ─────────────────
    let post_elim = apply_eliminations_to_tb(&pre_elim, &elim_jes)?;

    // ── 9. Per-entity translation (Task 6.2) ────────────────────────────
    //      v5.31 C1: translated_tbs was built during the streaming walk
    //      above (no second pass over contributing_tbs).

    // ── 10. CTA rollforward (Task 6.3) ─────────────────────────────────
    let cta_rolls = build_cta_rollforwards(
        &translated_tbs,
        &manifest.presentation_currency,
        opts.prior_period_aggregate.as_deref(),
    )?;
    let cta_path = write_cta_rollforward(&cta_rolls, out_dir)?;

    // ── 11. Translation worksheet (Task 6.4) ───────────────────────────
    let worksheet_path = write_translation_worksheet(&translated_tbs, out_dir)?;

    // ── 12. NCI rollforward (Tasks 7.1 + 7.2) ──────────────────────────
    // v5.2: walk the contributing shards' BusinessCombination files
    // up-front so the NCI rollforward can seed period-1 opening from
    // the IFRS 3.19(a) acquisition-date fair value when present.
    let acquisition_fv_map = ingest_acquisition_date_nci_fair_values(manifest, shards_dir);
    // v5.2: read each entity's `intercompany/ownership_change_events.json`
    // (PR #155) so the NCI rollforward can apply IFRS 10.23 equity-
    // transaction adjustments for `ControlIncreased` / `ControlDecreased`
    // events.  Empty map when no entity declares ownership changes —
    // preserves v5.0–v5.1 byte-identical NCI behaviour.
    let ownership_changes_map = ingest_ownership_change_events(manifest, shards_dir);
    let nci_rolls = build_nci_rollforwards(
        manifest,
        &translated_tbs,
        opts.prior_period_aggregate.as_deref(),
        &acquisition_fv_map,
        &ownership_changes_map,
    )?;
    let nci_path = write_nci_rollforward(&nci_rolls, out_dir)?;

    // ── 13. Equity-method investments (Task 7.3) ───────────────────────
    let eq_method_invs = build_equity_method_investments(
        manifest,
        &deferred_tbs,
        framework,
        opts.prior_period_aggregate.as_deref(),
    )?;
    let eq_method_path = write_equity_method_investments(&eq_method_invs, out_dir)?;
    // v5.1: side-by-side IAS 28.38 suppressed-loss memorandum
    // (filtered to records with closing_suppressed_loss > 0).
    let _ = write_suppressed_losses(&eq_method_invs, out_dir)?;

    // ── 14. Apply NCI + equity-method overlay (Task 7.4) ───────────────
    let post_overlay = apply_nci_and_equity_method(&post_elim, &nci_rolls, &eq_method_invs)?;

    // ── 15. Build consolidated FS (Tasks 8.1–8.4) ──────────────────────
    // v5.1: thread the engagement's CoA-master labels through the FS
    // builders so localised account names (SKR04 / PCG / custom client
    // chart) flow into the consolidated BS / IS line labels.  Falls
    // back to the canonical built-in English labels for any code the
    // master doesn't carry, then to the bare code itself.
    let account_names = AccountNameDictionary::from_coa_master(
        &manifest.chart_of_accounts_master,
        &manifest.chart_of_accounts_master.primary_framework,
    );
    let bs = build_consolidated_balance_sheet_with_names(
        &post_overlay,
        &manifest.group_id,
        manifest.period.end,
        &account_names,
    )?;
    let is = build_consolidated_income_statement_with_names(
        &post_overlay,
        &nci_rolls,
        &manifest.group_id,
        manifest.period.end,
        &account_names,
    )?;
    // For v5.0 the cash flow statement gets a minimal input set: net
    // income from the IS, no non-cash adjustments / capex / financing
    // activity.  The fully derived cash flow statement is on the v5.1
    // roadmap.  `post_elim_tb_prior = None` because the driver does not
    // re-run the prior-period aggregation pipeline (see module rustdoc
    // step 15 in the plan — leave at `None`).
    let cf_inputs = CashFlowInputs {
        post_elim_tb_current: &post_overlay,
        post_elim_tb_prior: None,
        net_income: is.net_income,
        depreciation_amortization: Decimal::ZERO,
        impairment: Decimal::ZERO,
        capex: Decimal::ZERO,
        debt_issuance: Decimal::ZERO,
        debt_repayment: Decimal::ZERO,
        dividends_paid_to_owners: Decimal::ZERO,
        dividends_paid_to_nci: Decimal::ZERO,
        equity_issuance: Decimal::ZERO,
    };
    let cf = build_consolidated_cash_flow(
        &cf_inputs,
        &manifest.group_id,
        manifest.period.start,
        manifest.period.end,
    )?;
    let eq_changes_inputs = EquityChangesInputs {
        opening_owners_equity: Decimal::ZERO,
        opening_nci: nci_rolls
            .iter()
            .map(|rf| rf.opening_nci)
            .fold(Decimal::ZERO, |acc, v| acc + v),
        net_income_to_owners: is.net_income_to_owners,
        net_income_to_nci: is.net_income_to_nci,
        oci_to_owners: cta_rolls
            .iter()
            .map(|rf| rf.period_cta)
            .fold(Decimal::ZERO, |acc, v| acc + v),
        oci_to_nci: nci_rolls
            .iter()
            .map(|rf| rf.nci_share_of_oci)
            .fold(Decimal::ZERO, |acc, v| acc + v),
        dividends_to_owners: Decimal::ZERO,
        dividends_to_nci: nci_rolls
            .iter()
            .map(|rf| rf.nci_dividends)
            .fold(Decimal::ZERO, |acc, v| acc + v),
        other_owners: Decimal::ZERO,
        other_nci: Decimal::ZERO,
    };
    let changes_in_equity = build_statement_of_changes_in_equity(
        &eq_changes_inputs,
        &manifest.group_id,
        manifest.period.start,
        manifest.period.end,
        &manifest.presentation_currency,
    );

    let fs_bundle = ConsolidatedFinancialStatements {
        balance_sheet: bs,
        income_statement: is,
        cash_flow: cf,
        changes_in_equity,
    };

    // ── 16. Build consolidation schedule (Task 8.5) ────────────────────
    //      v5.31 C1: pass the pre-built entity_contributions map (built
    //      during the streaming walk) instead of re-iterating all TBs.
    let schedule = build_consolidation_schedule_with_contributions(
        &pre_elim,
        &post_overlay,
        &entity_contributions,
        &manifest.group_id,
        manifest.period.end,
    )?;

    // ── 17. Build notes (Task 8.6) ─────────────────────────────────────
    let notes_inputs = NotesInputs {
        manifest,
        framework,
        ic_coverage: &coverage_report,
        nci_rollforwards: &nci_rolls,
        cta_rollforwards: &cta_rolls,
        equity_method_investments: &eq_method_invs,
    };
    let notes = build_notes_to_consolidated_fs(&notes_inputs, manifest.period.end);

    // ── 18. Write FS artefacts (Task 8.7) ──────────────────────────────
    let fs_paths = write_consolidated_fs(&fs_bundle, &schedule, &notes, out_dir)?;

    // ── 18b. CGU goodwill impairment tests (IAS 36 § 10) ───────────────
    // No-op when caller supplies no test inputs OR manifest has no
    // CGU plan — preserves backwards compatibility byte-for-byte.
    // Per-entity net assets (presentation currency) = sum of balance-sheet
    // lines (assets are debit-side, liabilities credit-side) from the translated
    // TBs, so a CGU's carrying derives from — and reconciles to — the generated
    // financials (IAS 36 CGU carrying <-> balance sheet). Consumed by
    // `run_cgu_impairment_tests` for any CGU whose `other_carrying` is `None`.
    let entity_net_assets: std::collections::BTreeMap<String, Decimal> = {
        use crate::aggregate::translation::TranslationAccountType as T;
        use crate::DrCr;
        let mut m = std::collections::BTreeMap::new();
        for t in &translated_tbs {
            let net: Decimal = t
                .lines
                .iter()
                .filter(|l| matches!(l.account_type, T::BsMonetary | T::BsNonMonetary))
                .map(|l| match l.local_dr_cr {
                    DrCr::Debit => l.translated_amount,
                    DrCr::Credit => -l.translated_amount,
                })
                .sum();
            m.insert(t.entity_code.clone(), net);
        }
        m
    };
    let cgu_results = crate::aggregate::cgu_impairment::run_cgu_impairment_tests(
        &manifest.cgu_plan,
        &opts.cgu_test_inputs,
        &entity_net_assets,
        manifest.period.end,
        &manifest.presentation_currency,
    )?;
    let cgu_path =
        crate::aggregate::cgu_impairment::write_cgu_impairment_tests(out_dir, &cgu_results)?;

    // ── 19. Build summary ───────────────────────────────────────────────
    let mut artifacts_written: Vec<PathBuf> = Vec::with_capacity(9);
    artifacts_written.push(coverage_path);
    artifacts_written.push(cta_path);
    artifacts_written.push(worksheet_path);
    artifacts_written.push(nci_path);
    artifacts_written.push(eq_method_path);
    artifacts_written.extend(fs_paths);
    if let Some(p) = cgu_path {
        artifacts_written.push(p);
    }

    let mut deferred_codes: Vec<String> = deferred_tbs.iter().map(|(c, _)| c.clone()).collect();
    deferred_codes.sort();
    deferred_codes.dedup();

    Ok(AggregateSummary {
        group_id: manifest.group_id.clone(),
        presentation_currency: manifest.presentation_currency.clone(),
        as_of_date: manifest.period.end,
        entities_processed: entities_processed_sorted,
        entities_missing,
        deferred_entities: deferred_codes,
        matched_pairs: match_result.matched.len(),
        coverage: match_result.coverage,
        total_assets: fs_bundle.balance_sheet.total_assets,
        total_liabilities: fs_bundle.balance_sheet.total_liabilities,
        total_equity: fs_bundle.balance_sheet.total_equity,
        total_nci: fs_bundle.balance_sheet.total_nci,
        artifacts_written,
    })
}

// ── Internal helpers ──────────────────────────────────────────────────────────

/// Outcome of the per-entity archive walk: balanced TBs partitioned by
/// consolidation method, paired with the JEs the runner emitted, plus
/// the entities that were missing from disk.
///
/// **v5.31 C1**: superseded by [`StreamingWalkOutcome`] + [`walk_aggregate_streaming`].
/// Kept for now so the legacy path can be revived for differential-test fixtures
/// without re-introducing the symbol; mark as `#[allow(dead_code)]` to silence the
/// dead-code lint while the streaming path bakes.
#[allow(dead_code)]
struct WalkOutcome {
    /// Parent + Full entity TBs, paired with their entity codes.  Fed
    /// into the pre-elim aggregator.
    contributing_tbs: Vec<(String, TrialBalance)>,
    /// Parent + Full entity JEs, paired with their entity codes.  Fed
    /// into the IC matcher.
    contributing_jes: Vec<(String, Vec<JournalEntry>)>,
    /// Equity-method / fair-value / proportional entity TBs, kept
    /// separate so Chunk 7 (equity-method rollforward) can consume them
    /// without re-walking the disk.
    deferred_tbs: Vec<(String, TrialBalance)>,
    /// Entity codes the driver expected but could not find on disk
    /// (only populated when `tolerate_missing_shards == true`).
    entities_missing: Vec<String>,
}

/// Walk every entity in `manifest.ownership_graph.entities` and load
/// its `period_close/trial_balances.json` + `journal_entries.json` from
/// `shards_dir/entities/{code}/`.
///
/// Per the plan, equity-method / fair-value / proportional entities
/// **are** expected to have shards on disk — the runner generated them
/// — but their TBs go into `deferred_tbs` rather than `contributing_tbs`
/// so the pre-elim aggregator only sees Parent + Full inputs.
///
/// A missing entity directory or `period_close/trial_balances.json` is
/// either fatal (`tolerate_missing_shards = false`) or pushed to
/// `entities_missing` with a `tracing::warn!`.  A missing
/// `journal_entries.json` for a present entity is treated as "no JEs"
/// (empty Vec) — the runner always writes the file even when the
/// entity emits zero JEs (the orchestrator at minimum produces opening-
/// balance entries for any seeded TB), so the missing-file case is
/// effectively unreachable under v5.0 but defended here defensively.
///
/// **v5.31 C1**: superseded by [`walk_aggregate_streaming`]. Kept for
/// the differential-test fixture path; marked dead-code-allowed.
#[allow(dead_code)]
fn walk_entity_archives(
    manifest: &GroupManifest,
    shards_dir: &Path,
    tolerate_missing_shards: bool,
) -> GroupResult<WalkOutcome> {
    let mut contributing_tbs: Vec<(String, TrialBalance)> = Vec::new();
    let mut contributing_jes: Vec<(String, Vec<JournalEntry>)> = Vec::new();
    let mut deferred_tbs: Vec<(String, TrialBalance)> = Vec::new();
    let mut entities_missing: Vec<String> = Vec::new();

    for entity in &manifest.ownership_graph.entities {
        let entity_dir = shards_dir.join("entities").join(&entity.code);
        let tb_path = entity_dir.join("period_close").join("trial_balances.json");

        if !tb_path.exists() {
            if tolerate_missing_shards {
                tracing::warn!(
                    entity = %entity.code,
                    path = %tb_path.display(),
                    "missing shard archive — continuing in tolerate_missing_shards mode",
                );
                entities_missing.push(entity.code.clone());
                continue;
            }
            return Err(GroupError::Aggregate(format!(
                "run_aggregate: missing shard archive for `{}` at `{}`",
                entity.code,
                tb_path.display()
            )));
        }

        let tb = load_entity_trial_balance(&entity_dir)?;
        let jes = load_entity_journal_entries(&entity_dir, &entity.code)?;

        match entity.consolidation_method {
            ConsolidationMethod::Parent | ConsolidationMethod::Full => {
                contributing_tbs.push((entity.code.clone(), tb));
                contributing_jes.push((entity.code.clone(), jes));
            }
            ConsolidationMethod::EquityMethod
            | ConsolidationMethod::Proportional
            | ConsolidationMethod::FairValue => {
                // Capture the deferred TB for Chunk 7 consumption; we
                // intentionally do *not* push the JEs into the
                // contributing slice because IC matching only spans
                // line-by-line consolidated entities.  v5.0 IC pair plans
                // for equity-method investees are out of scope; v5.3
                // will revisit.
                deferred_tbs.push((entity.code.clone(), tb));
            }
        }
    }

    entities_missing.sort();
    Ok(WalkOutcome {
        contributing_tbs,
        contributing_jes,
        deferred_tbs,
        entities_missing,
    })
}

/// Read every JE the orchestrator emitted for `entity_code` from
/// `entity_dir/journal_entries.json`.  Treats a missing file as zero
/// JEs (defensive — see `walk_entity_archives` rustdoc).
#[allow(dead_code)]
fn load_entity_journal_entries(
    entity_dir: &Path,
    entity_code: &str,
) -> GroupResult<Vec<JournalEntry>> {
    let path = entity_dir.join("journal_entries.json");
    if !path.exists() {
        tracing::warn!(
            entity = %entity_code,
            path = %path.display(),
            "no journal_entries.json found — treating as empty",
        );
        return Ok(Vec::new());
    }
    let bytes = std::fs::read(&path).map_err(GroupError::Io)?;
    let jes: Vec<JournalEntry> = serde_json::from_slice(&bytes)?;
    Ok(jes)
}

/// v5.31 C1 Phase 6 — streaming JSON-array parse for journal_entries.json.
///
/// Reads `entity_dir/journal_entries.json` (a JSON array of [`JournalEntry`])
/// via `serde_json::Deserializer::from_reader` and yields one JE at a time
/// through the `on_je` callback. **Never materialises the full
/// `Vec<JournalEntry>`** — peak per-iteration allocation is one JE
/// (~25 KB) rather than the legacy `load_entity_journal_entries`'s
/// ~25-300 MB.
///
/// Why this matters: the Phase 5 diagnostic
/// (`docs/baselines/2026-05-27-v5.31-c1-phase5-diagnostic/COMPARISON.md`)
/// found that 178 MB of RSS climbed per entity even though tracked
/// data structures only grew ~75 KB per entity — a 2 400× gap.
/// Allocator-fragmentation from the per-iteration spike of JSON-parse
/// String allocations was eating ~178 MB / entity even with mimalloc
/// (since the small String allocations leave the heap too fragmented
/// for the OS to reclaim pages). Streaming the parse eliminates the
/// spike → no fragmentation → RSS stays flat.
///
/// Treats a missing file as zero JEs (defensive — see
/// `walk_entity_archives` rustdoc).
///
/// # Errors
///
/// - [`GroupError::Io`] if the file cannot be opened.
/// - [`GroupError::Serde`] if the JSON is malformed at any point.
/// - Whatever error `on_je` returns (propagated through serde's
///   error-conversion machinery).
fn stream_entity_journal_entries<F>(
    entity_dir: &Path,
    entity_code: &str,
    mut on_je: F,
) -> GroupResult<usize>
where
    F: FnMut(JournalEntry) -> GroupResult<()>,
{
    let path = entity_dir.join("journal_entries.json");
    if !path.exists() {
        tracing::warn!(
            entity = %entity_code,
            path = %path.display(),
            "no journal_entries.json found — treating as empty",
        );
        return Ok(0);
    }
    let file = std::fs::File::open(&path).map_err(GroupError::Io)?;
    let reader = std::io::BufReader::with_capacity(64 * 1024, file);
    let mut de = serde_json::Deserializer::from_reader(reader);

    // Custom Visitor that drives the array iteration without ever
    // materialising the whole sequence into a Vec.
    struct JeStreamVisitor<F> {
        on_je: F,
        count: usize,
    }
    impl<'de, F> serde::de::Visitor<'de> for JeStreamVisitor<F>
    where
        F: FnMut(JournalEntry) -> GroupResult<()>,
    {
        type Value = usize;
        fn expecting(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
            fmt.write_str("a JSON array of JournalEntry")
        }
        fn visit_seq<A>(mut self, mut seq: A) -> Result<usize, A::Error>
        where
            A: serde::de::SeqAccess<'de>,
        {
            while let Some(je) = seq.next_element::<JournalEntry>()? {
                (self.on_je)(je).map_err(serde::de::Error::custom)?;
                self.count += 1;
            }
            Ok(self.count)
        }
    }

    let count = serde::de::Deserializer::deserialize_seq(
        &mut de,
        JeStreamVisitor {
            on_je: &mut on_je,
            count: 0,
        },
    )?;
    Ok(count)
}

/// v5.31 C1 — translate a **single** entity's TB.
///
/// v5.31 C1 Phase 5 — read this process's resident-set size from
/// `/proc/self/statm`. Returns `None` on non-Linux or read failure;
/// the caller logs `0` in that case so the tracing line is still
/// emitted. Used by the walk-loop RSS checkpoints to attribute the
/// Phase 4 OOM trajectory to a specific accumulator.
fn read_self_rss_kb() -> Option<u64> {
    // `/proc/self/statm` fields: size resident shared text lib data dt
    // — all in pages. We want field index 1 (resident).
    let s = std::fs::read_to_string("/proc/self/statm").ok()?;
    let pages: u64 = s.split_ascii_whitespace().nth(1)?.parse().ok()?;
    // Page size is 4 KB on x86_64 Linux. Could call `sysconf(_SC_PAGESIZE)`
    // for portability but the only platform we run on is x86_64 Linux.
    Some(pages * 4)
}

/// Factored out of [`translate_all_contributing`] so the streaming
/// aggregate walk can call it per-entity without materialising a
/// `Vec<(String, TrialBalance)>`.  The signature mirrors the inner
/// loop body verbatim — the only structural change is the explicit
/// `entity: &ManifestEntity` parameter (the caller already has the
/// lookup in hand, no need to repeat it inside the helper).
fn translate_one_entity(
    code: &str,
    tb: &TrialBalance,
    entity: &ManifestEntity,
    manifest: &GroupManifest,
    framework: AccountingFramework,
    cpi_opt: Option<&BTreeMap<String, datasynth_core::models::hyperinflation::GeneralPriceIndex>>,
) -> GroupResult<TranslatedTb> {
    let path = select_restatement_path(
        entity.hyperinflation_status,
        entity.functional_currency.as_str(),
        cpi_opt,
        manifest.period.start,
        manifest.period.end,
    );
    let translated = match &path {
        RestatementPath::Indexed(ir) => translate_entity_tb_with_indexed_restatement(
            tb,
            entity.functional_currency.as_str(),
            &manifest.fx_rate_master,
            manifest.period.end,
            &manifest.presentation_currency,
            framework,
            entity.hyperinflation_status,
            Some(ir),
        )?,
        RestatementPath::Standard | RestatementPath::ClosingRate => {
            if matches!(path, RestatementPath::ClosingRate) && cpi_opt.is_some() {
                tracing::warn!(
                    entity = %code,
                    functional_currency = %entity.functional_currency,
                    period_start = %manifest.period.start,
                    period_end = %manifest.period.end,
                    "hyperinflationary entity has no matching CPI series — falling back to IAS 21 § 42(b) closing-rate translation only (no IAS 29 § 12 indexed restatement)",
                );
            }
            translate_entity_tb_with_hyperinflation(
                tb,
                entity.functional_currency.as_str(),
                &manifest.fx_rate_master,
                manifest.period.end,
                &manifest.presentation_currency,
                framework,
                entity.hyperinflation_status,
            )?
        }
    };
    Ok(translated)
}

/// v5.31 C1 — outcome of the streaming aggregate walk.
///
/// Eliminates the `contributing_tbs: Vec<(String, TrialBalance)>` hold
/// (100-200 GB on a 2000-entity run) by fusing the walk, pre-elimination
/// accumulation, translation, and consolidation-contribution-map build
/// into one pass per entity directory.  After the walk completes, only
/// the **derived** structures live in memory:
///
/// - `pre_elim` — small (~50 GL accounts × few KB each)
/// - `translated_tbs` — ~20 MB × N entities (one per contributing
///   entity); needed by Chunk 6 + Chunk 7 + Chunk 8 downstream
/// - `entity_contributions` — `account_code → entity_code → net` map
///   used by Chunk 8 consolidation_schedule (~200 MB at 2000 entities)
/// - `deferred_tbs` — small (~200 entities, kept for equity-method
///   in Chunk 7; refactoring this is a future phase)
/// - `ic_journal_entries` — **v5.31 C1 Phase 2**: only IC-pair-tagged
///   JEs (typically <5 % of total emitted JEs). All non-IC JEs are
///   dropped inside the walk after their edges flow through the
///   streaming JE-network writer.
/// - `je_network_summary` — running tally from
///   `JeNetworkStreamingWriter` (the writer is closed at the end of
///   the walk, so `consolidated_csv_path` is populated).
/// - `entities_missing` — mirror of [`WalkOutcome`]
///
/// The source TBs and the full JE vectors are **never materialised
/// across entities** — each entity's TB + JE vec drops at the end of
/// its loop iteration; only the IC-filtered subset persists.
struct StreamingWalkOutcome {
    pre_elim: AggregatedTb,
    translated_tbs: Vec<TranslatedTb>,
    /// `account_code → entity_code → net balance`
    entity_contributions: BTreeMap<String, BTreeMap<String, Decimal>>,
    deferred_tbs: Vec<(String, TrialBalance)>,
    /// v5.31 C1 Phase 2: only IC-pair-tagged JEs survive the walk.
    /// Non-IC JEs flow through the JE-network writer + drop.
    ic_journal_entries: Vec<(String, Vec<JournalEntry>)>,
    /// v5.31 C1 Phase 2: the streaming JE-network writer. Stays open
    /// past the walk so the driver can write the post-walk elimination
    /// edges before calling `finalize()` to flush + close.
    je_network_writer: crate::aggregate::je_network::JeNetworkStreamingWriter,
    entities_missing: Vec<String>,
}

/// v5.31 C1 — streaming aggregate walk.
///
/// Combines the three former passes (walk → pre-elim → translate)
/// into one pass per entity, dropping each source TB immediately after
/// it has been accumulated into the running [`AggregatedTb`] and
/// translated into a [`TranslatedTb`].
///
/// Memory profile on a 2000-entity run:
/// - Peak source-TB hold: **1 TB at a time** (was 1800–2000 TBs)
/// - Net hold after walk: `pre_elim + translated_tbs +
///   entity_contributions + deferred_tbs + contributing_jes`
/// - Eliminates the ~200 GB that drove the OOM on the 2026-05-26 run.
///
/// Behaviour is byte-equivalent to the legacy three-step path: the
/// pre-elim aggregator, the translation function, and the consolidation
/// schedule's contribution-map all see identical inputs in identical
/// order; `pre_elim` is `finalise`d at the end to match the legacy
/// deterministic sort.
fn walk_aggregate_streaming(
    manifest: &GroupManifest,
    shards_dir: &Path,
    out_dir: &Path,
    framework: AccountingFramework,
    cpi_series_by_currency: &BTreeMap<
        String,
        datasynth_core::models::hyperinflation::GeneralPriceIndex,
    >,
    tolerate_missing_shards: bool,
) -> GroupResult<StreamingWalkOutcome> {
    let entity_lookup_map = entity_lookup(manifest);
    let cpi_opt = if cpi_series_by_currency.is_empty() {
        None
    } else {
        Some(cpi_series_by_currency)
    };

    let mut pre_elim = empty_aggregate(manifest);
    let mut translated_tbs: Vec<TranslatedTb> =
        Vec::with_capacity(manifest.ownership_graph.entities.len());
    let mut entity_contributions: BTreeMap<String, BTreeMap<String, Decimal>> = BTreeMap::new();
    let mut deferred_tbs: Vec<(String, TrialBalance)> = Vec::new();
    let mut ic_journal_entries: Vec<(String, Vec<JournalEntry>)> = Vec::new();
    let mut entities_missing: Vec<String> = Vec::new();

    // v5.31 C1 Phase 2: open the streaming JE-network writer once;
    // each per-entity loop iteration writes its edges directly to
    // disk (per-entity + consolidated CSV). No consolidated edge vec
    // ever materialises in memory.
    let mut je_network_writer =
        crate::aggregate::je_network::JeNetworkStreamingWriter::open(out_dir)?;

    // v5.31 C1 Phase 5 — RSS profiling.
    //
    // Periodically log peak RSS + the sizes of every long-lived
    // accumulator. The Phase 4 OOM analysis surfaced a 218 GB linear
    // climb at ~6 GB/min over 38 min, but the IC subset (the holding
    // structure I theorised) is only 0.03 % of edges; ic_journal_entries
    // is ~100 MB, not the 50 GB I'd estimated. The actual hotspot is
    // unknown. These checkpoint logs let the post-OOM log inspection
    // tell us which structure grows fastest.
    let entity_count = manifest.ownership_graph.entities.len();
    let log_every = (entity_count / 20).max(1); // ~20 samples across the walk

    for (idx, entity) in manifest.ownership_graph.entities.iter().enumerate() {
        let entity_dir = shards_dir.join("entities").join(&entity.code);
        let tb_path = entity_dir.join("period_close").join("trial_balances.json");

        if !tb_path.exists() {
            if tolerate_missing_shards {
                tracing::warn!(
                    entity = %entity.code,
                    path = %tb_path.display(),
                    "missing shard archive — continuing in tolerate_missing_shards mode",
                );
                entities_missing.push(entity.code.clone());
                continue;
            }
            return Err(GroupError::Aggregate(format!(
                "run_aggregate: missing shard archive for `{}` at `{}`",
                entity.code,
                tb_path.display()
            )));
        }

        let tb = load_entity_trial_balance(&entity_dir)?;

        match entity.consolidation_method {
            ConsolidationMethod::Parent | ConsolidationMethod::Full => {
                // 1. Accumulate into running pre-elim aggregate (small).
                accumulate_entity_into_aggregate(&mut pre_elim, manifest, &entity.code, &tb)?;

                // 2. Build per-account contribution map for the
                //    consolidation schedule (Chunk 8.5). Replaces the
                //    O(N) re-scan of all TBs that used to live inside
                //    `build_consolidation_schedule`.
                for line in &tb.lines {
                    let net = line.debit_balance - line.credit_balance;
                    entity_contributions
                        .entry(line.account_code.clone())
                        .or_default()
                        .insert(entity.code.clone(), net);
                }

                // 3. Translate to presentation currency (Chunk 6).
                //    `translated_tbs` stays in memory — downstream Chunks
                //    6 + 7 + 8 iterate it; refactoring those to lazy
                //    iteration is Phase 2.
                let entity_meta = entity_lookup_map.get(&entity.code).ok_or_else(|| {
                    GroupError::Aggregate(format!(
                        "run_aggregate: entity `{}` not in manifest's ownership graph",
                        entity.code,
                    ))
                })?;
                let translated = translate_one_entity(
                    &entity.code,
                    &tb,
                    entity_meta,
                    manifest,
                    framework,
                    cpi_opt,
                )?;
                translated_tbs.push(translated);

                // 4. v5.31 C1 Phase 6 — stream the entity's
                //    journal_entries.json one JE at a time, building
                //    edges incrementally + filtering the IC subset
                //    inline. NEVER materialises the full
                //    `Vec<JournalEntry>` — Phase 5 diagnostic showed
                //    that the per-iteration JSON-parse spike (~150-
                //    300 MB per entity) fragmented the heap to 218 GB
                //    by entity ~1200 at 2k scale. Streaming drops the
                //    peak per-iteration alloc to ~25 KB (one JE), so
                //    the same loop runs at ~10 GB RSS.
                //
                //    The `JeNetworkEdgeBuilder` maintains the
                //    `line_id → edge_id` map across `push_je` calls
                //    so cross-JE predecessor chains within this
                //    entity (PO → invoice → payment doc chains)
                //    resolve identically to the legacy
                //    `build_je_network_edges(&jes, …)` slice call.
                let mut edge_builder = datasynth_runtime::je_network::JeNetworkEdgeBuilder::new(
                    datasynth_config::JeNetworkMethod::A,
                );
                let mut ic_jes: Vec<JournalEntry> = Vec::new();
                let _je_count = stream_entity_journal_entries(&entity_dir, &entity.code, |je| {
                    edge_builder.push_je(&je);
                    if je.header.ic_pair_id.is_some() {
                        ic_jes.push(je);
                    }
                    // `je` drops here on the non-IC path —
                    // single-JE allocation reclaimed before the
                    // next stream iteration → no fragmentation
                    // spike.
                    Ok(())
                })?;
                let edges = edge_builder.into_edges();
                je_network_writer.write_entity_edges_prebuilt(&entity.code, &edges)?;

                // 5. IC subset retained for downstream IC matcher
                //    (per v5.30 SOTA defaults this is ~0.03 % of all
                //    JEs ≈ ~5-10 GB across all 2k entities, vs the
                //    100-400 GB of holding every JE).
                ic_journal_entries.push((entity.code.clone(), ic_jes));

                // 6. `tb`, `edges` drop here — never held past this
                //    iteration.
            }
            ConsolidationMethod::EquityMethod
            | ConsolidationMethod::Proportional
            | ConsolidationMethod::FairValue => {
                // accumulate_entity_into_aggregate already recorded the
                // DeferredEntity in pre_elim; we still keep the source
                // TB in `deferred_tbs` for equity-method downstream
                // (~200 entities × ~20 MB = ~4 GB max; manageable).
                accumulate_entity_into_aggregate(&mut pre_elim, manifest, &entity.code, &tb)?;
                deferred_tbs.push((entity.code.clone(), tb));
                // Deferred-method entities do not feed IC matching or
                // the consolidated je_network (per v5.0 contract — IC
                // pair plans only span Parent/Full entities). v5.31 C1
                // Phase 6 — we no longer load JEs for these entities
                // at all, saving the JSON-parse cost (~10-30 MB / entity).
            }
        }

        // v5.31 C1 Phase 5 RSS checkpoint — every ~5 % of entities,
        // emit a structured log so the Phase 4 OOM trajectory can be
        // attributed to a specific accumulator.
        if (idx + 1) % log_every == 0 || idx + 1 == entity_count {
            let rss_kb = read_self_rss_kb().unwrap_or(0);
            let ic_je_count: usize = ic_journal_entries.iter().map(|(_, v)| v.len()).sum();
            let pre_elim_accounts = pre_elim.account_totals.len();
            tracing::info!(
                target: "datasynth_group::c1_phase5",
                entity_idx = idx + 1,
                total_entities = entity_count,
                rss_gb = rss_kb / 1024 / 1024,
                pre_elim_accounts,
                pre_elim_contributing = pre_elim.contributing_entities.len(),
                pre_elim_deferred = pre_elim.deferred_entities.len(),
                translated_tbs_n = translated_tbs.len(),
                translated_tbs_total_lines = translated_tbs.iter().map(|t| t.lines.len()).sum::<usize>(),
                entity_contributions_keys = entity_contributions.len(),
                entity_contributions_total_pairs = entity_contributions.values().map(|m| m.len()).sum::<usize>(),
                deferred_tbs_n = deferred_tbs.len(),
                ic_journal_entries_entries = ic_journal_entries.len(),
                ic_journal_entries_total_jes = ic_je_count,
                "c1_phase5 checkpoint"
            );
        }
    }

    entities_missing.sort();
    finalise_streaming_aggregate(&mut pre_elim);

    // v5.31 C1 Phase 5 — final per-component sizes after walk completes
    // (before IC matching runs).
    let final_rss_kb = read_self_rss_kb().unwrap_or(0);
    let final_ic_jes: usize = ic_journal_entries.iter().map(|(_, v)| v.len()).sum();
    tracing::info!(
        target: "datasynth_group::c1_phase5",
        rss_gb = final_rss_kb / 1024 / 1024,
        rss_mb = final_rss_kb / 1024,
        translated_tbs_n = translated_tbs.len(),
        translated_tbs_lines = translated_tbs.iter().map(|t| t.lines.len()).sum::<usize>(),
        entity_contributions_keys = entity_contributions.len(),
        entity_contributions_total_pairs = entity_contributions.values().map(|m| m.len()).sum::<usize>(),
        deferred_tbs_n = deferred_tbs.len(),
        ic_journal_entries_n = ic_journal_entries.len(),
        ic_journal_entries_total = final_ic_jes,
        pre_elim_accounts = pre_elim.account_totals.len(),
        "c1_phase5 walk-complete summary"
    );

    // v5.31 C1 Phase 2: the writer stays open — the caller writes the
    // elim edges + finalises after IC matching produces them. We
    // hand the writer out of the walk via the outcome struct.
    Ok(StreamingWalkOutcome {
        pre_elim,
        translated_tbs,
        entity_contributions,
        deferred_tbs,
        ic_journal_entries,
        je_network_writer,
        entities_missing,
    })
}

/// Translate every contributing entity's TB to the presentation
/// currency.  Returns one [`TranslatedTb`] per `(entity_code, tb)` in
/// the contributing slice.
///
/// **v5.5.2 routing:** when `cpi_series_by_currency` is non-empty,
/// hyperinflationary entities whose functional currency matches an
/// entry in the map are translated via the indexed-restatement path
/// (IAS 29 § 12 + IAS 21 § 42(b)); hyperinflationary entities without a
/// matching series log a warning and fall back to the closing-rate-only
/// path. Non-hyperinflationary entities always use the standard IAS 21
/// multi-rate path regardless of the map.
///
/// **v5.31 C1**: superseded by [`translate_one_entity`] + the inline
/// per-entity translation in [`walk_aggregate_streaming`]. Kept for
/// the differential-test fixture path; marked dead-code-allowed.
#[allow(dead_code)]
fn translate_all_contributing(
    contributing_tbs: &[(String, TrialBalance)],
    manifest: &GroupManifest,
    framework: AccountingFramework,
    entity_lookup: &BTreeMap<String, ManifestEntity>,
    cpi_series_by_currency: &BTreeMap<
        String,
        datasynth_core::models::hyperinflation::GeneralPriceIndex,
    >,
) -> GroupResult<Vec<TranslatedTb>> {
    let mut out: Vec<TranslatedTb> = Vec::with_capacity(contributing_tbs.len());
    let cpi_opt = if cpi_series_by_currency.is_empty() {
        None
    } else {
        Some(cpi_series_by_currency)
    };
    for (code, tb) in contributing_tbs {
        let entity = entity_lookup.get(code).ok_or_else(|| {
            GroupError::Aggregate(format!(
                "run_aggregate: entity `{code}` not in manifest's ownership graph",
            ))
        })?;
        // v5.2 / v5.5.2: pick the IAS 21 / IAS 29 translation path per
        // entity based on its hyperinflation status + the optional CPI
        // series map.  See [`select_restatement_path`] for the routing
        // logic.
        let path = select_restatement_path(
            entity.hyperinflation_status,
            entity.functional_currency.as_str(),
            cpi_opt,
            manifest.period.start,
            manifest.period.end,
        );
        let translated = match &path {
            RestatementPath::Indexed(ir) => translate_entity_tb_with_indexed_restatement(
                tb,
                entity.functional_currency.as_str(),
                &manifest.fx_rate_master,
                manifest.period.end,
                &manifest.presentation_currency,
                framework,
                entity.hyperinflation_status,
                Some(ir),
            )?,
            RestatementPath::Standard | RestatementPath::ClosingRate => {
                if matches!(path, RestatementPath::ClosingRate) && cpi_opt.is_some() {
                    // Hyperinflationary entity but the CPI map missed
                    // (no entry for currency, or lookup outside the
                    // observation range).  Warn so the operator can
                    // notice the data gap, but continue with the
                    // closing-rate fallback rather than failing the
                    // aggregate run.
                    tracing::warn!(
                        entity = %code,
                        functional_currency = %entity.functional_currency,
                        period_start = %manifest.period.start,
                        period_end = %manifest.period.end,
                        "hyperinflationary entity has no matching CPI series — falling back to IAS 21 § 42(b) closing-rate translation only (no IAS 29 § 12 indexed restatement)",
                    );
                }
                translate_entity_tb_with_hyperinflation(
                    tb,
                    entity.functional_currency.as_str(),
                    &manifest.fx_rate_master,
                    manifest.period.end,
                    &manifest.presentation_currency,
                    framework,
                    entity.hyperinflation_status,
                )?
            }
        };
        out.push(translated);
    }
    Ok(out)
}

/// Build a [`CtaRollforward`] for every non-presentation-currency
/// entity.  Reads opening CTA from
/// `prior_period.consolidated/cta_rollforward.json` when supplied,
/// otherwise defaults to zero.
fn build_cta_rollforwards(
    translated_tbs: &[TranslatedTb],
    presentation_currency: &str,
    prior_period_aggregate: Option<&Path>,
) -> GroupResult<Vec<CtaRollforward>> {
    let opening_map = ingest_opening_cta_balances(prior_period_aggregate)?;

    let mut rolls: Vec<CtaRollforward> = Vec::new();
    for t in translated_tbs {
        if t.functional_currency == presentation_currency {
            // No CTA for entities already in the presentation currency.
            continue;
        }
        let opening = opening_map
            .get(&t.entity_code)
            .copied()
            .unwrap_or(Decimal::ZERO);
        rolls.push(cta_rollforward(
            &t.entity_code,
            &t.functional_currency,
            &t.presentation_currency,
            opening,
            t.cta,
        ));
    }
    Ok(rolls)
}

/// Read prior-period closing CTA balances by entity code.  Mirrors
/// [`crate::aggregate::nci::opening::ingest_opening_nci_balances`]
/// semantics (missing file → empty map + warn, duplicate entity →
/// error).
fn ingest_opening_cta_balances(
    prior_period_aggregate: Option<&Path>,
) -> GroupResult<BTreeMap<String, Decimal>> {
    let Some(prior) = prior_period_aggregate else {
        return Ok(BTreeMap::new());
    };
    let path = prior
        .join(CONSOLIDATED_SUBDIR)
        .join(CTA_ROLLFORWARD_FILENAME);
    if !path.exists() {
        tracing::warn!(
            path = %path.display(),
            "opening CTA file not found; defaulting to zero opening balance per entity",
        );
        return Ok(BTreeMap::new());
    }
    let bytes = std::fs::read(&path).map_err(GroupError::Io)?;
    let rolls: Vec<CtaRollforward> = serde_json::from_slice(&bytes)?;
    let mut map: BTreeMap<String, Decimal> = BTreeMap::new();
    for rf in rolls {
        if map.contains_key(&rf.entity_code) {
            return Err(GroupError::Aggregate(format!(
                "ingest_opening_cta_balances: duplicate entity `{}` in opening CTA file {}",
                rf.entity_code,
                path.display(),
            )));
        }
        map.insert(rf.entity_code, rf.closing_cta);
    }
    Ok(map)
}

/// Walk every contributing shard's
/// `accounting_standards/business_combinations.json` and build a
/// map of `acquiree_entity_code → acquisition_date_nci_fair_value`
/// for use as period-1 opening NCI in the rollforward (v5.2 IFRS
/// 3.19(a) full-goodwill basis).
///
/// - Missing per-shard files are tolerated (most engagements don't
///   have an acquisition every period).
/// - BC records without `acquiree_entity_code` or without a fair
///   value contribute nothing — those acquisitions either have no
///   consolidated counterpart or use the proportionate basis.
/// - When two shards both list a BC for the same acquiree, the
///   first one wins (deterministic — entities iterate in manifest
///   order).
fn ingest_acquisition_date_nci_fair_values(
    manifest: &GroupManifest,
    shards_dir: &Path,
) -> BTreeMap<String, Decimal> {
    let mut map: BTreeMap<String, Decimal> = BTreeMap::new();
    for entity in &manifest.ownership_graph.entities {
        let path = shards_dir
            .join("entities")
            .join(&entity.code)
            .join("accounting_standards")
            .join("business_combinations.json");
        if !path.exists() {
            continue;
        }
        let bytes = match fs::read(&path) {
            Ok(b) => b,
            Err(e) => {
                tracing::debug!(
                    path = %path.display(),
                    error = %e,
                    "failed to read business_combinations.json — skipping",
                );
                continue;
            }
        };
        let combinations: Vec<BusinessCombination> = match serde_json::from_slice(&bytes) {
            Ok(v) => v,
            Err(e) => {
                tracing::debug!(
                    path = %path.display(),
                    error = %e,
                    "failed to parse business_combinations.json — skipping",
                );
                continue;
            }
        };
        for bc in combinations {
            if let (Some(acquiree), Some(fv)) = (
                bc.acquiree_entity_code.as_deref(),
                bc.acquisition_date_nci_fair_value,
            ) {
                map.entry(acquiree.to_string()).or_insert(fv);
            }
        }
    }
    map
}

/// **v5.2** — Walk every entity's per-shard archive, read its
/// `intercompany/ownership_change_events.json` (PR #155 emission),
/// and build an `entity_code → events` map for the NCI rollforward
/// to consume.
///
/// Best-effort: malformed or missing files are skipped with a
/// `tracing::debug!` rather than failing the aggregate — engagements
/// without ownership-change events have no file at all, and that's
/// the v5.0–v5.1 byte-identical baseline.  A malformed events file
/// downgrades to "no events for this entity" rather than a hard
/// error, mirroring the BC ingestor pattern.
fn ingest_ownership_change_events(
    manifest: &GroupManifest,
    shards_dir: &Path,
) -> BTreeMap<String, Vec<datasynth_core::models::intercompany::OwnershipChangeEvent>> {
    let mut map: BTreeMap<String, Vec<datasynth_core::models::intercompany::OwnershipChangeEvent>> =
        BTreeMap::new();
    for entity in &manifest.ownership_graph.entities {
        let path = shards_dir
            .join("entities")
            .join(&entity.code)
            .join("intercompany")
            .join("ownership_change_events.json");
        if !path.exists() {
            continue;
        }
        let bytes = match fs::read(&path) {
            Ok(b) => b,
            Err(e) => {
                tracing::debug!(
                    path = %path.display(),
                    error = %e,
                    "failed to read ownership_change_events.json — skipping",
                );
                continue;
            }
        };
        let events: Vec<datasynth_core::models::intercompany::OwnershipChangeEvent> =
            match serde_json::from_slice(&bytes) {
                Ok(v) => v,
                Err(e) => {
                    tracing::debug!(
                        path = %path.display(),
                        error = %e,
                        "failed to parse ownership_change_events.json — skipping",
                    );
                    continue;
                }
            };
        if !events.is_empty() {
            map.insert(entity.code.clone(), events);
        }
    }
    map
}

/// Build an [`NciRollforward`] for every Full-method, non-wholly-owned
/// entity.  v5.0 sources the period P&L / OCI numbers from the entity's
/// translated TB; dividends paid is left at zero (the manifest does not
/// yet model it).
fn build_nci_rollforwards(
    manifest: &GroupManifest,
    translated_tbs: &[TranslatedTb],
    prior_period_aggregate: Option<&Path>,
    acquisition_fv_map: &BTreeMap<String, Decimal>,
    ownership_changes_map: &BTreeMap<
        String,
        Vec<datasynth_core::models::intercompany::OwnershipChangeEvent>,
    >,
) -> GroupResult<Vec<NciRollforward>> {
    let opening_map = match prior_period_aggregate {
        Some(p) => ingest_opening_nci_balances(p)?,
        None => BTreeMap::new(),
    };

    let translated_lookup: BTreeMap<&str, &TranslatedTb> = translated_tbs
        .iter()
        .map(|t| (t.entity_code.as_str(), t))
        .collect();

    let mut rolls: Vec<NciRollforward> = Vec::new();
    for entity in &manifest.ownership_graph.entities {
        if entity.consolidation_method != ConsolidationMethod::Full {
            continue;
        }
        let Some(ownership) = entity.ownership_percent else {
            continue;
        };
        if ownership >= Decimal::ONE {
            // Wholly-owned `Full` entities have no NCI to measure.  We
            // skip them silently here rather than surfacing an error
            // because mini_acme.yaml has both 100%-owned (ACME_USA,
            // ACME_BR) and partially-owned (ACME_DE 80%) Full
            // subsidiaries.  The NCI rollforward computer would reject
            // 100% ownership as a caller bug, so the filter must happen
            // at the driver level.
            continue;
        }
        let translated = translated_lookup.get(entity.code.as_str()).ok_or_else(|| {
            GroupError::Aggregate(format!(
                "run_aggregate: NCI computation needs translated TB for `{}` but none was produced",
                entity.code,
            ))
        })?;

        let inputs = NciInputs {
            entity,
            period_net_income: net_income_from_translated(translated),
            period_oci: oci_from_translated(translated),
            total_dividends_paid: Decimal::ZERO,
            opening_nci: opening_map
                .get(&entity.code)
                .copied()
                .unwrap_or(Decimal::ZERO),
            // v5.2: acquisition-date NCI fair value (IFRS 3.19(a))
            // looked up from the per-shard
            // `accounting_standards/business_combinations.json` files
            // by `acquiree_entity_code`.  Engagements without an
            // acquisition record produce `None`, preserving v5.0–v5.1
            // proportionate-basis behaviour byte-for-byte.
            acquisition_date_nci_fair_value: acquisition_fv_map.get(&entity.code).copied(),
            ownership_changes: ownership_changes_map
                .get(&entity.code)
                .map(|v| v.as_slice())
                .unwrap_or(&[]),
            period_start: manifest.period.end,
            period_end: manifest.period.end,
            currency: manifest.presentation_currency.clone(),
        };
        rolls.push(compute_nci_rollforward(&inputs)?);
    }
    Ok(rolls)
}

/// Build an [`EquityMethodInvestment`] for every EquityMethod investee
/// that has a deferred shard TB on disk.  Sources `investee_net_income`
/// and `investee_dividends_paid` from the deferred TB; impairment is
/// left at zero (the manifest does not yet model it).
fn build_equity_method_investments(
    manifest: &GroupManifest,
    deferred_tbs: &[(String, TrialBalance)],
    framework: AccountingFramework,
    prior_period_aggregate: Option<&Path>,
) -> GroupResult<Vec<EquityMethodInvestment>> {
    let (opening_map, opening_suppressed_map) = match prior_period_aggregate {
        Some(p) => (
            ingest_opening_equity_method_carrying_values(p)?,
            ingest_opening_suppressed_losses(p)?,
        ),
        None => (BTreeMap::new(), BTreeMap::new()),
    };

    let deferred_lookup: BTreeMap<&str, &TrialBalance> = deferred_tbs
        .iter()
        .map(|(c, tb)| (c.as_str(), tb))
        .collect();

    let mut invs: Vec<EquityMethodInvestment> = Vec::new();
    for entity in &manifest.ownership_graph.entities {
        if entity.consolidation_method != ConsolidationMethod::EquityMethod {
            continue;
        }
        let Some(parent_code) = &entity.parent_code else {
            // No parent declared — equity-method requires an investor
            // entity.  Skip silently rather than erroring since v5.0's
            // upstream validators already reject ownership graphs
            // missing parent declarations.
            continue;
        };

        // The investee's deferred TB may be absent if its shard was
        // missing and `tolerate_missing_shards` was enabled.  Treat
        // that as "no income / dividends this period" (zero) so the
        // rollforward still produces a stable opening → opening record.
        let (investee_net_income, investee_dividends_paid) =
            match deferred_lookup.get(entity.code.as_str()) {
                Some(tb) => (net_income_from_tb(tb, framework), dividends_from_tb(tb)),
                None => (Decimal::ZERO, Decimal::ZERO),
            };

        let inputs = EquityMethodInputs {
            investee: entity,
            investor_entity_code: parent_code.clone(),
            investee_net_income,
            investee_dividends_paid,
            opening_carrying_value: opening_map
                .get(&entity.code)
                .copied()
                .unwrap_or(Decimal::ZERO),
            opening_suppressed_loss: opening_suppressed_map
                .get(&entity.code)
                .copied()
                .unwrap_or(Decimal::ZERO),
            impairment: Decimal::ZERO,
            period_end: manifest.period.end,
            currency: manifest.presentation_currency.clone(),
        };
        invs.push(compute_equity_method_investment(&inputs)?);
    }
    Ok(invs)
}

/// Resolve the presentation framework once.  Maps the manifest's
/// primary CoA framework string back to the typed
/// [`AccountingFramework`].  Anything unrecognised falls through to the
/// default (UsGaap) — matches the [`AccountingFramework::Default`] impl.
fn resolve_primary_framework(manifest: &GroupManifest) -> AccountingFramework {
    let label = manifest
        .chart_of_accounts_master
        .primary_framework
        .to_lowercase();
    match label.as_str() {
        "ifrs" => AccountingFramework::Ifrs,
        "us_gaap" | "usgaap" | "us-gaap" => AccountingFramework::UsGaap,
        "dual_reporting" | "dual" => AccountingFramework::DualReporting,
        "french_gaap" | "frenchgaap" | "pcg" => AccountingFramework::FrenchGaap,
        "german_gaap" | "germangaap" | "hgb" => AccountingFramework::GermanGaap,
        _ => AccountingFramework::default(),
    }
}

/// O(1) lookup of `entity_code → ManifestEntity` from the manifest's
/// ownership graph.  Built once per `run_aggregate` call and shared
/// across the per-entity translation step.
fn entity_lookup(manifest: &GroupManifest) -> BTreeMap<String, ManifestEntity> {
    manifest
        .ownership_graph
        .entities
        .iter()
        .map(|e| (e.code.clone(), e.clone()))
        .collect()
}

/// Sum a translated TB's revenue minus expense (P&L lines) to derive
/// the entity's translated period net income.  Sign convention: revenue
/// (credit-natural) is added, expenses (debit-natural) subtracted, so
/// the result is positive on profit, negative on loss.
fn net_income_from_translated(translated: &TranslatedTb) -> Decimal {
    use crate::aggregate::translation::classify::TranslationAccountType as T;
    let mut net = Decimal::ZERO;
    for line in &translated.lines {
        let signed = match line.local_dr_cr {
            DrCr::Debit => line.translated_amount,
            DrCr::Credit => -line.translated_amount,
        };
        match line.account_type {
            // Revenue is credit-natural → contributes positively to net
            // income.  We negate the signed (debit positive, credit
            // negative) value so a CR revenue posting increments NI.
            T::PlRevenue => net -= signed,
            // Expense is debit-natural → subtracts from net income.  A
            // DR expense posting reduces NI; we subtract the (positive)
            // signed value.
            T::PlExpense => net -= signed,
            _ => {}
        }
    }
    net
}

/// Sum the OCI lines from a translated TB.  Sign convention follows
/// IAS 1 — OCI gains are credit-natural so we negate the signed amount.
fn oci_from_translated(translated: &TranslatedTb) -> Decimal {
    use crate::aggregate::translation::classify::TranslationAccountType as T;
    let mut oci = Decimal::ZERO;
    for line in &translated.lines {
        if line.account_type == T::PlOci {
            let signed = match line.local_dr_cr {
                DrCr::Debit => line.translated_amount,
                DrCr::Credit => -line.translated_amount,
            };
            oci -= signed;
        }
    }
    oci
}

/// Equivalent of [`net_income_from_translated`] but for an untranslated
/// [`TrialBalance`] — used for equity-method investees whose deferred
/// TB is consumed before translation.
///
/// Classifies each line by GL account → translation type so we can
/// pick out P&L revenue / expense lines without relying on the
/// AccountType enum (which the orchestrator may have populated with
/// different values for non-IFRS frameworks).
fn net_income_from_tb(tb: &TrialBalance, framework: AccountingFramework) -> Decimal {
    use crate::aggregate::translation::classify::{classify_account, TranslationAccountType as T};
    let mut net = Decimal::ZERO;
    for line in &tb.lines {
        let ty = classify_account(&line.account_code, framework);
        match ty {
            T::PlRevenue => net += line.credit_balance - line.debit_balance,
            T::PlExpense => net -= line.debit_balance - line.credit_balance,
            _ => {}
        }
    }
    net
}

/// Sum the dividends-paid balance from a TB.  Dividends paid lives in
/// the equity sub-account `equity_accounts::DIVIDENDS_PAID` (3500 in
/// some charts; the named constant is the source of truth).  We look up
/// that exact code rather than scan a range to avoid pulling in
/// unrelated equity movements.
fn dividends_from_tb(tb: &TrialBalance) -> Decimal {
    use datasynth_core::accounts::equity_accounts;
    tb.lines
        .iter()
        .filter(|l| l.account_code == equity_accounts::DIVIDENDS_PAID)
        .map(|l| l.debit_balance - l.credit_balance)
        .fold(Decimal::ZERO, |acc, v| acc + v)
}

// ── Unit tests ────────────────────────────────────────────────────────────────
//
// The driver itself is end-to-end tested by `tests/aggregate_e2e.rs`
// (which exercises the full pipeline against a Mini-Acme fixture).
// `resolve_primary_framework` is the only branch worth exercising at
// the unit-test layer; the remainder of the helpers are linear glue
// over already-tested sub-modules.

// ── v5.3 multi-period chain ────────────────────────────────────────────────

/// One period's input to [`run_aggregate_chain`].  Each `PeriodSpec`
/// describes a single aggregate run — its own manifest, the shards
/// dir to read from, and the out_dir to write to.
///
/// The chain runner inserts `prior_period_aggregate` automatically:
/// the N-th period's `out_dir` becomes the (N+1)-th's
/// `prior_period_aggregate`.  The first period uses the value the
/// caller set on its `AggregateOptions` (typically `None`).
#[derive(Debug, Clone)]
pub struct PeriodSpec {
    /// Manifest for this period.  Each period has its own manifest
    /// because IAS 21 closing rates, scoping, and entity ownership
    /// can change period-on-period.
    pub manifest: GroupManifest,
    /// Per-entity shard directory root for this period.
    pub shards_dir: PathBuf,
    /// Where to write this period's consolidated artefacts.
    pub out_dir: PathBuf,
    /// Per-period options.  `prior_period_aggregate` is overridden
    /// by the chain runner — set it on the FIRST period to seed the
    /// chain, and leave it `None` on subsequent periods.
    pub options: AggregateOptions,
}

/// **v5.3** — drive multi-period consolidation by chaining
/// [`run_aggregate`] calls in sequence.  Period N's `out_dir`
/// becomes period N+1's `prior_period_aggregate`, so opening NCI /
/// CTA / equity-method carrying values stitch automatically across
/// the chain.
///
/// v5.0–v5.2 already supported manual chaining via the
/// `--prior-period-aggregate` flag (PRs #141, #146 used it for the
/// acquisition-date NCI rollforward).  This helper formalises that
/// pattern so callers don't have to weave the paths by hand.
///
/// # Behaviour
///
/// - Periods run in the order they appear in `periods`.
/// - The first period uses its own `options.prior_period_aggregate`
///   (typically `None` for a fresh engagement, `Some(path)` when
///   resuming from an external archive).
/// - Each subsequent period gets `options.prior_period_aggregate`
///   **forcibly overridden** to the previous period's `out_dir` —
///   any value the caller set on `periods[n].options` for n ≥ 1 is
///   silently ignored.  This keeps the chain semantics intuitive.
/// - On any per-period failure the chain stops, returning the error
///   from `run_aggregate` for that period.  Output already written
///   for prior periods is preserved on disk.
///
/// # Returns
///
/// A `Vec<AggregateSummary>` matching `periods` in input order.
///
/// # Errors
///
/// Whatever [`run_aggregate`] returns for the first failing period.
pub fn run_aggregate_chain(periods: Vec<PeriodSpec>) -> GroupResult<Vec<AggregateSummary>> {
    let mut summaries: Vec<AggregateSummary> = Vec::with_capacity(periods.len());
    let mut prior_out: Option<PathBuf> = None;
    for (idx, mut spec) in periods.into_iter().enumerate() {
        // For periods 1..N, override prior_period_aggregate with the
        // previous period's out_dir.  Period 0 keeps whatever the
        // caller set.
        if idx > 0 {
            spec.options.prior_period_aggregate = prior_out.clone();
        }
        let summary = run_aggregate(
            &spec.manifest,
            &spec.shards_dir,
            &spec.out_dir,
            &spec.options,
        )?;
        prior_out = Some(spec.out_dir.clone());
        summaries.push(summary);
    }
    Ok(summaries)
}

#[cfg(test)]
mod tests {
    use super::resolve_primary_framework;
    use super::PeriodSpec;
    use std::path::PathBuf;

    #[test]
    fn period_spec_construction_uses_owned_paths() {
        // PeriodSpec must be constructible with owned PathBuf fields
        // and a real `AggregateOptions` value — the public contract
        // the chain runner consumes.  The actual chain execution
        // lives in `tests/aggregate_e2e.rs` once a multi-period
        // fixture is loaded; here we just pin the field shape so
        // refactors that move fields surface as a compile break.
        let opts = super::AggregateOptions {
            prior_period_aggregate: Some(PathBuf::from("/tmp/seed-from-engagement")),
            tolerate_missing_shards: false,
            cgu_test_inputs: Vec::new(),
            cpi_series_by_currency: std::collections::BTreeMap::new(),
        };
        assert!(opts.prior_period_aggregate.is_some());
        // PeriodSpec stays unconstructed in a runnable form because
        // `GroupManifest` requires a fully-populated graph; we keep
        // the type imported so a refactor that hides the type breaks
        // the test.
        let _: fn() -> PeriodSpec;
    }
    use crate::aggregate::driver::AggregateOptions;
    use datasynth_standards::framework::AccountingFramework;

    #[test]
    fn aggregate_options_default_is_fail_fast_no_prior() {
        let opts = AggregateOptions::default();
        assert!(!opts.tolerate_missing_shards, "default fails fast");
        assert!(opts.prior_period_aggregate.is_none(), "no prior period");
    }

    #[test]
    fn _resolve_primary_framework_is_pure_string_match() {
        // Type-only sanity: keep `resolve_primary_framework` in scope so
        // a refactor that drops a variant breaks the build instead of
        // silently falling through to `AccountingFramework::default()`.
        // The full label-vs-enum mapping is exercised in
        // `tests/aggregate_e2e.rs` once a fixture is loaded.
        let _ = resolve_primary_framework;
        let _ = AccountingFramework::default();
    }
}