fix(budget): 年度行(budget_month=0)不回落 1 月 + 同族 R1/R2/R3

- R-A2 sync-cash-plans: 年度行形态判定(_is_annual_budget_row)跳过 + skipped_annual 披露
- R2 POST /plans 期间形态校验(YYYY-MM / YYYY-00 显式接受)
- R3 budget_roll_forward 排除年度行
- R1 cash_classify 归类路径年度行判据 + 披露
- 新增 2 个回归件(14+10 例)

仅含本族 hunk(禁整文件提交): 开发区其余 14 hunk 属他会话在途件,未纳入。

backend/app/api/budget.py

@@ -1,6 +1,8 @@
 """预算管理 API — 管理会计OS
 预算值的CRUD、自动分解、版本管理
 """
+import re
+
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 from sqlalchemy import func, or_
@@ -96,7 +98,19 @@ def create_budget_plan(
     if not kpi:
         raise HTTPException(404, "KPI不存在")
 
-    year, month = period.split("-")
+    # 期间形态校验（年度行同族残余 2026-10-13：YYYY-00 显式接受为年度行，YYYY 提示补全形态）
+    if not isinstance(period, str):
+        raise HTTPException(400, "预算期间格式不正确：请使用 YYYY-MM（月 01–12）或 YYYY-00（年度）")
+    if re.fullmatch(r"\d{4}-(0[1-9]|1[0-2])", period):
+        # 月度形态
+        year, month = period.split("-")[0], int(period.split("-")[1])
+    elif re.fullmatch(r"\d{4}-00", period):
+        # 年度形态（系统内部既有年度行，如 apply-method / 年度分解写入）：显式接受
+        year, month = period.split("-")[0], 0
+    elif re.fullmatch(r"\d{4}", period):
+        raise HTTPException(400, f"预算期间缺少月份：年度预算请使用 {period}-00 形态（例：2027-00），月度预算使用 YYYY-MM")
+    else:
+        raise HTTPException(400, "预算期间格式不正确：请使用 YYYY-MM（月 01–12）或 YYYY-00（年度）")
     version = data.get("version", "v1.0")
     map_id = data.get("map_id")  # 预算归属地图（2026-08-27: 按地图隔离）
 
@@ -590,10 +604,14 @@ def budget_roll_forward(
              .filter(BudgetPlan.status == "active", KPIDefinition.entity_id == entity_id)
              .all())
 
-    # 按KPI分组
+    # 按KPI分组（年度行排除 2026-10-13：budget_month=0 哨兵不参与月度滚动，仅年度行的 KPI 不删不产新行）
     from collections import defaultdict
     kpi_plans = defaultdict(list)
+    annual_rows_ignored = 0
     for p in plans:
+        if _is_annual_budget_row(p.period, p.budget_month):
+            annual_rows_ignored += 1
+            continue
         kpi_plans[p.kpi_id].append(p)
 
     rolled_kpis = []
@@ -653,6 +671,7 @@ def budget_roll_forward(
         "rolled_kpis": rolled_kpis,
         "current_month": f"{current_year}-{current_month:02d}",
         "rolling_months": val.get("rolling_months", 12),
+        "annual_rows_ignored": annual_rows_ignored,
     }
 
 
@@ -1631,6 +1650,17 @@ def diff_budget_versions(
 # 预算↔现金流联动（断点修复#1, 2026-08-27）
 # ════════════════════════════════════════════════════════════
 
+def _is_annual_budget_row(period: Optional[str], budget_month: Optional[int]) -> bool:
+    """年度预算行判定（形态驱动）：budget_month == 0（年度哨兵），
+    或 period 命中年度形态 YYYY / YYYY-00。"""
+    if budget_month == 0:
+        return True
+    if period is None:
+        return False
+    p = str(period).strip()
+    return re.fullmatch(r"\d{4}(?:-00)?", p) is not None
+
+
 @router.post("/sync-cash-plans")
 def sync_cash_plans(
     entity_id: int = Depends(get_entity_id),
@@ -1683,7 +1713,13 @@ def sync_cash_plans(
     kpis = {k.id: k for k in db.query(KPIDefinition).filter(KPIDefinition.id.in_(kpi_ids)).all()} if kpi_ids else {}
 
     created, updated, unclassified_count = 0, 0, 0
+    skipped_annual = 0
     for b in budgets:
+        # 年度汇总行（budget_month=0 哨兵 / period=YYYY 或 YYYY-00）不联动现金流:
+        # CashPlan 无年度维度, 跳过以免全年额回落成 YYYY-01 污染按月分桶
+        if _is_annual_budget_row(b.period, b.budget_month):
+            skipped_annual += 1
+            continue
         kpi = kpis.get(b.kpi_id)
         if not kpi:
             continue
@@ -1736,8 +1772,12 @@ def sync_cash_plans(
             ))
             created += 1
     db.commit()
+    message = f"现金流联动完成: 新建{created}条, 更新{updated}条, 待分类{unclassified_count}条"
+    if skipped_annual:
+        message += f", 跳过年度行{skipped_annual}条"
     return {
-        "message": f"现金流联动完成: 新建{created}条, 更新{updated}条, 待分类{unclassified_count}条",
+        "message": message,
         "created": created, "updated": updated,
         "unclassified_count": unclassified_count,
+        "skipped_annual": skipped_annual,
     }

backend/app/api/cash_classify.py

@@ -196,40 +196,53 @@ def classify_unclassified(
     item.status = "classified"
     item.resolved_at = datetime.now()
 
-    # ③ 联动生成 CashPlan（有期间和预算值时）
+    # ③ 联动生成 CashPlan（有期间和预算值时；年度行跳过，不落月度计划）
+    from app.api.budget import _is_annual_budget_row  # 函数内局部导入，规避 import 顺序/循环风险
     plan_created = False
+    skipped_annual = False
     if item.period and item.budget_value is not None:
-        try:
-            year, month = int(item.period.split("-")[0]), int(item.period.split("-")[1])
-            plan_date = datetime(year, month, 1)
-        except Exception:
-            plan_date = None
-        if plan_date:
-            existing_plan = db.query(CashPlan).filter(
-                CashPlan.entity_id == entity_id,
-                CashPlan.related_kpi_id == item.kpi_id,
-                CashPlan.plan_type == plan_type,
-                CashPlan.plan_date == plan_date,
-            ).first()
-            if not existing_plan:
-                db.add(CashPlan(
-                    entity_id=entity_id,
-                    plan_type=plan_type,
-                    related_kpi_id=item.kpi_id,
-                    amount=item.budget_value,
-                    plan_date=plan_date,
-                    description=f"待分类队列归类: {item.kpi_name or ''}",
-                    status="pending",
-                    source="budget_sync",
-                ))
-                plan_created = True
+        # 先按形态解析月份（不裸索引 split，避免 'YYYY' / 'YYYY-00' 抛异常被静默吞掉）
+        parts = item.period.split("-") if isinstance(item.period, str) else []
+        month = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else None
+        if _is_annual_budget_row(item.period, month):
+            # 年度行（YYYY / YYYY-00 / budget_month=0）不得落成月度现金流计划
+            skipped_annual = True
+        else:
+            try:
+                year, month = int(item.period.split("-")[0]), int(item.period.split("-")[1])
+                plan_date = datetime(year, month, 1)
+            except Exception:
+                plan_date = None
+            if plan_date:
+                existing_plan = db.query(CashPlan).filter(
+                    CashPlan.entity_id == entity_id,
+                    CashPlan.related_kpi_id == item.kpi_id,
+                    CashPlan.plan_type == plan_type,
+                    CashPlan.plan_date == plan_date,
+                ).first()
+                if not existing_plan:
+                    db.add(CashPlan(
+                        entity_id=entity_id,
+                        plan_type=plan_type,
+                        related_kpi_id=item.kpi_id,
+                        amount=item.budget_value,
+                        plan_date=plan_date,
+                        description=f"待分类队列归类: {item.kpi_name or ''}",
+                        status="pending",
+                        source="budget_sync",
+                    ))
+                    plan_created = True
 
     db.commit()
+    message = f"已归类为 {plan_type}" + (" 并生成现金流计划" if plan_created else "")
+    if skipped_annual:
+        message += "（年度行，跳过月度现金流计划）"
     return {
-        "message": f"已归类为 {plan_type}" + (" 并生成现金流计划" if plan_created else ""),
+        "message": message,
         "plan_type": plan_type,
         "rule_created": not existing_rule,
         "plan_created": plan_created,
+        "skipped_annual": skipped_annual,
     }
 
 

backend/tests/test_budget_annual_row_residuals_r1r2r3_20261013.py

@@ -0,0 +1,187 @@
+"""年度行同族残余 R1/R2/R3 修复回归
+
+来源单：yanxue-20261013-budget-family-residuals-r1r2r3-fullstack（A 侧派单 P2；
+上游 = yanxue-20261013-budget-sync-cashplan-annual-row-fix-fullstack 回执 §五）
+
+约定（形态驱动，不读 remark/version；不改 DDL；月度行行为不变）：
+  R1 `POST /cash-unclassified/{id}/classify`：队列条目 period 为年度形态（YYYY / YYYY-00）
+     ⇒ 不生成月度 CashPlan，且**显式披露**（响应 `skipped_annual: true` + message 含「年度行」）；
+     月度形态行为不变（仍生成 CashPlan）。
+  R2 `POST /budget/plans`：`YYYY-MM`（月 01–12）与年度形态（`YYYY` / `YYYY-00` ⇒ budget_month=0）
+     ⇒ 200；其余形态（`2026-13` / 无 `-` 的非法串 / 非数字）⇒ **400 友好提示**（不得 500）。
+  R3 `POST /budget/roll-forward`：年度行**不参与**排序与「近三月均值」，也**不被删除**；
+     仅年度行的 KPI 不产出新行（fail-closed）。
+"""
+from datetime import datetime
+
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
+from app.models.budget_plan import BudgetPlan
+from app.models import CashPlan, CashPlanUnclassified
+
+
+BASE = "/api/cma/budget"
+
+
+def _row(db: Session, kpi_id: int, period: str, year: int, month: int, value: float) -> BudgetPlan:
+    bp = BudgetPlan(entity_id=1, kpi_id=kpi_id, period=period, budget_value=value,
+                    budget_year=year, budget_month=month, version="v1.0", status="active",
+                    remark="E1-ANNUAL" if month == 0 else "")
+    db.add(bp)
+    db.commit()
+    db.refresh(bp)
+    return bp
+
+
+# ── R1：待分类队列归类不得把年度行落成月度 CashPlan，且须披露 ──────────────
+
+class TestR1ClassifyAnnualRow:
+    def _queue_item(self, db: Session, kpi_id: int, period: str, value: float = 9999.0) -> CashPlanUnclassified:
+        it = CashPlanUnclassified(entity_id=1, kpi_id=kpi_id, kpi_name="神秘专项",
+                                  period=period, budget_value=value,
+                                  reason="未匹配任何分类规则", status="pending")
+        db.add(it)
+        db.commit()
+        db.refresh(it)
+        return it
+
+    def test_annual_period_00_not_materialized_and_disclosed(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R1_ANN_00", kpi_name="神秘专项")
+        item = self._queue_item(db, kpi.id, "2027-00")
+
+        resp = client.post(f"{BASE}/cash-unclassified/{item.id}/classify",
+                           headers=auth_header(token), json={"plan_type": "receive"})
+        assert resp.status_code == 200, resp.text[:300]
+        data = resp.json()
+        assert data.get("skipped_annual") is True, f"未披露年度行跳过: {data}"
+        assert "年度行" in data["message"]
+        assert db.query(CashPlan).count() == 0
+
+    def test_year_only_period_not_materialized_no_500(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R1_ANN_YR", kpi_name="神秘专项")
+        item = self._queue_item(db, kpi.id, "2028")
+
+        resp = client.post(f"{BASE}/cash-unclassified/{item.id}/classify",
+                           headers=auth_header(token), json={"plan_type": "pay"})
+        assert resp.status_code == 200, resp.text[:300]
+        assert resp.json().get("skipped_annual") is True
+        assert db.query(CashPlan).count() == 0
+
+    def test_monthly_period_behavior_unchanged(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R1_MON", kpi_name="神秘专项")
+        item = self._queue_item(db, kpi.id, "2026-06", value=300.0)
+
+        resp = client.post(f"{BASE}/cash-unclassified/{item.id}/classify",
+                           headers=auth_header(token), json={"plan_type": "receive"})
+        assert resp.status_code == 200
+        assert resp.json()["plan_created"] is True
+        plans = db.query(CashPlan).all()
+        assert len(plans) == 1
+        assert plans[0].plan_date == datetime(2026, 6, 1)
+        assert plans[0].amount == 300.0
+
+
+# ── R2：POST /plans period 形态校验（非法 400，年度形态显式接受） ─────────────
+
+class TestR2PlanPeriodForm:
+    def _post(self, client: TestClient, token: str, kpi_id: int, period: str, value: float = 100.0):
+        return client.post(f"{BASE}/plans", headers=auth_header(token),
+                           json={"kpi_id": kpi_id, "period": period, "budget_value": value})
+
+    def test_year_only_period_400_not_500(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R2_KPI")
+        resp = self._post(client, token, kpi.id, "2027")
+        assert resp.status_code in (200, 400), f"不得 500/抛异常: {resp.status_code}"
+        assert resp.status_code == 400, f"仅年份形态按年度行规则应显式处理，现: {resp.status_code} {resp.text[:200]}"
+
+    def test_month_13_400(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R2_KPI13")
+        resp = self._post(client, token, kpi.id, "2026-13")
+        assert resp.status_code == 400, f"月 13 应 400，现 {resp.status_code}"
+
+    def test_bad_form_400(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R2_KPIBAD")
+        resp = self._post(client, token, kpi.id, "2026/06")
+        assert resp.status_code == 400, f"非法形态应 400（不得 500），现 {resp.status_code}"
+
+    def test_monthly_period_unchanged(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R2_KPIOK")
+        resp = self._post(client, token, kpi.id, "2026-06", value=1000.0)
+        assert resp.status_code == 200
+        bp = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).one()
+        assert (bp.budget_year, bp.budget_month, bp.period) == (2026, 6, "2026-06")
+
+    def test_annual_form_00_accepted_with_month_zero(self, client: TestClient, db: Session):
+        """年度形态 `YYYY-00` 显式接受（系统内部年度行既有形态），落 budget_month=0"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R2_KPIANN")
+        resp = self._post(client, token, kpi.id, "2027-00", value=1800.0)
+        assert resp.status_code == 200, resp.text[:300]
+        bp = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).one()
+        assert (bp.budget_year, bp.budget_month, bp.period) == (2027, 0, "2027-00")
+
+
+# ── R3：滚动延展不得把年度行当月份（排序 / 近三月均值 / 删除面） ─────────────
+
+class TestR3RollForwardAnnualRows:
+    def _rolling(self, client: TestClient, token: str) -> dict:
+        client.post(f"{BASE}/config", headers=auth_header(token),
+                    json={"mode": "rolling", "rolling_months": 12})
+        resp = client.post(f"{BASE}/roll-forward", headers=auth_header(token))
+        assert resp.status_code == 200, resp.text[:300]
+        return resp.json()
+
+    def test_annual_rows_excluded_from_mean_and_deletion(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R3_MIX")
+        for m, v in ((1, 100.0), (2, 200.0), (3, 300.0)):
+            _row(db, kpi.id, f"2026-{m:02d}", 2026, m, v)
+        _row(db, kpi.id, "2027-00", 2027, 0, 5000.0)
+        _row(db, kpi.id, "2028-00", 2028, 0, 6000.0)
+
+        self._rolling(client, token)
+
+        annual = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id,
+                                            BudgetPlan.period.in_(["2027-00", "2028-00"])).all()
+        assert len(annual) == 2, "年度行被删除/篡改"
+        now = datetime.now()
+        exp_y, exp_m = now.year, now.month + 12
+        while exp_m > 12:
+            exp_m -= 12
+            exp_y += 1
+        new = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id,
+                                          BudgetPlan.period == f"{exp_y}-{exp_m:02d}").first()
+        assert new is not None, "未产出延展行"
+        assert new.budget_value == 200.0, f"近三月均值被年度行污染: {new.budget_value}（应为月度 100/200/300 ⇒ 200）"
+
+    def test_kpi_with_only_annual_rows_untouched(self, client: TestClient, db: Session):
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="R3_ANNONLY")
+        _row(db, kpi.id, "2027-00", 2027, 0, 5000.0)
+        _row(db, kpi.id, "2028-00", 2028, 0, 6000.0)
+        before = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).count()
+
+        self._rolling(client, token)
+
+        after = db.query(BudgetPlan).filter(BudgetPlan.kpi_id == kpi.id).all()
+        assert len(after) == before == 2, f"仅年度行的 KPI 被篡改: {[(b.period, b.budget_value) for b in after]}"
+        assert sorted(b.period for b in after) == ["2027-00", "2028-00"]

backend/tests/test_budget_sync_cashplan_annual_row_20261013.py

@@ -0,0 +1,154 @@
+"""sync-cash-plans 年度行（budget_month=0）回落 1 月 修复回归
+
+来源单：yanxue-20261013-budget-sync-cashplan-annual-row-fix-fullstack（A 侧派单 P1）
+
+背景：cma.budget_plans 含年度汇总行（period='YYYY-00' / budget_month=0，年度哨兵；
+生产列 budget_month int NOT NULL，禁 1–12 以免污染按月聚合）。
+旧实现 `year, month = b.budget_year or 2026, b.budget_month or 1` 把 0 回落成 1 月，
+按全年额生成 YYYY-01 现金流计划 —— 生产数据污染风险。
+
+实现口径（本单选择：跳过，理由见报告件）：
+  - 年度行（budget_month == 0，或 period 命中年度形态 YYYY / YYYY-00）在 sync-cash-plans 中
+    **跳过**，不生成 / 不更新任何 CashPlan；响应以 skipped_annual 计数披露。
+  - 月度行行为**不变**（含 upsert 幂等语义）。
+  - 判据**形态驱动**（只看 budget_month / period），不读 remark / version。
+
+边界说明：生产列 budget_year / budget_month 均 NOT NULL，NULL 行不可落库，故
+`X or 1` / `X or 2026` 的回落分支只在兼容语义层面存在 —— 用纯判据函数 `_is_annual_budget_row`
+在无 DB 条件下钉住其语义（显式改判：None 不视为年度行，仍走既有回落）。
+"""
+from datetime import datetime
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
+from app.models.budget_plan import BudgetPlan
+from app.models import CashPlan
+
+
+BASE = "/api/cma/budget"
+ANNUAL_REMARK = "BUDGET-2027-2029-E1"
+
+
+def _mk_budget_row(db: Session, kpi_id: int, period: str, year: int, month: int, value: float) -> BudgetPlan:
+    """直接落库预算行（年度行 month=0；POST /plans 不接受年度形态 period）"""
+    bp = BudgetPlan(
+        entity_id=1,
+        kpi_id=kpi_id,
+        period=period,
+        budget_value=value,
+        budget_year=year,
+        budget_month=month,
+        version="v1.0",
+        status="active",
+        remark=ANNUAL_REMARK if month == 0 else "",
+    )
+    db.add(bp)
+    db.commit()
+    db.refresh(bp)
+    return bp
+
+
+class TestSyncCashPlansAnnualRow:
+    def test_annual_row_not_fallen_back_to_january(self, client: TestClient, db: Session):
+        """年度行 budget_month=0 → 不得落成 2027-01，且不得生成任何现金流计划"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="ANN_REV_KPI", kpi_name="营业收入")
+        _mk_budget_row(db, kpi.id, "2027-00", 2027, 0, 9999.0)
+
+        resp = client.post(f"{BASE}/sync-cash-plans", headers=auth_header(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["created"] == 0
+        assert data["updated"] == 0
+        assert data.get("skipped_annual") == 1, f"响应未披露年度行跳过数: {data}"
+
+        plans = db.query(CashPlan).all()
+        assert plans == [], f"年度行生成了现金流计划: {[(p.plan_date, p.amount) for p in plans]}"
+        # 显式钉死污染形态
+        assert db.query(CashPlan).filter(CashPlan.plan_date == datetime(2027, 1, 1)).count() == 0
+
+    def test_annual_row_year_only_form_also_skipped(self, client: TestClient, db: Session):
+        """年度形态 period='2028'（无 -00 后缀）同样跳过；判据形态驱动、不依赖 remark"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="ANN_REV_KPI2", kpi_name="营业收入")
+        bp = _mk_budget_row(db, kpi.id, "2028", 2028, 0, 8888.0)
+        bp.remark = ""  # 抹掉批次 remark，证明判据不读 remark
+        db.commit()
+
+        resp = client.post(f"{BASE}/sync-cash-plans", headers=auth_header(token))
+        assert resp.status_code == 200
+        assert resp.json()["skipped_annual"] == 1
+        assert db.query(CashPlan).count() == 0
+
+    def test_monthly_row_behavior_unchanged(self, client: TestClient, db: Session):
+        """月度行行为不变：生成 2026-06-01 计划 + 二次执行走 upsert 更新"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="MON_REV_KPI", kpi_name="营业收入")
+        _mk_budget_row(db, kpi.id, "2026-06", 2026, 6, 1000.0)
+
+        resp = client.post(f"{BASE}/sync-cash-plans", headers=auth_header(token))
+        assert resp.status_code == 200
+        assert resp.json()["created"] == 1
+        assert resp.json().get("skipped_annual", 0) == 0
+
+        plans = db.query(CashPlan).all()
+        assert len(plans) == 1
+        assert plans[0].plan_date == datetime(2026, 6, 1)
+        assert plans[0].amount == 1000.0
+        assert plans[0].plan_type == "receive"
+
+        resp2 = client.post(f"{BASE}/sync-cash-plans", headers=auth_header(token))
+        assert resp2.json()["created"] == 0
+        assert resp2.json()["updated"] == 1
+
+    def test_mixed_rows_annual_amount_not_contaminating_monthly(self, client: TestClient, db: Session):
+        """年度行与月度行并存：只生成月度计划，全年额不进入任何计划"""
+        create_test_user(db)
+        token = get_token_for_user(client)
+        kpi = create_test_kpi(db, kpi_code="MIX_REV_KPI", kpi_name="营业收入")
+        _mk_budget_row(db, kpi.id, "2027-00", 2027, 0, 9999.0)
+        _mk_budget_row(db, kpi.id, "2026-06", 2026, 6, 1000.0)
+
+        resp = client.post(f"{BASE}/sync-cash-plans", headers=auth_header(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["created"] == 1
+        assert data["skipped_annual"] == 1
+
+        plans = db.query(CashPlan).all()
+        assert len(plans) == 1
+        assert plans[0].plan_date == datetime(2026, 6, 1)
+        assert plans[0].amount == 1000.0
+
+
+class TestAnnualRowPredicate:
+    """判据形态驱动（纯函数，无 DB）——含 NULL 兼容语义钉死"""
+
+    @pytest.mark.parametrize("period,month,expected", [
+        ("2027-00", 0, True),    # 生产年度哨兵形态
+        ("2028", 0, True),       # 仅年份形态
+        ("2029-00", 0, True),
+        ("2027", None, True),    # 仅年份 + 月缺省 → 年度形态
+        ("2026-06", 6, False),   # 月度行
+        ("2026-06", None, False),  # 月缺省但 period 是月度形态 → 走既有回落（非年度）
+        (None, None, False),     # 双缺省 → 非年度（既有 `or 2026/or 1` 回落语义保持）
+        ("", 0, True),           # 月哨兵优先于 period 形态
+        ("bad-form", 3, False),
+    ])
+    def test_is_annual_budget_row(self, period, month, expected):
+        from app.api.budget import _is_annual_budget_row
+
+        assert _is_annual_budget_row(period, month) is expected
+
+    def test_null_branch_keeps_legacy_fallback_semantics(self):
+        """NULL 边界显式改判：非年度行 → 保持 `budget_year or 2026` / `budget_month or 1` 回落"""
+        from app.api.budget import _is_annual_budget_row
+
+        assert _is_annual_budget_row(None, None) is False
+        assert _is_annual_budget_row("2026-06", None) is False