Add reconstruction similarity scoring

Author: Michal Warda

README.md

@@ -84,6 +84,13 @@ curl -f http://localhost:8000/v1/separate \
   -o voice-and-residual.zip
 ```
 
+The durable pipeline also sums the stored stereo-mapped stems after separation,
+reassembles the chunk joins into one joined stereo WAV per original record, and
+records a 0–100 sample-aligned reconstruction similarity score plus correlation,
+level, error, SNR, and per-channel metrics. These artifacts, chunk diagnostics,
+and the dataset-wide source-score distribution are available in the **Split
+data** dashboard; the direct model ZIP API remains unchanged.
+
 Generation, prompt, batching, and TF32 policy are configured with the
 `SAM_AUDIO_*` environment variables demonstrated in
 [`deploy/start-sam-audio-api.sh`](deploy/start-sam-audio-api.sh). A supplied

docs/pipeline/ARCHITECTURE.md

@@ -46,25 +46,38 @@ rule are persisted under `adaptive_routing`.
    loudness transfer is broadband so it cannot re-EQ the stem. Both are smoothed
    with a bidirectional EMA (`alpha=0.03`) before equal-power panning. An SFX-only
    pass-through bypasses mapping and keeps the original stereo PCM exactly.
-8. Raw and stereo-mapped WAVs, the 120-point pan/loudness trajectory, inference
-   metadata, timings, and review assertions are stored in S3 and indexed in
-   DynamoDB. A stereo mapping failure never discards or delays the raw result.
-9. Description and transcription tasks share one Audio Flamingo queue so that a
+8. After the output sound gate, the stored stereo-mapped stems are summed into
+   one joined WAV per chunk. Once every chunk is terminal, those chunk joins are
+   placed back on their source timeline (overlaps are averaged) to produce one
+   `source.joined.stereo.wav` for the original record. The stored source PCM is
+   compared sample-for-sample with the normalized stereo original. A 0–100
+   symmetric waveform-agreement score penalizes phase, timing, polarity, and
+   level error; correlation, left/right scores, level delta, normalized RMSE,
+   SNR, coverage, and limiter gain remain alongside it for diagnosis. Chunk
+   scores remain available for locating a problem. An identity SFX pass-through
+   reconstructs exactly at 100.
+9. Raw and stereo-mapped WAVs, the joined reconstruction, similarity metrics,
+   the 120-point pan/loudness trajectory, inference metadata, timings, and
+   review assertions are stored in S3 and indexed in DynamoDB. A stereo mapping
+   failure never discards or delays the raw result.
+10. Description and transcription tasks share one Audio Flamingo queue so that a
    single loaded model processes one item at a time. It creates one source-scene
    analysis per audible source, one description per music stem, and one
    speaker-labelled transcript per voice stem. Tasks exist in DynamoDB before
    SQS delivery and are reconciled after worker/message loss.
-10. A browser dashboard shows job/queue progress and exact remaining review
+11. A browser dashboard shows job/queue progress and exact remaining review
    counts. Its reviewer mode pulls the next `uncertain` or `failure` assertion,
    starts audio automatically, displays the prompt/assertion, and accepts
    one-key decisions.
-11. The dataset explorer reports source count, duration, input/chunk/stem storage,
+12. The dataset explorer reports source count, duration, input/chunk/stem storage,
    processing counts, stem mix, and review backlog. Each source displays the
    original recording first, then only the music/voice/SFX stems that passed the
    output sound gate, grouped by chunk, followed by its processing route and
-   model annotations. Raw mono playback is the default; a `Stereo mapped`
-   toggle swaps only the stem players and leaves the original recording
-   untouched. Each stem also shows its smoothed left/right trajectory.
+   model annotations. Stereo-mapped playback is the default; a `Raw` toggle
+   swaps only the stem players and leaves the original recording untouched.
+   Each stem also shows its smoothed left/right trajectory. The explorer plays
+   the full joined result beside its original, retains each chunk join and its
+   diagnostics, and graphs the dataset-wide source-score distribution.
 
 ## AWS resources
 
@@ -103,14 +116,18 @@ jobs/{job_id}/sources/{source_id}/{filename}
 jobs/{job_id}/chunks/{source_id}/{chunk_index}.wav
 jobs/{job_id}/stems/{source_id}/{chunk_index}/{music|voice|sfx}.wav
 jobs/{job_id}/stems/{source_id}/{chunk_index}/{music|voice|sfx}.stereo.wav
+jobs/{job_id}/reconstructions/{source_id}/{chunk_index}.joined.stereo.wav
+jobs/{job_id}/reconstructions/{source_id}/source.joined.stereo.wav
 jobs/{job_id}/metadata/{source_id}/{chunk_index}.json
 jobs/{job_id}/metadata/{source_id}/{chunk_index}.stereo.json
 ```
 
 Source and stem records include SHA-256 hashes. A stem record includes the
 source chunk hash, prompt, model/settings fingerprint, automated status,
 reviewer status, raw and stereo S3 keys/hashes/sizes, stereo mapping summary,
-timings, and raw Judge/CLAP evidence. The raw `s3_key` is never replaced by its
+chunk reconstruction key/hash/size and waveform similarity metrics, timings,
+and raw Judge/CLAP evidence. The source record stores the corresponding full
+record reconstruction and metrics. The raw `s3_key` is never replaced by its
 mapped companion.
 
 ## Review contract

docs/pipeline/OPERATIONS.md

@@ -97,6 +97,39 @@ ssh ubuntu@ec2-44-211-31-38.compute-1.amazonaws.com \
 Use `--max-chunks N` for a canary. Mapping is CPU-only and does not require a
 SAM or Audio Flamingo restart.
 
+## Joined reconstruction and similarity
+
+After stereo mapping and the output sound gate, the pipeline sums only the
+stored stereo variants into `{chunk}.joined.stereo.wav`. After all chunks for a
+record are terminal, it places those joins back on the source timeline and
+overlap-averages them into `source.joined.stereo.wav`. The primary dataset score
+compares that stored source PCM directly with the normalized stereo original,
+without gain fitting or time alignment:
+
+`100 × max(0, 2 × dot(original, joined) / (energy(original) + energy(joined)))`
+
+This makes 100 an exact reconstruction; missing-output silence against an
+audible original, opposite polarity, or another non-positive match scores 0.
+The source record also retains waveform correlation,
+left/right scores, level delta, normalized RMSE, SNR, coverage, and final limiter
+gain. Chunk records retain the same diagnostics plus the included stem types so
+a low source score can be localized. In **Split data**, the dataset histogram
+shows the source-score distribution, the source detail plays Original beside
+Joined, and each chunk exposes its normalized original and diagnostic join.
+
+Backfill completed jobs from existing stereo companions without rerunning SAM or
+Audio Flamingo:
+
+```bash
+ssh ubuntu@ec2-44-211-31-38.compute-1.amazonaws.com \
+  "sudo bash -c 'set -a; source /etc/sam-audio-pipeline.env; set +a; \
+   cd /home/ubuntu/sam-audio-deploy/pipeline; \
+   .venv/bin/sam-pipeline-reconstruction-backfill --job-id JOB_ID'"
+```
+
+The command is idempotent; use `--force` to regenerate an existing joined
+artifact or `--max-chunks N` for a canary.
+
 ## AudioSet validation batches
 
 Acquire a reproducible random sample from the official AudioSet segment CSVs.

docs/pipeline/PROGRESS.md

@@ -32,6 +32,8 @@ was an example batch size, not a workflow limit.
   original-plus-stems source explorer.
 - [x] Add frequency-aware stereo/loudness reconstruction while preserving every
   raw mono stem, plus a dashboard playback toggle and smoothed trajectory plots.
+- [x] Join stored stereo stems, persist phase/level-sensitive waveform fidelity
+  metrics, and graph the dataset-wide reconstruction score distribution.
 - [x] Require stereo input, gate SAM targets with source-scene presence, support
   single-stage inference, and use identity pass-through for pure SFX sources.
 - [x] Add persistent datasets, successive upload jobs, and reconciliation of
@@ -93,6 +95,13 @@ was an example batch size, not a workflow limit.
   source was repaired from its stored scene evidence and now has one SFX-only
   stem; raw, mapped, and original normalized PCM are sample-identical. The
   dataset now indexes 282 stems (93 music, 90 voice, 99 SFX).
+- Joined-stereo reconstruction is backfilled for all 100 stereo sources in seed
+  `20260713` and all 98 stereo sources in seed `20260714`. Their mean waveform
+  similarity is 82.79% (7.11–100) and 88.02% (24.92–100), respectively. The
+  repaired writing/SFX-only record scores exactly 100 with correlation 1.0 and
+  0 dB level error. Four additional reconstructable legacy sources (five
+  chunks) were also migrated, for 202 full-source joins from 203 chunk joins;
+  mono and sound-gated sources correctly have no joined artifact.
 - Live mono-filter job `5ec26b1bb28b420a9245e21d475d26be` completed with
   `non_stereo_input`, one input channel, zero chunks, zero stems, and zero model
   tasks. A live `targets=voice` API canary produced only voice and residual
@@ -106,7 +115,7 @@ was an example batch size, not a workflow limit.
 - Recovery job `1d07979ee96a45a98cf220b102fd7006` was uploaded without the
   completion callback. Reconciliation found the S3 object, enqueued ingestion,
   and completed it as sound-gated. All three queues were empty afterward.
-- Local and remote validation: 36 pipeline tests pass; the existing 12 SAM
+- Local and remote validation: 39 pipeline tests pass; the existing 12 SAM
   batching tests also pass.
 - Docker build execution remains unverified because neither the local Mac nor
   the current H100 host has a Docker daemon. Both Dockerfiles remain model-free.

pipeline/pyproject.toml

@@ -19,6 +19,7 @@ sam-pipeline-worker = "sam_audio_pipeline.worker:main"
 sam-pipeline-audioset = "sam_audio_pipeline.audioset:main"
 sam-pipeline-submit = "sam_audio_pipeline.submit:main"
 sam-pipeline-stereo-backfill = "sam_audio_pipeline.stereo:main"
+sam-pipeline-reconstruction-backfill = "sam_audio_pipeline.reconstruction:main"
 sam-audio-flamingo-api = "sam_audio_pipeline.flamingo_api:main"
 
 [dependency-groups]

pipeline/src/sam_audio_pipeline/api.py

@@ -4,6 +4,7 @@
 
 import mimetypes
 import re
+import statistics
 import uuid
 from collections import Counter, defaultdict
 from contextlib import asynccontextmanager
@@ -65,6 +66,33 @@ def _review_counts(aws: PipelineAWS) -> dict[str, int]:
     }
 
 
+def _percentile(values: list[float], quantile: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    position = (len(ordered) - 1) * quantile
+    lower = int(position)
+    upper = min(lower + 1, len(ordered) - 1)
+    weight = position - lower
+    return ordered[lower] * (1.0 - weight) + ordered[upper] * weight
+
+
+def _similarity_summary(scores: list[float]) -> dict[str, Any]:
+    return {
+        "count": len(scores),
+        "mean": round(statistics.fmean(scores), 4) if scores else None,
+        "median": round(statistics.median(scores), 4) if scores else None,
+        "minimum": round(min(scores), 4) if scores else None,
+        "maximum": round(max(scores), 4) if scores else None,
+        "p10": round(value, 4)
+        if (value := _percentile(scores, 0.1)) is not None
+        else None,
+        "p90": round(value, 4)
+        if (value := _percentile(scores, 0.9)) is not None
+        else None,
+    }
+
+
 def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
     dataset = aws.get(f"DATASET#{dataset_id}", "META")
     if not dataset:
@@ -121,6 +149,21 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
                 ),
                 None,
             )
+            chunk_reconstruction_scores = [
+                float(score)
+                for chunk in source_chunks
+                if (
+                    score := chunk.get("reconstruction", {})
+                    .get("metrics", {})
+                    .get("similarity_score")
+                )
+                is not None
+            ]
+            source_similarity = (
+                source.get("reconstruction", {})
+                .get("metrics", {})
+                .get("similarity_score")
+            )
             all_sources.append(
                 {
                     **source,
@@ -146,6 +189,22 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
                         )
                     ),
                     "selected_order": route,
+                    "reconstruction_count": (
+                        1 if source_similarity is not None else 0
+                    ),
+                    "reconstructed_chunk_count": len(
+                        chunk_reconstruction_scores
+                    ),
+                    "similarity_score": (
+                        round(float(source_similarity), 4)
+                        if source_similarity is not None
+                        else None
+                    ),
+                    "similarity_minimum": (
+                        round(min(chunk_reconstruction_scores), 4)
+                        if chunk_reconstruction_scores
+                        else None
+                    ),
                 }
             )
         job_summaries.append(
@@ -165,6 +224,41 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
     chunk_bytes = sum(int(chunk.get("bytes") or 0) for chunk in all_chunks)
     stem_bytes = sum(int(stem.get("bytes") or 0) for stem in all_stems)
     stereo_bytes = sum(int(stem.get("stereo_bytes") or 0) for stem in all_stems)
+    chunk_reconstruction_bytes = sum(
+        int(chunk.get("reconstruction", {}).get("bytes") or 0)
+        for chunk in all_chunks
+    )
+    source_reconstruction_bytes = sum(
+        int(source.get("reconstruction", {}).get("bytes") or 0)
+        for source in all_sources
+    )
+    reconstruction_bytes = (
+        chunk_reconstruction_bytes + source_reconstruction_bytes
+    )
+    reconstructions = []
+    for source in all_sources:
+        reconstruction = source.get("reconstruction", {})
+        metrics = reconstruction.get("metrics", {})
+        score = metrics.get("similarity_score")
+        if score is None:
+            continue
+        reconstructions.append(
+            {
+                "job_id": source.get("job_id"),
+                "source_id": source.get("source_id"),
+                "filename": source.get("filename"),
+                "chunk_count": source.get("chunk_count"),
+                "duration_seconds": source.get("duration_seconds"),
+                "similarity_score": float(score),
+                "waveform_correlation": metrics.get("waveform_correlation"),
+                "level_delta_db": metrics.get("level_delta_db"),
+                "snr_db": metrics.get("snr_db"),
+                "selected_order": source.get("selected_order"),
+            }
+        )
+    similarity_scores = [
+        float(item["similarity_score"]) for item in reconstructions
+    ]
     review_remaining = sum(
         stem.get("effective_status") in {"failure", "uncertain"} for stem in all_stems
     )
@@ -184,7 +278,16 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
             "chunk_bytes": chunk_bytes,
             "stem_bytes": stem_bytes,
             "stereo_bytes": stereo_bytes,
-            "total_bytes": input_bytes + chunk_bytes + stem_bytes + stereo_bytes,
+            "reconstruction_bytes": reconstruction_bytes,
+            "chunk_reconstruction_bytes": chunk_reconstruction_bytes,
+            "source_reconstruction_bytes": source_reconstruction_bytes,
+            "total_bytes": (
+                input_bytes
+                + chunk_bytes
+                + stem_bytes
+                + stereo_bytes
+                + reconstruction_bytes
+            ),
             "chunks": len(all_chunks),
             "audible_chunks": sum(
                 chunk.get("status") != "skipped" for chunk in all_chunks
@@ -194,6 +297,11 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
             ),
             "stems": len(all_stems),
             "stereo_stems": sum(bool(stem.get("stereo_s3_key")) for stem in all_stems),
+            "reconstructed_chunks": sum(
+                bool(chunk.get("reconstruction")) for chunk in all_chunks
+            ),
+            "reconstructed_sources": len(reconstructions),
+            "similarity": _similarity_summary(similarity_scores),
             "stems_by_type": dict(
                 Counter(str(stem.get("stem_type")) for stem in all_stems)
             ),
@@ -216,6 +324,10 @@ def _dataset_snapshot(aws: PipelineAWS, dataset_id: str) -> dict[str, Any]:
             key=lambda source: str(source.get("created_at", "")),
             reverse=True,
         ),
+        "reconstructions": sorted(
+            reconstructions,
+            key=lambda item: float(item["similarity_score"]),
+        ),
     }
 
 
@@ -432,13 +544,19 @@ def get_source(job_id: str, source_id: str, request: Request) -> dict[str, Any]:
         if not source:
             raise HTTPException(status_code=404, detail="Source not found")
         job = next(item for item in items if item["SK"] == "META")
+        source_reconstruction = source.get("reconstruction", {})
         source = {
             **source,
             "audio_url": (
                 store.presign_download(source["s3_key"])
                 if source.get("s3_key") and source.get("status") != "failed"
                 else None
             ),
+            "joined_audio_url": (
+                store.presign_download(source_reconstruction["s3_key"])
+                if source_reconstruction.get("s3_key")
+                else None
+            ),
         }
         chunks = sorted(
             [
@@ -499,6 +617,7 @@ def get_source(job_id: str, source_id: str, request: Request) -> dict[str, Any]:
         expanded_chunks = []
         for chunk in chunks:
             chunk_id = str(chunk["chunk_id"])
+            reconstruction = chunk.get("reconstruction", {})
             expanded_chunks.append(
                 {
                     **chunk,
@@ -514,6 +633,11 @@ def get_source(job_id: str, source_id: str, request: Request) -> dict[str, Any]:
                         ),
                     ),
                     "omitted_stems": sorted(omitted_by_chunk[chunk_id]),
+                    "joined_audio_url": (
+                        store.presign_download(reconstruction["s3_key"])
+                        if reconstruction.get("s3_key")
+                        else None
+                    ),
                 }
             )
         annotations = [