#!/usr/bin/env python3 """Fail-closed contract for server-generated consultation workflow plans.""" from __future__ import annotations from dataclasses import dataclass from typing import Any PLAN_VERSION = "consultation-plan-v2" _ALLOWED_DEPTHS = {"concise", "standard", "deep", "research"} _ALLOWED_HORIZONS = {"next_3_months", "next_12_months", "next_24_months", "long_term"} _ALLOWED_PRECISION_BOUNDARIES = {"server_evidence_required", "precise_timing_blocked"} _PLAN_METADATA_KEYS = frozenset({ "plan_version", "strict_workflow_route", "required_layers", "claim_boundary", "plan_depth", "requested_domains", "timing_horizon", "precision_boundary", "required_evidence_categories", }) @dataclass(frozen=True) class RouteContract: themes: tuple[str, ...] resolved_routes: tuple[str, ...] requested_domains: tuple[str, ...] required_evidence_categories: tuple[str, ...] required_layers: tuple[str, ...] claim_boundary: str _ROUTE_CONTRACTS = { "career": RouteContract( themes=("career",), resolved_routes=("career",), requested_domains=("career",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D10", "10th house/lord", "A10", "AmK", "Vimshottari", "Narayana", "Transit"), claim_boundary="career_direction_and_broad_timing_only", ), "marriage": RouteContract( themes=("marriage",), resolved_routes=("marriage", "relationship"), requested_domains=("marriage",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D9", "7th house/lord", "Venus/Jupiter", "DK", "UL", "A7", "Vimshottari", "Narayana", "Transit"), claim_boundary="relationship_pattern_and_broad_window_only", ), "wealth": RouteContract( themes=("wealth",), resolved_routes=("wealth", "finance"), requested_domains=("wealth",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D2", "D11", "2nd/11th/9th/5th houses", "Wealth Yogas", "Ashtakavarga", "Dasha"), claim_boundary="wealth_structure_not_financial_advice", ), "health": RouteContract( themes=("health",), resolved_routes=("health",), requested_domains=("health",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D6", "D8", "6th/8th houses", "Dasha", "non-medical boundary"), claim_boundary="wellbeing_pressure_patterns_not_medical_diagnosis", ), "education": RouteContract( themes=("education",), resolved_routes=("education",), requested_domains=("education",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D24", "5th/9th houses", "Mercury/Jupiter", "Dasha"), claim_boundary="learning_pattern_and_broad_timing_only", ), "migration": RouteContract( themes=("migration",), resolved_routes=("migration",), requested_domains=("migration",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D4", "D12", "4th/12th houses", "Dasha", "Narayana"), claim_boundary="migration_and_home_direction_broad_window_only", ), "family": RouteContract( themes=("family",), resolved_routes=("family",), requested_domains=("family",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D7", "D12", "4th/5th/9th houses", "Dasha"), claim_boundary="family_pattern_not_deterministic_event_claim", ), "annual": RouteContract( themes=("annual",), resolved_routes=("annual",), requested_domains=("annual",), required_evidence_categories=("natal_foundation", "timing", "validation"), required_layers=("D1", "Annual chart boundary", "Dasha", "Transit", "Tajika candidate"), claim_boundary="annual_report_broad_periods_only", ), "timing": RouteContract( themes=("timing",), resolved_routes=("timing",), requested_domains=("timing",), required_evidence_categories=("natal_foundation", "timing", "validation"), required_layers=("Vimshottari", "Narayana", "Transit", "Varga", "negative holdout gate"), claim_boundary="candidate_day_month_window_only_until_holdout_passes", ), "general": RouteContract( themes=("general",), resolved_routes=("general",), requested_domains=("general",), required_evidence_categories=("natal_foundation", "domain"), required_layers=("D1", "D9", "D10", "D2", "Dasha", "Narayana", "Transit", "Functional Benefic/Malefic"), claim_boundary="multi_domain_summary_with_missing_layers_disclosed", ), } class ConsultationPlanContractError(ValueError): """Raised when a versioned product plan does not match the server allowlist.""" def _require_list(body: dict[str, Any], key: str) -> tuple[str, ...]: value = body.get(key) if not isinstance(value, list) or not value or any(not isinstance(item, str) or not item for item in value): raise ConsultationPlanContractError(f"invalid consultation plan field: {key}") return tuple(value) def validate_consultation_plan_contract( body: dict[str, Any], *, themes: list[str], route_packet: dict[str, Any], ) -> dict[str, Any] | None: """Validate the optional versioned plan without accepting free-form workflow control. Older research/MCP callers may omit ``plan_version``. Product runtime calls include it and must match every server-owned route field exactly. """ present_plan_keys = _PLAN_METADATA_KEYS.intersection(body) if not present_plan_keys: return None missing_plan_keys = _PLAN_METADATA_KEYS.difference(body) if missing_plan_keys: missing = ", ".join(sorted(missing_plan_keys)) raise ConsultationPlanContractError(f"incomplete consultation plan metadata: {missing}") if body.get("plan_version") != PLAN_VERSION: raise ConsultationPlanContractError("unsupported consultation plan version") strict_route = body.get("strict_workflow_route") contract = _ROUTE_CONTRACTS.get(strict_route) if contract is None: raise ConsultationPlanContractError("unsupported consultation workflow route") if tuple(themes) != contract.themes: raise ConsultationPlanContractError("consultation plan theme mismatch") resolved_route = str(route_packet.get("question_type") or "") if resolved_route not in contract.resolved_routes: raise ConsultationPlanContractError("consultation plan route mismatch") if _require_list(body, "required_layers") != contract.required_layers: raise ConsultationPlanContractError("consultation plan required layers mismatch") if body.get("claim_boundary") != contract.claim_boundary: raise ConsultationPlanContractError("consultation plan claim boundary mismatch") if _require_list(body, "requested_domains") != contract.requested_domains: raise ConsultationPlanContractError("consultation plan requested domains mismatch") if _require_list(body, "required_evidence_categories") != contract.required_evidence_categories: raise ConsultationPlanContractError("consultation plan evidence categories mismatch") depth = body.get("plan_depth") if depth not in _ALLOWED_DEPTHS: raise ConsultationPlanContractError("unsupported consultation plan depth") horizon = body.get("timing_horizon") if horizon is not None and horizon not in _ALLOWED_HORIZONS: raise ConsultationPlanContractError("unsupported consultation timing horizon") if strict_route in {"timing", "annual"} and horizon is None: raise ConsultationPlanContractError("timing consultation requires a horizon") precision_boundary = body.get("precision_boundary") if precision_boundary not in _ALLOWED_PRECISION_BOUNDARIES: raise ConsultationPlanContractError("unsupported consultation precision boundary") return { "plan_version": PLAN_VERSION, "strict_workflow_route": strict_route, "plan_depth": depth, "requested_domains": list(contract.requested_domains), "timing_horizon": horizon, "required_evidence_categories": list(contract.required_evidence_categories), "required_layers": list(contract.required_layers), "claim_boundary": contract.claim_boundary, "precision_boundary": precision_boundary, "enforcement": "server_allowlist_validated", } def apply_plan_precision_boundary( consumer_context: dict[str, Any], plan_contract: dict[str, Any] | None, ) -> dict[str, Any]: """Apply only a restrictive plan boundary; evidence policy remains authoritative.""" if not plan_contract or plan_contract.get("precision_boundary") != "precise_timing_blocked": return consumer_context answer_policy = consumer_context.get("answer_policy") if not isinstance(answer_policy, dict): return consumer_context return { **consumer_context, "answer_policy": { **answer_policy, "can_answer_precise_timing": False, "should_lead_with_limitations": True, "plan_precision_boundary": "precise_timing_blocked", }, }