docs: ARCHITECTURE — фокусная сборка ответа на правку (AURORA-1850)

Co-Authored-By: Claude Haiku 5.5 <noreply@anthropic.com>

ARCHITECTURE.md

@@ -2,7 +2,7 @@
 
 Монорепозиторий: `api/` (FastAPI) отдаёт REST, WebSocket и собранный фронт; `web/` — SPA на React. Основная бизнес-логика — PL/pgSQL-функции схемы `v3` в Postgres. Python-слой проверяет роли, валидирует запрос, вызывает функцию и маппит её ошибки.
 
-Поток редактирования ячейки: `RealtimeTable` → WebSocket `/api/v1/ws/form/{id}/sheet/{sheet}` (`cell_edit_start` — блокировка, `cell_updated`) → `SheetService` → `v3.upd_form_cell(s)`. Функция проверяет роль и окно этапа и пишет аудит `v3.log_event`, после чего изменение рассылается остальным подключённым. Ответ `upd_form_cells` (HTTP и `cell_updated`) — полные строки (`data` как в GET листа, с явными `null`): изменённые INPUT и все их предки-агрегаты, в том числе неизменившиеся; при `sections` ключи исключённых разделов убираются. Выборку делает `v3._form_affected_rows` (миграция 0050).
+Поток редактирования ячейки: `RealtimeTable` → WebSocket `/api/v1/ws/form/{id}/sheet/{sheet}` (`cell_edit_start` — блокировка, `cell_updated`) → `SheetService` → `v3.upd_form_cell(s)`. Функция проверяет роль и окно этапа и пишет аудит `v3.log_event`, после чего изменение рассылается остальным подключённым. Ответ `upd_form_cells` (HTTP и `cell_updated`) — полные строки (`data` как в GET листа, с явными `null`): изменённые INPUT и все их предки-агрегаты, в том числе неизменившиеся; при `sections` ключи исключённых разделов убираются. Выборку делает `v3._form_affected_rows` (миграции 0050, 0051). Для FORM_1 и FORM_2 она не собирает лист целиком: построители (`v_form1_view`/`v_form2_view` → `*_sheet_jsonb` → `*_sheet_sections`) вызываются с фокусом `p_focus_lines` (NULL — весь лист, `'{}'` — ничего) и считают только изменённые строки и узлы их путей. `sort_order` в обоих режимах берётся из скелета листа (`_sort_rank`), поэтому совпадает с GET. FORM_4 до AURORA-1846 собирается полностью. Фоновая дельта АХР после правки сателлита (`get_form2_ahr_satellite_delta`) идёт тем же путём.
 
 ## Модули
 

api/alembic/versions/0051_upd_form_cells_subtree.py

@@ -0,0 +1,15 @@
+from sql_migration import run_sql_file_migration
+
+
+revision = "0051"
+down_revision = "0050"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    run_sql_file_migration("0051_upd_form_cells_subtree.sql")
+
+
+def downgrade() -> None:
+    pass

api/tests/integration/sheet_contract.py

@@ -20,3 +20,108 @@ def assert_patch_rows_match_sheet(patch_rows: list[dict], sheet_rows: list[dict]
         key = row_key(row)
         assert key in sheet_by_key, f"строки {key} нет в GET /sheet"
         assert row["data"] == sheet_by_key[key]["data"], f"data строки {key} расходится с GET"
+
+
+# --- AURORA-1850: эталон (фильтр 0050 поверх полного v_form_view) и хелперы SQL ---
+
+Row4 = tuple[str, int, int, str]  # (row_type, depth, sort_order, data::text)
+
+_STRIP_ALL = [
+    "plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking",
+    "contract_summary", "allocation", "collegial", "ckk",
+    "sequestration", "reserve", "contract_detail",
+    "q1", "q2", "q3", "q4", "totals",
+    "fact_q1", "fact_q2", "fact_q3", "fact_q4",
+]
+
+_SAT_SHEETS = ("AHR_RENT", "AHR_UTILITY", "AHR_SECURITY")
+
+
+def strip_keys(sections: list[str] | None) -> list[str]:
+    """Ключи верхнего уровня исключённых разделов (как v_strip в 0050)."""
+    if sections is None:
+        return []
+    return [k for k in _STRIP_ALL if k not in sections]
+
+
+def sheet_kind(sheet: str) -> str:
+    if sheet in _SAT_SHEETS:
+        return "sat"
+    if sheet == "AHR_LIMIT":
+        return "limit"
+    if sheet == "OTCH9F":
+        return "otch9f"
+    return "tree"
+
+
+_REF_FROM = """
+  SELECT v.row_type, v.depth, v.sort_order, (v.data - CAST(:strip AS text[]))::text
+    FROM v3.v_form_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar)) v
+   WHERE {where}
+   ORDER BY v.sort_order"""
+
+_REF_WHERE = {
+    "sat": "v.row_type = 'INPUT' AND (v.data->>'id')::int = ANY(CAST(:ids AS int[]))",
+    "limit": "(v.data->>'id')::int = ANY(v3._limit_template_path_set(CAST(:ids AS int[])))",
+    "otch9f": """EXISTS (SELECT 1 FROM v3.fixed_asset_report far
+                  WHERE far.id = ANY(CAST(:ids AS int[]))
+                    AND far.expense_item_id = (v.data->>'expense_item_id')::int
+                    AND far.equipment_name IS NOT DISTINCT FROM (v.data->>'equipment_name'))""",
+    "tree": """(v.row_type IN ('ROOT','GROUP','ITEM','SUB_ITEM','SUB_SUB_ITEM')
+                AND (v.data->'header'->>'expense_item_id')::int = ANY(v3._expense_item_path_set(
+                      ARRAY(SELECT DISTINCT bl.expense_item_id FROM v3.budget_line bl
+                             WHERE bl.id = ANY(CAST(:ids AS int[]))))))
+               OR (v.row_type = 'INPUT' AND (v.data->>'line_id')::int = ANY(CAST(:ids AS int[])))""",
+}
+
+
+def _rows(db_exec, sql: str, **params) -> list:
+    return [tuple(r) for r in db_exec(sql, **params)]
+
+
+def reference_affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None, strip=True):
+    sql = _REF_FROM.format(where=_REF_WHERE[sheet_kind(sheet)])
+    return _rows(
+        db_exec, sql, f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections,
+        strip=strip_keys(sections) if strip else [],
+    )
+
+
+def affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None):
+    return _rows(
+        db_exec,
+        """SELECT r.row_type, r.depth, r.sort_order, r.data::text
+             FROM v3._form_affected_rows(:f, :s, CAST(:ids AS int[]), CAST(:d AS varchar),
+                                         CAST(:sec AS text[])) r
+            ORDER BY r.sort_order""",
+        f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections,
+    )
+
+
+def focus_builder_rows(db_exec, form_type, form_id, sheet, ids, direction=None, sections=None):
+    if form_type == "FORM_2":
+        sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text
+                   FROM v3.v_form2_view(:f, :s, CAST(:sec AS text[]), CAST(:ids AS int[])) r
+                  ORDER BY r.sort_order"""
+        params = dict(f=form_id, s=sheet, sec=sections, ids=list(ids))
+    else:
+        sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text
+                   FROM v3.v_form1_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar),
+                                        CAST(:ids AS int[])) r
+                  ORDER BY r.sort_order"""
+        params = dict(f=form_id, s=sheet, sec=sections, d=direction, ids=list(ids))
+    return _rows(db_exec, sql, **params)
+
+
+def assert_rows_equal(actual: list, expected: list, label: str) -> None:
+    """Построчное сравнение; сообщение называет первую отличающуюся строку и длины."""
+    if actual == expected:
+        return
+    n = min(len(actual), len(expected))
+    first = next((i for i in range(n) if actual[i] != expected[i]), n)
+    a = actual[first] if first < len(actual) else None
+    e = expected[first] if first < len(expected) else None
+    raise AssertionError(
+        f"{label}: длины actual={len(actual)} expected={len(expected)}; "
+        f"первое расхождение в позиции {first}:\n  actual   = {a}\n  expected = {e}"
+    )

api/tests/integration/test_upd_form_cells_subtree.py

@@ -0,0 +1,469 @@
+"""AURORA-1850: эквивалентность пересчёта поддерева и полной сборки листа.
+
+Эталон — фильтр 0050 поверх полного v3.v_form_view. Новый путь (v3._form_affected_rows и
+сборщики v3.v_form2_view / v3.v_form1_view с фокусом по id) обязан давать те же строки.
+"""
+
+import uuid
+
+import pytest
+
+from sheet_contract import (
+    affected_rows,
+    assert_patch_rows_match_sheet,
+    assert_rows_equal,
+    focus_builder_rows,
+    reference_affected_rows,
+    row_key,
+)
+from test_upd_form_cells_full_rows import (  # noqa: F401  (фикстуры и хелперы)
+    API,
+    _add_line,
+    _get_sheet,
+    _new_form,
+    _patch_cell,
+    _qs,
+    db_exec,
+    headers,
+)
+
+TREE_TYPES = ("ROOT", "GROUP", "ITEM", "SUB_ITEM", "SUB_SUB_ITEM")
+SATELLITES = ("AHR_RENT", "AHR_UTILITY", "AHR_SECURITY")
+FORM2_SHEETS = ("AHR", "CAP", "OPER", "AHR_RENT", "AHR_UTILITY", "AHR_SECURITY", "AHR_LIMIT", "OTCH9F")
+
+# (form_type, sheet, direction, sections, год новой формы)
+CASES = {
+    "F1-AHR-Support": ("FORM_1", "AHR", "Support", None, 2041),
+    "F1-CAP-Support": ("FORM_1", "CAP", "Support", None, 2043),
+    "F1-OPER": ("FORM_1", "OPER", None, None, 2044),
+    "F2-AHR": ("FORM_2", "AHR", None, None, 2045),
+    "F2-CAP": ("FORM_2", "CAP", None, None, 2046),
+    "F2-OPER": ("FORM_2", "OPER", None, None, 2047),
+    "F2-AHR_RENT": ("FORM_2", "AHR_RENT", None, None, 0),
+    "F2-AHR_UTILITY": ("FORM_2", "AHR_UTILITY", None, None, 0),
+    "F2-AHR_SECURITY": ("FORM_2", "AHR_SECURITY", None, None, 0),
+    "F2-AHR_LIMIT": ("FORM_2", "AHR_LIMIT", None, None, 0),
+    "F2-OTCH9F": ("FORM_2", "OTCH9F", None, None, 0),
+    "F2-AHR-plan": ("FORM_2", "AHR", None, ["plan"], 2048),
+}
+SAT_FORM = 4
+
+
+def _ordered_leaves(db_exec, form_type, sheet):
+    """Листовые статьи, привязанные к типу формы, в порядке mv_expense_item_tree.path."""
+    return [
+        (r[0], r[1], list(r[2]))
+        for r in db_exec(
+            """
+            SELECT ei.id, ei.parent_id, t.path
+              FROM v3.expense_item ei
+              JOIN v3.mv_expense_item_tree t ON t.id = ei.id
+              JOIN v3.expense_item_form_type ft ON ft.expense_item_id = ei.id AND ft.form_type_code = :ft
+             WHERE ei.sheet = :sheet AND ei.direction = 'Support'
+               AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = ei.id)
+             ORDER BY t.path
+            """,
+            ft=form_type, sheet=sheet,
+        )
+    ]
+
+
+def _pick_items(db_exec, form_type, sheet):
+    """первая, последняя, две сестры, строка из другого ROOT (без повторов)."""
+    leaves = _ordered_leaves(db_exec, form_type, sheet)
+    first, last = leaves[0], leaves[-1]
+    chosen = [first, last]
+    by_parent = {}
+    for lf in leaves:
+        by_parent.setdefault(lf[1], []).append(lf)
+    sib = next(
+        (g for g in by_parent.values() if len(g) >= 2 and all(x not in chosen for x in g[:2])), None
+    )
+    sibs = sib[:2] if sib else []
+    assert len(sibs) == 2, "нет пары сестёр, отличных от первой/последней"
+    n_roots = len({lf[2][0] for lf in leaves})
+    other = None
+    if n_roots >= 2:
+        assert first[2][0] != last[2][0], "первая и последняя статьи в одном ROOT при нескольких ROOT"
+        # строка из ROOT, отличного от ROOT сестёр
+        other = next((lf for lf in leaves
+                      if lf[2][0] != sibs[0][2][0] and lf not in chosen + sibs), None)
+        assert other is not None, "нет строки из другого ROOT"
+    items = [first[0], last[0]] + [s[0] for s in sibs] + ([other[0]] if other else [])
+    return items, {"first": first[0], "last": last[0], "sibs": [s[0] for s in sibs],
+                   "other": other[0] if other else None}
+
+
+def _input_line_ids(rows):
+    return [r["data"]["line_id"] for r in rows if r["row_type"] == "INPUT"]
+
+
+def _prepare(client, headers, db_exec, case):
+    """Создаёт данные листа. Возвращает (form_id, units, batches)."""
+    form_type, sheet, direction, sections, year = CASES[case]
+    kind = "tree" if sheet in ("AHR", "CAP", "OPER") else sheet
+    if kind == "tree":
+        form_id = _new_form(client, headers, form_type, year)
+        items, names = _pick_items(db_exec, form_type, sheet)
+        lines = []
+        for n, item in enumerate(items):
+            lines.append(_add_line(client, headers, form_id, sheet, item, direction))
+        for n, line in enumerate(lines):
+            _patch_cell(client, headers, form_id, sheet, line, "plan.q1", 10 + n, direction)
+        _patch_cell(client, headers, form_id, sheet, lines[0], "header.justification", "обоснование", direction)
+        batch = list(lines)
+        if case == "F1-AHR-Support":
+            dev = _add_line(client, headers, form_id, sheet, items[0], "Development")
+            _patch_cell(client, headers, form_id, sheet, dev, "plan.q1", 77, "Development")
+            batch.append(dev)
+        get_rows = _get_sheet(client, headers, form_id, sheet, direction)
+        units = _input_line_ids(get_rows)
+        assert set(lines) <= set(units)
+        return form_id, units, [batch, batch + [lines[0]]]
+    if sheet in SATELLITES:
+        ids = []
+        for n in range(3):
+            r = client.post(
+                f"{API}/{SAT_FORM}/sheet/{sheet}/line",
+                json={"vsp_id": 1 + n % 3, "contract_number": f"C-{uuid.uuid4().hex[:6]}"},
+                headers=headers,
+            )
+            assert r.status_code == 200, r.text
+            sid = r.json()["result"][0]["data"]["id"]
+            _patch_cell(client, headers, SAT_FORM, sheet, sid, "plan.q1", 5 + n)
+            ids.append(sid)
+        get_rows = _get_sheet(client, headers, SAT_FORM, sheet)
+        units = [r["data"]["id"] for r in get_rows if r["row_type"] == "INPUT"]
+        assert set(ids) <= set(units)
+        return SAT_FORM, units, [ids, ids + [ids[0]]]
+    if sheet == "AHR_LIMIT":
+        extra = [r[0] for r in db_exec(
+            "SELECT id FROM v3.limit_template WHERE row_type = 'LEAF' AND id <> 13 ORDER BY id LIMIT 2")]
+        ids = [13] + extra
+        for n, i in enumerate(ids):
+            _patch_cell(client, headers, SAT_FORM, sheet, i, "qty_q1", 3 + n)
+        units = [r[0] for r in db_exec("SELECT id FROM v3.limit_template WHERE row_type = 'LEAF' ORDER BY id")]
+        return SAT_FORM, units, [ids, ids + [ids[0]]]
+    if sheet == "OTCH9F":
+        return SAT_FORM, *_prepare_otch9f(client, headers, db_exec)
+    raise AssertionError(case)
+
+
+def _prepare_otch9f(client, headers, db_exec):
+    items = [r[0] for r in db_exec(
+        """SELECT ei.id FROM v3.expense_item ei
+            WHERE ei.sheet = 'AHR'
+              AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = ei.id)
+              AND EXISTS (SELECT 1 FROM v3.expense_item_form_type t
+                           WHERE t.expense_item_id = ei.id AND t.form_type_code = 'FORM_2')
+              AND NOT EXISTS (SELECT 1 FROM v3.fixed_asset_report far
+                               WHERE far.budget_form_id = :f AND far.expense_item_id = ei.id)
+            ORDER BY ei.id LIMIT 2""", f=SAT_FORM)]
+    layout = [(items[0], "E1", 1), (items[0], "E1", 2), (items[0], "E1", 3), (items[0], "E2", 1),
+              (items[0], None, 1), (items[0], None, 2), (items[1], "E1", 1)]
+    ids = []
+    for item, name, month in layout:
+        ids.append(db_exec(
+            """INSERT INTO v3.fixed_asset_report
+                 (budget_form_id, expense_item_id, equipment_name, month,
+                  opening_qty_604, opening_amt_604, acquired_qty_604, acquired_amt_604,
+                  disposed_qty_604, disposed_amt_604,
+                  opening_qty_60415, opening_amt_60415, acquired_qty_60415, acquired_amt_60415)
+               VALUES (:f, :i, :n, :m, 10, 1000, 2, 200, 1, 100, 5, 500, 1, 50) RETURNING id""",
+            f=SAT_FORM, i=item, n=name, m=month)[0][0])
+    return ids, [ids, ids + [ids[0]]]
+
+
+def _check(db_exec, path, form_type, form_id, sheet, ids, direction, sections, label, must_have_input=False):
+    if path == "affected":
+        actual = affected_rows(db_exec, form_id, sheet, ids, direction, sections)
+        expected = reference_affected_rows(db_exec, form_id, sheet, ids, direction, sections, strip=True)
+    else:
+        actual = focus_builder_rows(db_exec, form_type, form_id, sheet, ids, direction, sections)
+        expected = reference_affected_rows(db_exec, form_id, sheet, ids, direction, sections, strip=False)
+    assert expected, f"{label}: empty reference"
+    assert_rows_equal(actual, expected, label)
+    if must_have_input:
+        assert any(_has_input(r, ids[0]) for r in actual), f"{label}: нет INPUT id={ids[0]} в результате"
+
+
+def _has_input(row, i):
+    import json
+    if row[0] != "INPUT":
+        return False
+    d = json.loads(row[3])
+    return d.get("line_id") == i or d.get("id") == i
+
+
+@pytest.mark.parametrize("path", ["affected", "builder"])
+@pytest.mark.parametrize("case", [c for c in CASES])
+def test_affected_rows_match_reference(client, headers, db_exec, path, case):
+    form_type, sheet, direction, sections, _ = CASES[case]
+    form_id, units, batches = _prepare(client, headers, db_exec, case)
+    assert units, "в листе нет строк"
+    is_tree = sheet in ("AHR", "CAP", "OPER")
+    for i in units:
+        _check(db_exec, path, form_type, form_id, sheet, [i], direction, sections, f"{case} id={i}",
+               must_have_input=sheet != "AHR_LIMIT" and sheet != "OTCH9F")
+    if case == "F1-AHR-Support" and path == "affected":
+        dev = batches[0][-1]
+        rows = affected_rows(db_exec, form_id, sheet, [dev], direction, sections)
+        assert not any(_has_input(r, dev) for r in rows), "Development INPUT попал в результат Support"
+    del is_tree
+    for n, batch in enumerate(batches):
+        _check(db_exec, path, form_type, form_id, sheet, batch, direction, sections, f"{case} batch#{n}")
+
+
+EDIT_CASES = ["F1-AHR-Support", "F2-AHR", "F2-AHR_RENT", "F2-AHR_LIMIT", "F2-OTCH9F", "delta"]
+VF_SIG = "v3.v_form_view(integer,character varying,text[],character varying,integer)"
+_CALLS = "pg_stat_get_xact_function_calls(CAST(CAST('" + VF_SIG + "' AS regprocedure) AS oid))"
+
+
+@pytest.mark.parametrize("case", EDIT_CASES)
+def test_edit_path_does_not_call_v_form_view(client, headers, db_exec, case):
+    # R-2: роль perf — суперпользователь, track_functions доступен; тест не пропускается.
+    assert db_exec("SELECT rolsuper FROM pg_roles WHERE rolname = current_user")[0][0] is True, (
+        "для track_functions нужна суперпользовательская роль"
+    )
+    if case == "delta":
+        r = client.post(
+            f"{API}/{SAT_FORM}/sheet/AHR_RENT/line",
+            json={"vsp_id": 1, "contract_number": f"C-{uuid.uuid4().hex[:6]}"},
+            headers=headers,
+        )
+        assert r.status_code == 200, r.text
+        rent_id = r.json()["result"][0]["data"]["id"]
+        _patch_cell(client, headers, SAT_FORM, "AHR_RENT", rent_id, "plan.q1", 100)
+        call, params = ("SELECT count(*) FROM v3.get_form2_ahr_satellite_delta(4, 'AHR_RENT', :i)"), {"i": rent_id}
+    else:
+        form_type, sheet, direction, sections, _ = CASES[case]
+        form_id, units, batches = _prepare(client, headers, db_exec, case)
+        call = ("SELECT count(*) FROM v3._form_affected_rows(:f, :s, CAST(:ids AS int[]), "
+                "CAST(:d AS varchar), CAST(:sec AS text[]))")
+        params = dict(f=form_id, s=sheet, ids=batches[0], d=direction, sec=sections)
+    db_exec("SET LOCAL track_functions = 'all'")
+    c0 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0]
+    db_exec("SELECT count(*) FROM v3.v_form_view(4, 'AHR_RENT', NULL, NULL)")
+    c1 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0]
+    assert c1 - c0 == 1, "положительный контроль: счётчик не отслеживает v_form_view"
+    n0 = c1
+    db_exec(call, **params)
+    n1 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0]
+    assert n1 - n0 == 0, f"v_form_view вызван {n1 - n0} раз на пути правки"
+
+
+def test_empty_focus_returns_nothing(db_exec):
+    for s in FORM2_SHEETS:
+        rows = db_exec(
+            "SELECT * FROM v3.v_form2_view(4, :s, CAST(NULL AS text[]), CAST('{}' AS int[]))", s=s)
+        assert rows == [], f"v_form2_view пуст для {s}: {len(rows)}"
+    rows = db_exec(
+        "SELECT * FROM v3.v_form1_view(1, 'AHR', CAST(NULL AS text[]), 'Support', CAST('{}' AS int[]))")
+    assert rows == []
+    assert db_exec("SELECT * FROM v3._form_affected_rows(4, 'AHR', CAST('{}' AS int[]))") == []
+    assert db_exec("SELECT * FROM v3.get_form2_ahr_satellite_delta(4, 'AHR_RENT', -1)") == []
+
+
+def _patch_cells(client, headers, form_id, sheet, changes, direction=None):
+    r = client.patch(
+        f"{API}/{form_id}/sheet/{sheet}/cells{_qs(direction)}", json={"changes": changes}, headers=headers)
+    assert r.status_code == 200, r.text
+    return r.json()["result"]
+
+
+def test_patch_cells_two_branches_matches_get(client, headers, db_exec):
+    form_id = _new_form(client, headers, "FORM_2", 2054)
+    items, names = _pick_items(db_exec, "FORM_2", "AHR")
+    l1 = _add_line(client, headers, form_id, "AHR", names["first"])
+    l2 = _add_line(client, headers, form_id, "AHR", names["last"])
+    patch = _patch_cells(client, headers, form_id, "AHR", [
+        {"line_id": l1, "column": "plan.q1", "value": 11},
+        {"line_id": l2, "column": "plan.q1", "value": 22},
+    ])
+    get_rows = _get_sheet(client, headers, form_id, "AHR")
+    assert_patch_rows_match_sheet(patch, get_rows)
+    assert {r["data"]["line_id"] for r in patch if r["row_type"] == "INPUT"} == {l1, l2}
+    keys = [row_key(r) for r in patch]
+    assert len(keys) == len(set(keys))
+    roots = {
+        db_exec("SELECT path[1] FROM v3.mv_expense_item_tree WHERE id = :i", i=i)[0][0]
+        for i in (names["first"], names["last"])
+    }
+    assert len(roots) == 2, "строки l1/l2 в одном ROOT"
+    root_eids = {r["data"]["header"]["expense_item_id"] for r in patch if r["row_type"] == "ROOT"}
+    assert len(root_eids) >= 2, root_eids
+
+
+@pytest.mark.parametrize(
+    "form_type,direction,year",
+    [("FORM_2", None, 2055), ("FORM_1", "Support", 2056)],
+    ids=["FORM_2-AHR", "FORM_1-AHR-Support"],
+)
+def test_patch_after_add_and_delete_matches_get(client, headers, db_exec, form_type, direction, year):
+    form_id = _new_form(client, headers, form_type, year)
+    leaves = _ordered_leaves(db_exec, form_type, "AHR")
+    step = max(len(leaves) // 5, 1)
+    items = [leaves[i * step][0] for i in range(5)]
+    lines = [_add_line(client, headers, form_id, "AHR", it, direction) for it in items]
+    get0 = _get_sheet(client, headers, form_id, "AHR", direction)
+    order = [r["data"]["line_id"] for r in get0 if r["row_type"] == "INPUT" and r["data"]["line_id"] in lines]
+    victim, below = order[2], order[3]
+    r = client.delete(f"{API}/{form_id}/sheet/AHR/line/{victim}{_qs(direction)}", headers=headers)
+    assert r.status_code == 200, r.text
+    patch = _patch_cell(client, headers, form_id, "AHR", below, "plan.q1", 42, direction)
+    get_rows = _get_sheet(client, headers, form_id, "AHR", direction)
+    assert_patch_rows_match_sheet(patch, get_rows)
+
+    def ident(row):
+        if row["row_type"] == "INPUT":
+            return ("L", row["data"]["line_id"])
+        return ("N", row["data"]["header"]["expense_item_id"], row["row_type"])
+
+    by_ident = {ident(r): r["sort_order"] for r in get_rows if r["row_type"] != "META"}
+    for row in patch:
+        assert row["sort_order"] == by_ident[ident(row)], ident(row)
+
+
+def test_patch_form1_development_matches_get(client, headers, db_exec):
+    form_id = _new_form(client, headers, "FORM_1", 2057)
+    items, names = _pick_items(db_exec, "FORM_1", "AHR")
+    sup = _add_line(client, headers, form_id, "AHR", names["first"], "Support")
+    dev1 = _add_line(client, headers, form_id, "AHR", names["first"], "Development")
+    dev2 = _add_line(client, headers, form_id, "AHR", names["last"], "Development")
+    _patch_cell(client, headers, form_id, "AHR", sup, "plan.q1", 5, "Support")
+    _patch_cell(client, headers, form_id, "AHR", dev2, "plan.q1", 6, "Development")
+    patch = _patch_cell(client, headers, form_id, "AHR", dev1, "plan.q1", 7, "Development")
+    assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR", "Development"))
+    assert {r["data"]["line_id"] for r in patch if r["row_type"] == "INPUT"} == {dev1}
+
+
+def test_patch_otch9f_returns_all_months_of_equipment(client, headers, db_exec):
+    ids, _ = _prepare_otch9f(client, headers, db_exec)
+    e1_m1 = ids[0]
+    item = db_exec("SELECT expense_item_id FROM v3.fixed_asset_report WHERE id = :i", i=e1_m1)[0][0]
+    before = _get_sheet(client, headers, SAT_FORM, "OTCH9F")
+    patch = _patch_cell(client, headers, SAT_FORM, "OTCH9F", e1_m1, "b_604.acquired_qty", 9)
+    get_rows = _get_sheet(client, headers, SAT_FORM, "OTCH9F")
+    assert_patch_rows_match_sheet(patch, get_rows)
+
+    def of_equipment(rows):
+        return {row_key(r): r for r in rows
+                if r["row_type"] != "META" and r["data"].get("expense_item_id") == item
+                and r["data"].get("equipment_name") == "E1"}
+
+    expected = of_equipment(get_rows)
+    assert len(expected) >= 3
+    assert {row_key(r) for r in patch} == set(expected)
+    prev = of_equipment(before)
+    changed = [k for k in expected if expected[k]["data"] != prev[k]["data"]]
+    assert len(changed) >= 2, "остатки следующих месяцев не пересчитаны"
+
+
+def _sort_key_candidate(db_exec):
+    rows = db_exec(
+        """
+        SELECT b.id, b.parent_id
+          FROM v3.mv_expense_item_tree b
+          JOIN v3.mv_expense_item_tree a ON a.id = b.parent_id
+          JOIN v3.expense_item_form_type fb ON fb.expense_item_id = b.id AND fb.form_type_code = 'FORM_2'
+          JOIN v3.expense_item_form_type fa ON fa.expense_item_id = a.id AND fa.form_type_code = 'FORM_2'
+         WHERE b.sheet = 'AHR' AND a.depth = 3 AND b.depth = 4
+           AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = b.id)
+           AND NOT EXISTS (SELECT 1 FROM v3.budget_line bl WHERE bl.id = b.id)
+         ORDER BY b.id LIMIT 1
+        """
+    )
+    return rows[0][0] if rows else None
+
+
+def test_sort_key_tie_node_before_line(client, headers, db_exec):
+    child = _sort_key_candidate(db_exec)
+    if child is None:
+        pytest.skip("нет статьи AHR глубины 3 с дочерней глубины 4 и свободным id в budget_line")
+    form_id = _new_form(client, headers, "FORM_2", 2058)
+    db_exec(
+        "INSERT INTO v3.budget_line (id, budget_form_id, expense_item_id, name) "
+        "VALUES (:i, :f, :i, 'tie')", i=child, f=form_id)
+    g1 = _get_sheet(client, headers, form_id, "AHR")
+    g2 = _get_sheet(client, headers, form_id, "AHR")
+    assert [row_key(r) for r in g1] == [row_key(r) for r in g2]
+    idx = next(n for n, r in enumerate(g1) if r["row_type"] == "INPUT" and r["data"]["line_id"] == child)
+    assert idx > 0
+    prev = g1[idx - 1]
+    assert prev["row_type"] in TREE_TYPES and prev["data"]["header"]["expense_item_id"] == child, prev
+    patch = _patch_cell(client, headers, form_id, "AHR", child, "plan.q1", 8)
+    node = next((r for r in patch if r["row_type"] in TREE_TYPES
+                 and r["data"]["header"]["expense_item_id"] == child), None)
+    assert node is not None, "узел дочерней статьи отсутствует в ответе"
+    line_row = next(r for r in patch if r["row_type"] == "INPUT" and r["data"]["line_id"] == child)
+    assert node["sort_order"] < line_row["sort_order"]
+    assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR"))
+
+
+def _delta_equals_get(db_exec, sheet, row_id):
+    delta = [tuple(r) for r in db_exec(
+        "SELECT row_type, depth, sort_order, data::text "
+        "FROM v3.get_form2_ahr_satellite_delta(4, :s, :i)", s=sheet, i=row_id)]
+    full = {r[2]: tuple(r) for r in db_exec(
+        "SELECT row_type, depth, sort_order, data::text "
+        "FROM v3.v_form_view(4, 'AHR', NULL, 'Support')")}
+    assert delta, "дельта пуста"
+    assert len({r[2] for r in delta}) == len(delta), "sort_order в дельте не уникален"
+    for row in delta:
+        assert row[2] in full, f"sort_order {row[2]} нет в полном AHR"
+        assert_rows_equal([row], [full[row[2]]], f"delta {sheet} sort_order={row[2]}")
+    return delta
+
+
+def test_ahr_delta_matches_get_after_rent_edit(client, headers, db_exec):
+    r = client.post(
+        f"{API}/{SAT_FORM}/sheet/AHR_RENT/line",
+        json={"vsp_id": 1, "contract_number": f"C-{uuid.uuid4().hex[:6]}"},
+        headers=headers,
+    )
+    assert r.status_code == 200, r.text
+    rent_id = r.json()["result"][0]["data"]["id"]
+    _patch_cell(client, headers, SAT_FORM, "AHR_RENT", rent_id, "plan.q1", 100)
+    delta = _delta_equals_get(db_exec, "AHR_RENT", rent_id)
+    import json
+    assert any(
+        row[0] == "INPUT" and (json.loads(row[3]).get("plan") or {}).get("q1") == 100 for row in delta
+    ), "в дельте нет INPUT с plan.q1 = 100"
+
+
+def test_ahr_delta_includes_sub_sub_item_for_depth4_limit(client, headers, db_exec):
+    rows = db_exec(
+        """
+        SELECT lt.id, t.id
+          FROM v3.limit_template lt
+          JOIN v3.mv_expense_item_tree t ON t.num_group_id = lt.expense_item_code
+          JOIN v3.expense_item_form_type ft ON ft.expense_item_id = t.id AND ft.form_type_code = 'FORM_2'
+         WHERE lt.row_type = 'LEAF' AND t.sheet = 'AHR' AND t.depth = 4
+         ORDER BY lt.id LIMIT 1
+        """
+    )
+    if not rows:
+        pytest.skip("нет LEAF шаблона лимитов с кодом статьи AHR глубины 4")
+    lt_id, eid = rows[0]
+    assert lt_id == 13, f"ожидался LEAF 13, найден {lt_id}"
+    _patch_cell(client, headers, SAT_FORM, "AHR_LIMIT", lt_id, "qty_q1", 4)
+    delta = _delta_equals_get(db_exec, "AHR_LIMIT", lt_id)
+    import json
+    assert any(
+        row[0] == "SUB_SUB_ITEM" and json.loads(row[3])["header"]["expense_item_id"] == eid for row in delta
+    ), "в дельте нет SUB_SUB_ITEM статьи глубины 4"
+
+
+JIT_OFF_FUNCS = ("v_form2_sheet_sections", "v_form1_sheet_sections",
+                 "_form_affected_rows", "get_form2_ahr_satellite_delta")
+
+
+def test_jit_off_on_builders_and_edit_path(db_exec):
+    rows = db_exec(
+        """SELECT p.proname, COALESCE(p.proconfig, '{}') FROM pg_proc p
+             JOIN pg_namespace n ON n.oid = p.pronamespace
+            WHERE n.nspname = 'v3' AND p.proname = ANY(CAST(:names AS text[]))""",
+        names=list(JIT_OFF_FUNCS))
+    assert {r[0] for r in rows} == set(JIT_OFF_FUNCS)
+    missing = [r[0] for r in rows if "jit=off" not in r[1]]
+    assert not missing, f"нет SET jit = off: {missing}"

api/tests/integration/test_upd_form_cells_subtree_0037_guard.py

@@ -0,0 +1,47 @@
+"""AURORA-1850: защита правки 0037 (LEAF-лимиты АХР и сопоставление num_group_id) после 0051.
+
+Откат 0037 — нарушение корректности: тесты должны падать, если тело функций вернулось к виду до 0037.
+"""
+
+import re
+from decimal import Decimal
+
+import pytest
+
+from test_upd_form_cells_full_rows import (  # noqa: F401  (фикстуры и хелперы)
+    _get_sheet,
+    _patch_cell,
+    db_exec,
+    headers,
+)
+
+
+def test_v_form2_view_keeps_0037_limit_formula(client, headers, db_exec):
+    # LEAF «- принтеры», limit_template.id = 13, код 204620030 (0013_form2_limit.sql:1707)
+    _patch_cell(client, headers, 4, "AHR_LIMIT", 13, "qty_q1", 4)
+    row = next(r for r in _get_sheet(client, headers, 4, "AHR_LIMIT") if r["data"].get("id") == 13)
+    assert Decimal(str(row["data"]["amount_q1"])) == Decimal(4) * Decimal("0.57") * 10900 / 4
+    src = db_exec("SELECT pg_get_functiondef(CAST(:s AS regprocedure))",
+                  s="v3.v_form2_view(integer,character varying,text[],integer[])")[0][0]
+    assert "_form2_ahr_limit_amount" in src
+
+
+OLD_DEPTH3_MATCHER = re.compile(r"ei\.depth\s*=\s*3\s+AND\s*\(")  # вид до 0037
+
+
+def test_ahr_delta_keeps_0037_num_group_matching(client, headers, db_exec):
+    found = db_exec(
+        """SELECT lt.id FROM v3.limit_template lt
+             JOIN v3.expense_item ei ON ei.num_group_id = lt.expense_item_code
+              AND ei.sheet = 'AHR' AND ei.direction = 'Support' AND ei.depth = 4
+            WHERE lt.row_type = 'LEAF' ORDER BY lt.id LIMIT 1""")
+    if not found:
+        pytest.skip("нет LEAF шаблона с кодом = num_group_id статьи AHR глубины 4")
+    lt_id = found[0][0]
+    _patch_cell(client, headers, 4, "AHR_LIMIT", lt_id, "qty_q1", 4)
+    delta = db_exec("SELECT row_type FROM v3.get_form2_ahr_satellite_delta(4, 'AHR_LIMIT', :i)", i=lt_id)
+    assert any(r[0] == "INPUT" for r in delta), "дельта АХР пуста — откат сопоставления 0037"
+    src = db_exec("SELECT pg_get_functiondef(CAST(:s AS regprocedure))",
+                  s="v3.get_form2_ahr_satellite_delta(integer,character varying,integer)")[0][0]
+    assert "ei.num_group_id = lt.expense_item_code" in src
+    assert not OLD_DEPTH3_MATCHER.search(src)

openspec/changes/aurora-1850-upd-form-cells-subtree/design.md

@@ -278,3 +278,11 @@
 | 2026-10-28 | План: `v_form1_view` принимает только AHR/CAP/OPER, на других листах — `RAISE EXCEPTION` | SMETA FORM_1 собирается `v_form_view` до вызова; явная ошибка лучше пустого результата | Нет: функция внутренняя |
 | 2026-10-28 | План: раскладка тестов — `test_upd_form_cells_subtree.py` (эквивалентность, `sort_order`, дельта, JIT) и `test_upd_form_cells_subtree_0037_guard.py`; SQL-хелперы в `sheet_contract.py`; HTTP-хелперы и фикстуры импортируются из `test_upd_form_cells_full_rows.py` | Оставлено исполнителю (design.md); решено в плане, чтобы задачи 2, 8, 9 ссылались на одни имена | Нет: только раскладка |
 | 2026-10-28 | План: защита 0037 для дельты проверяет `pg_get_functiondef` регуляркой `ei\.depth\s*=\s*3\s+AND\s*\(` (вид до 0037), а не литералом | В теле до 0037 между `3` и `AND (` стоят переводы строки и отступы; литерал не поймал бы откат | Нет |
+| 2026-10-28 | Контроллер (R-1, R-10): скелет сателлитов сохраняет те же JOIN (`budget_line`, `vsp`), что GET (0033); JOIN `vsp` убирать только при доказанных NOT NULL и FK. Одобрено главным контроллером | Рулинг уточняет D4: внутренний JOIN может отбрасывать строки и сдвигать ранг | Тихо неверный `sort_order` сателлитов; ловят тесты эквивалентности и снимок GET |
+| 2026-10-28 | Контроллер (R-2): роль `perf` в `perf-pg` и `dfip-pg15` — суперпользователь, `SET LOCAL track_functions` доступен; тест «0 вложенных `v_form_view`» не пропускается. Skip этого теста считается провалом | Иначе критерий «нет полной сборки на пути правки» остаётся непроверенным до замера | Если на другой роли тест скипнется, критерий проверяется только замером стенда |
+| 2026-10-28 | Контроллер (R-3): дополнительные БД `dfip_1850_base` и `dfip_1850_snap` в `perf-pg` одобрены; допустим только префикс `dfip_1850*`, БД `perf` не трогается | Быстрый сброс и снимок GET (см. решения плана) | Две лишние БД в контейнере |
+| 2026-10-28 | Контроллер (R-4): `before-0051.sql` коммитится в каталог change | База ревью и отката (D8) | Лишний файл в change |
+| 2026-10-28 | Контроллер (R-14): подпись коммитов `Co-Authored-By: Claude Opus 5.5` по плану и правилам волны 2, хотя системное напоминание сессии называет Sonnet 5.5 | Явная инструкция пользователя приоритетнее напоминания | Неверное имя модели в трейлере коммитов |
+| 2026-10-28 | Исполнитель Task 1 (T1.2): посев снимка GET создаёт строки FORM_1 AHR/CAP и для Support, и для Development; снимок включает и фикстурные формы 1 и 4 | `expense_item.direction` везде Support, иначе ветка Development в снимке пуста | Снимок содержит чуть больше строк, чем описано в плане; на проверку корректности не влияет |
+| 2026-10-28 | Исполнитель Task 3 (T3.1): ранг скелета присоединяется внешним `JOIN sk` поверх `UNION` построителя, а не внутри веток; ключ скелета один bigint (узел = `-id`, строка = `lid`) | Вариант с `JOIN sk` в каждой ветке обрушил оценку строк, планировщик выбрал nested loop, GET AHR вырос со 105 до 203 мс; внешний JOIN даёт около 114 мс (+8 %) | Хрупкость плана на других объёмах; ловит замер GET на стенде |
+| 2026-10-28 | Исполнитель Task 4 (T4.4): JOIN `vsp` в скелете сателлитов оставлен, хотя `vsp_id` и `line_id` NOT NULL с FK (доказано по схеме); OTCH9F: перестановка строк с равным `(expense_item_id, month)` по tie-break `id` в снимке GET — 4 строки формы 26 (id 4/1 месяц 1, id 6/2 месяц 2), содержимое строк одинаково | Рулинг R-1; порядок при равных ключах раньше не был определён (D6) | Если фронт опирался на прежний случайный порядок равных ключей, порядок двух пар строк изменится |