diff --git a/HANDOFF.md b/HANDOFF.md index 8ec5320..1e7150a 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -1,6 +1,6 @@ # Payment Intelligence Modules — handoff -_Last updated: 2026-05-07 20:40 +08_ +_Last updated: 2026-05-27 20:13 +08_ This is the next-session entry point for `Raafet57/payment-intelligence-modules`. Read this file first, then check live `git` / GitHub state before acting. @@ -9,15 +9,21 @@ Read this file first, then check live `git` / GitHub state before acting. - Repository: `https://github.com/Raafet57/payment-intelligence-modules` - Local path on Hermes: `/Users/Shared/AgentWork/repos/payment-intelligence-modules` -- Product baseline before the current branch: `3e71236` — merge commit for PR #12, which added the static root `/ssi` suite entry point. -- Active feature branch: `feat/ssi-public-evidence-control-tower` +- Product baseline before the current branch: `4eaf07d` — merge commit for PR #13, which added the SSI public evidence control tower foundation. +- Active feature branch: `feat/ssi-validation-catalogue` +- Local status: SSI validation catalogue and SSI shell cutover changes are committed locally; branch has not been pushed, and no PR, deploy, release, or publication has been done. - Deployment/release/tag/public announcement: **not done** ## Current branch scope -This branch improves the folded SSI Control Tower backend module under `apps/ssi-control-tower/` using public bank SSI / wire / settlement-instruction evidence while preserving the synthetic-only privacy boundary. +Current feature branch contains two committed slices: -Implemented shape: +1. SSI validation catalogue under `apps/ssi-control-tower/`. +2. SSI shell cutover in the root suite, which removes SSI from the Payment Intelligence nav/home launcher and keeps `/ssi` only as an unlinked static boundary pointer. + +The standalone SSI Control Tower adjacent app remains under `apps/ssi-control-tower/` using public bank SSI / wire / settlement-instruction evidence while preserving the synthetic-only privacy boundary. SSI Control Tower is a standalone adjacent app, not a Payment Intelligence module. + +Implemented shape from the public-evidence/control-tower slice: - Sanitized public evidence sidecar: `apps/ssi-control-tower/docs/public_ssi_field_evidence_sources.json`. - Evidence note: `apps/ssi-control-tower/docs/public_ssi_field_evidence.md`. @@ -40,29 +46,37 @@ Implemented shape: | BIC Validator | `/bic` | Demo/snapshot-bound only; not current-production BIC Directory validation. | | CBPR+ Readiness Checker | `/cbpr` | Local readiness/structure checks only; not certified/full XSD/MyStandards validation. | | Payment Insights Lite | `/insights` | Local lifecycle insight over files the user provides; not live payment tracking. | -| SSI Control Tower | `/ssi` | Static root entry page for the folded backend module under `apps/ssi-control-tower/`. | -## Verification evidence for this branch +## Adjacent standalone app pointer + +`/ssi` is retained only as an unlinked static boundary pointer to the standalone +SSI Control Tower app under `apps/ssi-control-tower/`. SSI Control Tower is not a +Payment Intelligence module/workflow, is absent from the suite nav and home +launcher, and still requires a separate repo-extraction/product-boundary +decision. + +## Verification evidence for current local work -Local gates run before handoff: +Latest local gates run for the SSI shell cutover on 2026-05-26: ```bash +pnpm test -- src/App.test.tsx src/pages/SsiPage.test.tsx +pnpm test:e2e -- e2e/smoke.spec.ts e2e/ssi.spec.ts +pnpm audit:privacy cd apps/ssi-control-tower && make test -python3 -m ruff check apps/ssi-control-tower pnpm verify -pnpm test:e2e git diff --check ``` Results: -- `cd apps/ssi-control-tower && make test` passed: **106 passed**. -- `python3 -m ruff check apps/ssi-control-tower` passed: all checks passed. +- Focused root Vitest command passed; Vitest ran **30 files / 230 tests passed**. +- Targeted Playwright smoke passed: **4 tests passed**. +- `pnpm audit:privacy` passed: `privacy-audit: clean (src/)`. +- `cd apps/ssi-control-tower && .venv/bin/python -m pytest -q` passed: **125 passed**. - `pnpm verify` passed: lint, format check, typecheck, Vitest, privacy audit, deterministic IBAN/BIC/CBPR data checks, and build. -- Root Vitest inside `pnpm verify`: 30 files passed / 230 tests passed. -- `pnpm test:e2e` passed: 21 Playwright tests passed. - `git diff --check` passed. -- Claude Code frontend/design review: final verdict `APPROVE`. +- Codex cold diff review for SSI-CUTOVER-1 returned `APPROVED` for implementation readiness; push/PR/deploy remain closed gates. ## Boundaries to preserve @@ -76,7 +90,7 @@ Root suite privacy/product boundaries: - The root Vite/React app remains static and browser-only. - Root `src/` must not add backend coupling, telemetry, analytics, remote logging, browser persistence, external fetches, or payment-data upload paths. -- `/ssi` is a static entry page only. It must remain fetch-free and iframe-free unless a separate architecture/security review explicitly changes the posture. +- `/ssi` is an unlinked static boundary pointer only. It must remain fetch-free, iframe-free, storage-free, telemetry-free, and free of clickable localhost links unless a separate architecture/security review explicitly changes the posture. SSI backend module boundaries: diff --git a/README.md b/README.md index 46d3aa9..803fa5c 100644 --- a/README.md +++ b/README.md @@ -6,25 +6,42 @@ data. For the latest repo/session handoff, start with [`HANDOFF.md`](./HANDOFF.md). -The root suite remains static and browser-only. Backend/product-control modules live under `apps/**` as separate, explicitly scoped applications and are not part of the root browser runtime. +The root suite remains static and browser-only. Standalone adjacent applications live under `apps/**` as separate, explicitly scoped apps; they are not Payment Intelligence modules and are not part of the root browser runtime. ## Modules -| Module | Route | Status | What it does | -| --------------------------- | -------------- | ---------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Scrubber** | `/scrubber` | Available | Strip personally identifying fields from `pacs.*` / `camt.*` XML before sharing with peers or vendors. Produces a privacy-safe mapping summary. | -| **Storyteller** | `/storyteller` | Available | Turn a `pacs.002 / 004 / 008 / 009` or `camt.052 / 053 / 054` message into a plain-language narrative plus a structured field projection and quick insights. | -| **IBAN Workbench** | `/iban` | Available | Validate, build, catalogue, and trace provenance for IBANs from bundled SWIFT IBAN Registry-derived data. Builder computes MOD-97 check digits from exact-length BBAN fields; no live BIC, VOP, account existence, or reachability checks. | -| **BIC Validator\*** | `/bic` | Demo | ISO 9362 syntax checks plus a tiny bundled snapshot lookup. **Demonstration only:** bundled BIC data is not accurate/current enough for production, routing, compliance, reachability, or payment decisions. | -| **CBPR+ Readiness Checker** | `/cbpr` | Available | Browser-only AppHdr / Document namespace / CBPR+ schema-profile coverage checks, plus UETR, BIC syntax, and IBAN syntax/checksum hints. Not a certified validator or MyStandards usage-rule engine. | -| **Payment Insights Lite** | `/insights` | Available | Local lifecycle insight over ACK/NACK, `pacs.*`, and `camt.*` files you provide. Groups files by identifiers in memory; not live payment tracking, VOP, reachability, or settlement monitoring. | -| **SSI Control Tower** | `/ssi` | Backend (folded) | Static root entry point for the separate FastAPI/Jinja SSI readiness and governance cockpit under `apps/ssi-control-tower/`. The backend module now includes public-evidence sourcing, command-centre panels, exception detail, and guided demo routes; run locally when needed. | -| Vault | — | Planned | Planned encrypted local export bundle: user-controlled download/import, no cloud vault, no server storage, and no persistent browser storage by default. Not built yet. | +| Module | Route | Status | What it does | +| --------------------------- | -------------- | --------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| **Scrubber** | `/scrubber` | Available | Strip personally identifying fields from `pacs.*` / `camt.*` XML before sharing with peers or vendors. Produces a privacy-safe mapping summary. | +| **Storyteller** | `/storyteller` | Available | Turn a `pacs.002 / 004 / 008 / 009` or `camt.052 / 053 / 054` message into a plain-language narrative plus a structured field projection and quick insights. | +| **IBAN Workbench** | `/iban` | Available | Validate, build, catalogue, and trace provenance for IBANs from bundled SWIFT IBAN Registry-derived data. Builder computes MOD-97 check digits from exact-length BBAN fields; no live BIC, VOP, account existence, or reachability checks. | +| **BIC Validator\*** | `/bic` | Demo | ISO 9362 syntax checks plus a tiny bundled snapshot lookup. **Demonstration only:** bundled BIC data is not accurate/current enough for production, routing, compliance, reachability, or payment decisions. | +| **CBPR+ Readiness Checker** | `/cbpr` | Available | Browser-only AppHdr / Document namespace / CBPR+ schema-profile coverage checks, plus UETR, BIC syntax, and IBAN syntax/checksum hints. Not a certified validator or MyStandards usage-rule engine. | +| **Payment Insights Lite** | `/insights` | Available | Local lifecycle insight over ACK/NACK, `pacs.*`, and `camt.*` files you provide. Groups files by identifiers in memory; not live payment tracking, VOP, reachability, or settlement monitoring. | +| Vault | — | Planned | Planned encrypted local export bundle: user-controlled download/import, no cloud vault, no server storage, and no persistent browser storage by default. Not built yet. | \*The BIC module intentionally does **not** perform live BIC Directory lookup, current bank-directory enrichment, current SEPA reachability checks, SWIFT FIN reachability checks, Verification of Payee, or account-owner/name matching. +## SSI Control Tower (standalone adjacent app) + +SSI Control Tower is **not** a Payment Intelligence module or workflow. It is a +standalone adjacent app under `apps/ssi-control-tower/`, with its own +FastAPI/Jinja runtime, Python dependencies, tests, and synthetic fixtures. It is +intentionally absent from the suite navigation and the home launcher. + +The root browser suite keeps `/ssi` only as an unlinked static boundary pointer: +it documents the boundary, stays fetch-free, iframe-free, storage-free, and +telemetry-free, and does not link to localhost. Repo extraction and the +product-boundary decision for the standalone app are still pending. + +The standalone app runs separately: + +```bash +cd apps/ssi-control-tower && make run # serves http://localhost:8000 +``` + ## Privacy boundary This is the load-bearing property of the suite, not a footer line: @@ -62,7 +79,8 @@ banned APIs or absolute-URL `fetch` calls appear under `src/`. `pacs.*`, and `camt.*` files the user provides in the browser. It is not live payment tracking, settlement monitoring, VOP, SEPA reachability, or live bank lookup. -- **SSI Control Tower:** the folded backend module uses synthetic fixtures plus a +- **SSI Control Tower:** the standalone adjacent app under `apps/ssi-control-tower/` + uses synthetic fixtures plus a sanitized public-evidence sidecar from official bank SSI/wire/settlement pages. The sidecar stores only bank name, URL, title, and abstract field categories; no account numbers, IBANs, BICs, routing codes, correspondent chains, excerpts, @@ -77,10 +95,10 @@ banned APIs or absolute-URL `fetch` calls appear under `src/`. - Node.js >= 20 - pnpm >= 9 -For the folded SSI Control Tower backend module only: +For the standalone SSI Control Tower adjacent app only: - Python 3.11 -- Run commands from `apps/ssi-control-tower/`; the module owns its own `pyproject.toml`, `Makefile`, tests, and SQLite development data. +- Run commands from `apps/ssi-control-tower/`; the standalone app owns its own `pyproject.toml`, `Makefile`, tests, and SQLite development data. ## Scripts @@ -144,7 +162,7 @@ src/ CbprPage.tsx IbanPage.tsx ScrubberPage.tsx - SsiPage.tsx # static entry point for the folded SSI backend module + SsiPage.tsx # unlinked static boundary pointer to the standalone SSI app StorytellerPage.tsx NotFoundPage.tsx lib/ @@ -169,11 +187,11 @@ scripts/ build-iban-registry.ts privacy-audit.sh apps/ - ssi-control-tower/ # Separate FastAPI/Jinja backend module; not part of root browser runtime + ssi-control-tower/ # Standalone adjacent FastAPI/Jinja SSI app; not a Payment Intelligence module app/ # Python app, routers, services, web templates data/ # Synthetic fixtures/reference YAML only; local DBs ignored docs/ # SSI docs including sanitized public-source evidence methodology - tests/ # pytest suite for the SSI backend module + tests/ # pytest suite for the standalone SSI app pyproject.toml # Python dependencies owned by the nested module Makefile # module-local test/install commands ``` diff --git a/ROADMAP.md b/ROADMAP.md index f51262a..830ab32 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -18,9 +18,9 @@ the suite is browser-only, static-host friendly, and privacy-first. ## Recommended next steps -### 0. SSI Control Tower folded backend module +### 0. SSI Control Tower standalone adjacent app boundary -SSI Control Tower now lives under `apps/ssi-control-tower/` as a separate FastAPI/Jinja backend module inside this repository. It is not part of the root static browser suite runtime, and it must not weaken the browser-only privacy boundary for `src/`. The root suite now includes `/ssi` as a static, fetch-free entry page that explains the backend module posture and local run path without coupling the browser app to the FastAPI runtime. +SSI Control Tower now lives under `apps/ssi-control-tower/` as a standalone adjacent FastAPI/Jinja app inside this repository. It is not a Payment Intelligence module or workflow, is not part of the root static browser suite runtime, and must not weaken the browser-only privacy boundary for `src/`. The root suite retains `/ssi` only as an unlinked static, fetch-free boundary pointer that records the app boundary and local run path without coupling the browser app to the SSI runtime. Repo extraction and the product-boundary decision remain pending. Merged SSI foundation includes: @@ -30,7 +30,7 @@ Merged SSI foundation includes: - instruction-level validation, instruction exceptions, and a unified control queue API; - approval, export, and account-privacy hardening for the synthetic public prototype. -Follow-up scope now includes a public-evidence command-centre slice inside the FastAPI app: sanitized public bank SSI/wire/settlement-instruction source pointers, `/public-evidence`, `/api/v1/dashboard/operator-summary`, five operator dashboard panels, rule-filtered exceptions, exception detail pages, and a read-only guided `/demo`. Deployment/publication decisions and real-data/private-lab workflows remain separate approval gates. +Follow-up scope remains inside the standalone FastAPI app: sanitized public bank SSI/wire/settlement-instruction source pointers, `/public-evidence`, `/api/v1/dashboard/operator-summary`, five operator dashboard panels, rule-filtered exceptions, exception detail pages, and a read-only guided `/demo`. Deployment/publication decisions, repo extraction, and real-data/private-lab workflows remain separate approval gates. ### 1. Deployment target decision @@ -48,7 +48,7 @@ Acceptance criteria: - Static build only; no backend or API. - SPA fallback works for `/scrubber`, `/storyteller`, `/iban`, `/bic`, `/cbpr`, - `/insights`, and `/ssi`. + `/insights`, and the unlinked `/ssi` boundary pointer. - No analytics injection, remote logging, third-party scripts, cookies, or browser persistence introduced by the host configuration. - README hosting notes are updated with the chosen target. diff --git a/apps/ssi-control-tower/app/services/instruction_validation.py b/apps/ssi-control-tower/app/services/instruction_validation.py index 62fe462..5153638 100644 --- a/apps/ssi-control-tower/app/services/instruction_validation.py +++ b/apps/ssi-control-tower/app/services/instruction_validation.py @@ -1,22 +1,45 @@ from __future__ import annotations +import re from dataclasses import dataclass from sqlalchemy.orm import Session from app.config import utc_now -from app.models import InstructionValidationResult, SsiInstruction +from app.models import InstructionValidationResult, SourceFile, SourceRecord, SsiInstruction, SsiSourceLink from app.services.audit import stable_id from app.services.instruction_exceptions import ( close_resolved_instruction_exceptions, upsert_instruction_exception, write_instruction_exception_audit, ) +from app.services.ssi_validation_catalogue import ( + SsiValidationRule, + allowed_asset_categories, + allowed_holder_types, + allowed_preferred_flags, + load_instruction_catalogue, +) RULE_VERSION = "instruction_rules_v1" _EXCEPTION_SEVERITIES = {"medium", "high", "critical"} _PREFERRED_TRUE = {"Y", "P", "TRUE", "1"} -_VALID_PREFERRED_FLAGS = _PREFERRED_TRUE | {"N", "FALSE", "0", ""} +_VALID_PREFERRED_FLAGS = allowed_preferred_flags() +_BIC_RE = re.compile(r"^[A-Z0-9]{8}([A-Z0-9]{3})?$") +_COUNTRY_RE = re.compile(r"^[A-Z]{2}$") +_CURRENCY_RE = re.compile(r"^[A-Z]{3}$") + +_LEGACY_RULE_IDS = { + "ssi.instrument.currency.required": "SSI.INSTRUCTION.CURRENCY_REQUIRED", + "ssi.instrument.asset.required": "SSI.INSTRUCTION.ASSET_REQUIRED", + "ssi.owner_bic.required": "SSI.INSTRUCTION.OWNER_BIC_REQUIRED", + "ssi.account_holder_bic.required": "SSI.INSTRUCTION.ACCOUNT_HOLDER_BIC_REQUIRED", + "ssi.lifecycle.active_start.required": "SSI.INSTRUCTION.START_DATE_REQUIRED", + "ssi.lifecycle.date_range": "SSI.INSTRUCTION.DATE_RANGE", + "ssi.preference.flag.reference": "SSI.INSTRUCTION.PREFERRED_FLAG", + "ssi.conflict.preferred_unique": "SSI.INSTRUCTION.PREFERRED_UNIQUE", + "ssi.conflict.active_interval_overlap": "SSI.INSTRUCTION.ACTIVE_INTERVAL_OVERLAP", +} @dataclass(frozen=True) @@ -30,25 +53,40 @@ class InstructionRuleOutcome: context_key: str | None = None -def _fail( - rule_id: str, - severity: str, - failed_field: str | None, - message: str, +def _outcome( + rule: SsiValidationRule, + status: str, + failed_field: str | None = None, + message: str | None = None, suggested_fix: str | None = None, context_key: str | None = None, ) -> InstructionRuleOutcome: return InstructionRuleOutcome( - rule_id=rule_id, - status="fail", - severity=severity, + rule_id=rule.rule_id, + status=status, + severity=rule.severity, failed_field=failed_field, - message=message, - suggested_fix=suggested_fix, + message=message or rule.message, + suggested_fix=suggested_fix or rule.suggested_fix, context_key=context_key, ) +def _legacy_alias(outcome: InstructionRuleOutcome) -> InstructionRuleOutcome | None: + legacy_id = _LEGACY_RULE_IDS.get(outcome.rule_id) + if not legacy_id or outcome.status != "fail": + return None + return InstructionRuleOutcome( + rule_id=legacy_id, + status=outcome.status, + severity=outcome.severity, + failed_field=outcome.failed_field, + message=outcome.message, + suggested_fix=outcome.suggested_fix, + context_key=outcome.context_key, + ) + + def _is_active(instruction: SsiInstruction) -> bool: return (instruction.status or "").strip().lower() == "active" @@ -74,6 +112,17 @@ def _safe_context(*parts: object) -> str: return stable_id("instruction-context", *parts) +def _safe_instruction_context(instruction: SsiInstruction, scope: str) -> str: + return _safe_context( + scope, + instruction.owner_country_code or "", + instruction.account_holder_country_code or "", + instruction.currency_code or "", + instruction.asset_category or "", + instruction.ssi_instruction_id, + ) + + def _date_range_invalid(instruction: SsiInstruction) -> bool: return bool(instruction.start_date and instruction.stop_date and instruction.stop_date < instruction.start_date) @@ -86,52 +135,125 @@ def _date_intervals_overlap(a: SsiInstruction, b: SsiInstruction) -> bool: return a.start_date <= b_stop and b.start_date <= a_stop -def _individual_outcomes(instruction: SsiInstruction) -> list[InstructionRuleOutcome]: - outcomes: list[InstructionRuleOutcome] = [] - required_rules = ( - ("SSI.INSTRUCTION.CURRENCY_REQUIRED", "currency_code", "Instruction currency is required."), - ("SSI.INSTRUCTION.ASSET_REQUIRED", "asset_category", "Instruction asset category is required."), - ("SSI.INSTRUCTION.OWNER_BIC_REQUIRED", "owner_bic", "Instruction owner BIC is required."), - ( - "SSI.INSTRUCTION.ACCOUNT_HOLDER_BIC_REQUIRED", - "account_holder_bic", - "Instruction account-holder BIC is required.", - ), - ) - for rule_id, field, message in required_rules: - if not getattr(instruction, field): - outcomes.append(_fail(rule_id, "high", field, message, f"Populate {field}.")) - if _is_active(instruction) and not instruction.start_date: - outcomes.append( - _fail( - "SSI.INSTRUCTION.START_DATE_REQUIRED", - "medium", - "start_date", - "Active instruction requires a start date.", - "Add an effective start date before activating the instruction.", - ) - ) - if _date_range_invalid(instruction): - outcomes.append( - _fail( - "SSI.INSTRUCTION.DATE_RANGE", - "high", - "stop_date", - "Instruction stop date must not be earlier than start date.", - "Correct the instruction lifecycle dates.", - ) +def _field_value(instruction: SsiInstruction, field: str) -> str: + return str(getattr(instruction, field, "") or "").strip() + + +def _source_link(session: Session, instruction: SsiInstruction) -> SsiSourceLink | None: + return session.query(SsiSourceLink).filter_by(ssi_instruction_id=instruction.ssi_instruction_id).first() + + +def _evaluate_rule(rule: SsiValidationRule, instruction: SsiInstruction, session: Session) -> InstructionRuleOutcome: + evaluator = rule.evaluator + fields = rule.fields + primary = fields[0] if fields else None + context = _safe_instruction_context(instruction, rule.rule_id) + + if evaluator == "source_lineage_present": + link = _source_link(session, instruction) + source_file = session.get(SourceFile, link.source_file_id) if link else None + failed = not ( + link + and source_file + and source_file.import_id == link.import_id + and source_file.source_file_id == link.source_file_id + and source_file.source_schema + and source_file.parser_version + and source_file.file_hash ) - if _preferred_flag(instruction) not in _VALID_PREFERRED_FLAGS: - outcomes.append( - _fail( - "SSI.INSTRUCTION.PREFERRED_FLAG", - "medium", - "preferred_flag", - "Preferred flag must use an approved value.", - "Use Y/P for preferred or N for non-preferred.", - ) + return _outcome(rule, "fail" if failed else "pass", "source_file" if failed else None, context_key=context if failed else None) + if evaluator == "source_record_lineage": + link = _source_link(session, instruction) + source_file = session.get(SourceFile, link.source_file_id) if link else None + source_record = session.get(SourceRecord, link.source_record_id) if link else None + failed = not ( + link + and source_file + and source_file.source_file_id == link.source_file_id + and source_record + and source_record.source_file_id == link.source_file_id + and source_record.import_id == link.import_id == source_file.import_id + and source_record.source_record_key == instruction.source_record_key + and source_record.source_record_key + and source_record.source_hash + and source_record.row_number is not None ) - return outcomes + return _outcome(rule, "fail" if failed else "pass", "source_record_key" if failed else None, context_key=context if failed else None) + if evaluator == "required": + missing = primary and not _field_value(instruction, primary) + return _outcome(rule, "fail" if missing else "pass", primary if missing else None, context_key=context if missing else None) + if evaluator == "bic_format": + value = _field_value(instruction, primary or "") + failed = not bool(_BIC_RE.fullmatch(value.upper())) + return _outcome(rule, "fail" if failed else "pass", primary if failed else None, context_key=context if failed else None) + if evaluator == "country_format_optional": + value = _field_value(instruction, primary or "") + failed = bool(value) and not bool(_COUNTRY_RE.fullmatch(value.upper())) + return _outcome(rule, "fail" if failed else "pass", primary if failed else None, context_key=context if failed else None) + if evaluator == "currency_format": + value = _field_value(instruction, "currency_code") + failed = not bool(_CURRENCY_RE.fullmatch(value.upper())) + return _outcome(rule, "fail" if failed else "pass", "currency_code" if failed else None, context_key=context if failed else None) + if evaluator == "asset_reference": + value = _field_value(instruction, "asset_category").upper() + failed = value not in allowed_asset_categories() + return _outcome(rule, "fail" if failed else "pass", "asset_category" if failed else None, context_key=context if failed else None) + if evaluator == "holder_type_reference": + value = _field_value(instruction, "account_holder_type").upper() + failed = value not in allowed_holder_types() + return _outcome(rule, "fail" if failed else "pass", "account_holder_type" if failed else None, context_key=context if failed else None) + if evaluator == "masked_account": + value = _field_value(instruction, "account_number_masked") + failed = not value or "****" not in value + return _outcome(rule, "fail" if failed else "pass", "account_number_masked" if failed else None, context_key=context if failed else None) + if evaluator == "preferred_flag": + failed = _preferred_flag(instruction) not in _VALID_PREFERRED_FLAGS + return _outcome(rule, "fail" if failed else "pass", "preferred_flag" if failed else None, context_key=context if failed else None) + if evaluator == "active_start_date": + failed = _is_active(instruction) and not instruction.start_date + return _outcome(rule, "fail" if failed else "pass", "start_date" if failed else None, context_key=context if failed else None) + if evaluator == "date_range": + failed = _date_range_invalid(instruction) + return _outcome(rule, "fail" if failed else "pass", "stop_date" if failed else None, context_key=context if failed else None) + if evaluator == "update_date_present": + failed = not instruction.update_date + return _outcome(rule, "fail" if failed else "pass", "update_date" if failed else None, context_key=context if failed else None) + if evaluator == "traffic_date_for_active_flag": + failed = (instruction.traffic_flag or "").upper() == "Y" and not instruction.traffic_date + return _outcome(rule, "fail" if failed else "pass", "traffic_date" if failed else None, context_key=context if failed else None) + if evaluator == "cross_reference_optional": + present = all(_field_value(instruction, field) for field in fields) + return _outcome(rule, "pass" if present else "skipped", None, "Reference dataset not loaded for this optional cross-reference gate." if not present else None) + if evaluator == "owner_profile_present": + return _outcome(rule, "pass" if instruction.owner_name or instruction.owner_city or instruction.group_key_owner else "skipped") + if evaluator == "account_holder_profile_present": + return _outcome(rule, "pass" if instruction.account_holder_name else "skipped") + if evaluator == "governance_route": + return _outcome(rule, "pass") + if evaluator in {"group_preferred_unique", "group_active_overlap", "export_eligibility"}: + return _outcome(rule, "pass") + return _outcome(rule, "skipped", message="Evaluator is not available for this rule.") + + +def _individual_outcomes(session: Session, instruction: SsiInstruction) -> list[InstructionRuleOutcome]: + outcomes: list[InstructionRuleOutcome] = [] + for rule in load_instruction_catalogue(): + if not rule.enabled or rule.evaluator in {"group_preferred_unique", "group_active_overlap", "export_eligibility"}: + continue + outcomes.append(_evaluate_rule(rule, instruction, session)) + return _with_legacy_aliases(outcomes) + + +def _group_members(session: Session, instruction: SsiInstruction) -> list[SsiInstruction]: + return [ + candidate + for candidate in session.query(SsiInstruction).all() + if _group_key(candidate) == _group_key(instruction) + ] + + +def _rule_by_id(rule_id: str) -> SsiValidationRule: + return next(rule for rule in load_instruction_catalogue() if rule.rule_id == rule_id) def _group_outcomes(instructions: list[SsiInstruction]) -> dict[str, list[InstructionRuleOutcome]]: @@ -141,38 +263,51 @@ def _group_outcomes(instructions: list[SsiInstruction]) -> dict[str, list[Instru continue grouped.setdefault(_group_key(instruction), []).append(instruction) + preferred_rule = _rule_by_id("ssi.conflict.preferred_unique") + overlap_rule = _rule_by_id("ssi.conflict.active_interval_overlap") outcomes: dict[str, list[InstructionRuleOutcome]] = {item.ssi_instruction_id: [] for item in instructions} + failed_rule_ids: dict[str, set[str]] = {item.ssi_instruction_id: set() for item in instructions} + for key, members in grouped.items(): preferred = [member for member in members if _is_preferred(member)] if len(preferred) > 1: context_key = _safe_context("preferred", *key) for member in preferred: outcomes[member.ssi_instruction_id].append( - _fail( - "SSI.INSTRUCTION.PREFERRED_UNIQUE", - "high", - "preferred_flag", - "Only one active preferred instruction is allowed per operational group.", - "Retain one preferred instruction and mark the others non-preferred.", - context_key, - ) + _outcome(preferred_rule, "fail", "preferred_flag", context_key=context_key) ) + failed_rule_ids[member.ssi_instruction_id].add(preferred_rule.rule_id) for index, first in enumerate(members): for second in members[index + 1 :]: if _date_intervals_overlap(first, second): context_key = _safe_context("overlap", *key) for member in (first, second): outcomes[member.ssi_instruction_id].append( - _fail( - "SSI.INSTRUCTION.ACTIVE_INTERVAL_OVERLAP", - "high", - "start_date", - "Active instruction intervals overlap within the same operational group.", - "Adjust lifecycle dates or close the superseded instruction.", - context_key, - ) + _outcome(overlap_rule, "fail", "start_date", context_key=context_key) ) - return outcomes + failed_rule_ids[member.ssi_instruction_id].add(overlap_rule.rule_id) + + for instruction in instructions: + for rule in (preferred_rule, overlap_rule): + if rule.rule_id not in failed_rule_ids[instruction.ssi_instruction_id]: + outcomes[instruction.ssi_instruction_id].append(_outcome(rule, "pass")) + return {key: _with_legacy_aliases(value) for key, value in outcomes.items()} + + +def _export_outcome(instruction: SsiInstruction, outcomes: list[InstructionRuleOutcome]) -> InstructionRuleOutcome: + rule = _rule_by_id("ssi.export.eligibility") + export_blocking_ids = {item.rule_id for item in load_instruction_catalogue() if item.export_blocking and item.rule_id != rule.rule_id} + failed = any(outcome.status == "fail" and outcome.rule_id in export_blocking_ids for outcome in outcomes) + return _outcome(rule, "fail" if failed else "pass", "export_eligibility" if failed else None, context_key=_safe_instruction_context(instruction, rule.rule_id) if failed else None) + + +def _with_legacy_aliases(outcomes: list[InstructionRuleOutcome]) -> list[InstructionRuleOutcome]: + expanded = list(outcomes) + for outcome in outcomes: + alias = _legacy_alias(outcome) + if alias is not None: + expanded.append(alias) + return expanded def _dedupe_outcomes(outcomes: list[InstructionRuleOutcome]) -> list[InstructionRuleOutcome]: @@ -200,6 +335,7 @@ def _persist_instruction_results( ).delete() results: list[InstructionValidationResult] = [] failed_keys: set[tuple[str, str | None, str | None]] = set() + catalogue_by_id = {rule.rule_id: rule for rule in load_instruction_catalogue()} for index, outcome in enumerate(outcomes): result = InstructionValidationResult( instruction_validation_result_id=stable_id( @@ -226,11 +362,14 @@ def _persist_instruction_results( results.append(result) if outcome.status == "fail": failed_keys.add((outcome.rule_id, outcome.failed_field, outcome.context_key)) - if outcome.severity in _EXCEPTION_SEVERITIES: + rule = catalogue_by_id.get(outcome.rule_id) + if rule is not None and rule.exception_eligible and outcome.severity in _EXCEPTION_SEVERITIES: + exception_rule_id = _LEGACY_RULE_IDS.get(outcome.rule_id, outcome.rule_id) + failed_keys.add((exception_rule_id, outcome.failed_field, outcome.context_key)) upsert_instruction_exception( session, ssi_instruction_id=instruction.ssi_instruction_id, - rule_id=outcome.rule_id, + rule_id=exception_rule_id, severity=outcome.severity, failed_field=outcome.failed_field, description=outcome.message, @@ -254,7 +393,10 @@ def validate_instruction( ) -> list[InstructionValidationResult]: """Validate one canonical SSI instruction with privacy-safe result messages.""" now = utc_now() - results = _persist_instruction_results(session, instruction, _individual_outcomes(instruction), actor, now) + outcomes = _individual_outcomes(session, instruction) + outcomes.extend(_group_outcomes(_group_members(session, instruction)).get(instruction.ssi_instruction_id, [])) + outcomes.append(_export_outcome(instruction, outcomes)) + results = _persist_instruction_results(session, instruction, outcomes, actor, now) session.commit() return results @@ -265,7 +407,8 @@ def validate_all_instructions(session: Session, actor: str = "system") -> list[I all_results: list[InstructionValidationResult] = [] now = utc_now() for instruction in instructions: - outcomes = _individual_outcomes(instruction) + grouped.get(instruction.ssi_instruction_id, []) + outcomes = _individual_outcomes(session, instruction) + grouped.get(instruction.ssi_instruction_id, []) + outcomes.append(_export_outcome(instruction, outcomes)) all_results.extend(_persist_instruction_results(session, instruction, outcomes, actor, now)) session.commit() return all_results diff --git a/apps/ssi-control-tower/app/services/ssi_validation_catalogue.py b/apps/ssi-control-tower/app/services/ssi_validation_catalogue.py new file mode 100644 index 0000000..7ddc36e --- /dev/null +++ b/apps/ssi-control-tower/app/services/ssi_validation_catalogue.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +from collections import Counter +from dataclasses import asdict, dataclass +from typing import Iterable + +from app.models import SsiInstruction + + +@dataclass(frozen=True) +class SsiValidationRule: + rule_id: str + version: str + family: str + component: str + fields: tuple[str, ...] + severity: str + enabled: bool + applies_to: str + evaluator: str + message: str + suggested_fix: str + privacy_class: str + exception_eligible: bool + export_blocking: bool + owner_team: str + sla_hours: int + + +REQUIRED_SSI_COMPONENTS = { + "source_file", + "source_record", + "owner", + "instrument", + "account_holder", + "account", + "preference", + "cross_reference", + "lifecycle", + "conflict", + "governance", + "export", +} + +REQUIRED_SOURCE_TRUTH_FIELDS = { + "source_record_key", + "owner_bic", + "owner_name", + "owner_city", + "owner_country_code", + "currency_code", + "asset_category", + "account_holder_bic", + "account_holder_name", + "account_holder_country_code", + "account_number_masked", + "preferred_flag", + "account_holder_type", + "group_key_owner", + "record_key_bdp_owner", + "eid_owner", + "record_key_bdp_account_holder", + "eid_account_holder", + "update_date", + "traffic_flag", + "traffic_date", + "start_date", + "stop_date", +} + +_ALLOWED_ASSET_CATEGORIES = { + "ANYY", + "WHLS", + "CASH", + "COLL", + "COMM", + "COPA", + "DERI", + "DOCC", + "FOEX", + "FUTU", + "GUAR", + "LETT", + "LOAN", + "MMKT", + "NDLF", + "OPTI", + "SECU", + "TFIN", + "TREA", + # existing synthetic fixtures use the display value below + "EQUITY", +} +_ALLOWED_HOLDER_TYPES = {"", "B", "C", "BANK", "CORRESPONDENT"} +_ALLOWED_TRAFFIC_FLAGS = {"", "Y", "N", "N/A"} +_ALLOWED_PREFERRED_FLAGS = {"", "Y", "P", "TRUE", "1", "N", "FALSE", "0"} + + +def _rule( + rule_id: str, + family: str, + component: str, + fields: Iterable[str], + severity: str, + evaluator: str, + message: str, + suggested_fix: str, + *, + privacy_class: str = "non_sensitive_code", + exception_eligible: bool = True, + export_blocking: bool = True, + owner_team: str = "ssi-operations", + sla_hours: int = 24, +) -> SsiValidationRule: + return SsiValidationRule( + rule_id=rule_id, + version="1.0.0", + family=family, + component=component, + fields=tuple(fields), + severity=severity, + enabled=True, + applies_to="SSIPLUS_V3", + evaluator=evaluator, + message=message, + suggested_fix=suggested_fix, + privacy_class=privacy_class, + exception_eligible=exception_eligible, + export_blocking=export_blocking, + owner_team=owner_team, + sla_hours=sla_hours, + ) + + +_CATALOGUE: tuple[SsiValidationRule, ...] = ( + _rule("ssi.schema.source_file", "schema", "source_file", ("source_file", "source_schema", "parser_version", "file_hash"), "critical", "source_lineage_present", "Source file lineage must identify the SSI Plus schema and parser.", "Import through the SSI Plus V3 source adapter with parser metadata."), + _rule("ssi.source_record.lineage", "schema", "source_record", ("source_record_key", "row_number"), "high", "source_record_lineage", "Source record lineage is required for auditability.", "Retain source row number and privacy-safe source fingerprint."), + _rule("ssi.owner_bic.required", "requiredness", "owner", ("owner_bic",), "high", "required", "Owner BIC is required.", "Populate the SSI owner BIC."), + _rule("ssi.owner_bic.format", "format", "owner", ("owner_bic",), "high", "bic_format", "Owner BIC is missing or malformed.", "Provide a valid 8 or 11 character BIC."), + _rule("ssi.owner_country.reference", "reference_data", "owner", ("owner_country_code",), "medium", "country_format_optional", "Owner country must use an ISO-like country code when present.", "Use a two-letter country code from the source reference data."), + _rule("ssi.owner_identity.profile", "cross_reference", "owner", ("owner_name", "owner_city", "group_key_owner"), "low", "owner_profile_present", "Owner profile fields support investigation routing.", "Retain owner name/city/group lineage internally; do not expose them in control payloads.", exception_eligible=False, export_blocking=False, privacy_class="sensitive_reference"), + _rule("ssi.instrument.currency.required", "requiredness", "instrument", ("currency_code",), "high", "required", "Instruction currency is required.", "Populate the SSI currency code."), + _rule("ssi.instrument.currency.reference", "reference_data", "instrument", ("currency_code",), "high", "currency_format", "Currency must use an ISO-like three-letter code.", "Use a three-letter currency code from the SWIFTRef code family."), + _rule("ssi.instrument.asset.required", "requiredness", "instrument", ("asset_category",), "high", "required", "Instruction asset category is required.", "Populate the SSI asset category."), + _rule("ssi.instrument.asset.reference", "reference_data", "instrument", ("asset_category",), "medium", "asset_reference", "Asset category must be in the SSI Directory relationship code family.", "Use a supported asset category such as CASH, SECU, FOEX, TREA, or an accepted synthetic fixture value."), + _rule("ssi.account_holder_bic.required", "requiredness", "account_holder", ("account_holder_bic",), "high", "required", "Account-holder BIC is required.", "Populate the account-holding institution BIC."), + _rule("ssi.account_holder_bic.format", "format", "account_holder", ("account_holder_bic",), "high", "bic_format", "Account-holder BIC is missing or malformed.", "Provide a valid 8 or 11 character BIC."), + _rule("ssi.account_holder_country.reference", "reference_data", "account_holder", ("account_holder_country_code",), "medium", "country_format_optional", "Account-holder country must use an ISO-like country code when present.", "Use a two-letter country code from the source reference data."), + _rule("ssi.account_holder.type.reference", "reference_data", "account_holder", ("account_holder_type",), "low", "holder_type_reference", "Account-holder type must use an approved code when present.", "Use a known account-holder type code or leave blank until reference data is available."), + _rule("ssi.account_holder_identity.profile", "cross_reference", "account_holder", ("account_holder_name",), "low", "account_holder_profile_present", "Account-holder profile fields support investigation routing.", "Retain account-holder name internally; do not expose it in control payloads.", exception_eligible=False, export_blocking=False, privacy_class="sensitive_reference"), + _rule("ssi.account.masked.required", "requiredness", "account", ("account_number_masked",), "critical", "required", "Masked account value is required.", "Provide a pre-masked account label before persistence.", privacy_class="secret_never_echo"), + _rule("ssi.account.masked.privacy", "privacy", "account", ("account_number_masked",), "critical", "masked_account", "Account value must remain masked and privacy-safe.", "Reject or tokenize raw account values before persistence.", privacy_class="secret_never_echo"), + _rule("ssi.preference.flag.reference", "reference_data", "preference", ("preferred_flag",), "medium", "preferred_flag", "Preferred flag must use an approved value.", "Use Y/P for preferred or N/blank for non-preferred."), + _rule("ssi.lifecycle.active_start.required", "lifecycle", "lifecycle", ("start_date", "status"), "medium", "active_start_date", "Active instruction requires a start date.", "Add an effective start date before activating the instruction."), + _rule("ssi.lifecycle.date_range", "lifecycle", "lifecycle", ("start_date", "stop_date"), "high", "date_range", "Instruction stop date must not be earlier than start date.", "Correct the instruction lifecycle dates."), + _rule("ssi.lifecycle.update_freshness", "freshness", "lifecycle", ("update_date",), "low", "update_date_present", "Update date is missing, so source freshness cannot be proven.", "Retain the source update date where available.", exception_eligible=False, export_blocking=False), + _rule("ssi.lifecycle.traffic_freshness", "freshness", "lifecycle", ("traffic_flag", "traffic_date"), "medium", "traffic_date_for_active_flag", "Traffic date is required when traffic flag indicates activity.", "Populate traffic date or correct traffic flag."), + _rule("ssi.cross_reference.owner_bdp_eid", "cross_reference", "cross_reference", ("record_key_bdp_owner", "eid_owner"), "low", "cross_reference_optional", "Owner BDP/EID cross-reference is not available.", "Load companion reference data or mark the gate skipped with evidence.", exception_eligible=False, export_blocking=False, privacy_class="sensitive_reference"), + _rule("ssi.cross_reference.holder_bdp_eid", "cross_reference", "cross_reference", ("record_key_bdp_account_holder", "eid_account_holder"), "low", "cross_reference_optional", "Account-holder BDP/EID cross-reference is not available.", "Load companion reference data or mark the gate skipped with evidence.", exception_eligible=False, export_blocking=False, privacy_class="sensitive_reference"), + _rule("ssi.conflict.preferred_unique", "preference_conflict", "conflict", ("preferred_flag", "currency_code", "asset_category", "account_holder_bic"), "high", "group_preferred_unique", "Only one active preferred instruction is allowed per operational group.", "Retain one preferred instruction and mark the others non-preferred."), + _rule("ssi.conflict.active_interval_overlap", "uniqueness_overlap", "conflict", ("start_date", "stop_date", "currency_code", "asset_category", "account_holder_bic"), "high", "group_active_overlap", "Active instruction intervals overlap within the same operational group.", "Adjust lifecycle dates or close the superseded instruction."), + _rule("ssi.governance.exception_route", "governance", "governance", ("owner_country_code", "currency_code", "asset_category"), "low", "governance_route", "Failed gates must be routeable to an accountable operations owner.", "Route exceptions by severity, country, currency, and asset family.", exception_eligible=False, export_blocking=False), + _rule("ssi.export.eligibility", "export_eligibility", "export", ("status",), "critical", "export_eligibility", "Instruction is not export-eligible while export-blocking gates fail.", "Resolve high/critical export-blocking validation failures before export.", exception_eligible=False), +) + + +def load_instruction_catalogue() -> tuple[SsiValidationRule, ...]: + return _CATALOGUE + + +def build_catalogue_matrix() -> dict[str, object]: + catalogue = load_instruction_catalogue() + return { + "total_rules": len(catalogue), + "by_family": dict(Counter(rule.family for rule in catalogue)), + "by_component": dict(Counter(rule.component for rule in catalogue)), + "by_severity": dict(Counter(rule.severity for rule in catalogue)), + "export_blocking": sum(1 for rule in catalogue if rule.export_blocking), + "rules": [asdict(rule) for rule in catalogue], + } + + +def existing_instruction_fields() -> set[str]: + return set(SsiInstruction.__mapper__.attrs.keys()) + + +def allowed_asset_categories() -> set[str]: + return set(_ALLOWED_ASSET_CATEGORIES) + + +def allowed_holder_types() -> set[str]: + return set(_ALLOWED_HOLDER_TYPES) + + +def allowed_traffic_flags() -> set[str]: + return set(_ALLOWED_TRAFFIC_FLAGS) + + +def allowed_preferred_flags() -> set[str]: + return set(_ALLOWED_PREFERRED_FLAGS) diff --git a/apps/ssi-control-tower/tests/test_instruction_exceptions.py b/apps/ssi-control-tower/tests/test_instruction_exceptions.py index eb77566..754d3e9 100644 --- a/apps/ssi-control-tower/tests/test_instruction_exceptions.py +++ b/apps/ssi-control-tower/tests/test_instruction_exceptions.py @@ -49,7 +49,11 @@ def test_failed_instruction_validation_creates_instruction_exception(db_session) validate_instruction(db_session, instruction, actor="tester@example.com") - case = db_session.query(InstructionExceptionCase).filter_by(ssi_instruction_id=instruction.ssi_instruction_id).one() + case = ( + db_session.query(InstructionExceptionCase) + .filter_by(ssi_instruction_id=instruction.ssi_instruction_id, rule_id="SSI.INSTRUCTION.START_DATE_REQUIRED") + .one() + ) assert case.rule_id == "SSI.INSTRUCTION.START_DATE_REQUIRED" assert case.status == "open" assert case.description @@ -66,7 +70,11 @@ def test_resolving_instruction_validation_closes_instruction_exception(db_sessio validate_instruction(db_session, instruction, actor="tester@example.com") - case = db_session.query(InstructionExceptionCase).filter_by(ssi_instruction_id=instruction.ssi_instruction_id).one() + case = ( + db_session.query(InstructionExceptionCase) + .filter_by(ssi_instruction_id=instruction.ssi_instruction_id, rule_id="SSI.INSTRUCTION.START_DATE_REQUIRED") + .one() + ) assert case.status == "closed" assert case.resolved_at is not None diff --git a/apps/ssi-control-tower/tests/test_ssi_instruction_validation_catalogue_engine.py b/apps/ssi-control-tower/tests/test_ssi_instruction_validation_catalogue_engine.py new file mode 100644 index 0000000..1cb4039 --- /dev/null +++ b/apps/ssi-control-tower/tests/test_ssi_instruction_validation_catalogue_engine.py @@ -0,0 +1,266 @@ +from __future__ import annotations + +from tests.test_instruction_validation import _add_instruction + + +def _add_lineage( + session, + instruction, + *, + source_record_key: str | None = None, + file_import_id: str = "import-validation-file", + record_import_id: str | None = None, + link_import_id: str | None = None, +): + from app.models import SourceFile, SourceRecord, SsiSourceLink + from app.services.audit import stable_id + + now = "2026-05-07T00:00:00Z" + source_file_id = stable_id("test-source-file", instruction.ssi_instruction_id) + source_record_id = stable_id("test-source-record", instruction.ssi_instruction_id) + record_import_id = record_import_id or file_import_id + link_import_id = link_import_id or file_import_id + session.add( + SourceFile( + source_file_id=source_file_id, + import_id=file_import_id, + file_name="synthetic-ssi-validation.csv", + file_hash=stable_id("test-source-file-hash", instruction.ssi_instruction_id), + source_system="synthetic-fixture", + source_schema="ssi-plus-v3", + parser_version="test-parser-v1", + uploaded_by="tester@example.com", + records_received=1, + records_accepted=1, + records_rejected=0, + created_at=now, + ) + ) + session.add( + SourceRecord( + source_record_id=source_record_id, + source_file_id=source_file_id, + import_id=record_import_id, + row_number=2, + source_record_key=source_record_key or instruction.source_record_key, + source_hash=stable_id("test-source-record-hash", instruction.ssi_instruction_id), + status="accepted", + created_at=now, + ) + ) + session.add( + SsiSourceLink( + ssi_source_link_id=stable_id("test-ssi-source-link", instruction.ssi_instruction_id), + ssi_instruction_id=instruction.ssi_instruction_id, + source_record_id=source_record_id, + source_file_id=source_file_id, + import_id=link_import_id, + created_at=now, + ) + ) + session.commit() + + +def test_validate_instruction_writes_result_for_every_enabled_catalogue_rule(db_session): + from app.services.instruction_validation import validate_instruction + from app.services.ssi_validation_catalogue import load_instruction_catalogue + + instruction = _add_instruction(db_session, source_record_key="fp-catalogue-results") + + results = validate_instruction(db_session, instruction, actor="tester@example.com") + + enabled_rule_ids = {rule.rule_id for rule in load_instruction_catalogue() if rule.enabled} + result_rule_ids = {result.rule_id for result in results} + assert enabled_rule_ids <= result_rule_ids + assert {result.status for result in results} <= {"pass", "fail", "skipped"} + + +def test_invalid_owner_bic_creates_privacy_safe_exception(db_session): + from app.models import InstructionExceptionCase + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction( + db_session, + source_record_key="SOURCE_RECORD_KEY_PLACEHOLDER", + owner_bic="BAD", + owner_name="INSTITUTION_LABEL_PLACEHOLDER", + owner_city="CITY_LABEL_PLACEHOLDER", + account_number_masked="MASKED_ACCOUNT_PLACEHOLDER", + ) + + validate_instruction(db_session, instruction, actor="tester@example.com") + + case = db_session.query(InstructionExceptionCase).filter_by(rule_id="ssi.owner_bic.format").one() + payload_text = " ".join( + item or "" + for item in ( + case.description, + case.suggested_fix, + case.context_key, + case.failed_field, + ) + ) + assert case.severity == "high" + assert case.failed_field == "owner_bic" + assert case.status in {"open", "assigned"} + assert case.sla_deadline is not None + for forbidden in ( + "SOURCE_RECORD_KEY_PLACEHOLDER", + "INSTITUTION_LABEL_PLACEHOLDER", + "CITY_LABEL_PLACEHOLDER", + "MASKED_ACCOUNT_PLACEHOLDER", + ): + assert forbidden not in payload_text + + +def test_export_eligibility_fails_when_export_blocking_gate_fails(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction( + db_session, + source_record_key="fp-export-blocked", + currency_code="US1", + ) + results = validate_instruction(db_session, instruction, actor="tester@example.com") + + export_result = next(result for result in results if result.rule_id == "ssi.export.eligibility") + assert export_result.status == "fail" + assert export_result.severity == "critical" + assert export_result.failed_field == "export_eligibility" + + +def test_missing_source_lineage_fails_export_blocking_catalogue_rules(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction(db_session, source_record_key="fp-missing-lineage") + results = validate_instruction(db_session, instruction, actor="tester@example.com") + by_rule = {result.rule_id: result for result in results} + + assert by_rule["ssi.schema.source_file"].status == "fail" + assert by_rule["ssi.schema.source_file"].failed_field == "source_file" + assert by_rule["ssi.source_record.lineage"].status == "fail" + assert by_rule["ssi.source_record.lineage"].failed_field == "source_record_key" + assert by_rule["ssi.export.eligibility"].status == "fail" + + +def test_present_source_lineage_passes_catalogue_lineage_rules(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction(db_session, source_record_key="fp-present-lineage") + _add_lineage(db_session, instruction) + + results = validate_instruction(db_session, instruction, actor="tester@example.com") + by_rule = {result.rule_id: result for result in results} + + assert by_rule["ssi.schema.source_file"].status == "pass" + assert by_rule["ssi.source_record.lineage"].status == "pass" + + +def test_mismatched_source_record_key_fails_lineage_rule(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction(db_session, source_record_key="fp-expected-lineage") + _add_lineage(db_session, instruction, source_record_key="fp-other-lineage") + + results = validate_instruction(db_session, instruction, actor="tester@example.com") + by_rule = {result.rule_id: result for result in results} + + assert by_rule["ssi.source_record.lineage"].status == "fail" + assert by_rule["ssi.export.eligibility"].status == "fail" + + +def test_mismatched_lineage_import_id_fails_lineage_rule(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction(db_session, source_record_key="fp-import-lineage") + _add_lineage(db_session, instruction, record_import_id="import-record-mismatch") + + results = validate_instruction(db_session, instruction, actor="tester@example.com") + by_rule = {result.rule_id: result for result in results} + + assert by_rule["ssi.source_record.lineage"].status == "fail" + assert by_rule["ssi.export.eligibility"].status == "fail" + + +def test_validate_instruction_detects_duplicate_preferred_peer(db_session): + from app.services.instruction_validation import validate_instruction + + first = _add_instruction(db_session, source_record_key="fp-single-preferred-a", preferred_flag="Y") + _add_instruction(db_session, source_record_key="fp-single-preferred-b", preferred_flag="Y") + + results = validate_instruction(db_session, first, actor="tester@example.com") + preferred_statuses = {result.status for result in results if result.rule_id == "ssi.conflict.preferred_unique"} + + assert preferred_statuses == {"fail"} + assert next(result for result in results if result.rule_id == "ssi.export.eligibility").status == "fail" + + +def test_validate_instruction_detects_active_interval_overlap_peer(db_session): + from app.services.instruction_validation import validate_instruction + + first = _add_instruction( + db_session, + source_record_key="fp-single-overlap-a", + start_date="2026-01-01", + stop_date="2026-06-30", + ) + _add_instruction( + db_session, + source_record_key="fp-single-overlap-b", + start_date="2026-05-01", + stop_date="2026-12-31", + ) + + results = validate_instruction(db_session, first, actor="tester@example.com") + overlap_statuses = {result.status for result in results if result.rule_id == "ssi.conflict.active_interval_overlap"} + + assert overlap_statuses == {"fail"} + assert next(result for result in results if result.rule_id == "ssi.export.eligibility").status == "fail" + + + +def test_validation_result_context_key_is_privacy_safe(db_session): + from app.services.instruction_validation import validate_instruction + + instruction = _add_instruction( + db_session, + source_record_key="BDP_RECORD_KEY_PLACEHOLDER", + account_holder_bic="BAD", + account_holder_name="ACCOUNT_HOLDER_LABEL_PLACEHOLDER", + account_number_masked="MASKED_ACCOUNT_PLACEHOLDER", + ) + + results = validate_instruction(db_session, instruction, actor="tester@example.com") + payload_text = " ".join( + " ".join(item or "" for item in (result.message, result.suggested_fix, result.context_key)) + for result in results + ) + + for forbidden in ( + "BDP_RECORD_KEY_PLACEHOLDER", + "ACCOUNT_HOLDER_LABEL_PLACEHOLDER", + "MASKED_ACCOUNT_PLACEHOLDER", + ): + assert forbidden not in payload_text + + +def test_group_conflict_rules_do_not_emit_both_pass_and_fail(db_session): + from app.models import InstructionValidationResult + from app.services.instruction_validation import validate_all_instructions + + first = _add_instruction(db_session, source_record_key="fp-catalogue-preferred-a", preferred_flag="Y") + second = _add_instruction(db_session, source_record_key="fp-catalogue-preferred-b", preferred_flag="Y") + + validate_all_instructions(db_session, actor="tester@example.com") + + for instruction in (first, second): + statuses = { + result.status + for result in db_session.query(InstructionValidationResult) + .filter_by( + ssi_instruction_id=instruction.ssi_instruction_id, + rule_id="ssi.conflict.preferred_unique", + ) + .all() + } + assert statuses == {"fail"} diff --git a/apps/ssi-control-tower/tests/test_ssi_validation_catalogue_contract.py b/apps/ssi-control-tower/tests/test_ssi_validation_catalogue_contract.py new file mode 100644 index 0000000..ba7beb3 --- /dev/null +++ b/apps/ssi-control-tower/tests/test_ssi_validation_catalogue_contract.py @@ -0,0 +1,81 @@ +from __future__ import annotations + + +def test_catalogue_loads_with_unique_stable_rule_ids(): + from app.services.ssi_validation_catalogue import load_instruction_catalogue + + catalogue = load_instruction_catalogue() + rule_ids = [rule.rule_id for rule in catalogue] + + assert catalogue + assert len(set(rule_ids)) == len(rule_ids) + assert all(rule_id.startswith("ssi.") for rule_id in rule_ids) + + +def test_catalogue_has_required_gate_families(): + from app.services.ssi_validation_catalogue import load_instruction_catalogue + + required_families = { + "schema", + "requiredness", + "format", + "reference_data", + "lifecycle", + "preference_conflict", + "uniqueness_overlap", + "cross_reference", + "freshness", + "privacy", + "governance", + "export_eligibility", + } + + assert required_families <= {rule.family for rule in load_instruction_catalogue()} + + +def test_catalogue_covers_every_ssi_instruction_component_and_source_truth_field(): + from app.models import SsiInstruction + from app.services.ssi_validation_catalogue import ( + REQUIRED_SSI_COMPONENTS, + REQUIRED_SOURCE_TRUTH_FIELDS, + load_instruction_catalogue, + ) + + catalogue = load_instruction_catalogue() + components = {rule.component for rule in catalogue} + fields = {field for rule in catalogue for field in rule.fields} + model_fields = set(SsiInstruction.__mapper__.attrs.keys()) + + assert REQUIRED_SSI_COMPONENTS <= components + assert REQUIRED_SOURCE_TRUTH_FIELDS <= fields + assert fields <= model_fields | {"source_file", "source_schema", "parser_version", "row_number", "file_hash"} + + +def test_catalogue_rules_have_governance_privacy_and_evaluator_metadata(): + from app.services.ssi_validation_catalogue import load_instruction_catalogue + + severities = {"info", "low", "medium", "high", "critical"} + for rule in load_instruction_catalogue(): + assert rule.version + assert rule.severity in severities + assert rule.evaluator + assert rule.message + assert rule.suggested_fix + assert rule.privacy_class in {"public_code", "non_sensitive_code", "sensitive_reference", "secret_never_echo"} + assert rule.owner_team + assert rule.sla_hours > 0 + assert isinstance(rule.export_blocking, bool) + assert isinstance(rule.exception_eligible, bool) + + +def test_catalogue_can_emit_machine_readable_demo_matrix(): + from app.services.ssi_validation_catalogue import build_catalogue_matrix + + matrix = build_catalogue_matrix() + + assert matrix["total_rules"] > 0 + assert matrix["by_family"]["privacy"] >= 1 + assert matrix["by_component"]["account"] >= 1 + assert matrix["export_blocking"] >= 1 + assert "rules" in matrix + assert all("rule_id" in rule for rule in matrix["rules"]) diff --git a/docs/ssi-control-tower-fold-plan.md b/docs/ssi-control-tower-fold-plan.md index 3810f73..5b93d94 100644 --- a/docs/ssi-control-tower-fold-plan.md +++ b/docs/ssi-control-tower-fold-plan.md @@ -4,15 +4,15 @@ _Last updated: 2026-05-07_ ## Decision -SSI Control Tower is folded into this repository as a self-contained backend module at `apps/ssi-control-tower/`. +SSI Control Tower is folded into this repository as a standalone adjacent FastAPI/Jinja app at `apps/ssi-control-tower/`. -This does **not** convert the existing Payment Intelligence browser suite into a backend app. The root Vite/React app remains browser-only and static-host friendly. Its privacy audit remains scoped to `src/` and must not be weakened by SSI Control Tower work. +This does **not** convert the existing Payment Intelligence browser suite into a backend app, and it does not make SSI Control Tower a Payment Intelligence module/workflow. The root Vite/React app remains browser-only and static-host friendly. Its privacy audit remains scoped to `src/` and must not be weakened by SSI Control Tower work. The retained `/ssi` route is only an unlinked static boundary pointer until repo extraction and product-boundary decisions are made. ## Selected implementation scope This branch implements the approved hybrid SSI instruction flow plan from the historical source repo commit `d6c9787`: -- Phase -1: repository fold into Payment Intelligence Modules. +- Phase -1: repository fold beside Payment Intelligence Modules, preserving SSI Control Tower as a standalone adjacent app. - Phase 0: baseline guardrails and SQLite `create_all` schema compatibility discipline. - Phase 1: unified source adapters and shared canonical instruction lineage ingestion. - Phase 2: instruction-level validation, instruction exceptions, and a unified control queue API. @@ -25,9 +25,9 @@ Phases 3-5, including instruction approval workflow, readiness/export expansion, - [x] Do not copy a nested `.git/` directory. - [x] Do not copy generated DBs, exports, caches, virtualenvs, or local private data. - [x] Keep the Python app runnable from `apps/ssi-control-tower/` with its own `Makefile` and `pyproject.toml`. -- [x] Keep root `src/` browser-only and leave the root route table untouched. +- [x] Keep root `src/` browser-only; any retained root `/ssi` route is a static boundary pointer, not a module launcher. - [x] Keep root `scripts/privacy-audit.sh src` unchanged and scoped to the browser suite. -- [x] Document SSI Control Tower as an adjacent backend module inside the repo, not as part of the static browser runtime. +- [x] Document SSI Control Tower as a standalone adjacent app inside the repo, not as part of the static browser runtime or the Payment Intelligence module set. ## Verification commands diff --git a/e2e/smoke.spec.ts b/e2e/smoke.spec.ts index 2121bc4..ce17083 100644 --- a/e2e/smoke.spec.ts +++ b/e2e/smoke.spec.ts @@ -20,6 +20,9 @@ test("home page lists module tiles and states the privacy boundary", async ({ pa await expect(page.getByRole("link", { name: /cbpr\+/i }).first()).toBeVisible(); await expect(page.getByRole("link", { name: /insights/i }).first()).toBeVisible(); + // SSI Control Tower is decoupled: zero SSI links in the primary nav or home launcher. + await expect(page.getByRole("link", { name: /ssi/i })).toHaveCount(0); + // The renamed module section heading is visible. await expect(page.getByRole("heading", { name: /choose a review workflow/i })).toBeVisible(); diff --git a/e2e/ssi.spec.ts b/e2e/ssi.spec.ts index 2eb815e..4524ab1 100644 --- a/e2e/ssi.spec.ts +++ b/e2e/ssi.spec.ts @@ -1,21 +1,52 @@ import { expect, test } from "@playwright/test"; -test("SSI Control Tower overview is a static backend-module entry point", async ({ page }) => { +test("retained /ssi is an unlinked static boundary pointer to a standalone adjacent app", async ({ + context, + page, +}) => { const requestedUrls: string[] = []; page.on("request", (request) => requestedUrls.push(request.url())); await page.goto("/ssi"); await expect(page.getByRole("heading", { level: 1, name: /ssi control tower/i })).toBeVisible(); - await expect(page.getByText(/backend module/i)).toBeVisible(); + await expect(page.getByText(/not a Payment Intelligence module/i).first()).toBeVisible(); + await expect(page.getByText(/standalone adjacent app/i).first()).toBeVisible(); await expect(page.getByText(/synthetic-data public prototype/i)).toBeVisible(); - await expect(page.getByText(/apps\/ssi-control-tower/)).toBeVisible(); - await expect(page.getByText(/make run/)).toBeVisible(); + await expect(page.getByText(/apps\/ssi-control-tower/).first()).toBeVisible(); + // Boundary pointer only: no suite-module framing and no clickable localhost link. + await expect(page.getByText(/backend module/i)).toHaveCount(0); + await expect(page.getByRole("link", { name: /localhost/i })).toHaveCount(0); + + // No external network requests — the page never calls the standalone app. const origin = new URL(page.url()).origin; const externalRequests = requestedUrls.filter((url) => { const parsed = new URL(url); return parsed.protocol.startsWith("http") && parsed.origin !== origin; }); expect(externalRequests).toEqual([]); + + // No browser persistence is written by the boundary pointer. + const persistenceState = await page.evaluate(async () => { + const indexedDbNames = + "databases" in indexedDB + ? (await indexedDB.databases()).map((database) => database.name).filter(Boolean) + : []; + + return { + localStorageLength: localStorage.length, + sessionStorageLength: sessionStorage.length, + cookie: document.cookie, + indexedDbNames, + }; + }); + + expect(persistenceState).toEqual({ + localStorageLength: 0, + sessionStorageLength: 0, + cookie: "", + indexedDbNames: [], + }); + await expect.poll(async () => (await context.cookies()).length).toBe(0); }); diff --git a/src/App.test.tsx b/src/App.test.tsx index 83e1d4e..5deb9bb 100644 --- a/src/App.test.tsx +++ b/src/App.test.tsx @@ -41,7 +41,8 @@ describe("App", () => { expect(screen.getAllByRole("link", { name: /iban/i }).length).toBeGreaterThan(0); expect(screen.getAllByRole("link", { name: /cbpr\+/i }).length).toBeGreaterThan(0); expect(screen.getAllByRole("link", { name: /insights/i }).length).toBeGreaterThan(0); - expect(screen.getAllByRole("link", { name: /ssi/i }).length).toBeGreaterThan(0); + // SSI is decoupled from the suite: no SSI link in the primary nav or home launcher. + expect(screen.queryAllByRole("link", { name: /ssi/i })).toHaveLength(0); }); it("renders the Scrubber page at /scrubber", async () => { @@ -76,11 +77,14 @@ describe("App", () => { await waitFor(() => expect(global.fetch).toHaveBeenCalled()); }); - it("renders the SSI Control Tower overview at /ssi", () => { + it("renders the SSI boundary pointer at /ssi", () => { renderAt("/ssi"); expect( screen.getByRole("heading", { level: 1, name: /ssi control tower/i }), ).toBeInTheDocument(); + // The retained /ssi route is a boundary pointer, not a suite module. + expect(screen.getAllByText(/not a Payment Intelligence module/i).length).toBeGreaterThan(0); + expect(screen.getAllByText(/standalone adjacent app/i).length).toBeGreaterThan(0); }); it("renders a not-found page for unknown routes", () => { diff --git a/src/components/layout/SuiteHeader.tsx b/src/components/layout/SuiteHeader.tsx index facc81a..6d1edc5 100644 --- a/src/components/layout/SuiteHeader.tsx +++ b/src/components/layout/SuiteHeader.tsx @@ -16,7 +16,6 @@ const navItems: NavItem[] = [ { to: "/bic", label: "BIC*", end: false }, { to: "/cbpr", label: "CBPR+", end: false }, { to: "/insights", label: "Insights", end: false }, - { to: "/ssi", label: "SSI", end: false }, ]; export function SuiteHeader() { diff --git a/src/pages/HomePage.tsx b/src/pages/HomePage.tsx index 2086665..ce9e912 100644 --- a/src/pages/HomePage.tsx +++ b/src/pages/HomePage.tsx @@ -19,7 +19,7 @@ interface ModuleTile { name: string; summary: string; icon: typeof Eraser; - status: "available" | "demo" | "backend" | "planned"; + status: "available" | "demo" | "planned"; } const modules: ModuleTile[] = [ @@ -71,14 +71,6 @@ const modules: ModuleTile[] = [ icon: GitBranch, status: "available", }, - { - to: "/ssi", - name: "SSI Control Tower", - summary: - "Static entry point for the folded FastAPI/Jinja SSI governance cockpit. Runs locally from apps/ssi-control-tower and stays outside the browser-only runtime.", - icon: ShieldCheck, - status: "backend", - }, { to: "#", name: "Vault", @@ -208,7 +200,6 @@ export function HomePage() { const Icon = module.icon; const isPlanned = module.status === "planned"; const isDemo = module.status === "demo"; - const isBackend = module.status === "backend"; const tileClass = cn( "group relative flex min-h-64 flex-col p-6 transition-all", "practice-card", @@ -216,20 +207,12 @@ export function HomePage() { ? "cursor-not-allowed opacity-65" : "hover:-translate-y-0.5 hover:border-accent/40 hover:shadow-md hover:shadow-slate-200/80", ); - const badgeLabel = isPlanned - ? "Planned" - : isDemo - ? "Demo" - : isBackend - ? "Backend" - : "Available"; + const badgeLabel = isPlanned ? "Planned" : isDemo ? "Demo" : "Available"; const badgeClass = isPlanned ? "border-border bg-muted text-muted-foreground" : isDemo ? "border-amber-300/60 bg-amber-100/70 text-amber-900" - : isBackend - ? "border-indigo-200 bg-indigo-50 text-indigo-900" - : "border-brand/25 bg-brand/10 text-primary"; + : "border-brand/25 bg-brand/10 text-primary"; const inner = ( <>
- SSI Control Tower is not part of the browser-only runtime. It lives as a folded
- FastAPI/Jinja application inside this repository, while this React page documents the
- entry point, posture, and local operator workflow without calling an API.
+ SSI Control Tower is not part of this browser-only suite and is not one of its modules.
+ It is a standalone adjacent app under apps/ssi-control-tower. This React
+ page only records the boundary and points to where the separate app runs — it never
+ calls an API and is not coupled to the SSI runtime.
+
+ Repo extraction and the product-boundary decision for the standalone app are still + pending. Until that decision is made, this page stays a backlog pointer rather than a + suite feature.
@@ -59,11 +72,7 @@ export function SsiPage() {
What it does
+What the standalone app does
Run it locally
{"cd apps/ssi-control-tower && make run\n# opens http://localhost:8000"}
- The local app owns its Python dependencies, SQLite development data, templates, APIs, - tests, and synthetic fixtures. Read the folded app README for command details and demo - flow. + The standalone app owns its Python dependencies, SQLite development data, templates, + APIs, tests, and synthetic fixtures. Read its README under apps/ssi-control-tower for + command details and demo flow.
@@ -116,7 +125,7 @@ export function SsiPage() {Rule catalogue