fix(consult): let the declared domain decide the route, not the question text

A staging consultation asked one question about two domains, career and
wealth. The model planned both, and all four attempts failed identically with
calculation_failed. The server resolves the workflow route from the question
text — an explicit-timing check, then keyword domain_tokens, falling back to
the themes argument only when the text yields nothing — while the frontend
declares strict_workflow_route from the domain it chose. The plan contract then
requires the text-derived route to equal the declared one, and each
RouteContract allows exactly one, so the mismatch became
ConsultationPlanContractError, BadRequest, HTTP 400, workflow_bad_request.
Here "事业" is in the career token list and "财运" is not in the wealth one, so
both calls resolved to career and the wealth call was rejected every time.

Widening the token list would only move the contradiction to the next
phrasing. The frontend sends one Python call per domain with the same question
text, so text routing can agree with at most one domain of a multi-domain plan
and every other domain is refused by construction. Now that 1955ba8c caps the
plan at three domains and merges the per-domain packets into one top-level
contract, two- and three-domain plans are expected to work end to end and this
is what stops them.

Make the server-issued declaration authoritative. declared_workflow_route()
returns the route a complete, version-supported, allowlisted plan declares, and
resolve_route() honours it instead of reading the text; a caller that sends no
plan metadata keeps the text heuristics verbatim, so the MCP strict_workflow
tool and the research callers behave exactly as before. The route packet
records which rule decided, because routing now has two legitimate sources.

This is not a way to silence the 400: the whole packet comes from the declared
domain's RouteDefinition, so the sync steps, the consumer_context required
layers, the evidence packet and the frontend's themes[primary_theme] lookup all
land on the domain that was declared. A call declaring wealth can no longer
execute career and label career evidence as wealth. The contract stays
fail-closed — a route off the allowlist is refused before execution, and
themes, layers, boundary, domains, categories, depth, horizon and precision are
still checked one by one. The surviving resolved_routes check changes meaning
rather than going away: it now asserts the workflow executed what was declared.

The timing prefix "应期与阶段问题：" existed only to inject 应期 so the text
router would agree with the declared route for one domain out of ten. With the
declaration authoritative it fixes nothing and still rewrites the question the
model's answer derives from, so it goes. It influences no other server
behaviour: it matches none of the consumer-context domain regexes, and the
timing route already sets precise_timing_requested.

test_consultation_workflow_domains.py deliberately sent no plan metadata, which
is why this was never caught — its per-domain question happened to route to its
own domain. It now sends the real plan for all ten canonical domains behind one
question whose text routes to career; nine of them fail without this change.

Refs BUG-259.

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_consultation_plan_contract.py

@@ -2,13 +2,35 @@ from __future__ import annotations
 
 import pytest
 
+from scripts.consultation_domain_registry import CANONICAL_DOMAINS
 from scripts.consultation_plan_contract import (
     ConsultationPlanContractError,
     apply_plan_precision_boundary,
+    declared_workflow_route,
+    plan_route_contract,
     validate_consultation_plan_contract,
 )
 
 
+def plan_body(route: str, **overrides):
+    """Build the metadata the product runtime sends for one declared domain."""
+    contract = plan_route_contract(route)
+    assert contract is not None
+    body = {
+        "plan_version": "consultation-plan-v2",
+        "strict_workflow_route": route,
+        "required_layers": list(contract.required_layers),
+        "claim_boundary": contract.claim_boundary,
+        "plan_depth": "standard",
+        "requested_domains": list(contract.requested_domains),
+        "timing_horizon": "next_12_months" if route in {"timing", "annual"} else None,
+        "precision_boundary": "server_evidence_required",
+        "required_evidence_categories": list(contract.required_evidence_categories),
+    }
+    body.update(overrides)
+    return body
+
+
 def timing_body(**overrides):
     body = {
         "plan_version": "consultation-plan-v2",
@@ -203,6 +225,88 @@ def test_blocked_plan_can_only_restrict_evidence_owned_precision_policy():
     assert evidence_owned == context
 
 
+def test_declared_workflow_route_is_only_read_from_a_complete_allowlisted_plan():
+    assert declared_workflow_route({"question": "事业如何"}) is None
+    assert declared_workflow_route(timing_body()) == "timing"
+    assert declared_workflow_route(plan_body("wealth")) == "wealth"
+
+    with pytest.raises(ConsultationPlanContractError, match="incomplete consultation plan metadata"):
+        declared_workflow_route({"strict_workflow_route": "wealth"})
+    with pytest.raises(ConsultationPlanContractError, match="version"):
+        declared_workflow_route(timing_body(plan_version="consultation-plan-v1"))
+    with pytest.raises(ConsultationPlanContractError, match="unsupported consultation workflow route"):
+        declared_workflow_route(timing_body(strict_workflow_route="free_script"))
+
+
+@pytest.mark.parametrize("domain", CANONICAL_DOMAINS)
+def test_plan_route_allowlist_covers_every_canonical_domain(domain: str):
+    contract = plan_route_contract(domain)
+
+    assert contract is not None
+    assert contract.themes == (domain,)
+    assert domain in contract.resolved_routes
+    assert plan_route_contract("free_script") is None
+
+
+def _consultation_body(**overrides):
+    body = {
+        "dry_run": True,
+        "entry_mode": "direct_chart",
+        "year": 1990,
+        "month": 1,
+        "day": 1,
+        "hour": 12,
+        "minute": 0,
+        "lat": 25,
+        "lon": 121,
+        "tz": 8,
+    }
+    body.update(overrides)
+    return body
+
+
+def test_declared_route_executes_even_when_question_text_names_another_domain():
+    """One question, several domains: text routing can only ever agree with one of them."""
+    from scripts.jyotish_api_server import JyotishAPIHandler
+
+    handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
+    question = "我的事业和财运接下来会怎么走，两者之间该怎么取舍"
+
+    wealth = handler._compute_consultation_workflow(_consultation_body(
+        **plan_body("wealth"), question=question, theme=["wealth"],
+    ))
+    career = handler._compute_consultation_workflow(_consultation_body(
+        **plan_body("career"), question=question, theme=["career"],
+    ))
+
+    assert wealth["routing"]["question_type"] == "wealth"
+    assert wealth["routing"]["primary_theme"] == "wealth"
+    assert "D2" in wealth["routing"]["focus_techniques"]
+    assert wealth["consultation_plan_contract"]["strict_workflow_route"] == "wealth"
+    assert career["routing"]["primary_theme"] == "career"
+
+
+def test_declared_route_cannot_smuggle_a_route_off_the_server_allowlist():
+    from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler
+
+    handler = JyotishAPIHandler.__new__(JyotishAPIHandler)
+
+    with pytest.raises(BadRequest, match="unsupported consultation workflow route"):
+        handler._compute_consultation_workflow(_consultation_body(
+            **plan_body("wealth", strict_workflow_route="free_script"),
+            question="财运如何",
+            theme=["wealth"],
+        ))
+    with pytest.raises(BadRequest, match="theme mismatch"):
+        handler._compute_consultation_workflow(_consultation_body(
+            **plan_body("wealth"), question="财运如何", theme=["career"],
+        ))
+    with pytest.raises(BadRequest, match="required layers mismatch"):
+        handler._compute_consultation_workflow(_consultation_body(
+            **plan_body("wealth", required_layers=["D1"]), question="财运如何", theme=["wealth"],
+        ))
+
+
 def test_api_dry_run_exposes_validated_plan_contract_and_rejects_tampering():
     from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler
 
@@ -211,7 +315,7 @@ def test_api_dry_run_exposes_validated_plan_contract_and_rejects_tampering():
         **timing_body(),
         "dry_run": True,
         "entry_mode": "direct_chart",
-        "question": "应期与阶段问题：未来哪些阶段值得把握？",
+        "question": "未来哪些阶段值得把握？",
         "theme": ["timing"],
         "year": 1990,
         "month": 1,

tests/test_consultation_workflow_domains.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import pytest
 
 from scripts.consultation_domain_registry import CANONICAL_DOMAINS
+from scripts.consultation_plan_contract import plan_route_contract
 from scripts.jyotish_api_server import BadRequest, JyotishAPIHandler
 from scripts.report_orchestrator import (
     INDEPENDENT_REPORT_THEMES,
@@ -98,6 +99,51 @@ def test_complete_consultation_workflow_handles_every_canonical_domain(
         assert "No general-theme fallback" in domain_report["boundary"]
 
 
+@pytest.mark.parametrize("domain", CANONICAL_DOMAINS)
+def test_versioned_plan_domains_execute_their_declared_route_not_the_question_text_route(
+    monkeypatch,
+    domain: str,
+) -> None:
+    """The product runtime sends one question per domain, so text routing agrees with at most one."""
+    handler = _handler()
+    _stub_consultation_runtime(monkeypatch, handler)
+    contract = plan_route_contract(domain)
+    assert contract is not None
+
+    result = handler._compute_consultation_workflow({
+        "entry_mode": "direct_chart",
+        "question": "我的事业和财运接下来会怎么走，两者之间该怎么取舍",
+        "theme": [domain],
+        "plan_version": "consultation-plan-v2",
+        "strict_workflow_route": domain,
+        "required_layers": list(contract.required_layers),
+        "claim_boundary": contract.claim_boundary,
+        "plan_depth": "standard",
+        "requested_domains": list(contract.requested_domains),
+        "timing_horizon": "next_12_months" if domain in {"timing", "annual"} else None,
+        "precision_boundary": "server_evidence_required",
+        "required_evidence_categories": list(contract.required_evidence_categories),
+        "year": 1997,
+        "month": 8,
+        "day": 8,
+        "hour": 5,
+        "minute": 0,
+        "lat": 36.420487,
+        "lon": 114.209936,
+        "tz": 8,
+        "western_mode": False,
+        "defer_optional_external_evidence": True,
+    })
+
+    primary_theme = result["routing"]["primary_theme"]
+    assert result["success"] is True
+    assert result["routing"]["question_type"] == domain
+    assert primary_theme == domain
+    assert result["consumer_context"]["route"] == domain
+    # The consumer reads thematic_report.themes[primary_theme] as this domain's evidence.
+    assert result["thematic_report"]["themes"][primary_theme]["theme"] == domain
+
+
 def test_consultation_thematic_capabilities_are_registry_derived_and_exclude_spirituality() -> None:
     capabilities = consultation_thematic_capabilities()
 

tests/test_unified_consultation_orchestrator.py

@@ -226,6 +226,41 @@ def test_unified_consultation_orchestrator_prefers_timing_when_career_question_a
     assert route["question_type"] == "timing"
 
 
+def test_declared_route_wins_over_question_text_keywords() -> None:
+    orchestrator = UnifiedConsultationOrchestrator()
+    question = "我的事业和财运接下来会怎么走，两者之间该怎么取舍"
+
+    declared = orchestrator.resolve_route(question, ["wealth"], declared_route="wealth")
+
+    assert declared["question_type"] == "wealth"
+    assert declared["primary_theme"] == "wealth"
+    assert "D2" in declared["focus_techniques"]
+    assert declared["route_source"] == "declared_plan"
+
+
+def test_callers_without_a_declared_route_keep_question_text_routing() -> None:
+    orchestrator = UnifiedConsultationOrchestrator()
+    question = "我的事业和财运接下来会怎么走，两者之间该怎么取舍"
+
+    text_routed = orchestrator.resolve_route(question, ["wealth"])
+
+    assert text_routed["question_type"] == "career"
+    assert text_routed["route_source"] == "question_text"
+
+
+@pytest.mark.parametrize("domain", CANONICAL_DOMAINS)
+def test_every_canonical_domain_is_executable_as_a_declared_route(domain: str) -> None:
+    route = UnifiedConsultationOrchestrator().resolve_route("综合看看", [domain], declared_route=domain)
+
+    assert route["question_type"] == domain
+    assert route["primary_theme"] == domain
+
+
+def test_unknown_declared_route_is_refused_instead_of_silently_text_routed() -> None:
+    with pytest.raises(ValueError, match="unknown declared consultation route"):
+        UnifiedConsultationOrchestrator().resolve_route("事业如何", ["career"], declared_route="free_script")
+
+
 def test_runtime_evidence_log_classifies_official_verified_local_fallback_and_blocked() -> None:
     orchestrator = UnifiedConsultationOrchestrator()
     route = {"question_type": "career", "primary_theme": "career"}