Add skill experience service contracts

mcp_server.py

@@ -45,6 +45,7 @@ from mcp.server.fastmcp import FastMCP
 from functional_benefics import derive_functional_benefic_malefic
 from vedastro_priority import official_snapshot_evidence
 from unified_consultation_orchestrator import UnifiedConsultationOrchestrator
+from skill_experience import build_skill_doctor, build_skill_onboarding, summarize_execution_status
 
 load_local_env(SCRIPT_DIR)
 
@@ -4353,6 +4354,22 @@ def life_event_graph(
     }
 
 
+# ============================================================================
+# Skill experience tools
+# ============================================================================
+
+@mcp.tool()
+def skill_onboarding(payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    """Return the minimal next input or active rectification question set."""
+    return build_skill_onboarding(payload)
+
+
+@mcp.tool()
+def skill_doctor() -> Dict[str, Any]:
+    """Check local Skill assets and external adapter readiness."""
+    return build_skill_doctor()
+
+
 # ============================================================================
 # Resources
 # ============================================================================

scripts/consultation_workflow_service.py

@@ -0,0 +1,32 @@
+#!/usr/bin/env python3
+"""Shared consultation workflow boundary for API and MCP callers."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS_DIR = ROOT / "scripts"
+for path in (ROOT, SCRIPTS_DIR):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+
+
+def execute_consultation_workflow(body: dict[str, Any], *, surface: str = "api") -> dict[str, Any]:
+    from jyotish_api_server import JyotishAPIHandler, execute_consultation_workflow as _execute
+
+    handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
+    return _execute(handler, body=body, surface=surface)
+
+
+def build_runtime_evidence_helpers(chart: dict[str, Any]) -> dict[str, Any]:
+    from jyotish_api_server import JyotishAPIHandler
+
+    handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
+    return {
+        "vedastro_official": handler._high_rigor_vedastro_official_summary(chart),
+        "vedastro_archive_manifest": handler._compute_vedastro_gateway_archives(),
+        "interpretation_coverage": handler._interpretation_source_runtime_coverage(chart),
+    }

scripts/skill_experience.py

@@ -0,0 +1,135 @@
+"""Stable user-facing contracts shared by Skill and MCP entry points."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from scripts.active_rectification_questions import build_questionnaire, score_answers
+from scripts.diagnose_external_engine_adapters import build_report as adapter_report
+
+
+ROOT = Path(__file__).resolve().parents[1]
+_REQUIRED_BIRTH_FIELDS = ("year", "month", "day", "hour", "minute", "lat", "lon")
+
+
+def _missing_birth_fields(payload: dict[str, Any]) -> list[str]:
+    return [field for field in _REQUIRED_BIRTH_FIELDS if payload.get(field) is None]
+
+
+def build_skill_onboarding(payload: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Return the next minimal user action; never infer missing birth inputs."""
+    payload = payload or {}
+    missing = _missing_birth_fields(payload)
+    if missing:
+        return {
+            "scope": "skill_onboarding",
+            "status": "needs_birth_data",
+            "entry_mode": "pending",
+            "missing_fields": missing,
+            "next_action": "collect_birth_data",
+            "input_template": {
+                "year": "YYYY", "month": "MM", "day": "DD",
+                "hour": "0-23", "minute": "0-59", "lat": "decimal", "lon": "decimal",
+                "time_uncertainty_minutes": "optional; use when birth time is approximate",
+                "question": "optional; career, relationship, wealth, health, general",
+            },
+        }
+
+    uncertainty = int(payload.get("time_uncertainty_minutes") or 0)
+    if uncertainty > 0:
+        birth_time = (
+            f"{int(payload['year']):04d}-{int(payload['month']):02d}-{int(payload['day']):02d} "
+            f"{int(payload['hour']):02d}:{int(payload['minute']):02d}"
+        )
+        questionnaire = build_questionnaire(birth_time, uncertainty_minutes=uncertainty)
+        first_question = questionnaire.get("questions", [{}])[0]
+        return {
+            "scope": "skill_onboarding",
+            "status": "ready",
+            "entry_mode": "rectification",
+            "next_action": "run_rectification_questionnaire",
+            "first_question": first_question,
+            "questionnaire": questionnaire,
+        }
+
+    return {
+        "scope": "skill_onboarding",
+        "status": "ready",
+        "entry_mode": "direct_chart",
+        "next_action": "run_consultation_workflow",
+        "question": str(payload.get("question") or ""),
+    }
+
+
+def build_rectification_questionnaire(payload: dict[str, Any]) -> dict[str, Any]:
+    """Build the active-choice questionnaire from a minimal approximate time."""
+    required = ("year", "month", "day", "hour", "minute")
+    missing = [field for field in required if payload.get(field) is None]
+    if missing:
+        raise ValueError(f"missing rectification fields: {', '.join(missing)}")
+    birth_time = (
+        f"{int(payload['year']):04d}-{int(payload['month']):02d}-{int(payload['day']):02d} "
+        f"{int(payload['hour']):02d}:{int(payload['minute']):02d}"
+    )
+    uncertainty = max(int(payload.get("time_uncertainty_minutes") or 30), 1)
+    step = max(int(payload.get("step_minutes") or 1), 1)
+    return build_questionnaire(birth_time, uncertainty_minutes=uncertainty, step_minutes=step)
+
+
+def score_rectification_answers(questionnaire: dict[str, Any], answers: dict[str, str]) -> dict[str, Any]:
+    """Score user choices; preserves the boundary against false minute precision."""
+    return score_answers(questionnaire, answers or {})
+
+
+def build_skill_doctor() -> dict[str, Any]:
+    """Expose readiness, not an unsupported promise that all engines are usable."""
+    assets = {
+        "skill_instructions": (ROOT / "SKILL.md").is_file(),
+        "mcp_server": (ROOT / "mcp_server.py").is_file(),
+        "native_engine": (ROOT / "scripts" / "jyotish_engine.py").is_file(),
+        "unified_orchestrator": (ROOT / "scripts" / "unified_consultation_orchestrator.py").is_file(),
+    }
+    adapters = adapter_report()
+    adapter_status = adapters.get("status", "blocked")
+    return {
+        "scope": "skill_doctor",
+        "status": "ready" if all(assets.values()) and adapter_status == "ready" else "degraded",
+        "core_assets": assets,
+        "external_engine_adapters": adapters,
+        "boundary": "Readiness only. An available adapter is not external raw-oracle verification.",
+    }
+
+
+def _vedastro_status(result: dict[str, Any]) -> str:
+    engines = result.get("external_engine_cross_validation")
+    if isinstance(engines, dict):
+        engines = engines.get("engines")
+    vedastro = engines.get("VedAstro") if isinstance(engines, dict) else None
+    if isinstance(vedastro, dict):
+        return str(vedastro.get("status") or "")
+    return ""
+
+
+def summarize_execution_status(result: dict[str, Any] | None) -> dict[str, Any]:
+    """Normalize official/local evidence state for every conversational surface."""
+    result = result or {}
+    fallback_reason = str(result.get("fallback_reason") or "")
+    vedastro = _vedastro_status(result)
+    raw_status = str(result.get("official_evidence_status") or "")
+    if raw_status == "official_verified" or vedastro == "official_verified":
+        official, source = "official_verified", "official_raw"
+    elif fallback_reason or vedastro in {"local_fallback", "official_blocked", "blocked"}:
+        official, source = "official_blocked", "local_fallback"
+    else:
+        official, source = "official_not_requested", "local_or_unverified"
+    return {
+        "scope": "execution_status",
+        "official_evidence_status": official,
+        "calculation_source": source,
+        "fallback_reason": fallback_reason or None,
+        "allowed_claims": ["official_verified", "official_blocked", "local_fallback"],
+        "claim_boundary": (
+            "Only official_verified permits claims that VedAstro official raw evidence was used."
+        ),
+    }

tests/test_skill_experience.py

@@ -0,0 +1,82 @@
+from scripts.skill_experience import (
+    build_rectification_questionnaire,
+    build_skill_doctor,
+    build_skill_onboarding,
+    score_rectification_answers,
+    summarize_execution_status,
+)
+
+
+def test_onboarding_requests_only_missing_birth_fields():
+    packet = build_skill_onboarding({"year": 1993, "month": 4, "day": 17})
+
+    assert packet["status"] == "needs_birth_data"
+    assert packet["entry_mode"] == "pending"
+    assert packet["missing_fields"] == ["hour", "minute", "lat", "lon"]
+    assert packet["next_action"] == "collect_birth_data"
+
+
+def test_onboarding_selects_rectification_for_uncertain_time():
+    packet = build_skill_onboarding({
+        "year": 1993,
+        "month": 4,
+        "day": 17,
+        "hour": 14,
+        "minute": 49,
+        "lat": 36.68,
+        "lon": 114.35,
+        "time_uncertainty_minutes": 20,
+    })
+
+    assert packet["status"] == "ready"
+    assert packet["entry_mode"] == "rectification"
+    assert packet["next_action"] == "run_rectification_questionnaire"
+    assert packet["first_question"]
+
+
+def test_execution_status_makes_official_fallback_machine_readable():
+    status = summarize_execution_status({
+        "fallback_reason": "VedAstro official snapshot blocked: official_snapshot_budget_exhausted",
+        "external_engine_cross_validation": {
+            "engines": {"VedAstro": {"status": "local_fallback"}}
+        },
+    })
+
+    assert status["official_evidence_status"] == "official_blocked"
+    assert status["calculation_source"] == "local_fallback"
+    assert status["fallback_reason"] == "VedAstro official snapshot blocked: official_snapshot_budget_exhausted"
+    assert "official_verified" in status["allowed_claims"]
+
+
+def test_doctor_has_machine_readable_core_and_adapter_state():
+    packet = build_skill_doctor()
+
+    assert packet["scope"] == "skill_doctor"
+    assert "core_assets" in packet
+    assert "external_engine_adapters" in packet
+    assert packet["status"] in {"ready", "degraded"}
+
+
+def test_mcp_exposes_skill_experience_tools():
+    import mcp_server
+
+    onboarding = mcp_server.skill_onboarding({})
+    doctor = mcp_server.skill_doctor()
+
+    assert onboarding["scope"] == "skill_onboarding"
+    assert doctor["scope"] == "skill_doctor"
+
+
+def test_rectification_contract_generates_and_scores_choice_answers():
+    questionnaire = build_rectification_questionnaire({
+        "year": 1993, "month": 4, "day": 17, "hour": 14, "minute": 49,
+        "time_uncertainty_minutes": 20,
+    })
+    scored = score_rectification_answers(questionnaire, {
+        "education_environment_shift": "A",
+        "health_crisis_or_low_period": "C",
+    })
+
+    assert questionnaire["scope"] == "active_birth_time_rectification_questionnaire"
+    assert scored["scope"] == "active_birth_time_rectification_scoring"
+    assert scored["candidate_cluster_rankings"]