From 7a2d06c22ba7d55e587936bec017b99ed516fe4d Mon Sep 17 00:00:00 2001 From: fangs311 Date: Sun, 1 Nov 2026 23:08:45 -0500 Subject: [PATCH] Add wafer comparison and stream analysis scans. Ship ANA-002 wafer-comparison over the shared adapter, stream per-source attempt reduces without multi-source buffering, and document the Dashboard spill skip with spike evidence. --- CHANGELOG.md | 6 +- docs/architecture/data-contracts.md | 9 +- docs/architecture/implementation-backlog.md | 2 +- docs/architecture/se-matrix-reassessment.md | 46 +++---- docs/cli.md | 3 +- docs/performance_baseline.md | 12 +- stdf-analytics/src/summary_ops.rs | 95 ++++++++++++++- stdf-cli/src/viewer/analysis.rs | 125 +++++++++++++------- stdf-cli/src/viewer/analysis_tests.rs | 73 ++++++++++++ 9 files changed, 292 insertions(+), 79 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 783f5a0..0574d02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,10 +58,14 @@ Shared retained-evidence adapter also ships: - `site-variation-request-v1` / `prr-site-variation-v1` - `lot-comparison-request-v1` / `prr-lot-comparison-v1` +- `wafer-comparison-request-v1` / `prr-wafer-comparison-v1` - `retest-impact-request-v1` / `prr-retest-impact-v1` Public entry: `zstdf-cli snapshot analyze ` (attempt-yield -remains available). Dashboard `coordinate-v1` merging is unchanged. +remains available). Dashboard `coordinate-v1` merging is unchanged. Absent and +empty wafer IDs stay distinct (`__missing__` / `__empty__`). Pareto remains +deferred. Analysis scans stream per evidence source after inventory collection +so corrupt indexes still fail before a complete receipt. ### Run telemetry diff --git a/docs/architecture/data-contracts.md b/docs/architecture/data-contracts.md index ef4ebd1..a1907db 100644 --- a/docs/architecture/data-contracts.md +++ b/docs/architecture/data-contracts.md @@ -169,13 +169,18 @@ distinct request/result versions: - `site-variation-request-v1` → per `(head, site)` yield; unresolved visible - `lot-comparison-request-v1` → per `lot_id` yield; unresolved visible +- `wafer-comparison-request-v1` → per wafer yield; absent wafer groups as + `__missing__`, empty-string wafer as `__empty__` (kept distinct); unresolved + visible - `retest-impact-request-v1` → per-source ECID first/last attempt impact, recovery/degradation among known verdicts; unresolved identity counted separately; `latest_device` keeps full history only for ECIDs with a selected latest flag -Pareto remains deferred. Site differences alone do not prove hardware causality -or GR&R; qualification/reliability decisions need study metadata. +Pareto / FTR pattern-attempt population remains deferred (needs execution +evidence beyond PRR companions). Site or wafer differences alone do not prove +hardware causality or GR&R; qualification/reliability decisions need study +metadata. ## Receipt and reproducibility diff --git a/docs/architecture/implementation-backlog.md b/docs/architecture/implementation-backlog.md index f45cbfc..5d876ad 100644 --- a/docs/architecture/implementation-backlog.md +++ b/docs/architecture/implementation-backlog.md @@ -85,7 +85,7 @@ or installer evidence. Do not alter current CI merely to hide extension failures | Priority / issue | Scope | Gate | | --- | --- | --- | -| P1 / ANA-002 | Latest yield, lot comparison, site variation, retest impact and Pareto | Latest yield + site/lot/retest shipped over retained evidence; Pareto still deferred; Dashboard merge untouched | +| P1 / ANA-002 | Latest yield, lot comparison, site variation, wafer comparison, retest impact and Pareto | Latest yield + site/lot/wafer/retest shipped over retained evidence; Pareto / FTR pattern population still deferred; Dashboard merge untouched | | P1 / API-001 | Read-only typed local tool adapter | Schema, authorization, cancellation, payload limits and denied operations tested; no LLM needed | | Optional / SQL-001 | Embedded DuckDB query adapter | Pinned version, fixed operations, restricted object access, resource tests, semantic parity; no install required for ordinary users | | Optional / AI-001 | Local or opted-in cloud explanation adapter | Authorized data scope, no arbitrary tool capability, provenance-grounded output, prompt-injection tests | diff --git a/docs/architecture/se-matrix-reassessment.md b/docs/architecture/se-matrix-reassessment.md index 33cf5a5..bbe054a 100644 --- a/docs/architecture/se-matrix-reassessment.md +++ b/docs/architecture/se-matrix-reassessment.md @@ -1,8 +1,7 @@ # SE matrix reassessment (post low-score fixes) -Status: evidence-based rescore after Waves 1–4 remainders plus adversarial -corner-case adapter tests on branch `feature/corner-case-tests-and-rescore` -(baseline main includes `#1`–`#5`). +Status: evidence-based rescore after TDD residuals on branch +`feature/se-tdd-residuals` (baseline main includes `#1`–`#6`). Companion canvas: `canvases/stdf-se-matrix-reassessment.canvas.tsx` (Cursor project canvases). @@ -11,37 +10,39 @@ Companion canvas: Same 1–5 rubric as the prior review. Scores move only when shipped code/tests/docs changed the evidence. Dashboard `coordinate-v1` yield math was not modified. -Pareto ANA-002 and dashboard SpillStore rewrite remain open (documented skip). -Identity policies stay distinct (attempt-yield / latest-device vs Dashboard merge). +Pareto ANA-002 and dashboard SpillStore rewrite remain open (documented skip with +spike evidence). Identity policies stay distinct (attempt-yield / latest-device vs +Dashboard merge). ## Before → prior → latest -| Dimension | Original | Prior (~Waves 1–4) | Latest | Why (latest delta) | +| Dimension | Original | Prior (~corner) | Latest | Why (latest delta) | | --- | ---: | ---: | ---: | --- | | Layering & modularity | 4.0 | 4.0 | 4.0 | Shared reducers in `stdf-analytics`; CLI still I/O host | | Cohesion of concerns | 3.0 | 3.5 | 3.5 | `summary_ops` + `attempt_yield_op` host analytics | -| Extensibility seams | 3.0 | 3.5 | 3.5 | Versioned request dispatch for site/lot/retest/yield | -| Domain fidelity | 4.5 | 4.5 | 4.5 | Tied/missing START_T + unresolved overlap asserted on adapter | -| Schema & versioning | 4.0 | 4.5 | 4.5 | `analysis-receipt-v1` + ANA-002 result versions | -| Provenance & reproducibility | 3.5 | 4.0 | 4.5 | `result_hash` stable under reorder; ANA-002 receipt replay | -| Reliability & failure model | 3.5 | 4.0 | 4.5 | Mid-scan cancel inject + corrupt ANA-002 empty-scope fail | -| Performance engineering | 3.5 | 3.5 | 3.5 | SpillStore not wired into dashboard (documented skip) | +| Extensibility seams | 3.0 | 3.5 | 4.0 | Wafer comparison joins versioned analyze dispatch | +| Domain fidelity | 4.5 | 4.5 | 4.5 | Missing vs empty wafer labels kept distinct | +| Schema & versioning | 4.0 | 4.5 | 4.5 | `wafer-comparison-*-v1` + existing ANA-002 versions | +| Provenance & reproducibility | 3.5 | 4.5 | 4.5 | Wafer receipt replay stable | +| Reliability & failure model | 3.5 | 4.5 | 4.5 | Stream scan still fails corrupt empty-scope evidence | +| Performance engineering | 3.5 | 3.5 | 4.0 | Per-source stream reduce; no multi-source attempt Vec | | Trust boundary | 3.5 | 4.0 | 4.0 | Path-traversal denial tests retained | | Observability | 2.0 | 3.0 | 3.0 | `run-telemetry-v1` on convert/view/analyze | | Operability & portability | 3.0 | 3.5 | 3.5 | Linux CI job (Rust/clippy/test; no Edge smoke) | -| Test strategy | 4.0 | 4.0 | 4.5 | 12 adversarial CLI corner cases over shipped adapter | +| Test strategy | 4.0 | 4.5 | 4.5 | Wafer + streaming corrupt fixtures; goldens held | | CI & automation | 3.0 | 3.5 | 3.5 | Clippy -D warnings + Linux job + cargo-deny | | Process & governance | 4.5 | 4.5 | 4.5 | Unchanged multi-agent / review policy | -| Maintainability | 3.0 | 3.5 | 4.0 | Analysis adapter + dashboard HTML/JSON render split (`#5`) | -| Analytics completeness | 2.5 | 3.5 | 3.5 | Latest yield + site/lot/retest; Pareto still deferred | +| Maintainability | 3.0 | 4.0 | 4.0 | Adapter + dashboard render split retained | +| Analytics completeness | 2.5 | 3.5 | 4.0 | Site/lot/wafer/retest + latest; Pareto still deferred | | API / release stability | 3.0 | 3.5 | 3.5 | Public `snapshot analyze` + CHANGELOG contracts | -Approximate weighted maturity: **~3.3 → ~3.8 → ~4.0** (same weights as prior canvas). +Approximate weighted maturity: **~3.3 → ~4.0 → ~4.1** (same weights as prior canvas). ## Residuals -1. Pareto / full 10F step analytics still deferred. -2. Dashboard SpillStore rewrite skipped to protect yield math. +1. Pareto / FTR pattern-attempt population still deferred (needs execution evidence). +2. Dashboard SpillStore rewrite skipped again after spike: SpillStore is an + external-sort store, not a map backend; no bitwise KPI golden suite yet. 3. Linux CI omits Playwright/Edge viewer smoke by design. 4. Loopback-only multi-user trust model unchanged. 5. Independent semantic review of this delta still required before calling the @@ -51,12 +52,11 @@ Approximate weighted maturity: **~3.3 → ~3.8 → ~4.0** (same weights as prior ```text cargo test -p stdf-analytics --locked -- summary_ops attempt_yield -cargo test -p stdf-cli --locked -- attempt_yield analysis_receipt latest_device site_variation lot_comparison retest_impact analysis_path +cargo test -p stdf-cli --locked -- attempt_yield analysis_receipt latest_device site_variation lot_comparison wafer_comparison retest_impact streaming_scan analysis_path cargo fmt --all cargo clippy -p stdf-analytics -p stdf-cli --locked -- -D warnings ``` -Corner cases added in this pass (CLI `analysis_tests.rs`): absent vs empty scope -`result_hash`, unresolved pass/fail/unknown overlap, tied/missing START_T latest, -reorder-stable `result_hash`, mid-scan cancel, site scope/excludes + zero-denom + -corrupt empty scope, multi-lot comparison, lot `latest_device`, retest receipt replay. +This pass added: `wafer-comparison-request-v1` reducer + CLI adapter coverage, +streaming inventory→scan reduce (inventory parity enforced), and documented +Dashboard spill skip spike evidence. diff --git a/docs/cli.md b/docs/cli.md index 61780f5..c25726d 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -153,6 +153,7 @@ versions and publish the same `analysis-receipt-v1` layout. Dispatch via | --- | --- | --- | | `site-variation-request-v1` | `site-variation-result-v1` / `prr-site-variation-v1` | Yield by `H{head} / S{site}` | | `lot-comparison-request-v1` | `lot-comparison-result-v1` / `prr-lot-comparison-v1` | Yield by `lot_id` | +| `wafer-comparison-request-v1` | `wafer-comparison-result-v1` / `prr-wafer-comparison-v1` | Yield by wafer; `__missing__` / `__empty__` reserved | | `retest-impact-request-v1` | `retest-impact-result-v1` / `prr-retest-impact-v1` | Per-source ECID first/last; unresolved identity counted separately | Populations remain `all_attempts` and `latest_device` (with @@ -161,7 +162,7 @@ Populations remain `all_attempts` and `latest_device` (with ```powershell .\target\release\zstdf-cli.exe snapshot analyze path\to\dataset site-request.json -cargo test -p stdf-cli site_variation lot_comparison retest_impact --locked +cargo test -p stdf-cli site_variation lot_comparison wafer_comparison retest_impact --locked ``` ## Multi-file Conversion (Phase 10B) diff --git a/docs/performance_baseline.md b/docs/performance_baseline.md index 4aea31c..3cf00b5 100644 --- a/docs/performance_baseline.md +++ b/docs/performance_baseline.md @@ -26,10 +26,14 @@ nonzero. Analysis callers can pass the same helper around `analysis::execute`. ## Dashboard spill rewrite (skipped) `stdf-analytics::SpillStore` remains available for future disk-backed dashboard -aggregations. Wiring it into `stdf-cli/src/dashboard.rs` in-memory maps was -skipped because changing aggregation order or intermediate materialization -risks Dashboard `coordinate-v1` yield math. Keep current in-memory KPIs until a -separate experiment proves bitwise-identical yields under spill. +aggregations. A 2026-11 residual spike reconfirmed the skip: Dashboard KPI / +wafer / site / bin yields are computed from in-memory `BTreeMap` part merges +(`coordinate-v1`), while `SpillStore` is an external-sort record store—not a +drop-in map backend. There is no locked bitwise-identical KPI golden suite that +would prove yield math under a rewrite. Keep current in-memory KPIs until a +separate experiment adds golden KPI fixtures and proves identical yields under +spill. Analysis adapters (attempt-yield / ANA-002) already stream per-source +attempt scans and do not depend on Dashboard spill. Run: diff --git a/stdf-analytics/src/summary_ops.rs b/stdf-analytics/src/summary_ops.rs index 7ee15d4..0d9f635 100644 --- a/stdf-analytics/src/summary_ops.rs +++ b/stdf-analytics/src/summary_ops.rs @@ -1,5 +1,5 @@ // Copyright 2026 zstdf contributors. SPDX-License-Identifier: Apache-2.0 -//! ANA-002 site / lot / retest summaries over physical PRR attempts. +//! ANA-002 site / lot / wafer / retest summaries over physical PRR attempts. //! //! These reducers reuse the attempt-yield request envelope and //! [`YieldAccumulator`] counting rules. Dashboard `coordinate-v1` merging is @@ -23,6 +23,14 @@ pub const LOT_REQUEST_VERSION: &str = "lot-comparison-request-v1"; pub const LOT_RESULT_VERSION: &str = "lot-comparison-result-v1"; pub const LOT_OPERATION: &str = "prr-lot-comparison-v1"; +pub const WAFER_REQUEST_VERSION: &str = "wafer-comparison-request-v1"; +pub const WAFER_RESULT_VERSION: &str = "wafer-comparison-result-v1"; +pub const WAFER_OPERATION: &str = "prr-wafer-comparison-v1"; +/// Group label when retained attempt wafer is absent (`None`). +pub const WAFER_MISSING_LABEL: &str = "__missing__"; +/// Group label when retained attempt wafer is present but empty. +pub const WAFER_EMPTY_LABEL: &str = "__empty__"; + pub const RETEST_REQUEST_VERSION: &str = "retest-impact-request-v1"; pub const RETEST_RESULT_VERSION: &str = "retest-impact-result-v1"; pub const RETEST_OPERATION: &str = "prr-retest-impact-v1"; @@ -31,6 +39,7 @@ pub const RETEST_OPERATION: &str = "prr-retest-impact-v1"; pub enum SummaryKind { SiteVariation, LotComparison, + WaferComparison, RetestImpact, } @@ -39,6 +48,7 @@ impl SummaryKind { match self { Self::SiteVariation => SITE_REQUEST_VERSION, Self::LotComparison => LOT_REQUEST_VERSION, + Self::WaferComparison => WAFER_REQUEST_VERSION, Self::RetestImpact => RETEST_REQUEST_VERSION, } } @@ -47,6 +57,7 @@ impl SummaryKind { match self { Self::SiteVariation => SITE_RESULT_VERSION, Self::LotComparison => LOT_RESULT_VERSION, + Self::WaferComparison => WAFER_RESULT_VERSION, Self::RetestImpact => RETEST_RESULT_VERSION, } } @@ -55,6 +66,7 @@ impl SummaryKind { match self { Self::SiteVariation => SITE_OPERATION, Self::LotComparison => LOT_OPERATION, + Self::WaferComparison => WAFER_OPERATION, Self::RetestImpact => RETEST_OPERATION, } } @@ -64,6 +76,7 @@ pub fn kind_of_version(version: &str) -> Option { match version { SITE_REQUEST_VERSION => Some(SummaryKind::SiteVariation), LOT_REQUEST_VERSION => Some(SummaryKind::LotComparison), + WAFER_REQUEST_VERSION => Some(SummaryKind::WaferComparison), RETEST_REQUEST_VERSION => Some(SummaryKind::RetestImpact), _ => None, } @@ -130,6 +143,14 @@ fn lot_key(row: &AttemptRecord) -> String { row.lot.clone() } +fn wafer_key(row: &AttemptRecord) -> String { + match &row.wafer { + None => WAFER_MISSING_LABEL.to_owned(), + Some(wafer) if wafer.is_empty() => WAFER_EMPTY_LABEL.to_owned(), + Some(wafer) => wafer.clone(), + } +} + fn disposition(row: &AttemptRecord, excludes: Option<&HashSet>) -> AttemptDisposition { AttemptDisposition { verdict: Verdict::from_prr_flags(row.flags), @@ -138,7 +159,7 @@ fn disposition(row: &AttemptRecord, excludes: Option<&HashSet>) -> AttemptD } } -/// Streaming site or lot aggregation. Caller owns cancel checks between rows. +/// Streaming site / lot / wafer aggregation. Caller owns cancel checks between rows. pub fn accumulate_groups( kind: SummaryKind, rows: impl IntoIterator, @@ -156,6 +177,7 @@ pub fn accumulate_groups( let key = match kind { SummaryKind::SiteVariation => site_key(&row), SummaryKind::LotComparison => lot_key(&row), + SummaryKind::WaferComparison => wafer_key(&row), SummaryKind::RetestImpact => { return Err(io::Error::new( io::ErrorKind::InvalidInput, @@ -172,7 +194,9 @@ pub fn accumulate_groups( fn row_in_summary(kind: SummaryKind, row: &AttemptRecord, population: Population) -> bool { match kind { - SummaryKind::SiteVariation | SummaryKind::LotComparison => in_population(row, population), + SummaryKind::SiteVariation | SummaryKind::LotComparison | SummaryKind::WaferComparison => { + in_population(row, population) + } // Retest needs full attempt history for selected devices; handled separately. SummaryKind::RetestImpact => true, } @@ -363,6 +387,7 @@ pub fn build_grouped_result( let group_field = match kind { SummaryKind::SiteVariation => "sites", SummaryKind::LotComparison => "lots", + SummaryKind::WaferComparison => "wafers", SummaryKind::RetestImpact => "groups", }; let mut items = Vec::with_capacity(groups.len()); @@ -484,12 +509,25 @@ mod tests { flags: u8, ecid: Option<&str>, latest: Option, + ) -> AttemptRecord { + row_wafer(attempt, head, site, lot, Some("W1"), flags, ecid, latest) + } + + fn row_wafer( + attempt: u64, + head: u8, + site: u8, + lot: &str, + wafer: Option<&str>, + flags: u8, + ecid: Option<&str>, + latest: Option, ) -> AttemptRecord { AttemptRecord { attempt, run: 1, lot: lot.into(), - wafer: Some("W1".into()), + wafer: wafer.map(str::to_owned), head, site, flags, @@ -545,6 +583,55 @@ mod tests { assert_eq!(groups["L2"].summary().fail, 1); } + #[test] + fn wafer_comparison_keeps_missing_and_empty_distinct() { + let scope = compile_scope(None, &HashMap::new()).unwrap(); + // Hand-computed: W1 pass, W2 fail, missing unknown, empty pass. + let rows = [ + row_wafer(1, 1, 1, "L1", Some("W1"), 0, Some("a"), Some(true)), + row_wafer(2, 1, 1, "L1", Some("W2"), 8, Some("b"), Some(true)), + row_wafer(3, 1, 1, "L1", None, 4, None, None), + row_wafer(4, 1, 1, "L1", Some(""), 0, Some("c"), Some(true)), + ]; + let (total, groups) = accumulate_groups( + SummaryKind::WaferComparison, + rows, + &"s".repeat(64), + Population::AllAttempts, + &scope, + None, + ) + .unwrap(); + let summary = total.summary(); + assert_eq!(summary.pass, 2); + assert_eq!(summary.fail, 1); + assert_eq!(summary.unknown, 1); + assert_eq!(summary.unresolved, 1); + assert_eq!(groups["W1"].summary().pass, 1); + assert_eq!(groups["W2"].summary().fail, 1); + assert_eq!(groups[WAFER_MISSING_LABEL].summary().unknown, 1); + assert_eq!(groups[WAFER_EMPTY_LABEL].summary().pass, 1); + let encoded = build_grouped_result( + SummaryKind::WaferComparison, + Population::AllAttempts, + &summary, + &summaries_of(&groups), + &Coverage { + snapshot_sources: 1, + selected_sources: 1, + missing_source_ids: Vec::new(), + partial: false, + expected_inventory: "unknown".into(), + content_alias_source_count: 0, + catalog_run_status: "complete".into(), + catalog_failure_count: 0, + }, + ); + assert_eq!(encoded["version"], WAFER_RESULT_VERSION); + assert_eq!(encoded["operation"], WAFER_OPERATION); + assert_eq!(encoded["wafers"].as_array().unwrap().len(), 4); + } + #[test] fn retest_impact_recovery_and_unresolved_are_visible() { let scope = compile_scope(None, &HashMap::new()).unwrap(); diff --git a/stdf-cli/src/viewer/analysis.rs b/stdf-cli/src/viewer/analysis.rs index c30a43f..d45a975 100644 --- a/stdf-cli/src/viewer/analysis.rs +++ b/stdf-cli/src/viewer/analysis.rs @@ -1,10 +1,11 @@ // Copyright 2026 zstdf contributors. SPDX-License-Identifier: Apache-2.0 //! Shared retained-evidence analysis adapter (not Dashboard coordinate-v1 merge). //! -//! Dispatches versioned request envelopes for attempt-yield, site variation, -//! lot comparison and retest impact. Attempt rows are scanned with cooperative -//! cancel checks; corrupt selected evidence always fails before a complete -//! result is published. +//! Dispatches versioned request envelopes for attempt-yield, site / lot / wafer +//! comparison and retest impact. Inventory is collected without buffering all +//! attempts; each evidence source is then stream-scanned with cooperative cancel +//! checks. Corrupt selected evidence always fails before a complete result is +//! published. Retest still materializes one source at a time for ECID history. use super::evidence::{self, EvidenceBundle, EvidenceLimits}; use super::{CliResult, Row}; use crate::telemetry::RunTelemetry; @@ -71,10 +72,9 @@ impl OpKind { } } -struct LoadedEvidence { +struct EvidenceMeta { bundle: EvidenceBundle, run_ids: BTreeSet, - rows: Vec, } /// Execute a versioned analysis request against retained evidence for `dataset`. @@ -105,34 +105,28 @@ pub(crate) fn execute_with_telemetry( t.counter("evidence_ids", request.evidence_ids.len() as u64); } - // Sequential open/scan so each evidence lock is released before the next - // acquire. Every selected ID is fully scanned so corrupt indexes fail even - // when the later scope would match zero rows. - let mut loaded = Vec::with_capacity(request.evidence_ids.len()); + // Phase 1: open each evidence for inventory/run IDs only (release lock + // before the next). Scope compiles from manifests; attempt bodies are + // not buffered across sources. + let mut metas = Vec::with_capacity(request.evidence_ids.len()); let mut seen_sources = HashSet::new(); for evidence_id in &request.evidence_ids { deny_traversal_id(evidence_id, "evidence_id")?; - let mut attempts = evidence::open_attempts(dataset, evidence_id, limits)?; + let attempts = evidence::open_attempts(dataset, evidence_id, limits)?; if attempts.bundle.snapshot_id != request.snapshot_id { return Err("evidence snapshot_id does not match request".into()); } if !seen_sources.insert(attempts.bundle.source_id.clone()) { return Err("duplicate catalog source_id among selected evidence".into()); } - let mut rows = Vec::with_capacity(attempts.bundle.attempts as usize); - while let Some(row) = attempts.next()? { - limits_check(limits)?; - rows.push(to_record(&row)); - } - loaded.push(LoadedEvidence { + metas.push(EvidenceMeta { bundle: attempts.bundle.clone(), run_ids: attempts.run_ids.clone(), - rows, }); drop(attempts); } - let selected: BTreeSet<_> = loaded + let selected: BTreeSet<_> = metas .iter() .map(|item| item.bundle.source_id.clone()) .collect(); @@ -148,7 +142,7 @@ pub(crate) fn execute_with_telemetry( return Err("selected evidence does not cover every snapshot catalog source".into()); } - let run_sets: HashMap> = loaded + let run_sets: HashMap> = metas .iter() .map(|item| { ( @@ -161,7 +155,7 @@ pub(crate) fn execute_with_telemetry( run_sets.iter().map(|(k, v)| (k.as_str(), v)).collect(); let compiled = compile_scope(request.scope.as_ref(), &run_refs).map_err(|e| e.to_string())?; - let attempt_inventory: HashMap<&str, u64> = loaded + let attempt_inventory: HashMap<&str, u64> = metas .iter() .map(|item| (item.bundle.source_id.as_str(), item.bundle.attempts)) .collect(); @@ -172,27 +166,30 @@ pub(crate) fn execute_with_telemetry( t.phase("reduce"); } - let metas: Vec<_> = loaded.iter().map(|i| i.bundle.clone()).collect(); + let bundles: Vec<_> = metas.iter().map(|i| i.bundle.clone()).collect(); + // Phase 2: re-open each evidence and stream-scan. Full scan always runs + // so corrupt indexes fail even when scope matches zero rows. Scanned + // count must equal retained attempt inventory. let mut result = match op { OpKind::AttemptYield => { let mut total = YieldAccumulator::default(); let mut per_source = BTreeMap::new(); - for item in &loaded { + for item in &metas { let source_excludes = excluded.get(item.bundle.source_id.as_str()); let mut acc = YieldAccumulator::default(); - for row in &item.rows { - limits_check(limits)?; - if !in_population(row, population) - || !in_scope(row, &item.bundle.source_id, &compiled) + stream_attempts(dataset, &item.bundle, limits, |row| { + if !in_population(&row, population) + || !in_scope(&row, &item.bundle.source_id, &compiled) { - continue; + return Ok(()); } acc.add(AttemptDisposition { verdict: Verdict::from_prr_flags(row.flags), identity_resolved: row.ecid.is_some(), excluded: source_excludes.is_some_and(|set| set.contains(&row.attempt)), })?; - } + Ok(()) + })?; total.merge(&acc)?; per_source.insert( item.bundle.source_id.clone(), @@ -212,7 +209,7 @@ pub(crate) fn execute_with_telemetry( t.counter("scoped_attempts", total.summary().scoped_attempts); } let coverage = - coverage_of(&snapshot, &selected, missing, catalog_incomplete, &metas); + coverage_of(&snapshot, &selected, missing, catalog_incomplete, &bundles); build_result(population, &total.summary(), &per_source, &coverage) } OpKind::Summary(SummaryKind::RetestImpact) => { @@ -227,11 +224,18 @@ pub(crate) fn execute_with_telemetry( let mut degraded = 0u64; let mut first_total = YieldAccumulator::default(); let mut latest_total = YieldAccumulator::default(); - for item in &loaded { + for item in &metas { limits_check(limits)?; let source_excludes = excluded.get(item.bundle.source_id.as_str()); + // One-source buffer only: ECID history needs the full source + // attempt set, but not every selected source at once. + let mut rows = Vec::with_capacity(item.bundle.attempts as usize); + stream_attempts(dataset, &item.bundle, limits, |row| { + rows.push(row); + Ok(()) + })?; let part = accumulate_retest( - item.rows.iter().cloned(), + rows, &item.bundle.source_id, population, &compiled, @@ -281,23 +285,30 @@ pub(crate) fn execute_with_telemetry( degraded, }; let coverage = - coverage_of(&snapshot, &selected, missing, catalog_incomplete, &metas); + coverage_of(&snapshot, &selected, missing, catalog_incomplete, &bundles); build_retest_result(population, &combined, &coverage) } OpKind::Summary(kind) => { let mut total = YieldAccumulator::default(); let mut groups: BTreeMap = BTreeMap::new(); - for item in &loaded { + for item in &metas { limits_check(limits)?; let source_excludes = excluded.get(item.bundle.source_id.as_str()); - let (part_total, part_groups) = accumulate_groups( - kind, - item.rows.iter().cloned(), - &item.bundle.source_id, - population, - &compiled, - source_excludes, - )?; + let mut part_total = YieldAccumulator::default(); + let mut part_groups: BTreeMap = BTreeMap::new(); + stream_attempts(dataset, &item.bundle, limits, |row| { + let (acc_total, acc_groups) = accumulate_groups( + kind, + std::iter::once(row), + &item.bundle.source_id, + population, + &compiled, + source_excludes, + )?; + part_total.merge(&acc_total)?; + merge_group_maps(&mut part_groups, acc_groups)?; + Ok(()) + })?; total.merge(&part_total)?; merge_group_maps(&mut groups, part_groups)?; } @@ -305,7 +316,7 @@ pub(crate) fn execute_with_telemetry( t.counter("scoped_attempts", total.summary().scoped_attempts); } let coverage = - coverage_of(&snapshot, &selected, missing, catalog_incomplete, &metas); + coverage_of(&snapshot, &selected, missing, catalog_incomplete, &bundles); build_grouped_result( kind, population, @@ -540,6 +551,34 @@ fn to_record(row: &Row) -> AttemptRecord { } } +/// Re-open retained attempts, stream every row, and enforce inventory parity. +fn stream_attempts( + dataset: &Path, + bundle: &EvidenceBundle, + limits: &EvidenceLimits, + mut on_row: impl FnMut(AttemptRecord) -> CliResult<()>, +) -> CliResult<()> { + let mut attempts = evidence::open_attempts(dataset, &bundle.evidence_id, limits)?; + if attempts.bundle.evidence_id != bundle.evidence_id + || attempts.bundle.source_id != bundle.source_id + || attempts.bundle.attempts != bundle.attempts + { + return Err("evidence inventory changed between analysis phases".into()); + } + let mut scanned = 0u64; + while let Some(row) = attempts.next()? { + limits_check(limits)?; + scanned = scanned + .checked_add(1) + .ok_or_else(|| "attempt scan count overflow".to_string())?; + on_row(to_record(&row))?; + } + if scanned != bundle.attempts { + return Err("scanned attempt count does not match retained inventory".into()); + } + Ok(()) +} + fn build_identity() -> BuildIdentity { BuildIdentity { package_version: env!("CARGO_PKG_VERSION").into(), diff --git a/stdf-cli/src/viewer/analysis_tests.rs b/stdf-cli/src/viewer/analysis_tests.rs index 98c6642..16ff0e6 100644 --- a/stdf-cli/src/viewer/analysis_tests.rs +++ b/stdf-cli/src/viewer/analysis_tests.rs @@ -1110,6 +1110,79 @@ fn lot_comparison_latest_device_population() { assert_eq!(lots[0]["counts"]["pass"], "1"); } +#[test] +fn wafer_comparison_splits_multiple_wafers() { + let f = Fixture::new(); + // Hand-computed: W1 pass + W2 fail across two MIR runs. + let mut bytes = start(); + prr_only(&mut bytes, 1, 1, 0, 1); + finish(&mut bytes, 200); + bytes.extend(run("LOT-A", "W2", 300)); + prr_only(&mut bytes, 1, 1, 8, 2); + finish(&mut bytes, 400); + let (snapshot, _, bundle) = f.setup(&bytes); + let response = f + .execute(&summary_request( + "wafer-comparison-request-v1", + &snapshot, + &[bundle], + )) + .unwrap(); + assert_eq!(response["version"], "wafer-comparison-result-v1"); + assert_eq!(response["operation"], "prr-wafer-comparison-v1"); + assert_eq!(response["counts"], counts(2, 2, 0, 1, 1, 0, 0)); + let wafers = response["wafers"].as_array().unwrap(); + assert_eq!(wafers.len(), 2); + assert_eq!(wafers[0]["label"], "W1"); + assert_eq!(wafers[0]["counts"], counts(1, 1, 0, 1, 0, 0, 0)); + assert_eq!(wafers[1]["label"], "W2"); + assert_eq!(wafers[1]["counts"], counts(1, 1, 0, 0, 1, 0, 0)); + assert_eq!(response["analysis"]["status"], "complete"); +} + +#[test] +fn wafer_comparison_latest_device_and_receipt_replay() { + let f = Fixture::new(); + let mut bytes = start(); + prr_only(&mut bytes, 1, 1, 8, 1); + finish(&mut bytes, 200); + bytes.extend(run("LOT-A", "W1", 300)); + prr_only(&mut bytes, 1, 1, 0, 1); + finish(&mut bytes, 400); + let (snapshot, _, bundle) = f.setup(&bytes); + let mut req = summary_request("wafer-comparison-request-v1", &snapshot, &[bundle]); + req["population"] = json!("latest_device"); + req["identity_policy"] = json!("latest-ecid-start-t-v1"); + let first = f.execute(&req).unwrap(); + let second = f.execute(&req).unwrap(); + assert_eq!(first["population"], "latest_device"); + assert_eq!(first["counts"], counts(1, 1, 0, 1, 0, 0, 0)); + assert_eq!(first["wafers"].as_array().unwrap().len(), 1); + assert_eq!( + first["analysis"]["result_hash"], + second["analysis"]["result_hash"] + ); + assert_eq!(first["counts"], second["counts"]); +} + +#[test] +fn streaming_scan_still_fails_corrupt_evidence_with_empty_scope() { + // Stream phase must fully read attempts even when scope matches nothing. + let f = Fixture::new(); + let (snapshot, bundles) = f.aliases(); + let mut req = summary_request("wafer-comparison-request-v1", &snapshot, &bundles); + req["scope"] = json!({"wafer_ids": ["NO-SUCH-WAFER"]}); + let corrupt = f + .dataset() + .join("evidence") + .join(bundles[1]["evidence_id"].as_str().unwrap()) + .join("attempts.idx"); + fs::write(corrupt, b"corrupt index").unwrap(); + assert!(f.execute(&req).is_err()); + req["allow_partial"] = json!(true); + assert!(f.execute(&req).is_err()); +} + #[test] fn retest_impact_receipt_replay_stable() { let f = Fixture::new();