HEL-174: 公司端往来确认完成态改为 success 绿

确认全部完成后，账期流程第 3 步从 doing（warn 黄）切到 done（success 绿），本月待办 pill 与阻断提示同步用 success；待确认仍保留 warn。补前端契约与 DOM 色值测试。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
2026-08-27 12:56:20 +00:00
8 changed files with 413 additions and 1751 deletions
@@ -0,0 +1,186 @@
 #!/usr/bin/env python3
 """HEL-174: 360 / 820 / 1440 截图——待确认黄 vs 已完成绿。"""
 from __future__ import annotations
 import re
 import threading
 from functools import partial
 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from playwright.sync_api import sync_playwright
 ROOT = Path(__file__).resolve().parents[1]
 WEB = ROOT / "web"
 OUT = Path(__file__).resolve().parent.parent / "hel174-shots"
 OUT.mkdir(parents=True, exist_ok=True)
 VIEWPORTS = [
    (1440, 900, "1440"),
    (820, 900, "820"),
    (360, 800, "360"),
 ]
 def extract_fn(source: str, name: str) -> str:
    marker = f"function {name}("
    start = source.index(marker)
    depth = 0
    for i, ch in enumerate(source[start:], start):
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return source[start : i + 1]
    raise RuntimeError(name)
 def main() -> None:
    app_js = (WEB / "app.js").read_text(encoding="utf-8")
    inject = "\n".join(
        [
            "const state = {};",
            "function $(sel, root) { return (root || document).querySelector(sel); }",
            "function $$(sel, root) { return Array.from((root || document).querySelectorAll(sel)); }",
            extract_fn(app_js, "formatWorkspaceAmount"),
            extract_fn(app_js, "applyCompanyWorkspace"),
        ]
    )
    handler = partial(SimpleHTTPRequestHandler, directory=str(WEB))
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    port = httpd.server_address[1]
    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
    thread.start()
    base = f"http://127.0.0.1:{port}"
    html = f"""<!doctype html>
 <html lang="zh-CN"><head>
 <meta charset="UTF-8"/>
 <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
 <link rel="stylesheet" href="{base}/design-system.css"/>
 <title>HEL-174 往来确认状态色</title>
 <style>
  body {{ margin: 0; background: var(--bg); font-family: var(--font-body); color: var(--fg); }}
  .shot-wrap {{ padding: 16px; max-width: 1100px; margin: 0 auto; }}
  .shot-label {{ font-size: 13px; color: var(--muted); margin: 0 0 10px; }}
 </style>
 </head>
 <body data-portal="company">
 <div class="shot-wrap">
  <p class="shot-label" id="shotLabel">状态预览</p>
  <button id="workspaceUnilateralCta" class="btn btn-primary">去确认单边流水 (0)</button>
  <div class="card" id="workspaceTodos" style="margin-top:12px;">
    <div class="card-head">
      <span class="card-title">本月待办<span class="sub" id="workspaceTodoSub">…</span></span>
      <span class="pill pill-warn" id="workspacePendingStatus">加载中</span>
    </div>
    <div id="workspaceTodoList"></div>
    <div class="table-foot" id="workspaceTodoFoot"><span>…</span></div>
  </div>
  <div class="card" style="margin-top:12px;">
    <div class="card-head">
      <span class="card-title">账期流程 · 2026-07<span class="sub" id="workspaceFlowSub">…</span></span>
    </div>
    <div class="flow">
      <a class="flow-step part" href="#upload">
        <div class="fs-top"><span class="fs-idx">01</span><span class="fs-dot"></span><span class="fs-name">流水导入</span></div>
        <div class="fs-state">部分完成</div>
      </a>
      <a class="flow-step done" href="#manual">
        <div class="fs-top"><span class="fs-idx">02</span><span class="fs-dot"></span><span class="fs-name">手工补录</span></div>
        <div class="fs-state">已完成</div>
      </a>
      <a class="flow-step doing" data-view-link="reconcile" href="#reconcile">
        <div class="fs-top"><span class="fs-idx">03</span><span class="fs-dot"></span><span class="fs-name">往来确认</span></div>
        <div class="fs-state" id="workspaceConfirmState">加载中…</div>
        <div class="fs-meta" id="workspaceConfirmMeta">…</div>
      </a>
      <div class="flow-step wait">
        <div class="fs-top"><span class="fs-idx">04</span><span class="fs-dot"></span><span class="fs-name">管理复核</span></div>
        <div class="fs-state">等待集团</div>
      </div>
    </div>
  </div>
  <div class="notice warn" id="blocking-notice" style="margin-top:12px;">
    <div>
      <div class="n-title" id="notice-title">…</div>
      <div class="n-body" id="notice-body">…</div>
    </div>
  </div>
  <div style="margin-top:12px;">
    <span class="pill pill-warn">待确认</span>
    <span class="pill pill-success" style="margin-left:8px;">已确认</span>
  </div>
  <span class="tab-count" id="count-match" hidden>0</span>
  <nav class="side-nav" hidden><a data-view="reconcile"><span class="nav-badge">0</span></a></nav>
 </div>
 </body></html>"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
        page = browser.new_page()
        for width, height, tag in VIEWPORTS:
            page.set_viewport_size({"width": width, "height": height})
            page.set_content(html, wait_until="domcontentloaded")
            page.add_script_tag(content=inject)
            page.evaluate(
                """() => {
                  document.getElementById('shotLabel').textContent = '待确认（应黄）';
                  applyCompanyWorkspace({
                    pending_unilateral: 3, pending_total: 3,
                    unilateral_events: [
                      {event_id:1, amount:'100000.00', currency:'CNY',
                       counterparty_company_name:'金牛贸易', effective_at:'2026-07-03'},
                      {event_id:2, amount:'200000.00', currency:'CNY',
                       counterparty_company_name:'金牛物流', effective_at:'2026-07-11'},
                      {event_id:3, amount:'300000.00', currency:'CNY',
                       counterparty_company_name:'金牛置业', effective_at:'2026-07-24'}
                    ]
                  });
                }"""
            )
            page.wait_for_timeout(200)
            page.screenshot(path=str(OUT / f"hel174-pending-{tag}.png"), full_page=True)
            page.evaluate(
                """() => {
                  document.getElementById('shotLabel').textContent = '已完成（应绿）';
                  applyCompanyWorkspace({
                    pending_unilateral: 0, pending_total: 0, unilateral_events: []
                  });
                }"""
            )
            page.wait_for_timeout(200)
            page.screenshot(path=str(OUT / f"hel174-done-{tag}.png"), full_page=True)
            colors = page.evaluate(
                """() => {
                  const step = document.querySelector('.flow-step[data-view-link="reconcile"]');
                  const state = document.getElementById('workspaceConfirmState');
                  const pill = document.getElementById('workspacePendingStatus');
                  return {
                    stepClass: step.className,
                    stateText: state.textContent,
                    stateColor: getComputedStyle(state).color,
                    pillClass: pill.className,
                    overflowX: document.documentElement.scrollWidth > document.documentElement.clientWidth + 1,
                  };
                }"""
            )
            assert "done" in colors["stepClass"], colors
            assert "pill-success" in colors["pillClass"], colors
            assert colors["stateText"] == "已完成", colors
            assert not colors["overflowX"], colors
            print(tag, colors)
        browser.close()
    httpd.shutdown()
    print("shots:", sorted(p.name for p in OUT.glob("*.png")))
 if __name__ == "__main__":
    main()
@@ -11,8 +11,8 @@ from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, urlparse
 from bank_importer import (
-    auth, company_transfers, dashboard, importing, ledger_events, manual_records,
+    auth, dashboard, importing, ledger_events, manual_records, master_data, matching,
-    master_data, matching, multipart, personal_transit, positions, settings, subjects,
+    multipart, personal_transit, positions, settings, subjects,
 )
 from bank_importer.db import connect, migrate, utc_now
@@ -155,14 +155,6 @@ class AppHandler(SimpleHTTPRequestHandler):
            self._handle_admin_intercompany_evidence(int(admin_evidence.group(1)))
            return
        # Company transfer-summary (HEL-169/HEL-175/HEL-176)
        if path == "/api/company/intercompany/summary":
            self._handle_company_intercompany_summary(query)
            return
        if path == "/api/company/intercompany/export.csv":
            self._handle_company_intercompany_export_csv(query)
            return
        # B-44 intercompany positions (company, own-company scope)
        if path == "/api/company/intercompany/balances":
            self._handle_company_intercompany_balances(query)
@@ -172,11 +164,6 @@ class AppHandler(SimpleHTTPRequestHandler):
            self._handle_company_intercompany_pair(int(company_pair.group(1)), query)
            return
        if path == "/api/company/intercompany/events":
            # HEL-176 transfer-detail list shares this path with B-44 ledger
            # events; dispatch by distinctive query params (to/direction/…).
            if company_transfers.is_transfer_summary_events_query(query):
                self._handle_company_transfer_summary_events(query)
            else:
            self._handle_company_intercompany_events(query)
            return
        if path == "/api/company/manual-records":
@@ -2871,137 +2858,6 @@ class AppHandler(SimpleHTTPRequestHandler):
            return None, None
        return user, user["company_id"]
    def _handle_company_intercompany_summary(self, query: dict[str, list[str]]) -> None:
        """Own-company transfer summary; company_id is session-only (HEL-175)."""
        connection = connect(DB_PATH)
        try:
            user, company_id = self._company_intercompany_scope(connection)
            if company_id is None:
                return
            # Front-end must never supply company_id; reject even matching values.
            if (query.get("company_id") or [None])[0] is not None:
                self._send_json(
                    400,
                    {"status": "error", "message": "不允许传入 company_id 参数。"},
                )
                return
            as_of = (query.get("as_of") or [None])[0]
            try:
                payload = company_transfers.company_intercompany_summary(
                    connection, company_id=int(company_id), as_of=as_of
                )
            except company_transfers.TransferSummaryInputError as exc:
                self._send_json(400, {"status": "error", "message": str(exc)})
                return
            self._send_json(200, {"status": "ok", **payload})
        finally:
            connection.close()
    def _reject_forged_company_id(self, query: dict[str, list[str]]) -> bool:
        """Return True when the handler already sent a 400 for company_id."""
        if (query.get("company_id") or [None])[0] is not None:
            self._send_json(
                400,
                {"status": "error", "message": "不允许传入 company_id 参数。"},
            )
            return True
        return False
    def _handle_company_transfer_summary_events(
        self, query: dict[str, list[str]]
    ) -> None:
        """HEL-176 filtered transfer-detail list (keyset pagination)."""
        connection = connect(DB_PATH)
        try:
            user, company_id = self._company_intercompany_scope(connection)
            if company_id is None:
                return
            if self._reject_forged_company_id(query):
                return
            try:
                payload = company_transfers.company_intercompany_events(
                    connection,
                    company_id=int(company_id),
                    from_=(query.get("from") or [None])[0],
                    to=(query.get("to") or [None])[0],
                    counterparty_id=(query.get("counterparty_id") or [None])[0],
                    direction=(query.get("direction") or [None])[0],
                    state=(query.get("state") or [None])[0],
                    limit=(query.get("limit") or [None])[0],
                    cursor=(query.get("cursor") or [None])[0],
                )
            except company_transfers.TransferSummaryInputError as exc:
                self._send_json(400, {"status": "error", "message": str(exc)})
                return
            self._send_json(200, {"status": "ok", **payload})
        finally:
            connection.close()
    def _handle_company_intercompany_export_csv(
        self, query: dict[str, list[str]]
    ) -> None:
        """Confirmed-only CSV export; session company scope + audit (HEL-176)."""
        connection = connect(DB_PATH)
        try:
            user, company_id = self._company_intercompany_scope(connection)
            if company_id is None:
                return
            if self._reject_forged_company_id(query):
                return
            # Pending must never leave via export, even if a client sends state=.
            if (query.get("state") or [None])[0] not in (None, "", "confirmed"):
                self._send_json(
                    400,
                    {
                        "status": "error",
                        "message": "导出仅支持已确认明细，不允许导出待确认。",
                    },
                )
                return
            try:
                items, meta = company_transfers.company_intercompany_export_rows(
                    connection,
                    company_id=int(company_id),
                    from_=(query.get("from") or [None])[0],
                    to=(query.get("to") or [None])[0],
                    counterparty_id=(query.get("counterparty_id") or [None])[0],
                    direction=(query.get("direction") or [None])[0],
                )
            except company_transfers.TransferSummaryInputError as exc:
                self._send_json(400, {"status": "error", "message": str(exc)})
                return
            content = company_transfers.render_intercompany_export_csv(items)
            detail_parts = [
                f"rows:{meta['row_count']}",
                f"from:{meta['start']}",
                f"to:{meta['end']}",
                "state:confirmed",
            ]
            if meta["counterparty_id"] is not None:
                detail_parts.append(f"counterparty_id:{meta['counterparty_id']}")
            if meta["direction"] is not None:
                detail_parts.append(f"direction:{meta['direction']}")
            auth.audit(
                connection,
                "export_intercompany_csv",
                actor=user,
                target=f"company:{company_id}",
                detail=";".join(detail_parts),
                ip=self._client_ip,
            )
            self.send_response(200)
            self.send_header("Content-Type", "text/csv; charset=utf-8")
            self.send_header(
                "Content-Disposition",
                'attachment; filename="intercompany-export.csv"',
            )
            self.send_header("Content-Length", str(len(content)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(content)
        finally:
            connection.close()
    def _handle_company_intercompany_balances(self, query: dict[str, list[str]]) -> None:
        connection = connect(DB_PATH)
        try:
@@ -1,740 +0,0 @@
 """Company-portal intercompany transfer summary / detail / export (HEL-175/176).
 Confirmed totals reuse the authoritative ``eligible_intercompany_events``
 view (intercompany + paired or locked). Pending counts/amounts are listed
 separately and never enter outflow, inflow or net. All money math uses
 ``Decimal`` on stored TEXT amounts — never float or SQLite SUM.
 Detail listing (HEL-176) applies counterparty / date / direction / state
 filters inside SQL before LIMIT, and uses keyset pagination on
 (effective_at, event_id) descending. Export only ships confirmed rows.
 """
 from __future__ import annotations
 import base64
 import csv
 import io
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 import re
 import sqlite3
 from . import matching, settings
 _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 _ZERO = Decimal("0.00")
 _DEFAULT_PAGE = 50
 _MAX_PAGE = 200
 _MAX_EXPORT_ROWS = 20000
 class TransferSummaryInputError(ValueError):
    """Invalid query parameters (mapped to HTTP 400)."""
 def today_shanghai() -> str:
    return datetime.now(timezone(timedelta(hours=8))).date().isoformat()
 def _validate_date(value: object, label: str) -> str:
    text = str(value or "").strip()
    if not text or not _DATE_RE.fullmatch(text):
        raise TransferSummaryInputError(f"{label}必须是 YYYY-MM-DD 格式。")
    try:
        datetime.strptime(text, "%Y-%m-%d")
    except ValueError as exc:
        raise TransferSummaryInputError(f"{label}不是有效日期。") from exc
    return text
 def _money(value: Decimal) -> str:
    return format(value.quantize(Decimal("0.01")), "f")
 def _as_decimal(raw: object) -> Decimal:
    try:
        return Decimal(str(raw))
    except (InvalidOperation, TypeError) as exc:
        raise TransferSummaryInputError("金额数据无效，无法汇总。") from exc
 def _company_row(connection: sqlite3.Connection, company_id: int) -> dict[str, object]:
    row = connection.execute(
        "SELECT id, name FROM companies WHERE id = ?", (company_id,)
    ).fetchone()
    if row is None:
        raise TransferSummaryInputError("本公司不存在。")
    return {"id": int(row["id"]), "name": row["name"]}
 def _window_bounds(
    connection: sqlite3.Connection, as_of: str | None
 ) -> tuple[str, str]:
    end = _validate_date(as_of, "as_of") if as_of else today_shanghai()
    start = settings.get_settings(connection).get("start_date") or "2026-01-01"
    start = _validate_date(start, "start_date")
    if start > end:
        # Opening / start-date plumbing may lag; clamp rather than 500.
        start = end
    return start, end
 def _load_confirmed(
    connection: sqlite3.Connection, company_id: int, start: str, end: str
 ) -> list[sqlite3.Row]:
    return connection.execute(
        """
        SELECT e.event_id, e.decision_id, e.effective_at, e.amount, e.currency,
               e.payer_company_id, e.payee_company_id, e.pairing,
               cpayer.name AS payer_company_name,
               cpayee.name AS payee_company_name
        FROM eligible_intercompany_events e
        JOIN companies cpayer ON cpayer.id = e.payer_company_id
        JOIN companies cpayee ON cpayee.id = e.payee_company_id
        WHERE (e.payer_company_id = ? OR e.payee_company_id = ?)
          AND e.effective_at >= ?
          AND e.effective_at <= ?
        ORDER BY e.event_id
        """,
        (company_id, company_id, start, end + "T23:59:59"),
    ).fetchall()
 def _load_pending(
    connection: sqlite3.Connection, company_id: int, start: str, end: str
 ) -> list[sqlite3.Row]:
    """Pending = not eligible confirmed, still company-visible for tip only.
    Includes unresolved / needs_review / internal_single, plus any
    intercompany decision that is neither paired nor locked.
    """
    return connection.execute(
        """
        SELECT c.event_id, d.id AS decision_id, d.effective_at, d.amount,
               d.currency, d.classification, d.pairing, d.locked,
               payer.company_id AS payer_company_id,
               payee.company_id AS payee_company_id,
               cpayer.name AS payer_company_name,
               cpayee.name AS payee_company_name
        FROM current_transfer_decisions c
        JOIN transfer_match_decisions d ON d.id = c.decision_id
        JOIN canonical_transfer_events e ON e.id = c.event_id
        LEFT JOIN transfer_decision_participants payer
             ON payer.decision_id = d.id AND payer.role = 'payer'
        LEFT JOIN transfer_decision_participants payee
             ON payee.decision_id = d.id AND payee.role = 'payee'
        LEFT JOIN companies cpayer ON cpayer.id = payer.company_id
        LEFT JOIN companies cpayee ON cpayee.id = payee.company_id
        WHERE e.lifecycle = 'active'
          AND (payer.company_id = ? OR payee.company_id = ?)
          AND d.effective_at >= ?
          AND d.effective_at <= ?
          AND (
                d.classification IN ('unresolved', 'needs_review', 'internal_single')
             OR (
                    d.classification = 'intercompany'
                AND d.pairing != 'paired'
                AND d.locked = 0
             )
          )
        ORDER BY c.event_id
        """,
        (company_id, company_id, start, end + "T23:59:59"),
    ).fetchall()
 def _counterparty_of(row: sqlite3.Row, company_id: int) -> tuple[int | None, str | None]:
    payer = row["payer_company_id"]
    payee = row["payee_company_id"]
    if payer is not None and int(payer) == int(company_id):
        if payee is None:
            return None, None
        return int(payee), row["payee_company_name"]
    if payee is not None and int(payee) == int(company_id):
        if payer is None:
            return None, None
        return int(payer), row["payer_company_name"]
    return None, None
 def _is_outflow(row: sqlite3.Row, company_id: int) -> bool:
    return row["payer_company_id"] is not None and int(row["payer_company_id"]) == int(
        company_id
    )
 def company_intercompany_summary(
    connection: sqlite3.Connection,
    *,
    company_id: int,
    as_of: str | None = None,
 ) -> dict[str, object]:
    """Build the HEL-169 summary payload for one company session."""
    own = _company_row(connection, company_id)
    start, end = _window_bounds(connection, as_of)
    confirmed_rows = _load_confirmed(connection, company_id, start, end)
    pending_rows = _load_pending(connection, company_id, start, end)
    seen_confirmed: set[int] = set()
    outflow = _ZERO
    inflow = _ZERO
    outflow_count = 0
    inflow_count = 0
    # counterparty_id -> bucket
    buckets: dict[int, dict[str, object]] = {}
    def bucket_for(cp_id: int, cp_name: str | None) -> dict[str, object]:
        bucket = buckets.get(cp_id)
        if bucket is None:
            bucket = {
                "company_id": cp_id,
                "company_name": cp_name or "",
                "confirmed_outflow": _ZERO,
                "confirmed_inflow": _ZERO,
                "pending_count": 0,
                "last_effective_at": None,
            }
            buckets[cp_id] = bucket
        elif cp_name and not bucket["company_name"]:
            bucket["company_name"] = cp_name
        return bucket
    def touch_last(bucket: dict[str, object], effective_at: str | None) -> None:
        if not effective_at:
            return
        previous = bucket["last_effective_at"]
        if previous is None or str(effective_at) > str(previous):
            bucket["last_effective_at"] = effective_at
    for row in confirmed_rows:
        event_id = int(row["event_id"])
        if event_id in seen_confirmed:
            continue
        seen_confirmed.add(event_id)
        amount = _as_decimal(row["amount"])
        cp_id, cp_name = _counterparty_of(row, company_id)
        if _is_outflow(row, company_id):
            outflow += amount
            outflow_count += 1
            if cp_id is not None:
                bucket = bucket_for(cp_id, cp_name)
                bucket["confirmed_outflow"] += amount
                touch_last(bucket, row["effective_at"])
        else:
            inflow += amount
            inflow_count += 1
            if cp_id is not None:
                bucket = bucket_for(cp_id, cp_name)
                bucket["confirmed_inflow"] += amount
                touch_last(bucket, row["effective_at"])
    pending_total = _ZERO
    seen_pending: set[int] = set()
    for row in pending_rows:
        event_id = int(row["event_id"])
        if event_id in seen_pending or event_id in seen_confirmed:
            continue
        seen_pending.add(event_id)
        amount = _as_decimal(row["amount"])
        pending_total += amount
        cp_id, cp_name = _counterparty_of(row, company_id)
        if cp_id is not None:
            bucket = bucket_for(cp_id, cp_name)
            bucket["pending_count"] = int(bucket["pending_count"]) + 1
            touch_last(bucket, row["effective_at"])
    net = outflow - inflow
    if net > 0:
        net_direction: str | None = "receivable"
    elif net < 0:
        net_direction = "payable"
    else:
        net_direction = None
    counterparties = []
    for cp_id in sorted(
        buckets.keys(),
        key=lambda i: (
            -(
                buckets[i]["confirmed_outflow"]  # type: ignore[operator]
                + buckets[i]["confirmed_inflow"]
            ),
            buckets[i]["company_name"] or "",
            i,
        ),
    ):
        bucket = buckets[cp_id]
        conf_out = bucket["confirmed_outflow"]
        conf_in = bucket["confirmed_inflow"]
        assert isinstance(conf_out, Decimal) and isinstance(conf_in, Decimal)
        counterparties.append(
            {
                "company_id": cp_id,
                "company_name": bucket["company_name"],
                "confirmed_outflow": _money(conf_out),
                "confirmed_inflow": _money(conf_in),
                "net": _money(conf_out - conf_in),
                "pending_count": int(bucket["pending_count"]),
                "last_effective_at": bucket["last_effective_at"],
            }
        )
    return {
        "own_company": own,
        "window": {
            "start": start,
            "end": end,
            "has_opening": False,
            # Reserved for opening-balance rollout; callers must not invent balances.
            "opening": None,
            "ending": None,
        },
        "confirmed": {
            "outflow_total": _money(outflow),
            "outflow_count": outflow_count,
            "inflow_total": _money(inflow),
            "inflow_count": inflow_count,
            "net_change": _money(net),
            "net_direction": net_direction,
        },
        "pending": {
            "count": len(seen_pending),
            "amount_total": _money(pending_total),
        },
        "counterparties": counterparties,
    }
 # ---------------------------------------------------------------------------
 # Detail list + CSV export (HEL-176)
 # ---------------------------------------------------------------------------
 def encode_cursor(effective_at: str, event_id: int) -> str:
    raw = f"{effective_at}|{event_id}"
    return base64.urlsafe_b64encode(raw.encode("utf-8")).decode("ascii")
 def decode_cursor(cursor: str | None) -> tuple[str, int] | None:
    if not cursor:
        return None
    try:
        raw = base64.urlsafe_b64decode(cursor.encode("ascii")).decode("utf-8")
        effective_at, event_id_text = raw.split("|", 1)
        return effective_at, int(event_id_text)
    except Exception as exc:
        raise TransferSummaryInputError("分页游标无效。") from exc
 def _parse_limit(raw: object | None, default: int = _DEFAULT_PAGE) -> int:
    if raw is None or str(raw).strip() == "":
        return default
    try:
        value = int(str(raw).strip())
    except (TypeError, ValueError) as exc:
        raise TransferSummaryInputError("limit 必须是正整数。") from exc
    if value < 1:
        raise TransferSummaryInputError("limit 必须是正整数。")
    return min(value, _MAX_PAGE)
 def _parse_optional_date(raw: object | None, label: str) -> str | None:
    if raw is None or str(raw).strip() == "":
        return None
    return _validate_date(raw, label)
 def _parse_direction(raw: object | None) -> str | None:
    if raw is None or str(raw).strip() == "":
        return None
    value = str(raw).strip().lower()
    if value not in ("out", "in"):
        raise TransferSummaryInputError("direction 只能是 out 或 in。")
    return value
 def _parse_state(raw: object | None) -> str | None:
    if raw is None or str(raw).strip() == "":
        return None
    value = str(raw).strip().lower()
    if value not in ("confirmed", "pending"):
        raise TransferSummaryInputError("state 只能是 confirmed 或 pending。")
    return value
 def _parse_counterparty_id(raw: object | None) -> int | None:
    if raw is None or str(raw).strip() == "":
        return None
    try:
        value = int(str(raw).strip())
    except (TypeError, ValueError) as exc:
        raise TransferSummaryInputError("counterparty_id 参数无效。") from exc
    if value < 1:
        raise TransferSummaryInputError("counterparty_id 参数无效。")
    return value
 def _event_window(
    connection: sqlite3.Connection,
    *,
    from_: str | None,
    to: str | None,
 ) -> tuple[str, str]:
    end = _parse_optional_date(to, "to") or today_shanghai()
    start_default = settings.get_settings(connection).get("start_date") or "2026-01-01"
    start = _parse_optional_date(from_, "from") or _validate_date(
        start_default, "start_date"
    )
    if start > end:
        raise TransferSummaryInputError("from 不能晚于 to。")
    return start, end
 # Confirmed = eligible_intercompany_events. Pending matches summary tip set.
 _CONFIRMED_PREDICATE = """
    d.classification = 'intercompany'
    AND (d.pairing = 'paired' OR d.locked = 1)
 """
 _PENDING_PREDICATE = """
    (
        d.classification IN ('unresolved', 'needs_review', 'internal_single')
     OR (
            d.classification = 'intercompany'
        AND d.pairing != 'paired'
        AND d.locked = 0
     )
    )
 """
 _BOTH_PREDICATE = f"""
    (
        ({_CONFIRMED_PREDICATE})
     OR ({_PENDING_PREDICATE})
    )
 """
 def _state_predicate(state: str | None) -> str:
    if state == "confirmed":
        return f"({_CONFIRMED_PREDICATE})"
    if state == "pending":
        return f"({_PENDING_PREDICATE})"
    return _BOTH_PREDICATE
 def _list_select_sql() -> str:
    return """
        SELECT c.event_id, d.id AS decision_id, d.revision, d.classification,
               d.pairing, d.amount, d.currency, d.effective_at, d.mode,
               d.locked, d.rule_version, d.created_at, d.reason,
               payer.company_id AS payer_company_id,
               payee.company_id AS payee_company_id,
               payer.bank_account_id AS payer_account_id,
               payee.bank_account_id AS payee_account_id,
               cpayer.name AS payer_company_name,
               cpayee.name AS payee_company_name,
               (SELECT COUNT(*) FROM transfer_decision_observations o
                WHERE o.decision_id = d.id) AS evidence_count,
               (SELECT r.summary
                FROM transfer_decision_observations o
                JOIN source_rows r ON r.id = o.source_row_id
                JOIN sheet_batches s ON s.id = r.sheet_batch_id
                JOIN import_batches b ON b.id = s.import_batch_id
                WHERE o.decision_id = d.id AND b.company_id = ?
                ORDER BY o.id
                LIMIT 1) AS summary
        FROM current_transfer_decisions c
        JOIN transfer_match_decisions d ON d.id = c.decision_id
        JOIN canonical_transfer_events e ON e.id = c.event_id
        LEFT JOIN transfer_decision_participants payer
             ON payer.decision_id = d.id AND payer.role = 'payer'
        LEFT JOIN transfer_decision_participants payee
             ON payee.decision_id = d.id AND payee.role = 'payee'
        LEFT JOIN companies cpayer ON cpayer.id = payer.company_id
        LEFT JOIN companies cpayee ON cpayee.id = payee.company_id
    """
 def _build_event_filters(
    *,
    company_id: int,
    start: str,
    end: str,
    counterparty_id: int | None,
    direction: str | None,
    state: str | None,
    cursor: tuple[str, int] | None,
 ) -> tuple[str, list[object]]:
    conditions = [
        "e.lifecycle = 'active'",
        "(payer.company_id = ? OR payee.company_id = ?)",
        "d.effective_at >= ?",
        "d.effective_at <= ?",
        _state_predicate(state),
    ]
    params: list[object] = [
        company_id,
        company_id,
        start,
        end + "T23:59:59",
    ]
    if counterparty_id is not None:
        # Counterparty is the other participant; own company stays forced above.
        conditions.append(
            """
            (
                (payer.company_id = ? AND payee.company_id = ?)
             OR (payee.company_id = ? AND payer.company_id = ?)
            )
            """
        )
        params.extend([company_id, counterparty_id, company_id, counterparty_id])
    if direction == "out":
        conditions.append("payer.company_id = ?")
        params.append(company_id)
    elif direction == "in":
        conditions.append("payee.company_id = ?")
        params.append(company_id)
    if cursor is not None:
        cursor_at, cursor_id = cursor
        conditions.append(
            """
            (
                d.effective_at < ?
             OR (d.effective_at = ? AND c.event_id < ?)
            )
            """
        )
        params.extend([cursor_at, cursor_at, cursor_id])
    where = " WHERE " + " AND ".join(conditions)
    return where, params
 def _row_state(row: sqlite3.Row) -> str:
    classification = row["classification"]
    pairing = row["pairing"]
    locked = bool(row["locked"])
    if classification == "intercompany" and (pairing == "paired" or locked):
        return "confirmed"
    return "pending"
 def _row_direction(row: sqlite3.Row, company_id: int) -> str | None:
    if row["payer_company_id"] is not None and int(row["payer_company_id"]) == int(
        company_id
    ):
        return "out"
    if row["payee_company_id"] is not None and int(row["payee_company_id"]) == int(
        company_id
    ):
        return "in"
    return None
 def _event_list_item(row: sqlite3.Row, company_id: int) -> dict[str, object]:
    direction = _row_direction(row, company_id)
    state = _row_state(row)
    if direction == "out":
        counterparty_company_id = row["payee_company_id"]
        counterparty_company_name = row["payee_company_name"]
    else:
        counterparty_company_id = row["payer_company_id"]
        counterparty_company_name = row["payer_company_name"]
    summary = row["summary"] or row["reason"] or ""
    return {
        "event_id": int(row["event_id"]),
        "decision_id": int(row["decision_id"]),
        "revision": row["revision"],
        "classification": row["classification"],
        "pairing": row["pairing"],
        "status": matching.exposed_status(row),
        "state": state,
        "direction": direction,
        "amount": row["amount"],
        "currency": row["currency"],
        "effective_at": row["effective_at"],
        "mode": row["mode"],
        "locked": bool(row["locked"]),
        "rule_version": row["rule_version"],
        "summary": summary,
        "own_company_id": company_id,
        "counterparty_company_id": (
            int(counterparty_company_id) if counterparty_company_id is not None else None
        ),
        "counterparty_company_name": counterparty_company_name,
        "evidence_count": row["evidence_count"],
    }
 def company_intercompany_events(
    connection: sqlite3.Connection,
    *,
    company_id: int,
    from_: str | None = None,
    to: str | None = None,
    counterparty_id: int | None | object = None,
    direction: str | None | object = None,
    state: str | None | object = None,
    limit: object | None = None,
    cursor: str | None = None,
 ) -> dict[str, object]:
    """Filtered keyset page of company-visible transfer events (HEL-176)."""
    start, end = _event_window(connection, from_=from_, to=to)
    cp_id = _parse_counterparty_id(counterparty_id)
    direction_value = _parse_direction(direction)
    state_value = _parse_state(state)
    page_size = _parse_limit(limit)
    cursor_tuple = decode_cursor(cursor)
    where, params = _build_event_filters(
        company_id=company_id,
        start=start,
        end=end,
        counterparty_id=cp_id,
        direction=direction_value,
        state=state_value,
        cursor=cursor_tuple,
    )
    # summary subquery binds own company_id first.
    sql = (
        _list_select_sql()
        + where
        + " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?"
    )
    rows = connection.execute(
        sql, (company_id, *params, page_size + 1)
    ).fetchall()
    has_more = len(rows) > page_size
    page = rows[:page_size]
    items = [_event_list_item(row, company_id) for row in page]
    next_cursor = None
    if has_more and page:
        last = page[-1]
        next_cursor = encode_cursor(str(last["effective_at"]), int(last["event_id"]))
    return {
        "window": {"start": start, "end": end},
        "events": items,
        "next_cursor": next_cursor,
        "has_more": has_more,
    }
 def company_intercompany_export_rows(
    connection: sqlite3.Connection,
    *,
    company_id: int,
    from_: str | None = None,
    to: str | None = None,
    counterparty_id: int | None | object = None,
    direction: str | None | object = None,
 ) -> tuple[list[dict[str, object]], dict[str, object]]:
    """Confirmed-only rows for CSV export; pending never included."""
    start, end = _event_window(connection, from_=from_, to=to)
    cp_id = _parse_counterparty_id(counterparty_id)
    direction_value = _parse_direction(direction)
    where, params = _build_event_filters(
        company_id=company_id,
        start=start,
        end=end,
        counterparty_id=cp_id,
        direction=direction_value,
        state="confirmed",
        cursor=None,
    )
    sql = (
        _list_select_sql()
        + where
        + " ORDER BY d.effective_at DESC, c.event_id DESC LIMIT ?"
    )
    rows = connection.execute(
        sql, (company_id, *params, _MAX_EXPORT_ROWS + 1)
    ).fetchall()
    if len(rows) > _MAX_EXPORT_ROWS:
        raise TransferSummaryInputError(
            f"导出行数超过上限 {_MAX_EXPORT_ROWS}，请缩小筛选范围。"
        )
    items = [_event_list_item(row, company_id) for row in rows]
    meta = {
        "start": start,
        "end": end,
        "counterparty_id": cp_id,
        "direction": direction_value,
        "state": "confirmed",
        "row_count": len(items),
    }
    return items, meta
 def render_intercompany_export_csv(items: list[dict[str, object]]) -> bytes:
    buffer = io.StringIO()
    writer = csv.writer(buffer)
    writer.writerow(
        [
            "日期",
            "方向",
            "对方公司",
            "金额",
            "币种",
            "摘要",
            "状态",
            "配对",
            "事件ID",
            "决策ID",
        ]
    )
    direction_label = {"out": "转出", "in": "转入"}
    for item in items:
        writer.writerow(
            [
                item.get("effective_at") or "",
                direction_label.get(str(item.get("direction") or ""), ""),
                item.get("counterparty_company_name") or "",
                item.get("amount") or "",
                item.get("currency") or "",
                item.get("summary") or "",
                "已确认",
                item.get("pairing") or "",
                item.get("event_id") or "",
                item.get("decision_id") or "",
            ]
        )
    # UTF-8 BOM so Excel opens the CSV with the right encoding.
    return (chr(0xFEFF) + buffer.getvalue()).encode("utf-8")
 def is_transfer_summary_events_query(query: dict[str, list[str]]) -> bool:
    """Discriminate HEL-176 transfer list from B-44 ledger ``/events``.
    B-44 uses ``cutoff`` / subject / posting_kind / source_kind / pending_subject.
    HEL-176 uses ``to`` / direction / counterparty_id / state=pending|confirmed
    (without ledger-only knobs).
    """
    if (query.get("direction") or [None])[0] is not None:
        return True
    if (query.get("counterparty_id") or [None])[0] is not None:
        return True
    if (query.get("to") or [None])[0] is not None:
        return True
    state = (query.get("state") or [None])[0]
    if state in ("pending", "confirmed") and (query.get("cutoff") or [None])[0] is None:
        # Bare state=confirmed without cutoff is the transfer-summary list;
        # B-44 confirmed always pairs with cutoff in existing callers/tests.
        if (query.get("subject") or [None])[0] is not None:
            return False
        if (query.get("posting_kind") or [None])[0] is not None:
            return False
        if (query.get("source_kind") or [None])[0] is not None:
            return False
        return True
    return False
@@ -0,0 +1,210 @@
 """HEL-174: 公司端往来确认完成态用 success 绿，待确认保留 warn 黄。"""
 from __future__ import annotations
 import re
 import threading
 import unittest
 from functools import partial
 from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 ROOT = Path(__file__).resolve().parents[1]
 WEB = ROOT / "web"
 try:
    from playwright.sync_api import sync_playwright
 except ImportError:  # pragma: no cover - 默认测试环境无浏览器依赖
    sync_playwright = None
 class ConfirmStatusSourceContractTests(unittest.TestCase):
    """不依赖浏览器：锁住完成态切绿 / 待确认仍黄的实现契约。"""
    def test_app_js_toggles_success_green_when_pending_zero(self) -> None:
        js = (WEB / "app.js").read_text(encoding="utf-8")
        self.assertIn('pill ${total ? "pill-warn" : "pill-success"}', js)
        self.assertIn('flowStep.classList.toggle("doing", pending > 0)', js)
        self.assertIn('flowStep.classList.toggle("done", pending === 0)', js)
        self.assertIn('notice.classList.toggle("warn", pending > 0)', js)
        self.assertIn('notice.classList.toggle("success", pending === 0)', js)
        # 文案：完成=已完成，不把待确认一并改绿
        self.assertRegex(js, r'status\.textContent = total \? `\$\{total\} 项待处理` : "已完成"')
        self.assertRegex(js, r'flowState\.textContent = pending \? `待处理 \$\{pending\} 笔` : "已完成"')
    def test_design_tokens_map_done_to_success_doing_to_warn(self) -> None:
        css = (WEB / "design-system.css").read_text(encoding="utf-8")
        self.assertIn("--success:", css)
        self.assertIn("--warn:", css)
        self.assertIn(".flow-step.done .fs-dot { background: var(--success); }", css)
        self.assertIn(".flow-step.done .fs-state { color: var(--success); }", css)
        self.assertIn(".flow-step.doing .fs-dot { background: var(--warn);", css)
        self.assertIn(".pill-success { background: var(--success-soft); color: var(--success); }", css)
        self.assertIn(".pill-warn { background: var(--warn-soft);", css)
        self.assertIn(".notice.success { background: var(--success-soft);", css)
        self.assertIn(".notice.warn { background: var(--warn-soft);", css)
    def test_company_html_exposes_flow_step_and_cache_bust(self) -> None:
        html = (WEB / "company.html").read_text(encoding="utf-8")
        self.assertIn('id="workspaceConfirmState"', html)
        self.assertIn('id="workspacePendingStatus"', html)
        self.assertIn('id="workspaceFlowSub"', html)
        self.assertIn('data-view-link="reconcile"', html)
        self.assertIn("app.js?v=11", html)
        # 静态初值仍为进行中（黄），由 JS 在 pending=0 时切 done
        self.assertRegex(html, r'class="flow-step doing"[^>]*data-view-link="reconcile"')
 def _extract_fn(source: str, name: str) -> str:
    marker = f"function {name}("
    start = source.index(marker)
    depth = 0
    for i, ch in enumerate(source[start:], start):
        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return source[start : i + 1]
    raise AssertionError(f"未能截取 function {name}")
@unittest.skipUnless(sync_playwright, "playwright 未安装，跳过真实 DOM 色值校验")
 class ConfirmStatusDomTests(unittest.TestCase):
    """真实浏览器：pending>0 为黄，pending=0 为绿。"""
    @classmethod
    def setUpClass(cls) -> None:
        handler = partial(SimpleHTTPRequestHandler, directory=str(WEB))
        cls.httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
        cls.port = cls.httpd.server_address[1]
        cls.thread = threading.Thread(target=cls.httpd.serve_forever, daemon=True)
        cls.thread.start()
        cls.base = f"http://127.0.0.1:{cls.port}"
        app_js = (WEB / "app.js").read_text(encoding="utf-8")
        cls.inject_js = "\n".join(
            [
                "const state = {};",
                "function $(sel, root) { return (root || document).querySelector(sel); }",
                "function $$(sel, root) { return Array.from((root || document).querySelectorAll(sel)); }",
                _extract_fn(app_js, "formatWorkspaceAmount"),
                _extract_fn(app_js, "applyCompanyWorkspace"),
            ]
        )
    @classmethod
    def tearDownClass(cls) -> None:
        cls.httpd.shutdown()
        cls.httpd.server_close()
    def _open_fixture(self, page):
        page.goto(f"{self.base}/design-system.css", wait_until="domcontentloaded")
        page.set_content(
            f"""<!doctype html>
 <html lang="zh-CN"><head>
 <meta charset="UTF-8"/>
 <link rel="stylesheet" href="{self.base}/design-system.css"/>
 </head>
 <body data-portal="company">
  <button id="workspaceUnilateralCta" class="btn btn-primary">去确认单边流水 (0)</button>
  <div class="card" id="workspaceTodos">
    <div class="card-head">
      <span class="card-title">本月待办<span class="sub" id="workspaceTodoSub">…</span></span>
      <span class="pill pill-warn" id="workspacePendingStatus">加载中</span>
    </div>
    <div id="workspaceTodoList"></div>
    <div class="table-foot" id="workspaceTodoFoot"><span>…</span></div>
  </div>
  <div class="card">
    <div class="card-head">
      <span class="card-title">账期流程<span class="sub" id="workspaceFlowSub">…</span></span>
    </div>
    <div class="flow">
      <a class="flow-step doing" data-view-link="reconcile" href="#reconcile">
        <div class="fs-top"><span class="fs-idx">03</span><span class="fs-dot"></span><span class="fs-name">往来确认</span></div>
        <div class="fs-state" id="workspaceConfirmState">加载中…</div>
        <div class="fs-meta" id="workspaceConfirmMeta">…</div>
      </a>
    </div>
  </div>
  <div class="notice warn" id="blocking-notice">
    <div class="n-title" id="notice-title">…</div>
    <div class="n-body" id="notice-body">…</div>
  </div>
  <span class="tab-count" id="count-match">0</span>
  <nav class="side-nav"><a data-view="reconcile"><span class="nav-badge">0</span></a></nav>
 </body></html>""",
            wait_until="domcontentloaded",
        )
        page.add_script_tag(content=self.inject_js)
        page.wait_for_function("() => typeof applyCompanyWorkspace === 'function'")
    @staticmethod
    def _colors(page) -> dict:
        return page.evaluate(
            """() => {
              const step = document.querySelector('.flow-step[data-view-link="reconcile"]');
              const state = document.getElementById('workspaceConfirmState');
              const pill = document.getElementById('workspacePendingStatus');
              const notice = document.getElementById('blocking-notice');
              const cs = (el) => getComputedStyle(el);
              return {
                stepClass: step.className,
                stateText: state.textContent,
                stateColor: cs(state).color,
                pillClass: pill.className,
                pillColor: cs(pill).color,
                pillText: pill.textContent,
                noticeClass: notice.className,
              };
            }"""
        )
    def test_pending_stays_warn_completed_turns_success(self) -> None:
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True, args=["--no-sandbox"])
            page = browser.new_page(viewport={"width": 1440, "height": 900})
            self._open_fixture(page)
            page.evaluate(
                """() => applyCompanyWorkspace({
                  pending_unilateral: 2,
                  pending_total: 2,
                  unilateral_events: [
                    {event_id: 1, amount: '100.00', currency: 'CNY',
                     counterparty_company_name: '乙', effective_at: '2026-07-01'}
                  ]
                })"""
            )
            pending = self._colors(page)
            self.assertIn("doing", pending["stepClass"])
            self.assertNotIn("done", pending["stepClass"].split())
            self.assertIn("pill-warn", pending["pillClass"])
            self.assertNotIn("pill-success", pending["pillClass"])
            self.assertIn("warn", pending["noticeClass"].split())
            self.assertIn("待处理", pending["stateText"])
            self.assertIn("待处理", pending["pillText"])
            page.evaluate(
                """() => applyCompanyWorkspace({
                  pending_unilateral: 0,
                  pending_total: 0,
                  unilateral_events: []
                })"""
            )
            done = self._colors(page)
            self.assertIn("done", done["stepClass"])
            self.assertNotIn("doing", done["stepClass"].split())
            self.assertIn("pill-success", done["pillClass"])
            self.assertNotIn("pill-warn", done["pillClass"])
            self.assertIn("success", done["noticeClass"].split())
            self.assertEqual("已完成", done["stateText"])
            self.assertEqual("已完成", done["pillText"])
            self.assertNotEqual(pending["stateColor"], done["stateColor"])
            self.assertNotEqual(pending["pillColor"], done["pillColor"])
            browser.close()
 if __name__ == "__main__":
    unittest.main()
@@ -1,500 +0,0 @@
 """HTTP tests for company intercompany events list + CSV export (HEL-176).
 Covers combined filters, keyset pagination, empty state, lateral access,
 forged company_id, ID guessing, export isolation and audit logging.
 """
 from __future__ import annotations
 import io
 import json
 import os
 from pathlib import Path
 import tempfile
 import threading
 import unittest
 from openpyxl import Workbook
 from bank_importer.db import connect, migrate
 import server
 from test_server_auth import Client, as_json
 BOOTSTRAP_PASSWORD = "BootAdmin123"
 ADMIN_PASSWORD = "AdminPass123"
 CASHIER_PASSWORD = "Cashier123"
 CCB_HEADER = [
    "客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）",
    "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注",
 ]
 ACCOUNT_A = "6222000000000001"
 ACCOUNT_B = "6222000000000002"
 ACCOUNT_C = "6222000000000003"
 def workbook_bytes(rows) -> bytes:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "正常流水"
    sheet.append(CCB_HEADER)
    for row in rows:
        sheet.append(row)
    buffer = io.BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()
 def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00"):
    return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""]
 def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00"):
    return [own, "测试公司", at, "", amount, "50000.00", "RMB", "对方", cp, "某银行", "收款", ""]
 class CompanyIntercompanyEventsTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        root = Path(self.temp_dir.name)
        self.db_path = root / "app.db"
        self.storage = root / "files"
        self._old_db_path = server.DB_PATH
        self._old_storage = server.STORAGE_DIR
        server.DB_PATH = self.db_path
        server.STORAGE_DIR = self.storage
        os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
        connection = connect(self.db_path)
        migrate(connection)
        assert server.ensure_bootstrap_admin(connection) is None
        connection.close()
        class QuietHandler(server.AppHandler):
            def log_message(self, *args) -> None:
                pass
        self.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
        self.port = self.httpd.server_address[1]
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()
        self.admin = Client("127.0.0.1", self.port)
        status, _, data = self.admin.post_json(
            "/api/login",
            {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
        )
        assert status == 200, data
        status, _, data = self.admin.post_json(
            "/api/password/change",
            {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
        )
        assert status == 200, data
        self.initial_passwords: dict[str, str] = {}
        self.company_a = self._create_company("甲公司", "cashier-a")
        self.company_b = self._create_company("乙公司", "cashier-b")
        self.company_c = self._create_company("丙公司", "cashier-c")
        self.cashier_a = self._login_company("cashier-a")
        self.cashier_b = self._login_company("cashier-b")
        self.cashier_c = self._login_company("cashier-c")
        self._approve_account(self.company_a, ACCOUNT_A, self.cashier_a)
        self._approve_account(self.company_b, ACCOUNT_B, self.cashier_b)
        self._approve_account(self.company_c, ACCOUNT_C, self.cashier_c)
    def tearDown(self) -> None:
        self.httpd.shutdown()
        self.httpd.server_close()
        server.DB_PATH = self._old_db_path
        server.STORAGE_DIR = self._old_storage
        os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
    def _create_company(self, name: str, username: str) -> int:
        status, _, data = self.admin.post_json(
            "/api/admin/companies", {"name": name, "username": username}
        )
        assert status == 200, data
        self.initial_passwords[username] = as_json(data)["initial_password"]
        return as_json(data)["company_id"]
    def _login_company(self, username: str) -> Client:
        client = Client("127.0.0.1", self.port)
        initial = self.initial_passwords[username]
        status, _, data = client.post_json(
            "/api/login", {"username": username, "password": initial, "portal": "company"}
        )
        assert status == 200, data
        status, _, data = client.post_json(
            "/api/password/change",
            {"old_password": initial, "new_password": CASHIER_PASSWORD},
        )
        assert status == 200, data
        return client
    def _approve_account(self, company_id: int, number: str, client: Client) -> int:
        status, _, data = client.post_json(
            "/api/company/accounts",
            {
                "bank_name": "中信银行",
                "account_type": "基本户",
                "account_number": number,
                "start_date": "2026-01-01",
            },
        )
        assert status == 200, data
        account_id = as_json(data)["account"]["id"]
        status, _, data = self.admin.post_json(
            f"/api/admin/accounts/{account_id}/review",
            {
                "decision": "approve",
                "reason": "测试启用",
                "effective_from": "2026-01-01",
            },
        )
        assert status == 200, data
        return account_id
    def _upload_and_confirm(self, client: Client, company_id: int, rows) -> int:
        content = workbook_bytes(rows)
        status, _, data = self.admin.post_multipart(
            "/api/parse", {"company_id": str(company_id)}, "账单.xlsx", content
        )
        assert status == 200, data
        batch_id = as_json(data)["batch_id"]
        status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
        names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
        status, _, data = client.post_json(
            f"/api/batches/{batch_id}/confirm", {"sheets": names}
        )
        assert status == 200, data
        return batch_id
    def _lock_single(self, amount: str, at: str = "2026-03-01 10:00:00") -> dict:
        self._upload_and_confirm(
            self.cashier_a,
            self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)],
        )
        status, _, data = self.admin.get("/api/admin/transfer-events")
        self.assertEqual(200, status, data)
        single = next(
            e
            for e in as_json(data)["events"]
            if e["status"] == "internal_single" and e["amount"] == amount
        )
        status, _, data = self.admin.get(f"/api/admin/transfer-events/{single['event_id']}")
        self.assertEqual(200, status, data)
        revision = as_json(data)["event"]["revision"]
        status, _, data = self.admin.post_json(
            f"/api/admin/transfer-events/{single['event_id']}/decisions",
            {
                "action": "assign_participant",
                "reason": "函证确认",
                "expected_revision": revision,
                "request_key": f"lock-{amount}-{at}",
                "participant": {"role": "payee", "company_id": self.company_b},
            },
        )
        self.assertEqual(200, status, data)
        return as_json(data)["decision"]
    def _seed_mixed(self) -> None:
        """Paired A→B 100, B→A 40, locked A→B 25, pending A→B 7, B→C 200."""
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, "100.00", "2026-01-05 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [incoming(ACCOUNT_B, ACCOUNT_A, "100.00", "2026-01-05 11:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [outgoing(ACCOUNT_B, ACCOUNT_A, "40.00", "2026-01-10 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [incoming(ACCOUNT_A, ACCOUNT_B, "40.00", "2026-01-10 11:00:00")],
        )
        self._lock_single("25.00", "2026-02-01 10:00:00")
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, "7.00", "2026-02-15 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [outgoing(ACCOUNT_B, ACCOUNT_C, "200.00", "2026-01-20 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_c, self.company_c,
            [incoming(ACCOUNT_C, ACCOUNT_B, "200.00", "2026-01-20 11:00:00")],
        )
    def _events(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
        status, _, data = client.get(f"/api/company/intercompany/events?{query}")
        return status, as_json(data) if data else {}
    def _export(self, client: Client, query: str = "from=2026-01-01&to=2026-12-31"):
        return client.get(f"/api/company/intercompany/export.csv?{query}")
    # ------------------------------------------------------------------
    # Auth / parameter guards
    # ------------------------------------------------------------------
    def test_requires_company_role(self) -> None:
        status, payload = self._events(self.admin)
        self.assertEqual(403, status, payload)
    def test_rejects_forged_company_id_on_events_and_export(self) -> None:
        status, payload = self._events(
            self.cashier_a,
            f"from=2026-01-01&to=2026-12-31&company_id={self.company_b}",
        )
        self.assertEqual(400, status, payload)
        status, _, data = self._export(
            self.cashier_a,
            f"from=2026-01-01&to=2026-12-31&company_id={self.company_a}",
        )
        self.assertEqual(400, status, data)
    def test_rejects_bad_filters(self) -> None:
        status, payload = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=sideways"
        )
        self.assertEqual(400, status, payload)
        status, payload = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&state=maybe"
        )
        self.assertEqual(400, status, payload)
        status, payload = self._events(
            self.cashier_a, "from=2026-13-40&to=2026-12-31"
        )
        self.assertEqual(400, status, payload)
    def test_empty_window(self) -> None:
        status, payload = self._events(self.cashier_a)
        self.assertEqual(200, status, payload)
        self.assertEqual([], payload["events"])
        self.assertFalse(payload["has_more"])
        self.assertIsNone(payload["next_cursor"])
    # ------------------------------------------------------------------
    # Combined filters + confirmed/pending separation
    # ------------------------------------------------------------------
    def test_combined_filters_and_state_split(self) -> None:
        self._seed_mixed()
        status, payload = self._events(self.cashier_a)
        self.assertEqual(200, status, payload)
        events = payload["events"]
        # A sees: out 100, in 40, locked out 25, pending out 7 — not B↔C 200
        self.assertEqual(4, len(events))
        amounts = {e["amount"] for e in events}
        self.assertEqual({"100.00", "40.00", "25.00", "7.00"}, amounts)
        for event in events:
            self.assertNotEqual(self.company_c, event["counterparty_company_id"])
            self.assertIn(event["state"], ("confirmed", "pending"))
            self.assertIn(event["direction"], ("out", "in"))
        status, confirmed = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&state=confirmed"
        )
        self.assertEqual(200, status, confirmed)
        self.assertEqual(3, len(confirmed["events"]))
        self.assertTrue(all(e["state"] == "confirmed" for e in confirmed["events"]))
        self.assertNotIn("7.00", {e["amount"] for e in confirmed["events"]})
        status, pending = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
        )
        self.assertEqual(200, status, pending)
        self.assertEqual(1, len(pending["events"]))
        self.assertEqual("7.00", pending["events"][0]["amount"])
        self.assertEqual("pending", pending["events"][0]["state"])
        status, outs = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&direction=out&state=confirmed"
        )
        self.assertEqual(200, status, outs)
        self.assertEqual({"100.00", "25.00"}, {e["amount"] for e in outs["events"]})
        self.assertTrue(all(e["direction"] == "out" for e in outs["events"]))
        status, by_cp = self._events(
            self.cashier_a,
            f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_b}&state=confirmed",
        )
        self.assertEqual(200, status, by_cp)
        self.assertEqual(3, len(by_cp["events"]))
        # Date window excludes Feb locked/pending
        status, jan = self._events(
            self.cashier_a, "from=2026-01-01&to=2026-01-31&state=confirmed"
        )
        self.assertEqual(200, status, jan)
        self.assertEqual({"100.00", "40.00"}, {e["amount"] for e in jan["events"]})
    def test_keyset_pagination_no_dup_no_gap(self) -> None:
        # Three confirmed A→B outs on distinct days
        for i, amount in enumerate(("11.00", "12.00", "13.00", "14.00", "15.00")):
            day = 5 + i
            self._upload_and_confirm(
                self.cashier_a, self.company_a,
                [outgoing(ACCOUNT_A, ACCOUNT_B, amount, f"2026-01-{day:02d} 10:00:00")],
            )
            self._upload_and_confirm(
                self.cashier_b, self.company_b,
                [incoming(ACCOUNT_B, ACCOUNT_A, amount, f"2026-01-{day:02d} 11:00:00")],
            )
        status, page1 = self._events(
            self.cashier_a,
            "from=2026-01-01&to=2026-12-31&state=confirmed&direction=out&limit=2",
        )
        self.assertEqual(200, status, page1)
        self.assertEqual(2, len(page1["events"]))
        self.assertTrue(page1["has_more"])
        self.assertIsNotNone(page1["next_cursor"])
        status, page2 = self._events(
            self.cashier_a,
            "from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
            f"&limit=2&cursor={page1['next_cursor']}",
        )
        self.assertEqual(200, status, page2)
        self.assertEqual(2, len(page2["events"]))
        self.assertTrue(page2["has_more"])
        status, page3 = self._events(
            self.cashier_a,
            "from=2026-01-01&to=2026-12-31&state=confirmed&direction=out"
            f"&limit=2&cursor={page2['next_cursor']}",
        )
        self.assertEqual(200, status, page3)
        self.assertEqual(1, len(page3["events"]))
        self.assertFalse(page3["has_more"])
        self.assertIsNone(page3["next_cursor"])
        ids = [e["event_id"] for e in page1["events"] + page2["events"] + page3["events"]]
        self.assertEqual(5, len(ids))
        self.assertEqual(len(ids), len(set(ids)))
        # Descending by effective_at then event_id
        amounts = [e["amount"] for e in page1["events"] + page2["events"] + page3["events"]]
        self.assertEqual(["15.00", "14.00", "13.00", "12.00", "11.00"], amounts)
    # ------------------------------------------------------------------
    # Isolation / ID guess / detail reuse
    # ------------------------------------------------------------------
    def test_lateral_isolation_and_id_guess_404(self) -> None:
        self._seed_mixed()
        status, payload_a = self._events(self.cashier_a)
        self.assertEqual(200, status, payload_a)
        a_ids = {e["event_id"] for e in payload_a["events"]}
        status, payload_c = self._events(self.cashier_c)
        self.assertEqual(200, status, payload_c)
        # C only participates in B↔C 200
        self.assertTrue(payload_c["events"])
        for event in payload_c["events"]:
            self.assertEqual("200.00", event["amount"])
            self.assertNotIn(event["event_id"], a_ids)
        # C guessing A's event id via transfer-events detail → 404
        a_event_id = next(iter(a_ids))
        status, _, data = self.cashier_c.get(
            f"/api/company/transfer-events/{a_event_id}"
        )
        self.assertEqual(404, status, data)
        # A can open own event; counterparty account masked; only own observations
        status, _, data = self.cashier_a.get(
            f"/api/company/transfer-events/{a_event_id}"
        )
        self.assertEqual(200, status, data)
        detail = as_json(data)["event"]
        if detail.get("counterparty") and "account_number_masked" in detail["counterparty"]:
            masked = detail["counterparty"]["account_number_masked"]
            self.assertTrue(str(masked).startswith("****"))
            self.assertNotIn(ACCOUNT_B, masked)
        for obs in detail["observations"]:
            self.assertEqual(self.company_a, obs["batch_company_id"])
            self.assertNotIn(ACCOUNT_B, json.dumps(obs, ensure_ascii=False))
        # Filtering by counterparty C still cannot leak B↔C into A's list
        status, filtered = self._events(
            self.cashier_a,
            f"from=2026-01-01&to=2026-12-31&counterparty_id={self.company_c}",
        )
        self.assertEqual(200, status, filtered)
        self.assertEqual([], filtered["events"])
    def test_b44_events_path_still_works_with_cutoff(self) -> None:
        # Without HEL-176 discriminators, /events stays on B-44 ledger list.
        status, _, data = self.cashier_a.get(
            "/api/company/intercompany/events?from=2026-01-01&cutoff=2026-12-31"
        )
        self.assertEqual(200, status, data)
        payload = as_json(data)
        self.assertIn("items", payload)
        self.assertNotIn("events", payload)
    # ------------------------------------------------------------------
    # Export: confirmed only + audit + isolation
    # ------------------------------------------------------------------
    def test_export_confirmed_only_isolated_and_audited(self) -> None:
        self._seed_mixed()
        status, headers, data = self._export(self.cashier_a)
        self.assertEqual(200, status, data)
        self.assertEqual("text/csv; charset=utf-8", headers.get("content-type"))
        text = data.decode("utf-8-sig")
        lines = [line for line in text.splitlines() if line]
        self.assertGreaterEqual(len(lines), 2)
        body = "\n".join(lines[1:])
        self.assertIn("100.00", body)
        self.assertIn("40.00", body)
        self.assertIn("25.00", body)
        self.assertNotIn("7.00", body)  # pending excluded
        self.assertNotIn("200.00", body)  # B↔C excluded
        # Explicit pending state rejected
        status, _, data = self._export(
            self.cashier_a, "from=2026-01-01&to=2026-12-31&state=pending"
        )
        self.assertEqual(400, status, data)
        # C export must not contain A's amounts
        status, _, data = self._export(self.cashier_c)
        self.assertEqual(200, status, data)
        text_c = data.decode("utf-8-sig")
        self.assertNotIn("100.00", text_c)
        self.assertNotIn("25.00", text_c)
        self.assertIn("200.00", text_c)
        status, _, data = self.admin.get("/api/admin/audit-log?limit=50")
        self.assertEqual(200, status, data)
        actions = [row["action"] for row in as_json(data)["entries"]]
        self.assertIn("export_intercompany_csv", actions)
        export_rows = [
            row
            for row in as_json(data)["entries"]
            if row["action"] == "export_intercompany_csv"
        ]
        self.assertTrue(export_rows)
        self.assertTrue(
            any(f"company:{self.company_a}" == row.get("target") for row in export_rows)
        )
        self.assertTrue(any("state:confirmed" in (row.get("detail") or "") for row in export_rows))
 if __name__ == "__main__":
    unittest.main()
@@ -1,362 +0,0 @@
 """HTTP tests for GET /api/company/intercompany/summary (HEL-175).
 Covers session-scoped company_id, forged company_id rejection, confirmed vs
 pending separation, Decimal net math, dual-company isolation and empty data.
 """
 from __future__ import annotations
 import io
 import json
 import os
 from decimal import Decimal
 from pathlib import Path
 import tempfile
 import threading
 import unittest
 from openpyxl import Workbook
 from bank_importer.db import connect, migrate
 import server
 from test_server_auth import Client, as_json
 BOOTSTRAP_PASSWORD = "BootAdmin123"
 ADMIN_PASSWORD = "AdminPass123"
 CASHIER_PASSWORD = "Cashier123"
 CCB_HEADER = [
    "客户账号", "账户名称", "交易时间", "借方发生额（支取）", "贷方发生额（收入）",
    "余额", "币种", "对方户名", "对方账号", "对方开户机构", "摘要", "备注",
 ]
 ACCOUNT_A = "6222000000000001"
 ACCOUNT_B = "6222000000000002"
 ACCOUNT_C = "6222000000000003"
 def workbook_bytes(rows) -> bytes:
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "正常流水"
    sheet.append(CCB_HEADER)
    for row in rows:
        sheet.append(row)
    buffer = io.BytesIO()
    workbook.save(buffer)
    return buffer.getvalue()
 def outgoing(own: str, cp: str, amount: str, at: str = "2026-01-05 10:00:00"):
    return [own, "测试公司", at, amount, "", "50000.00", "RMB", "对方", cp, "某银行", "货款", ""]
 def incoming(own: str, cp: str, amount: str, at: str = "2026-01-05 11:00:00"):
    return [own, "测试公司", at, "", amount, "50000.00", "RMB", "对方", cp, "某银行", "收款", ""]
 class CompanyIntercompanySummaryTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp_dir.cleanup)
        root = Path(self.temp_dir.name)
        self.db_path = root / "app.db"
        self.storage = root / "files"
        self._old_db_path = server.DB_PATH
        self._old_storage = server.STORAGE_DIR
        server.DB_PATH = self.db_path
        server.STORAGE_DIR = self.storage
        os.environ["APP_BOOTSTRAP_ADMIN_PASSWORD"] = BOOTSTRAP_PASSWORD
        connection = connect(self.db_path)
        migrate(connection)
        assert server.ensure_bootstrap_admin(connection) is None
        connection.close()
        class QuietHandler(server.AppHandler):
            def log_message(self, *args) -> None:
                pass
        self.httpd = server.ThreadingHTTPServer(("127.0.0.1", 0), QuietHandler)
        self.port = self.httpd.server_address[1]
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()
        self.admin = Client("127.0.0.1", self.port)
        status, _, data = self.admin.post_json(
            "/api/login",
            {"username": "group-admin", "password": BOOTSTRAP_PASSWORD, "portal": "admin"},
        )
        assert status == 200, data
        status, _, data = self.admin.post_json(
            "/api/password/change",
            {"old_password": BOOTSTRAP_PASSWORD, "new_password": ADMIN_PASSWORD},
        )
        assert status == 200, data
        self.initial_passwords: dict[str, str] = {}
        self.company_a = self._create_company("甲公司", "cashier-a")
        self.company_b = self._create_company("乙公司", "cashier-b")
        self.company_c = self._create_company("丙公司", "cashier-c")
        self.cashier_a = self._login_company("cashier-a")
        self.cashier_b = self._login_company("cashier-b")
        self.cashier_c = self._login_company("cashier-c")
        self._approve_account(self.company_a, ACCOUNT_A, self.cashier_a)
        self._approve_account(self.company_b, ACCOUNT_B, self.cashier_b)
        self._approve_account(self.company_c, ACCOUNT_C, self.cashier_c)
    def tearDown(self) -> None:
        self.httpd.shutdown()
        self.httpd.server_close()
        server.DB_PATH = self._old_db_path
        server.STORAGE_DIR = self._old_storage
        os.environ.pop("APP_BOOTSTRAP_ADMIN_PASSWORD", None)
    def _create_company(self, name: str, username: str) -> int:
        status, _, data = self.admin.post_json(
            "/api/admin/companies", {"name": name, "username": username}
        )
        assert status == 200, data
        self.initial_passwords[username] = as_json(data)["initial_password"]
        return as_json(data)["company_id"]
    def _login_company(self, username: str) -> Client:
        client = Client("127.0.0.1", self.port)
        initial = self.initial_passwords[username]
        status, _, data = client.post_json(
            "/api/login", {"username": username, "password": initial, "portal": "company"}
        )
        assert status == 200, data
        status, _, data = client.post_json(
            "/api/password/change",
            {"old_password": initial, "new_password": CASHIER_PASSWORD},
        )
        assert status == 200, data
        return client
    def _approve_account(self, company_id: int, number: str, client: Client) -> int:
        status, _, data = client.post_json(
            "/api/company/accounts",
            {
                "bank_name": "中信银行",
                "account_type": "基本户",
                "account_number": number,
                "start_date": "2026-01-01",
            },
        )
        assert status == 200, data
        account_id = as_json(data)["account"]["id"]
        status, _, data = self.admin.post_json(
            f"/api/admin/accounts/{account_id}/review",
            {
                "decision": "approve",
                "reason": "测试启用",
                "effective_from": "2026-01-01",
            },
        )
        assert status == 200, data
        return account_id
    def _upload_and_confirm(self, client: Client, company_id: int, rows) -> int:
        content = workbook_bytes(rows)
        status, _, data = self.admin.post_multipart(
            "/api/parse", {"company_id": str(company_id)}, "账单.xlsx", content
        )
        assert status == 200, data
        batch_id = as_json(data)["batch_id"]
        status, _, data = client.get(f"/api/batches/{batch_id}/sheets")
        names = [s["sheet_name"] for s in as_json(data)["sheets"] if s["outcome"] == "parsed"]
        status, _, data = client.post_json(
            f"/api/batches/{batch_id}/confirm", {"sheets": names}
        )
        assert status == 200, data
        return batch_id
    def _lock_single(self, amount: str, at: str = "2026-03-01 10:00:00") -> dict:
        """A-side only upload → admin locks as intercompany single."""
        self._upload_and_confirm(
            self.cashier_a,
            self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, amount, at)],
        )
        status, _, data = self.admin.get("/api/admin/transfer-events")
        self.assertEqual(200, status, data)
        single = next(
            e
            for e in as_json(data)["events"]
            if e["status"] == "internal_single" and e["amount"] == amount
        )
        status, _, data = self.admin.get(f"/api/admin/transfer-events/{single['event_id']}")
        self.assertEqual(200, status, data)
        revision = as_json(data)["event"]["revision"]
        status, _, data = self.admin.post_json(
            f"/api/admin/transfer-events/{single['event_id']}/decisions",
            {
                "action": "assign_participant",
                "reason": "函证确认",
                "expected_revision": revision,
                "request_key": f"lock-{amount}-{at}",
                "participant": {"role": "payee", "company_id": self.company_b},
            },
        )
        self.assertEqual(200, status, data)
        return as_json(data)["decision"]
    def _summary(self, client: Client, query: str = "as_of=2026-12-31"):
        status, _, data = client.get(f"/api/company/intercompany/summary?{query}")
        return status, as_json(data) if data else {}
    # ------------------------------------------------------------------
    # Auth / parameter guards
    # ------------------------------------------------------------------
    def test_requires_company_role(self) -> None:
        status, payload = self._summary(self.admin)
        self.assertEqual(403, status, payload)
    def test_rejects_forged_company_id(self) -> None:
        status, payload = self._summary(
            self.cashier_a, f"as_of=2026-12-31&company_id={self.company_b}"
        )
        self.assertEqual(400, status, payload)
        self.assertIn("company_id", payload.get("message", ""))
    def test_rejects_own_company_id_param(self) -> None:
        # Even matching the session company is forbidden.
        status, payload = self._summary(
            self.cashier_a, f"as_of=2026-12-31&company_id={self.company_a}"
        )
        self.assertEqual(400, status, payload)
    def test_rejects_bad_as_of(self) -> None:
        status, payload = self._summary(self.cashier_a, "as_of=2026-13-40")
        self.assertEqual(400, status, payload)
    def test_anonymous_is_unauthorized(self) -> None:
        anon = Client("127.0.0.1", self.port)
        status, _, data = anon.get("/api/company/intercompany/summary?as_of=2026-12-31")
        self.assertIn(status, (401, 403))
    # ------------------------------------------------------------------
    # Empty / confirmed math / pending isolation
    # ------------------------------------------------------------------
    def test_empty_window_returns_zeros(self) -> None:
        status, payload = self._summary(self.cashier_a)
        self.assertEqual(200, status, payload)
        self.assertEqual(self.company_a, payload["own_company"]["id"])
        self.assertFalse(payload["window"]["has_opening"])
        self.assertIsNone(payload["window"]["opening"])
        self.assertEqual("0.00", payload["confirmed"]["outflow_total"])
        self.assertEqual("0.00", payload["confirmed"]["inflow_total"])
        self.assertEqual("0.00", payload["confirmed"]["net_change"])
        self.assertEqual(0, payload["pending"]["count"])
        self.assertEqual([], payload["counterparties"])
        # Amounts must be strings, never floats.
        self.assertIsInstance(payload["confirmed"]["net_change"], str)
        self.assertNotIsInstance(payload["confirmed"]["net_change"], float)
    def test_paired_locked_pending_math_and_isolation(self) -> None:
        # Paired A→B 100 + B→A 40 → A net outflow 60
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, "100.00", "2026-01-05 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [incoming(ACCOUNT_B, ACCOUNT_A, "100.00", "2026-01-05 11:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [outgoing(ACCOUNT_B, ACCOUNT_A, "40.00", "2026-01-10 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [incoming(ACCOUNT_A, ACCOUNT_B, "40.00", "2026-01-10 11:00:00")],
        )
        # Locked single A→B 25 (confirmed)
        self._lock_single("25.00", "2026-02-01 10:00:00")
        # Pending unilateral A→B 7 (internal_single, not confirmed)
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, "7.00", "2026-02-15 10:00:00")],
        )
        # B↔C paired 200 — must not appear in A's summary
        self._upload_and_confirm(
            self.cashier_b, self.company_b,
            [outgoing(ACCOUNT_B, ACCOUNT_C, "200.00", "2026-01-20 10:00:00")],
        )
        self._upload_and_confirm(
            self.cashier_c, self.company_c,
            [incoming(ACCOUNT_C, ACCOUNT_B, "200.00", "2026-01-20 11:00:00")],
        )
        status, payload = self._summary(self.cashier_a)
        self.assertEqual(200, status, payload)
        confirmed = payload["confirmed"]
        self.assertEqual("125.00", confirmed["outflow_total"])  # 100 + 25
        self.assertEqual(2, confirmed["outflow_count"])
        self.assertEqual("40.00", confirmed["inflow_total"])
        self.assertEqual(1, confirmed["inflow_count"])
        self.assertEqual("85.00", confirmed["net_change"])  # 125 - 40
        self.assertEqual("receivable", confirmed["net_direction"])
        # Pending tip only — never folded into confirmed totals
        self.assertEqual(1, payload["pending"]["count"])
        self.assertEqual("7.00", payload["pending"]["amount_total"])
        # Decimal identity: net = outflow - inflow, no float drift
        net = Decimal(confirmed["net_change"])
        self.assertEqual(
            Decimal(confirmed["outflow_total"]) - Decimal(confirmed["inflow_total"]),
            net,
        )
        counterparties = {row["company_id"]: row for row in payload["counterparties"]}
        self.assertIn(self.company_b, counterparties)
        self.assertNotIn(self.company_c, counterparties)
        row_b = counterparties[self.company_b]
        self.assertEqual("125.00", row_b["confirmed_outflow"])
        self.assertEqual("40.00", row_b["confirmed_inflow"])
        self.assertEqual("85.00", row_b["net"])
        self.assertEqual(1, row_b["pending_count"])
        # B must not see C-only? B sees C; A must not see B's C totals via forgery
        status_b, payload_b = self._summary(self.cashier_b)
        self.assertEqual(200, status_b, payload_b)
        cps_b = {row["company_id"] for row in payload_b["counterparties"]}
        self.assertIn(self.company_c, cps_b)
        # A's view still excludes C
        self.assertNotIn(self.company_c, counterparties)
        # Same event counted once: eligible event count equals outflow+inflow counts
        self.assertEqual(
            confirmed["outflow_count"] + confirmed["inflow_count"],
            3,
        )
    def test_company_b_cannot_see_a_only_pending(self) -> None:
        self._upload_and_confirm(
            self.cashier_a, self.company_a,
            [outgoing(ACCOUNT_A, ACCOUNT_B, "9.00", "2026-04-01 10:00:00")],
        )
        status_a, payload_a = self._summary(self.cashier_a)
        self.assertEqual(200, status_a, payload_a)
        self.assertEqual(1, payload_a["pending"]["count"])
        status_c, payload_c = self._summary(self.cashier_c)
        self.assertEqual(200, status_c, payload_c)
        self.assertEqual(0, payload_c["pending"]["count"])
        self.assertEqual("0.00", payload_c["confirmed"]["outflow_total"])
        self.assertEqual([], payload_c["counterparties"])
 if __name__ == "__main__":
    unittest.main()
@@ -2619,6 +2619,18 @@ function applyCompanyWorkspace(payload) {
      ? `单边流水 ${pending} 笔待确认`
      : "已全部确认，等待集团结账";
  }
  // 完成态必须切到 success 绿（.flow-step.done），待确认保留 warn 黄（.doing）
  const flowStep = flowState?.closest(".flow-step");
  if (flowStep) {
    flowStep.classList.toggle("doing", pending > 0);
    flowStep.classList.toggle("done", pending === 0);
  }
  const flowSub = $("#workspaceFlowSub");
  if (flowSub) {
    flowSub.textContent = pending
      ? "当前停在第 3 步「往来确认」，完成后即可等待集团结账"
      : "第 3 步「往来确认」已完成，等待集团复核与结账";
  }
  const countMatch = $("#count-match");
  if (countMatch) countMatch.textContent = String(pending);
@@ -63,7 +63,7 @@
            <div class="card">
              <div class="card-head">
-                <span class="card-title">账期流程 · 2026-07<span class="sub">当前停在第 3 步「往来确认」，完成后即可等待集团结账</span></span>
+                <span class="card-title">账期流程 · 2026-07<span class="sub" id="workspaceFlowSub">当前停在第 3 步「往来确认」，完成后即可等待集团结账</span></span>
                <span class="pill pill-warn">结账日顺延至 08-29</span>
              </div>
              <div class="flow">
@@ -814,6 +814,6 @@
      </div>
    </div>
    <div class="toast-region" id="toastRegion" aria-live="polite"></div>
-    <script src="app.js?v=10"></script>
+    <script src="app.js?v=11"></script>
  </body>
 </html>