feat(api): Track 3.4 — additive publicStatus field on every stage

apps/backend/server_phase1.py

@@ -9,6 +9,7 @@
 
 from analysis_runtime import AnalysisRuntime
 from server_upload import ERROR_PHASE_LOCAL_DSP
+from stage_status import to_public_status
 
 
 # ── Constants ────────────────────────────────────────────────────────────────
@@ -226,6 +227,30 @@ def _build_phase1(payload: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _annotate_public_status(stages: dict[str, Any]) -> dict[str, Any]:
+    """Attach ``publicStatus`` to every stage whose ``status`` is set.
+
+    Returns a shallow-copy of ``stages`` with each stage dict updated.
+    The original ``status`` field is preserved untouched; ``publicStatus``
+    is the additive 5-state collapse documented in
+    :mod:`stage_status`.
+
+    Defensive against non-dict stage entries (e.g. nulls in legacy
+    snapshots) — those pass through unchanged.
+    """
+    annotated: dict[str, Any] = {}
+    for stage_name, stage_value in stages.items():
+        if not isinstance(stage_value, dict):
+            annotated[stage_name] = stage_value
+            continue
+        stage_copy = dict(stage_value)
+        # `status` may legitimately be absent on legacy/partial snapshots;
+        # to_public_status(None) yields None which we expose as null.
+        stage_copy["publicStatus"] = to_public_status(stage_copy.get("status"))
+        annotated[stage_name] = stage_copy
+    return annotated
+
+
 def _normalize_run_snapshot(
     snapshot: dict[str, Any], runtime: AnalysisRuntime | None = None
 ) -> dict[str, Any]:
@@ -236,22 +261,27 @@ def _normalize_run_snapshot(
     _build_phase1 produces for the legacy /api/analyze endpoint — notably,
     top-level stereoWidth/stereoCorrelation extracted from stereoDetail.
 
+    Each stage in ``stages`` is also annotated with a ``publicStatus`` field
+    that collapses the 8 internal stage statuses to 5 public ones (see
+    :mod:`stage_status`). The original ``status`` field is preserved; this
+    is purely additive.
+
     When *runtime* is provided, spectral visualization artifacts (spectrogram
     PNGs and time-series JSON) are attached under ``artifacts.spectral``.
     """
     stages = snapshot.get("stages")
     if not isinstance(stages, dict):
         return snapshot
-    measurement = stages.get("measurement")
-    if not isinstance(measurement, dict):
-        return snapshot
-    raw_result = measurement.get("result")
-    if not isinstance(raw_result, dict):
-        return snapshot
     snapshot = dict(snapshot)
-    snapshot["stages"] = dict(stages)
-    snapshot["stages"]["measurement"] = dict(measurement)
-    snapshot["stages"]["measurement"]["result"] = _build_phase1(raw_result)
+    snapshot["stages"] = _annotate_public_status(stages)
+
+    measurement = snapshot["stages"].get("measurement")
+    if isinstance(measurement, dict):
+        raw_result = measurement.get("result")
+        if isinstance(raw_result, dict):
+            # Reuse the dict we already copied in _annotate_public_status;
+            # no need to clone twice.
+            measurement["result"] = _build_phase1(raw_result)
 
     if runtime is not None:
         run_id = snapshot.get("runId")

apps/backend/stage_status.py

@@ -0,0 +1,88 @@
+"""Public-facing collapse of the internal stage-status state machine.
+
+ASA's run-state machine has eight internal statuses
+(``queued``, ``running``, ``blocked``, ``ready``, ``completed``,
+``failed``, ``interrupted``, ``not_requested``). Most of these are
+runtime-scheduling concerns the client does not need to act on:
+
+- ``blocked`` and ``ready`` are transient states between "waiting on
+  a dependency" and "scheduled to run." Both look identical to a
+  client polling the snapshot — there's nothing to display or do
+  differently.
+- ``not_requested`` means the caller did not ask for this stage to
+  run (e.g. ``pitch_note_mode="off"``). Conceptually distinct from
+  "queued"; the stage is simply absent from the requested pipeline.
+
+This module exposes the five-state vocabulary intended for external
+consumers and the mapping from the internal eight to that public
+five. The internal ``status`` field on each stage stays untouched;
+this is an *additive* collapse — the route layer attaches a parallel
+``publicStatus`` field next to ``status`` on each stage.
+
+Mapping:
+
+==============  ===============================
+Internal        Public (publicStatus)
+==============  ===============================
+queued          queued
+running         running
+blocked         queued       (transient internal)
+ready           queued       (scheduled, not yet running)
+completed       completed
+failed          failed
+interrupted     interrupted
+not_requested   None         (not in the pipeline)
+==============  ===============================
+
+A ``None`` mapping is exposed in JSON as ``"publicStatus": null``.
+This is deliberate — explicit ``null`` is easier for clients than
+checking key presence, especially in strongly-typed languages where
+optional vs missing are different shapes.
+
+See ``docs/adr/0001-phase1-json-schema-v1.md`` for the schema-version
+treatment of this field.
+"""
+
+from __future__ import annotations
+
+from typing import Final
+
+
+PUBLIC_STATUS_VALUES: Final[frozenset[str]] = frozenset(
+    {"queued", "running", "completed", "failed", "interrupted"}
+)
+
+
+_INTERNAL_TO_PUBLIC: Final[dict[str, str | None]] = {
+    "queued": "queued",
+    "running": "running",
+    "blocked": "queued",
+    "ready": "queued",
+    "completed": "completed",
+    "failed": "failed",
+    "interrupted": "interrupted",
+    "not_requested": None,
+}
+
+
+def to_public_status(internal_status: str | None) -> str | None:
+    """Map an internal stage status to its public-facing equivalent.
+
+    Returns ``None`` when:
+    - ``internal_status`` is ``None`` (stage exists but no status set)
+    - ``internal_status`` is ``"not_requested"`` (stage is not in the pipeline)
+    - ``internal_status`` is an unrecognized value (defensive — should not
+      happen with the current state machine, but a forwards-compat
+      guard against accidental internal-only additions leaking out)
+    """
+    if internal_status is None:
+        return None
+    return _INTERNAL_TO_PUBLIC.get(internal_status)
+
+
+def public_status_values() -> frozenset[str]:
+    """The complete set of non-null public status values.
+
+    Useful for type generation, validation, and documentation.
+    """
+    return PUBLIC_STATUS_VALUES

apps/backend/tests/test_server.py

@@ -1289,6 +1289,48 @@ def test_get_analysis_run_returns_persisted_stage_snapshot(self) -> None:
         self.assertEqual(payload["runId"], created["runId"])
         self.assertIn("artifacts", payload)
 
+    def test_get_analysis_run_attaches_public_status_to_every_stage(self) -> None:
+        """Track 3.4: every stage in the response carries publicStatus
+        alongside the internal status. Pitch-note and interpretation
+        stages here are not_requested (publicStatus must be null);
+        measurement stage is queued (publicStatus must be queued).
+        """
+        from analysis_runtime import AnalysisRuntime
+
+        with tempfile.TemporaryDirectory(prefix="asa_server_runtime_") as temp_dir:
+            runtime = AnalysisRuntime(Path(temp_dir) / "runtime")
+            created = runtime.create_run(
+                filename="track.mp3",
+                content=b"fake-audio",
+                mime_type="audio/mpeg",
+                pitch_note_mode="off",
+                pitch_note_backend="auto",
+                interpretation_mode="off",
+                interpretation_profile="producer_summary",
+                interpretation_model=None,
+            )
+            with patch.object(server, "get_analysis_runtime", return_value=runtime):
+                response = asyncio.run(server.get_analysis_run(created["runId"]))
+
+        payload = self._decode_json_response(response)
+        stages = payload["stages"]
+
+        # measurement starts queued; publicStatus should match
+        self.assertEqual(stages["measurement"]["status"], "queued")
+        self.assertEqual(stages["measurement"]["publicStatus"], "queued")
+
+        # not_requested stages must surface publicStatus: null so a
+        # client can distinguish "you didn't ask for this" from
+        # "queued and waiting" without checking a specific string.
+        self.assertEqual(
+            stages["pitchNoteTranslation"]["status"], "not_requested"
+        )
+        self.assertIsNone(stages["pitchNoteTranslation"]["publicStatus"])
+        self.assertEqual(
+            stages["interpretation"]["status"], "not_requested"
+        )
+        self.assertIsNone(stages["interpretation"]["publicStatus"])
+
     def test_interrupt_analysis_run_marks_stages_interrupted_and_reports_terminated_children(self) -> None:
         from analysis_runtime import AnalysisRuntime
 

apps/backend/tests/test_stage_status.py

@@ -0,0 +1,140 @@
+"""Unit tests for stage_status — the 8 → 5 public-status collapse.
+
+The collapse is the documented contract for ``publicStatus`` on every
+stage object in the run snapshot. Tests here pin the mapping cell-by-cell
+so a quiet edit to the lookup table is caught.
+"""
+
+import sys
+import unittest
+from pathlib import Path
+
+
+_BACKEND_ROOT = Path(__file__).resolve().parents[1]
+if str(_BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_ROOT))
+
+
+import stage_status  # noqa: E402 — load after sys.path is set
+
+
+class PublicStatusValuesTests(unittest.TestCase):
+    """The five public-facing values, exhaustively."""
+
+    def test_exact_five_values(self):
+        self.assertEqual(
+            stage_status.PUBLIC_STATUS_VALUES,
+            frozenset({"queued", "running", "completed", "failed", "interrupted"}),
+        )
+
+    def test_helper_returns_same_set(self):
+        self.assertEqual(
+            stage_status.public_status_values(),
+            stage_status.PUBLIC_STATUS_VALUES,
+        )
+
+
+class ToPublicStatusTests(unittest.TestCase):
+    """Cell-by-cell mapping pins for the eight internal states."""
+
+    def test_queued_maps_to_queued(self):
+        self.assertEqual(stage_status.to_public_status("queued"), "queued")
+
+    def test_running_maps_to_running(self):
+        self.assertEqual(stage_status.to_public_status("running"), "running")
+
+    def test_blocked_maps_to_queued(self):
+        """``blocked`` is transient internal scheduling; clients see it
+        the same as ``queued``."""
+        self.assertEqual(stage_status.to_public_status("blocked"), "queued")
+
+    def test_ready_maps_to_queued(self):
+        """``ready`` means scheduled but not yet started; same as queued
+        from the client's perspective."""
+        self.assertEqual(stage_status.to_public_status("ready"), "queued")
+
+    def test_completed_maps_to_completed(self):
+        self.assertEqual(stage_status.to_public_status("completed"), "completed")
+
+    def test_failed_maps_to_failed(self):
+        self.assertEqual(stage_status.to_public_status("failed"), "failed")
+
+    def test_interrupted_maps_to_interrupted(self):
+        self.assertEqual(
+            stage_status.to_public_status("interrupted"), "interrupted"
+        )
+
+    def test_not_requested_maps_to_none(self):
+        """``not_requested`` means the caller didn't ask for this stage;
+        conceptually distinct from queued. Exposed as ``null`` in JSON."""
+        self.assertIsNone(stage_status.to_public_status("not_requested"))
+
+
+class ToPublicStatusDefensiveTests(unittest.TestCase):
+    """Defensive cases: nones, unknowns, and the empty string."""
+
+    def test_none_returns_none(self):
+        self.assertIsNone(stage_status.to_public_status(None))
+
+    def test_unknown_internal_status_returns_none(self):
+        """A new internal-only state added without a public mapping
+        should fall through as None rather than leak through. This is
+        the forwards-compat guard."""
+        self.assertIsNone(stage_status.to_public_status("does_not_exist"))
+
+    def test_empty_string_returns_none(self):
+        self.assertIsNone(stage_status.to_public_status(""))
+
+    def test_case_sensitive(self):
+        """The internal vocabulary is lowercase. Uppercase shouldn't
+        match — the mapping table is case-sensitive by design."""
+        self.assertIsNone(stage_status.to_public_status("QUEUED"))
+        self.assertIsNone(stage_status.to_public_status("Running"))
+
+
+class MappingCompletenessTests(unittest.TestCase):
+    """Pins the set of internal states the mapping covers.
+
+    If a new internal status is added without a corresponding entry
+    here, this test fails. If an old one is removed, this test fails.
+    Either way, the developer must make a deliberate choice about the
+    public mapping rather than silently drift.
+    """
+
+    EXPECTED_INTERNAL_STATES = frozenset(
+        {
+            "queued",
+            "running",
+            "blocked",
+            "ready",
+            "completed",
+            "failed",
+            "interrupted",
+            "not_requested",
+        }
+    )
+
+    def test_internal_states_unchanged(self):
+        actual = frozenset(stage_status._INTERNAL_TO_PUBLIC.keys())
+        self.assertEqual(actual, self.EXPECTED_INTERNAL_STATES)
+
+    def test_all_non_null_mappings_are_public_values(self):
+        """Every non-None target of the mapping must be in
+        PUBLIC_STATUS_VALUES — no smuggling new public states in via
+        the mapping table without updating the canonical set."""
+        for internal, public in stage_status._INTERNAL_TO_PUBLIC.items():
+            if public is None:
+                continue
+            self.assertIn(
+                public,
+                stage_status.PUBLIC_STATUS_VALUES,
+                msg=(
+                    f"{internal!r} maps to {public!r}, which is not in "
+                    f"PUBLIC_STATUS_VALUES. Either add it to the public "
+                    f"set or change the mapping."
+                ),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()