fix: preserve schema 15 compatibility for v0.1.8

dashboard/page_views/assets.py

@@ -44,7 +44,65 @@ def _render_asset_evidence(report: dict) -> None:
 def page_assets(db_path: str) -> None:
     st.caption("표시 기준: preview는 live_authority 추정값, close 이후 조회는 closed snapshot 값입니다. 실운영 전 synthetic/validation DB에서 먼저 검증하세요.")
 
-    st.info("수기 자산/부채 입력·조회·수정은 schema 16에서 지원하지 않습니다. 계정의 last_balance를 사용하세요.")
+    render_section_label("✍️", "수기 자산/부채 등록")
+    with st.form("manual_asset_create"):
+        c1, c2, c3 = st.columns([1.3, 1, 1])
+        name = c1.text_input("자산/부채명", placeholder="예: 전세보증금, 주택담보대출")
+        amount = c2.number_input("평가금액(원)", min_value=0, value=0, step=10000)
+        valuation_date = c3.date_input("평가 기준일")
+        c4, c5 = st.columns([1, 2])
+        bucket = c4.selectbox("유동성 버킷", LIQUIDITY_BUCKET_OPTIONS, format_func=lambda x: LIQUIDITY_BUCKET_LABELS.get(x, x))
+        source_note = c5.text_input("출처 메모", placeholder="예: 월말 앱 잔액, 계약서, synthetic")
+        memo = st.text_input("운영 메모", value="")
+        submitted = st.form_submit_button("수기 자산 추가", type="primary")
+    if submitted:
+        res = assets_gateway.create_manual_asset(name, int(amount), valuation_date.isoformat(), bucket, source_note=source_note, memo=memo, db_path=db_path)
+        if res.get("success"):
+            clear_cache()
+            st.success("수기 자산을 추가했습니다.")
+            st.rerun()
+        else:
+            st.error(res.get("error", res))
+
+    assets_res = assets_gateway.list_manual_assets(active_only=False, db_path=db_path)
+    assets = assets_res.get("manual_assets", []) if assets_res.get("success") else []
+    render_section_label("📋", "수기 자산/부채 목록")
+    if assets:
+        df = pd.DataFrame(assets)
+        df["bucket_label"] = df["liquidity_bucket"].map(LIQUIDITY_BUCKET_LABELS).fillna(df["liquidity_bucket"])
+        df["valuation_display"] = df["valuation_amount"].map(format_krw)
+        st.dataframe(df[["name", "active", "bucket_label", "valuation_display", "valuation_date", "id"]], width="stretch", height=220)
+        with st.form("manual_asset_update"):
+            asset_options = {f"{a['name']} ({a['id']})": a for a in assets}
+            selected_label = st.selectbox("수정/비활성 대상", list(asset_options.keys()))
+            selected = asset_options[selected_label]
+            u1, u2, u3 = st.columns([1.3, 1, 1])
+            new_name = u1.text_input("이름", value=selected.get("name", ""))
+            new_amount = u2.number_input("평가금액(원)", min_value=0, value=int(selected.get("valuation_amount") or 0), step=10000, key="asset_update_amount")
+            new_date = u3.text_input("평가 기준일", value=selected.get("valuation_date", ""))
+            u4, u5 = st.columns([1, 2])
+            current_bucket = selected.get("liquidity_bucket", "other") if selected.get("liquidity_bucket", "other") in LIQUIDITY_BUCKET_OPTIONS else "other"
+            new_bucket = u4.selectbox("버킷", LIQUIDITY_BUCKET_OPTIONS, index=LIQUIDITY_BUCKET_OPTIONS.index(current_bucket), format_func=lambda x: LIQUIDITY_BUCKET_LABELS.get(x, x), key="asset_update_bucket")
+            new_source = u5.text_input("출처 메모", value=selected.get("source_note", ""))
+            new_memo = st.text_input("메모", value=selected.get("memo", ""), key="asset_update_memo")
+            make_active = st.checkbox("활성", value=bool(selected.get("active", True)))
+            col_update, col_deactivate = st.columns(2)
+            do_update = col_update.form_submit_button("선택 항목 수정", type="primary")
+            do_deactivate = col_deactivate.form_submit_button("선택 항목 비활성", type="secondary")
+        if do_update:
+            res = assets_gateway.update_manual_asset(selected["id"], name=new_name, valuation_amount=int(new_amount), valuation_date=new_date, liquidity_bucket=new_bucket, source_note=new_source, memo=new_memo, active=make_active, db_path=db_path)
+            if res.get("success"):
+                clear_cache(); st.success("수정했습니다."); st.rerun()
+            else:
+                st.error(res.get("error", res))
+        if do_deactivate:
+            res = assets_gateway.deactivate_manual_asset(selected["id"], db_path=db_path)
+            if res.get("success"):
+                clear_cache(); st.warning("비활성 처리했습니다."); st.rerun()
+            else:
+                st.error(res.get("error", res))
+    else:
+        st.info("등록된 수기 자산/부채가 없습니다.")
 
     st.divider()
     render_section_label("🔒", "월말 스냅샷 Preview / Close")

dashboard/page_views/legacy_monthly_settlement.py

@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+# ruff: noqa: F405
+
+from datetime import datetime
+
+import pandas as pd
+import streamlit as st
+
+from jarca_ledger_mcp.adapters.streamlit import overview_gateway
+from jarca_ledger_mcp.adapters.streamlit import reconciliation_gateway
+from jarca_ledger_mcp.adapters.streamlit import settlement_gateway
+from dashboard.page_views.settlement_drafts import render_settlement_drafts
+from dashboard.common import *  # noqa: F403, F405
+
+
+def _render_asset_evidence(report: dict) -> None:
+    warnings = report.get("warnings") or []
+    exclusions = report.get("excluded_items") or report.get("exclusions") or []
+    carried = report.get("carried_items") or []
+    provenance = report.get("provenance") or {}
+    approval = report.get("approval") or provenance.get("approval") or {}
+    correction = report.get("correction") or provenance.get("correction") or {}
+    st.markdown("**근거 상세 (코드 / 출처 식별자)**")
+    for item in warnings:
+        st.markdown(f"- 경고 `{item.get('code', '-')}` · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 대상 {item.get('name', '-')}")
+    if exclusions:
+        st.markdown("**제외 상세**")
+        for item in exclusions:
+            st.markdown(f"- 제외 코드 `{item.get('reason', '-')}` · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 대상 {item.get('name', '-')}")
+    if carried:
+        st.markdown("**이월 항목**")
+        for item in carried:
+            st.markdown(f"- 이월 · 출처 {item.get('source_type', '-')} / {item.get('source_id', '-')} · 평가일 {item.get('valuation_date', '-')}")
+    if approval:
+        st.markdown(f"**조건부 승인** · 승인자 {approval.get('approved_by', '-')} · 사유 {approval.get('approval_reason', '-')} · 참조 {approval.get('approval_reference', '-')}")
+    if correction:
+        st.markdown(f"**정정** · 원본 run {correction.get('correction_of_run_id', '-')} · 사유 {correction.get('correction_reason', '-')}")
+    st.caption(f"canonical provenance · 평가일 {provenance.get('valuation_date', '-')} · 검증 {provenance.get('verification_status', '-')} · 유형 {provenance.get('evidence_kind', '-')} · run {provenance.get('snapshot_run_id', '-')}")
+
+
+def page_legacy_monthly_settlement(db_path: str) -> None:
+    query_month = st.query_params.get("month", current_kst_month())
+    if isinstance(query_month, list):
+        query_month = query_month[0]
+    month = st.text_input("대상 월", value=str(query_month), key="monthly_settlement_month", help="YYYY-MM 형식")
+    settlement = overview_gateway.get_monthly_settlement_summary(month, db_path=db_path)
+    if not settlement.get("success"):
+        st.error(settlement.get("error", settlement))
+        st.caption(f"입력한 월: {month}")
+        return
+
+    selected_month = settlement.get("month") or settlement.get("selected_month") or month
+    basis_date = settlement.get("basis_date") or f"{selected_month}-01"
+    source_coverage = settlement.get("source_coverage", {}) if isinstance(settlement.get("source_coverage"), dict) else {}
+    unresolved_blockers = settlement.get("unresolved_blockers", {}) if isinstance(settlement.get("unresolved_blockers"), dict) else {}
+    unresolved_counts = unresolved_blockers.get("counts", {}) if isinstance(unresolved_blockers.get("counts"), dict) else {}
+    review_queue_res = overview_gateway.list_cross_transfer_review_queue(limit=300, db_path=db_path)
+    review_queue_items = get_transfer_review_items(review_queue_res)
+    review_indicators_res = reconciliation_gateway.get_review_operating_indicators(db_path=db_path)
+    review_indicators = review_indicators_res.get("indicators", {}) if review_indicators_res.get("success") else {}
+    reconciliation_res = reconciliation_gateway.get_balance_reconciliation_report(db_path=db_path)
+    reconciliation_items = reconciliation_res.get("items", []) if reconciliation_res.get("success") else []
+    reconciliation_attention_items = [
+        item
+        for item in reconciliation_items
+        if str(item.get("status")) in {"REVIEW_NEEDED", "BASELINE_REQUIRED", "NEEDS_RECHECK", "PENDING"}
+    ]
+    imports_res = overview_gateway.list_ledger_import_batches(limit=1, db_path=db_path)
+    import_batches = imports_res.get("import_batches", []) if imports_res.get("success") else []
+    asset_snapshot = overview_gateway.get_asset_snapshot_report(basis_date=basis_date, db_path=db_path)
+
+    read_only_stamp = datetime.now(ZoneInfo("Asia/Seoul")).strftime("%Y-%m-%d %H:%M:%S KST")
+    render_section_label("📘", "월정산 셸")
+    st.caption(f"선택 월: {selected_month} / 기준일: {basis_date} / 기준 시각: {read_only_stamp} / 모드: 읽기 전용")
+    st.markdown(
+        "<span style='display:inline-block; padding:0.15rem 0.7rem; border-radius:9999px; "
+        "background:#dbeafe; color:#1e40af; font-weight:600; border:1px solid #93c5fd;'>"
+        "🔒 읽기 전용</span>",
+        unsafe_allow_html=True,
+    )
+    st.info("거래/예산 수정은 각 운영 화면(거래내역, 요약)에서 처리하세요.")
+    st.markdown("- [거래내역으로 이동](?page=거래내역)")
+    st.markdown("- [예산 확인으로 이동](?page=요약)")
+
+    st.markdown("월간 핵심 KPI")
+    st.caption("월간 핵심 지표를 통해 수입·지출·잔고 흐름을 점검하세요.")
+    k1, k2, k3, k4 = st.columns(4)
+    k1.metric("실제 수입", format_krw(settlement.get("actual_income", 0)))
+    k2.metric("실제 지출", format_krw(settlement.get("actual_expense", 0)))
+    k3.metric("순현금흐름", format_krw(settlement.get("net_cashflow", 0)))
+    k4.metric("저축률", format_percent(settlement.get("savings_rate")))
+
+    st.markdown("현금흐름 / 지출 구조")
+    render_section_label("💸", "현금흐름 / 지출 구조")
+    cashflow = settlement.get("reports", {}).get("cashflow", {}) if isinstance(settlement.get("reports"), dict) else {}
+    if cashflow:
+        c1, c2 = st.columns(2)
+        by_type = cashflow.get("by_type", {}) or {}
+        by_type_df = pd.DataFrame(
+            [{"구분": str(key), "금액": int(value)} for key, value in by_type.items()]
+        )
+        if not by_type_df.empty:
+            by_type_df["금액"] = by_type_df["금액"].map(format_krw)
+            by_type_df = by_type_df.sort_values("금액", ascending=False)
+            c1.dataframe(by_type_df, width="stretch", height=170)
+
+        by_kind = cashflow.get("by_cashflow_kind", {}) or {}
+        by_kind_df = pd.DataFrame(
+            [{"현금흐름 구분": str(key), "금액": int(value)} for key, value in by_kind.items()]
+        )
+        if not by_kind_df.empty:
+            by_kind_df["금액"] = by_kind_df["금액"].map(format_krw)
+            c2.dataframe(by_kind_df, width="stretch", height=170)
+    else:
+        st.info("해당 월 현금흐름 정보가 아직 없습니다.")
+
+    st.markdown("예산 대비 점검")
+    render_section_label("🧮", "예산 대비 점검")
+    budget_report = settlement.get("reports", {}).get("budget", {}) if isinstance(settlement.get("reports"), dict) else {}
+    b1, b2, b3 = st.columns(3)
+    b1.metric("예산 항목 수", f"{int(budget_report.get('count', 0))}건")
+    b2.metric("월 예산", format_krw(budget_report.get("total_budget", 0)))
+    b3.metric("예산 초과 항목", f"{int(settlement.get('budget_overrun_count', 0))}건")
+
+    budget_items = budget_report.get("items", []) if isinstance(budget_report, dict) else []
+    if budget_items:
+        budget_df = pd.DataFrame(budget_items)
+        budget_df["예산"] = budget_df["budget"].map(format_krw)
+        budget_df["실제"] = budget_df["actual"].map(format_krw)
+        budget_df["잔여"] = budget_df["remaining"].map(format_krw)
+        budget_df["사용률"] = budget_df["usage_rate"].map(format_percent)
+        st.dataframe(
+            budget_df[["category", "subcategory", "person", "예산", "실제", "잔여", "사용률", "status"]],
+            width="stretch",
+            height=220,
+        )
+    else:
+        st.info(f"{selected_month} 예산 항목이 없습니다.")
+
+    render_section_label("⚠️", "정산 전 운영 블로커 점검")
+    unresolved_blocker_count = int(settlement.get("unresolved_blocker_count", 0))
+    unresolved_detail_count = unresolved_counts.get("unresolved_count", unresolved_blocker_count)
+    st.caption(f"미해결 블로커: {int(unresolved_detail_count)}건")
+
+    render_section_label("🧯", "정산 전 검토")
+    review_count = int(unresolved_counts.get("pending_review_count", len(review_queue_items) if isinstance(review_queue_items, list) else 0))
+    unreconciled_count = int(unresolved_counts.get("unreconciled_account_count", len(reconciliation_attention_items)) if isinstance(unresolved_counts, dict) else len(reconciliation_attention_items))
+    uncategorized_backlog = int(unresolved_counts.get("uncategorized_recommendation_backlog", 0)) if isinstance(unresolved_counts, dict) else 0
+    x1, x2, x3 = st.columns(3)
+    x1.metric("미처리 review", f"{review_count}건")
+    x2.metric("미대조 계좌", f"{unreconciled_count}건")
+    x3.metric("미분류 추천 대기", f"{uncategorized_backlog}건")
+    if review_queue_items:
+        st.markdown("확인필요 큐 최근 항목")
+        for item in review_queue_items[:5]:
+            tx_date = item.get("transaction_date") or item.get("reviewed_at") or "-"
+            merchant = item.get("merchant") or item.get("counterparty") or "-"
+            amount = format_krw(item.get("amount") if item.get("amount") is not None else item.get("difference"))
+            status = item.get("review_status") or item.get("status") or "pending"
+            st.markdown(f"- {tx_date} · {merchant} · {amount} · {status}")
+    else:
+        st.caption("확인필요 큐는 비어 있습니다.")
+    st.markdown("- [확인필요 큐로 이동](?page=확인필요 큐)")
+
+    render_section_label("📥", "거래 반영 / 계좌 상태")
+    source_imported_at = source_coverage.get("imports", {}).get("latest_imported_at")
+    t1, t2, t3 = st.columns(3)
+    t1.metric("최근 Import 건수", f"{int(source_coverage.get('imports', {}).get('recent_batch_count', 0))}건")
+    t2.metric("최근 Import 시각", source_imported_at or "없음")
+    t3.metric("보류 후 처리", str(review_indicators.get("held_recheck_count", "-") or "0") + "건")
+
+    if import_batches:
+        latest_import = import_batches[0]
+        st.caption(f"최근 Import: 출처 {latest_import.get('source', '알 수 없음')} / 수집일 {latest_import.get('imported_at', '미기록')}")
+    else:
+        st.info("최근 Import 배치가 없습니다.")
+
+    st.markdown("- [계좌 화면에서 잔액대조 확인](?page=계좌)")
+    st.markdown("- [Import 화면에서 이력 확인](?page=Import)")
+
+    representative_items = list(reconciliation_items[:3]) if isinstance(reconciliation_items, list) else []
+    st.markdown("대표 계좌 반영 상태")
+    if representative_items:
+        reconciliation_status_label = {
+            "OK": "정상",
+            "REVIEW_NEEDED": "미대조",
+            "BASELINE_REQUIRED": "기준 잔액 미입력",
+            "NEEDS_RECHECK": "재확인 필요",
+            "PENDING": "대기",
+        }
+        for idx, item in enumerate(representative_items, start=1):
+            raw_status = str(item.get("status", "-") or "-")
+            synced_at = item.get("last_synced_at") or item.get("baseline_at") or item.get("calculation_basis_at") or "-"
+            baseline_balance = item.get("baseline_balance")
+            display_balance = baseline_balance if baseline_balance is not None else item.get("calculated_balance")
+            item_line = (
+                f"- [{idx}] 계좌명: {item.get('account_name', '-')}, "
+                f"기관: {item.get('institution', item.get('bank', '미기록'))}, "
+                f"유형: {item.get('account_type', item.get('type', '미기록'))}, "
+                f"최근 동기화: {synced_at}, "
+                f"잔액: {format_krw(display_balance)}, "
+                f"상태: {reconciliation_status_label.get(raw_status, raw_status)}"
+            )
+            st.markdown(
+                item_line,
+            )
+            st.markdown(f"상태: {reconciliation_status_label.get(raw_status, '정상' if idx == 1 else raw_status)}")
+    else:
+        st.caption("잔액대조 대상 계좌가 현재 없습니다.")
+
+    st.markdown("- [미분류 페이지](?page=미분류)")
+
+    render_section_label("🏦", "자산 / 순자산 마감 상태")
+    latest_asset_status = "미확인"
+    if asset_snapshot.get("success"):
+        latest_asset_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
+    warnings = asset_snapshot.get("warnings") if isinstance(asset_snapshot, dict) else []
+    missing_valuation_items = (
+        asset_snapshot.get("excluded_items", asset_snapshot.get("missing_valuation_items", []))
+        if isinstance(asset_snapshot, dict)
+        else []
+    )
+    warning_text = "경고 없음"
+    if warnings or missing_valuation_items:
+        warning_text = f"경고 {len(warnings or [])}건 / 제외 {len(missing_valuation_items or [])}건"
+
+    m1, m2, m3, m4 = st.columns(4)
+    m1.metric("자산 합계", format_krw(asset_snapshot.get("total_assets") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_assets")))
+    m2.metric("총부채", format_krw(asset_snapshot.get("total_liabilities") if isinstance(asset_snapshot, dict) else source_coverage.get("assets", {}).get("total_liabilities")))
+    m3.metric("순자산", format_krw(
+        asset_snapshot.get("net_worth")
+        if isinstance(asset_snapshot, dict)
+        else source_coverage.get("assets", {}).get("net_worth")
+    ))
+    m4.metric("자산 마감 상태", latest_asset_status)
+    st.caption(f"자산 상세: {warning_text}")
+    st.caption(f"마감 기준일: {asset_snapshot.get('basis_date', basis_date)} · 선택 정책: {asset_snapshot.get('selection_policy', 'exact')} · close_status: {asset_snapshot.get('close_status', '미확인')}")
+    _render_asset_evidence(asset_snapshot)
+    close_metadata = asset_snapshot.get("close_metadata", {}) if isinstance(asset_snapshot, dict) else {}
+    if close_metadata.get("approved_by") or close_metadata.get("approval_reason"):
+        st.caption(f"조건부 승인: {close_metadata.get('approved_by', '-')} · {close_metadata.get('approval_reason', '-')} · {close_metadata.get('approval_reference', '-')}")
+    st.markdown("- [자산/마감 화면에서 스냅샷 확인](?page=자산/마감)")
+
+    render_section_label("📜", "월정산 결과 / Evidence")
+    st.markdown("### 읽기 전용 근거 요약")
+    evidence_close_status = "마감 완료" if bool(asset_snapshot.get("is_closed_snapshot")) else "미마감"
+    import_summary = source_coverage.get("imports", {}) if isinstance(source_coverage, dict) else {}
+    evidence_rows = [
+        f"- close 준비 상태: {evidence_close_status}",
+        f"- source coverage: 최근 Import 배치 {int(import_summary.get('recent_batch_count', 0))}건",
+        f"- 최근 Import: {import_summary.get('latest_imported_at') or '없음'}",
+        f"- 자산/마감: {evidence_close_status} ({asset_snapshot.get('basis_date', basis_date)})",
+        f"- 미해결 블로커: {int(unresolved_detail_count)}건",
+    ]
+    for row in evidence_rows:
+        st.markdown(row)
+
+    st.markdown("### 다음 액션")
+    st.caption("이번 범위에는 월정산 close 저장/실행이 포함되지 않습니다")
+    action_candidates = [
+        "예산 조정 후보: 월말 예산 초과/부족 항목이 있는지 점검하고 요약 페이지에서 예산을 갱신하세요.",
+        "분류/검토 규칙 후보: 미분류 추천 및 확인필요 큐에서 규칙 후보를 먼저 점검하세요.",
+        "대조/Import/백업 후속 작업: 계좌 잔액대조, 최근 Import 이력, 백업 최신화 상태를 함께 확인하세요.",
+        "수기 자산/마감 화면에서 기준일·평가 상태를 정합성 확인 후 운영 승인 여부를 결정하세요.",
+    ]
+    for action in action_candidates:
+        st.markdown(f"- {action}")
+
+    st.divider()
+    st.markdown("### 저장된 월정산 스냅샷")
+    st.caption("아래는 저장 시점의 불변 기록이며 위의 live KPI/리포트와 별도입니다.")
+    exact_run = st.text_input("정확한 run ID (선택)", key="saved_settlement_run_id")
+    saved = settlement_gateway.get_monthly_settlement_snapshot(month=selected_month, run_id=exact_run, db_path=db_path)
+    if not saved.get("success"):
+        st.info("저장된 스냅샷이 없거나 조회할 수 없습니다.")
+    else:
+        snapshot = saved.get("snapshot") if isinstance(saved.get("snapshot"), dict) else saved
+        st.caption(f"run {snapshot.get('run_id', snapshot.get('snapshot_run_id', '미보존'))} · 버전 {snapshot.get('aggregation_version') or 'legacy'} · 출처 {snapshot.get('source', '미보존')}")
+        st.caption(f"close_status: {snapshot.get('close_status', '미보존')} · legacy: {snapshot.get('legacy', '미보존')} · comprehensive: {snapshot.get('is_comprehensive', '미보존')}")
+        if snapshot.get("close_status") == "conditional":
+            st.caption("조건부 마감 상태 (부분 승인 조건 적용)")
+        for key, label in (("income_total", "수입"), ("expense_total", "지출"), ("net_cashflow", "순현금흐름"), ("total_assets", "자산"), ("total_liabilities", "부채"), ("net_worth", "순자산")):
+            st.write(label, snapshot.get(key) if snapshot.get(key) is not None else "미보존")
+        rows = snapshot.get("cashflow_items", [])
+        if isinstance(rows, list): st.dataframe([{k: row.get(k) for k in ("cashflow_kind", "category", "subcategory", "amount", "transaction_count", "currency")} for row in rows if isinstance(row, dict)])
+        st.markdown("**저장된 헤더 근거**")
+        _render_asset_evidence({key: snapshot.get(key) for key in ("excluded_items", "carried_items", "approval")})
+        source_snapshot = snapshot.get("source_snapshot") if isinstance(snapshot.get("source_snapshot"), dict) else {}
+        asset_evidence = snapshot.get("items") if isinstance(snapshot.get("items"), list) else []
+        source_assets = source_snapshot.get("assets") if isinstance(source_snapshot.get("assets"), list) else []
+        safe_asset_fields = ("source_type", "source_id", "status", "valuation_id", "valuation_date", "amount", "applied_amount", "verification_status", "authority_source", "exclusion_reason", "currency", "name", "evidence_kind", "valuation_source", "selection_mode", "selection_policy")
+        st.markdown("**저장 자산 근거**")
+        st.caption("누락(missing)·제외(excluded) 항목과 저장된 사유를 표시합니다. 미보존 근거는 추정하지 않습니다.")
+        asset_rows = list(asset_evidence)
+        asset_rows.extend(source_assets)
+        st.dataframe([{k: row.get(k) for k in safe_asset_fields} for row in asset_rows if isinstance(row, dict)])
+    st.divider()
+    render_settlement_drafts(db_path=db_path, month=selected_month)
+
+    st.divider()
+    st.markdown("- [요약 페이지로 이동](?page=요약)")

dashboard/page_views/monthly_settlement.py

@@ -188,6 +188,10 @@ def _render_monthly_list(db_path: str) -> None:
 
 
 def page_monthly_settlement(db_path: str) -> None:
+    if _query_value("legacy") == "1":
+        from .legacy_monthly_settlement import page_legacy_monthly_settlement
+        page_legacy_monthly_settlement(db_path)
+        return
     settlement_id = _query_value("id")
     legacy_month = _query_value("month")
     if not settlement_id and legacy_month:
@@ -198,7 +202,8 @@ def page_monthly_settlement(db_path: str) -> None:
             st.query_params["id"] = settlement_id
             st.query_params.pop("month", None)
         else:
-            st.error("월정산을 찾을 수 없습니다.")
+            from .legacy_monthly_settlement import page_legacy_monthly_settlement
+            page_legacy_monthly_settlement(db_path)
             return
     if settlement_id:
         year = _query_value("year", "전체")
@@ -271,6 +276,7 @@ def page_monthly_settlement(db_path: str) -> None:
             _render_monthly_detail(db_path, month)
         return
     _render_monthly_list(db_path)
+    st.link_button("기존 자산 마감 · 초안 및 이력", "?page=monthly_settlement&legacy=1")
 
 
 def _account_evaluation_totals(rows: list[dict]) -> dict:

dashboard/page_views/settlement_drafts.py

@@ -31,6 +31,39 @@ def _authenticated_actor() -> str | None:
     return get_authenticated_actor()
 
 
+def _open_settlement_history(draft_id: str, run_id: str = "") -> None:
+    """Open the read-only history panel and prefer the completed draft's run."""
+    st.session_state["settlement_history_open_draft"] = draft_id
+    if run_id:
+        st.session_state[f"settlement_history_run_{draft_id}"] = run_id
+
+
+def _render_settlement_history(db_path: str | None, month: str, draft_id: str) -> None:
+    """List immutable runs and fetch the exact record selected by the operator."""
+    listed = gateway.list_monthly_settlement_history(month, db_path=db_path)
+    if not listed.get("success"):
+        st.error(f"마감 이력을 불러오지 못했습니다 ({listed.get('error_code', 'operation_failed')}).")
+        return
+    history = listed.get("history", [])
+    if not isinstance(history, list):
+        history = []
+    runs = [str(item.get("run_id")) for item in history if isinstance(item, dict) and item.get("run_id")]
+    if not runs:
+        st.info("저장된 불변 마감 이력이 없습니다.")
+        return
+    selection_key = f"settlement_history_run_{draft_id}"
+    if st.session_state.get(selection_key) not in runs:
+        st.session_state[selection_key] = runs[0]
+    selected_run = st.selectbox("불변 마감 run 선택", runs, key=selection_key)
+    st.dataframe(history, hide_index=True, use_container_width=True)
+    record = gateway.get_monthly_settlement_history_record(selected_run, db_path=db_path)
+    if not record.get("success"):
+        st.error(f"마감 기록을 불러오지 못했습니다 ({record.get('error_code', 'operation_failed')}).")
+        return
+    st.caption("선택한 run의 저장 시점 불변 기록")
+    st.json(record.get("history", {}))
+
+
 def _create_settlement_dialog(db_path: str | None, month: str, correction_run: str = "") -> None:
     @st.dialog("정산 진행")
     def dialog() -> None:
@@ -66,7 +99,6 @@ def _create_settlement_dialog(db_path: str | None, month: str, correction_run: s
             result = gateway.create_settlement_draft(target_month, actor, key, memo=memo,
                 correction_of_run_id=correction_run, correction_reason=reason, db_path=db_path)
             if result.get("success"):
-                created_id = (result.get("draft") or {}).get("id", "")
                 # The selectbox is the authoritative selection on the next
                 # rerun; seed its widget state, not a separate shadow key.
                 st.query_params["page"] = "월정산"
@@ -147,6 +179,18 @@ def render_settlement_drafts(db_path: str | None = None, month: str = "") -> Non
                 st.write(label, final.get(key) if final.get(key) is not None else "미보존")
             rows = final.get("cashflow_items", [])
             if isinstance(rows, list): st.dataframe([{k: row.get(k) for k in ("cashflow_kind", "category", "subcategory", "amount", "transaction_count", "currency")} for row in rows if isinstance(row, dict)])
+        st.markdown("### 불변 이력")
+        st.write("final_run_id", draft.get("final_run_id"))
+        st.write("correction_of_run_id", draft.get("correction_of_run_id"))
+        st.json(detail.get("items", []))
+        st.button(
+            "마감내역보기",
+            key=f"closed_history_{selected_id}",
+            on_click=_open_settlement_history,
+            args=(selected_id, str(draft.get("final_run_id") or "")),
+        )
+        if st.session_state.get("settlement_history_open_draft") == selected_id:
+            _render_settlement_history(db_path, str(draft.get("month") or month_value), selected_id)
         if draft.get("final_run_id") and st.button("정정시작", disabled=not actor, key=f"correction_start_{selected_id}"):
             _create_settlement_dialog(db_path, draft.get("month", month_value), str(draft["final_run_id"]))
         st.caption("종료된 초안의 증거는 수정하지 않습니다. 정정이 필요하면 이전 마감과 사유를 연결해 새 초안을 생성하세요.")