test: pin public settlement response and liability registration contracts

tests/test_settlement_draft_docs.py

@@ -104,6 +104,15 @@ def test_legacy_close_fields_are_scoped_separately_from_draft_finalize() -> None
     assert not re.search(r"new.*(?:identical|same).*shape", draft, re.I)
 
 
+def test_unregistered_liability_contract_separates_registration_from_unknown_debt() -> None:
+    interface = text("docs/interface-contract.md")
+    clause = interface.split("### Draft evidence public fields", 1)[1].split("## ", 1)[0]
+    assert re.search(r"unregistered_liabilities.*(?:no|not).*registered.*liabilit", clause, re.I | re.S)
+    assert re.search(r"not.*(?:unknown|real).*debt", clause, re.I | re.S)
+    assert re.search(r"not.*(?:proof.*debt.free|prove.*debt.free)", clause, re.I | re.S)
+    assert re.search(r"exclu.*(?:missing|completeness)|(?:missing|completeness).*exclu", clause, re.I | re.S)
+
+
 def test_requirement_authority_scope_distinguishes_approved_drafts_and_automatic_selection() -> None:
     req = REQ.read_text(encoding="utf-8")
     constraints = req.split("## 6. 제약과 불변조건", 1)[1].split("## 7.", 1)[0]
@@ -117,3 +126,12 @@ def test_requirement_authority_scope_distinguishes_approved_drafts_and_automatic
     assert re.search(r"unverified.*rejected.*unknown.*(?:any|all).*path", constraints, re.I | re.S), \
         "unverified, rejected, or unknown amounts are never authority on any path"
     assert "future_manual_asset" not in constraints
+
+
+def test_legacy_metadata_field_lists_are_required_retained_fields_not_exhaustive_response_keys() -> None:
+    interface = text("docs/interface-contract.md")
+    legacy = interface.split("## D04 snapshot close output contract", 1)[1].split("## settlement-draft-service", 1)[0]
+    assert re.search(r"legacy direct close", legacy, re.I)
+    assert re.search(r"required top-level metadata fields", legacy, re.I)
+    assert re.search(r"not (?:an )?exhaustive", legacy, re.I)
+    assert "additive settlement-draft finalize response" in legacy

tests/test_settlement_draft_liability_registration.py

@@ -0,0 +1,56 @@
+"""Registration semantics for draft unregistered-liability status."""
+
+from jarca_ledger_mcp import service as s
+from jarca_ledger_mcp.db import init_db
+
+
+def ok(result):
+    assert result["success"], result
+    return result
+
+
+def test_unregistered_liabilities_tracks_registered_targets_not_evidence_completeness(tmp_path):
+    path = str(tmp_path / "synthetic-liability-registration.sqlite3")
+    init_db(path)
+
+    empty = ok(s.create_settlement_draft("2026-09", "synthetic", "empty", db_path=path))["draft"]
+    empty_review = ok(s.review_settlement_draft(empty["id"], db_path=path))
+    assert empty_review["unregistered_liabilities"] is True
+
+    loan = ok(s.create_ledger_account("Synthetic loan", type="loan", liquidity_bucket="loan", db_path=path))["account"]["id"]
+    loan_draft = ok(s.create_settlement_draft("2026-10", "synthetic", "loan", db_path=path))["draft"]
+    loan_review = ok(s.review_settlement_draft(loan_draft["id"], db_path=path))
+    loan_item = next(item for item in loan_review["items"] if item["source_id"] == loan)
+    assert loan_item["amount"] is None
+    assert loan_review["unregistered_liabilities"] is False
+
+    manual = ok(s.create_manual_asset("Synthetic manual liability", 50, "2026-10-31", "loan", db_path=path))["manual_asset"]["id"]
+    refreshed = ok(s.refresh_settlement_draft_targets(loan_draft["id"], loan_draft["version"], "synthetic", db_path=path))["draft"]
+    manual_item = next(item for item in ok(s.get_settlement_draft(refreshed["id"], db_path=path))["items"] if item["source_id"] == manual)
+    excluded = ok(s.set_settlement_draft_item(refreshed["id"], "manual_asset", manual, refreshed["version"], "synthetic", included=False, exclusion_reason="synthetic excluded", db_path=path))["draft"]
+    excluded_review = ok(s.review_settlement_draft(excluded["id"], db_path=path))
+    assert manual_item["amount"] == 50
+    assert next(item for item in excluded_review["excluded_items"] if item["source_id"] == manual)
+    assert excluded_review["unregistered_liabilities"] is False
+    assert excluded_review["missing_items"]
+
+
+def test_manual_only_frozen_liability_registration_and_missing_exclusion_are_independent(tmp_path):
+    path = str(tmp_path / "manual-only-frozen-liability.sqlite3")
+    init_db(path)
+    manual = ok(s.create_manual_asset("Manual-only debt", 75, "2026-10-31", "loan", db_path=path))["manual_asset"]["id"]
+    draft = ok(s.create_settlement_draft("2026-10", "synthetic", "manual-only", db_path=path))["draft"]
+    # Freeze the target list without any registered loan account.
+    frozen = ok(s.refresh_settlement_draft_targets(draft["id"], draft["version"], "synthetic", db_path=path))["draft"]
+    frozen_items = ok(s.get_settlement_draft(frozen["id"], db_path=path))["items"]
+    item = next(item for item in frozen_items if item["source_id"] == manual)
+    assert item["amount"] == 75
+    assert not any(item.get("source_type") == "account" for item in frozen_items)
+    review = ok(s.review_settlement_draft(frozen["id"], db_path=path))
+    assert review["unregistered_liabilities"] is False
+    assert isinstance(review["missing_items"], list)
+    excluded = ok(s.set_settlement_draft_item(frozen["id"], "manual_asset", manual, frozen["version"], "synthetic", included=False, exclusion_reason="not registered", db_path=path))["draft"]
+    excluded_review = ok(s.review_settlement_draft(excluded["id"], db_path=path))
+    assert any(item["source_id"] == manual for item in excluded_review["excluded_items"])
+    assert excluded_review["unregistered_liabilities"] is False
+    assert isinstance(excluded_review["missing_items"], list)

tests/test_settlement_draft_mcp_runtime.py

@@ -29,6 +29,20 @@ def payload(result):
     return data
 
 
+def require_fields(value, fields, context):
+    missing = set(fields) - set(value)
+    assert not missing, f'{context} missing required public fields: {sorted(missing)}'
+
+
+def test_required_public_field_assertion_detects_omission():
+    synthetic_contract_result = {'success': True, 'draft': {}}
+    require_fields(synthetic_contract_result, {'success', 'draft'}, 'stdio result')
+    dropped_field_copy = dict(synthetic_contract_result)
+    dropped_field_copy.pop('draft')
+    with pytest.raises(AssertionError, match='missing required public fields.*draft'):
+        require_fields(dropped_field_copy, {'success', 'draft'}, 'dropped-field mutation control')
+
+
 def rows(db, table):
     assert table in {'asset_snapshot_runs', 'asset_snapshots', 'settlement_drafts', 'settlement_draft_items'}
     with closing(sqlite3.connect(db)) as connection:
@@ -74,31 +88,79 @@ async def _exercise_stdio_lifecycle(tmp_path):
 
             made = await call('create_settlement_draft', {'month':'2026-09','created_by':'stdio-test','idempotency_key':'mcp-draft'})
             assert made['success']
+            require_fields(made, {'success', 'draft', 'items', 'review'}, 'create stdio result')
+            assert made['draft']['status'] == 'active' and made['draft']['version'] == 1
             draft_id = made['draft']['id']
-            assert (await call('list_settlement_drafts', {'month':'2026-09'}))['drafts'][0]['id'] == draft_id
+            listed_drafts = await call('list_settlement_drafts', {'month':'2026-09'})
+            require_fields(listed_drafts, {'success', 'drafts'}, 'list stdio result')
+            assert listed_drafts['success'] and listed_drafts['drafts'][0]['id'] == draft_id
             fetched = await call('get_settlement_draft', {'draft_id':draft_id})
+            require_fields(fetched, {'success', 'draft', 'items', 'review'}, 'get stdio result')
+            assert fetched['success'] and fetched['draft']['id'] == draft_id and fetched['draft']['version'] == 1
             draft = fetched['draft']
             updated = await call('update_settlement_draft', {'draft_id':draft_id,'expected_version':draft['version'],'updated_by':'stdio-test','memo':'transport lifecycle'})
+            require_fields(updated, {'success', 'draft', 'items', 'review'}, 'update stdio result')
+            assert updated['success'] and updated['draft']['memo'] == 'transport lifecycle' and updated['draft']['version'] == 2
             draft = updated['draft']
             refreshed = await call('refresh_settlement_draft_targets', {'draft_id':draft_id,'expected_version':draft['version'],'updated_by':'stdio-test'})
+            require_fields(refreshed, {'success', 'draft', 'items', 'review'}, 'refresh stdio result')
+            assert refreshed['success'] and refreshed['draft']['version'] == 3
             draft = refreshed['draft']
             changed = await call('set_settlement_draft_item', {'draft_id':draft_id,'source_type':'account','source_id':account,'expected_version':draft['version'],'selected_by':'stdio-test','valuation_id':service.list_account_valuations(account, db_path=str(db))['valuations'][0]['id']})
+            require_fields(changed, {'success', 'draft', 'items', 'review'}, 'set-item stdio result')
+            assert changed['success'] and changed['draft']['version'] == 4
             draft = changed['draft']
             reviewed = await call('review_settlement_draft', {'draft_id':draft_id})
             assert reviewed['success'], reviewed
+            require_fields(reviewed, {'success', 'draft_id', 'version', 'review_fingerprint', 'items', 'included_items', 'excluded_items', 'missing_items', 'warnings', 'approval_required', 'totals', 'complete', 'contains_substitutions', 'contains_carried', 'has_approved_missing', 'unregistered_liabilities'}, 'review stdio result')
+            selected = reviewed['included_items'][0]
+            for field in ('valuation_date', 'recorded_at', 'date_difference_days', 'date_difference_direction', 'age_days', 'verification_status', 'evidence_kind', 'selection_mode', 'selection_reason', 'selected_by', 'selected_at', 'item_version'):
+                assert field in selected, f'review selected evidence missing {field}'
+            assert selected['valuation_date'] == '2026-09-30' and selected['age_days'] == 0 and selected['date_difference_days'] == 0
             fingerprint = reviewed['review_fingerprint']
             approved = await call('approve_settlement_draft', {'draft_id':draft_id,'expected_version':draft['version'],'review_fingerprint':fingerprint,'approved_by':'stdio-test','approval_reason':'transport acceptance','approval_reference':'test'})
+            require_fields(approved, {'success', 'draft', 'items', 'review'}, 'approve stdio result')
+            require_fields(approved['draft'], {'approved_by', 'approved_at', 'approval_reason', 'approval_reference', 'approval_flags'}, 'approved draft')
             assert approved['success']
+            assert approved['draft']['approved_by'] == 'stdio-test' and approved['draft']['approval_reason'] == 'transport acceptance' and approved['draft']['approval_reference'] == 'test'
             closed = await call('finalize_settlement_draft', {'draft_id':draft_id,'expected_version':draft['version'],'review_fingerprint':fingerprint,'idempotency_key':'mcp-close'})
+            require_fields(closed, {'success', 'draft_id', 'draft_version', 'review_fingerprint', 'snapshot_run_id', 'run_id', 'monthly_source_snapshot_id', 'basis_date', 'close_status', 'closed_at', 'source', 'is_closed_snapshot', 'selection_policy', 'inserted_count', 'inserted_snapshot_ids', 'items', 'excluded_items', 'totals', 'contains_substitutions', 'contains_carried', 'has_approved_missing', 'settlement_label', 'warnings', 'carried_items', 'approval', 'correction', 'is_correction', 'correction_of_run_id', 'correction_reason', 'provenance', 'total_assets', 'total_liabilities', 'net_worth'}, 'finalize stdio result')
             assert closed['success'] and closed['totals']['net_worth'] == 100
+            assert closed['source'] == 'snapshot' and closed['is_closed_snapshot'] is True
+            assert closed['snapshot_run_id'] == closed['run_id'] == closed['monthly_source_snapshot_id']
+            assert closed['approval']['approved_by'] == 'stdio-test'
+            assert closed['approval']['allow_missing_exclusions'] is False and closed['approval']['allow_carried_before'] is False and closed['approval']['allow_manual_substitute'] is False
+            item = closed['items'][0]
+            for field in ('valuation_date', 'recorded_at', 'date_difference_days', 'date_difference_direction', 'age_days', 'verification_status', 'evidence_kind', 'approval', 'draft_id', 'draft_version', 'review_fingerprint'):
+                assert field in item, f'finalized selected evidence missing {field}'
+            provenance = closed['provenance']
+            for field in ('valuation_date', 'age_days', 'carried_forward', 'verification_status', 'evidence_kind', 'approval', 'correction', 'warnings', 'exclusions'):
+                assert field in provenance, f'final provenance missing {field}'
+            assert provenance['valuation_date'] == item['valuation_date'] and provenance['age_days'] == item['age_days']
             before = (await call('get_settlement_draft', {'draft_id':draft_id}))
             tables = ('asset_snapshot_runs', 'asset_snapshots', 'settlement_drafts', 'settlement_draft_items')
             before_replay = {table: rows(db, table) for table in tables}
             assert await call('finalize_settlement_draft', {'draft_id':draft_id,'expected_version':draft['version'],'review_fingerprint':fingerprint,'idempotency_key':'mcp-close'}) == closed
-            assert await call('get_settlement_draft', {'draft_id':draft_id}) == before
+            archived = await call('get_settlement_draft', {'draft_id':draft_id})
+            require_fields(archived, {'success', 'draft', 'items', 'final_response'}, 'archived get stdio result')
+            assert archived['final_response'] == closed
+            assert archived == before
             assert {table: rows(db, table) for table in tables} == before_replay
 
             second = await call('create_settlement_draft', {'month':'2026-10','created_by':'stdio-test','idempotency_key':'mcp-cancel'})
             assert second['success']
             cancelled = await call('cancel_settlement_draft', {'draft_id':second['draft']['id'],'expected_version':second['draft']['version'],'cancelled_by':'stdio-test','reason':'synthetic cancellation'})
+            require_fields(cancelled, {'success', 'draft', 'items', 'review'}, 'cancel stdio result')
             assert cancelled['success'] and cancelled['draft']['status'] == 'cancelled'
+            assert cancelled['draft']['cancelled_by'] == 'stdio-test' and cancelled['draft']['cancellation_reason'] == 'synthetic cancellation'
+
+            # Legacy close retains required contract fields; this is a subset check, not exact whole-key equality.
+            legacy_account = service.create_ledger_account('Synthetic legacy cash', db_path=str(db))['account']['id']
+            service.register_account_valuation(legacy_account, '2026-11-30', 31, verification_status='verified', idempotency_key='legacy-value', db_path=str(db))
+            preview = await call('preview_asset_snapshot_close', {'basis_date':'2026-11-30','selection_policy':'exact'})
+            assert preview['success'] and preview['source'] == 'live_authority' and preview['is_closed_snapshot'] is False
+            legacy = await call('close_asset_snapshot', {'basis_date':'2026-11-30','selection_policy':'exact','idempotency_key':'legacy-close','allow_missing_exclusions':True,'approved_by':'stdio-test','approval_reason':'legacy transport test','approval_reference':'legacy-test'})
+            legacy_fields = {'snapshot_run_id', 'close_status', 'approval', 'correction', 'warnings', 'excluded_items', 'carried_items', 'provenance', 'valuation_date', 'age_days', 'verification_status', 'evidence_kind', 'monthly_source_snapshot_id'}
+            require_fields(legacy, {'success'} | legacy_fields, 'legacy close stdio result')
+            assert not ({'draft_id', 'draft_version', 'review_fingerprint', 'contains_substitutions', 'contains_carried', 'has_approved_missing'} & set(legacy))
+            assert legacy['monthly_source_snapshot_id'] == legacy['snapshot_run_id']