fix(workbook): preserve stable column IDs across display aliases

apps/api/services/workbook/cell_scope.py

@@ -39,15 +39,18 @@ def allows_automatic_retry(column: dict) -> bool:
 def row_execution_data(row, columns: list[dict] | None = None) -> dict:
     """Hydrate saved dependencies, then assert database execution identities."""
     data = dict(row.data or {})
+    # Failed computed snapshots invalidate materialized values in data, while
+    # inputs read their original stored fields regardless of configuration order.
+    input_data = dict(data)
     values, aliases = {}, {}
     for column in columns or []:
         cid = column.get("id")
         if column.get("type") in ("lead_field", "input"):
             field = column.get("lead_field") or cid
-            if field in data:
-                values[cid] = data[field]
+            if field in input_data:
+                values[cid] = input_data[field]
                 if column.get("name"):
-                    aliases[column["name"]] = data[field]
+                    aliases[column["name"]] = input_data[field]
             continue
         cell = (row.enrichments or {}).get(cid)
         if isinstance(cell, dict) and (cell.get("status") != "complete" or cell.get("value") is None):

apps/api/services/workbook/conditions.py

@@ -108,14 +108,16 @@ def _get_values(row_cells: dict, columns_config: list) -> Dict[str, str]:
     """Extract flat {col_id: value, col_name: value} dict."""
     col_name_map = {c["id"]: c.get("name", c["id"]) for c in columns_config}
     values = {}
+    aliases = {}
 
     for col_id, cell in row_cells.items():
         val = cell.get("value", "") if isinstance(cell, dict) else cell
         val_str = str(val) if val is not None else ""
         values[col_id] = val_str
-        values[col_name_map.get(col_id, col_id)] = val_str
+        aliases[col_name_map.get(col_id, col_id)] = val_str
 
-    return values
+    # IDs remain authoritative when a display name collides with another ID.
+    return {**aliases, **values}
 
 
 def _resolve_placeholders(condition: str, values: Dict[str, str]) -> str:

tests/test_workbook_views.py

@@ -2739,3 +2739,109 @@ def test_saved_view_empty_comparison_still_matches_missing_values(client, compar
     rows = tc.get(f"/api/workbooks/{wid}", params={"view_id": created.json()["id"]})
     assert rows.status_code == 200, rows.text
     assert [row["row_id"] for row in rows.json()["rows"]] == [blank]
+
+@pytest.mark.parametrize("computed_first", [True, False])
+@pytest.mark.parametrize("collision", [True, False])
+def test_failed_display_alias_preserves_public_input_id(client, computed_first, collision):
+    tc, Session, _ = client
+    wid = _mk_workbook(Session, [])
+    source = {"id": "b", "name": "Input B", "type": "input"}
+    computed = {"id": "a", "name": "b" if collision else "Computed A",
+                "type": "formula", "formula": "'OTHER'"}
+    columns = [computed, source] if computed_first else [source, computed]
+    columns.append({"id": "result", "name": "Result", "type": "formula", "formula": "{b}"})
+    for column in columns:
+        added = tc.post(f"/api/workbooks/{wid}/columns", json={"column": column})
+        assert added.status_code == 200, added.text
+    original = {"b": "500", "a": "STALE"}
+    failed = {"status": "error", "value": None, "error": "fixture_failure"}
+    row_id = _mk_row(Session, wid, original, {"a": failed})
+    response = tc.post(f"/api/workbooks/{wid}/rows/{row_id}/cells/result/run",
+                       json={"force": True})
+    assert response.status_code == 200, response.text
+    assert response.json()["status"] == "complete", response.json()
+    assert response.json()["value"] == "500", response.json()
+    with Session() as db:
+        row = db.get(WorkbookRow, row_id)
+        assert row.data == original
+        assert row.enrichments["a"] == failed
+        assert row.enrichments["result"]["value"] == "500"
+    if not collision:
+        alias_column = {"id": "alias_result", "name": "Alias result",
+                        "type": "formula", "formula": "{Input B}"}
+        added = tc.post(f"/api/workbooks/{wid}/columns", json={"column": alias_column})
+        assert added.status_code == 200, added.text
+        alias = tc.post(f"/api/workbooks/{wid}/rows/{row_id}/cells/alias_result/run",
+                        json={"force": True})
+        assert alias.status_code == 200, alias.text
+        assert alias.json()["status"] == "complete", alias.json()
+        assert alias.json()["value"] == "500", alias.json()
+
+
+@pytest.mark.parametrize("computed_first", [True, False])
+@pytest.mark.parametrize("collision", [True, False])
+def test_completed_display_alias_cannot_replace_public_condition_id(client, computed_first, collision):
+    tc, Session, _ = client
+    wid = _mk_workbook(Session, [])
+    source = {"id": "b", "name": "Input B", "type": "input"}
+    computed = {"id": "a", "name": "b" if collision else "Computed A",
+                "type": "formula", "formula": "'OTHER'"}
+    columns = [computed, source] if computed_first else [source, computed]
+    columns.append({"id": "result", "name": "Result", "type": "formula",
+                    "formula": "{b}", "condition": '{b} == "500"'})
+    for column in columns:
+        added = tc.post(f"/api/workbooks/{wid}/columns", json={"column": column})
+        assert added.status_code == 200, added.text
+    original = {"b": "500", "a": "STALE"}
+    completed = {"status": "complete", "value": "OTHER"}
+    row_id = _mk_row(Session, wid, original, {"a": completed})
+    response = tc.post(f"/api/workbooks/{wid}/rows/{row_id}/cells/result/run",
+                       json={"force": True})
+    assert response.status_code == 200, response.text
+    assert response.json()["status"] == "complete", response.json()
+    assert response.json()["value"] == "500", response.json()
+    with Session() as db:
+        row = db.get(WorkbookRow, row_id)
+        assert row.data == original
+        assert row.enrichments["a"] == completed
+        assert row.enrichments["result"]["value"] == "500"
+    if not collision:
+        alias_column = {"id": "alias_result", "name": "Alias result",
+                        "type": "formula", "formula": "{Computed A}"}
+        added = tc.post(f"/api/workbooks/{wid}/columns", json={"column": alias_column})
+        assert added.status_code == 200, added.text
+        alias = tc.post(f"/api/workbooks/{wid}/rows/{row_id}/cells/alias_result/run",
+                        json={"force": True})
+        assert alias.status_code == 200, alias.text
+        assert alias.json()["status"] == "complete", alias.json()
+        assert alias.json()["value"] == "OTHER", alias.json()
+
+@pytest.mark.parametrize("computed_first", [True, False])
+def test_duplicate_display_alias_rejects_ambiguous_public_reference(client, computed_first):
+    tc, Session, _ = client
+    wid = _mk_workbook(Session, [])
+    source = {"id": "b", "name": "Shared", "type": "input"}
+    computed = {"id": "a", "name": "Shared", "type": "formula",
+                "formula": "'OTHER'"}
+    columns = [computed, source] if computed_first else [source, computed]
+    columns.append({"id": "result", "name": "Result", "type": "formula",
+                    "formula": "{Shared}"})
+    for column in columns:
+        added = tc.post(f"/api/workbooks/{wid}/columns", json={"column": column})
+        assert added.status_code == 200, added.text
+    original = {"b": "500", "a": "STALE"}
+    completed = {"status": "complete", "value": "OTHER"}
+    row_id = _mk_row(Session, wid, original, {"a": completed})
+    response = tc.post(f"/api/workbooks/{wid}/rows/{row_id}/cells/result/run",
+                       json={"force": True})
+    assert response.status_code == 200, response.text
+    assert response.json()["status"] == "error", response.json()
+    assert response.json()["error"] == "ambiguous_column_reference", response.json()
+    assert response.json()["value"] is None, response.json()
+    with Session() as db:
+        row = db.get(WorkbookRow, row_id)
+        assert row.data == original
+        assert row.enrichments["a"] == completed
+        assert row.enrichments["result"]["status"] == "error"
+        assert row.enrichments["result"]["error"] == "ambiguous_column_reference"
+        assert row.enrichments["result"]["value"] is None