Jyotisha/scripts/consultation_plan_contract.py

#!/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 plan_route_contract(route: Any) -> RouteContract | None:
    """Expose the server-owned contract for a route; callers never supply their own."""

    return _ROUTE_CONTRACTS.get(route) if isinstance(route, str) else None


def _declared_route_and_contract(body: dict[str, Any]) -> tuple[str, RouteContract] | None:
    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 = plan_route_contract(strict_route)
    if contract is None:
        raise ConsultationPlanContractError("unsupported consultation workflow route")
    return str(strict_route), contract


def declared_workflow_route(body: dict[str, Any]) -> str | None:
    """Return the allowlisted route a versioned plan declares, before any text heuristic runs.

    The route the product declares is server-issued, so it decides execution instead of
    competing with keyword routing. ``None`` keeps question-text routing authoritative for
    legacy research/MCP callers that send no plan metadata.
    """

    resolved = _declared_route_and_contract(body)
    return None if resolved is None else resolved[0]


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. ``route_packet`` must already be
    resolved from :func:`declared_workflow_route`, so the route check below confirms the
    workflow executed the declared route rather than arbitrating between two routers.
    """

    resolved = _declared_route_and_contract(body)
    if resolved is None:
        return None
    strict_route, contract = resolved
    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",
        },
    }