"""Independent demand/role/expression axes and deterministic business suggestions.""" from __future__ import annotations import asyncio import copy import pytest from fastapi import HTTPException from apps.server.analysis import PROMPT_VERSION, RESULT_VERSION, SYSTEM_PROMPT, checked_results from apps.server.db import digest, encoded, identifier, now from apps.server.model_client import ModelCallError from apps.server.tests.test_analysis import analysis_client, answer, attach, client, pg_client, seed, wire CONDITIONS = {"service": "装修服务", "target_customer": "有装修需求的业主", "region": None, "region_scope": "unrestricted", "exclude_service_promotions": True} def source(text, ref="note", media=False): return {"source_ref": ref, "text": text, "media_uninterpreted": media} def result(text, *, demand="explicit", role="suspected_demand_side", expressions=("self_need",), relevance="relevant", region=None): return {"source_ref": "note", "demand": demand, "evidence": [text], "role": role, "role_evidence": [] if role == "unknown" else [text], "expressions": [{"type": expression, "evidence": [] if expression == "unknown" else [text]} for expression in expressions], "business_relevance": {"value": relevance, "evidence": [] if relevance == "unclear" else [text]}, "stage": None, "region": {"value": region, "evidence": text} if region else None, "budget": None} def check(item, text, conditions=None): return checked_results(encoded({"results": [item]}), [source(text)], "completed", conditions or CONDITIONS)[0] def test_axes_self_need_coexists_with_question_advice_and_service_provider(): text = "我家准备装修，问大家家装顺序。" current = check(result(text, expressions=("self_need", "question", "seek_advice")), text) assert current["demand"] == "explicit" and current["match"] == "matched" and current["next_step"] == "review_author_need" assert {entry["type"] for entry in current["expressions"]} == {"self_need", "question", "seek_advice"} text = "我是设计师可以提供装修服务，同时我家也准备装修。" provider = check(result(text, role="suspected_service_provider", expressions=("service_offer", "self_need")), text) assert provider["match"] == "matched" and provider["next_step"] == "review_author_need" assert "service_promotion_excluded" not in provider["match_reasons"] @pytest.mark.parametrize("text,role,expressions,next_step", [ ("你是哪里的嘛？", "unknown", ("question",), "skip"), ("开始装修了吗？", "unknown", ("question",), "skip"), ("我有设计师朋友，可以推荐给你", "suspected_service_provider", ("service_offer",), "inspect_comments"), ("展示一下我们装修公司的案例", "suspected_service_provider", ("service_offer", "case_share"), "inspect_comments"), ("建议先确定需求，再考虑半包", "participant", ("advice",), "skip"), ]) def test_axes_questions_offers_advice_do_not_inherit_authors_need(text, role, expressions, next_step): current = check(result(text, demand="discussion", role=role, expressions=expressions), text) assert current["match"] == "not_matched" and current["next_step"] == next_step assert "no_self_need_signal" in current["match_reasons"] if "service_offer" in expressions: assert "service_promotion_excluded" in current["match_reasons"] @pytest.mark.parametrize("scope,condition_region,own_region,match,reason", [ ("unrestricted", None, None, "matched", "region_unrestricted"), ("unspecified", None, None, "insufficient", "region_unspecified"), ("specified", "杭州", None, "insufficient", "region_missing"), ("specified", "杭州", "杭州", "matched", "region_exact"), ("specified", "杭州", "余杭", "insufficient", "region_potential_mismatch"), ("specified", "杭州", "上海", "insufficient", "region_potential_mismatch"), ]) def test_axes_region_missing_and_different_words_are_not_guessed(scope, condition_region, own_region, match, reason): text = "我家准备装修" + ("，地点" + own_region if own_region else "") current = check(result(text, region=own_region), text, {**CONDITIONS, "region_scope": scope, "region": condition_region}) assert current["match"] == match and reason in current["match_reasons"] assert current["next_step"] == ("review_author_need" if match == "matched" else "need_more_context") def test_axes_possible_need_comment_binding_irrelevant_unclear_and_uninterpreted_media(): text = "我也考虑装修，预算还没定" item = result(text, demand="possible") item["source_ref"] = "comment_2" current = checked_results(encoded({"results": [item]}), [source(text, ref="comment_2")], "partial", CONDITIONS)[0] assert current["match"] == "possible" and current["next_step"] == "review_comment_need" assert "reading_partial" in current["match_reasons"] assert check(result(text, relevance="irrelevant"), text)["match"] == "not_matched" assert check(result(text, relevance="unclear"), text)["match"] == "insufficient" media = checked_results('{"results":[]}', [source("", media=True)], "completed", CONDITIONS)[0] assert media["role"] == "unknown" and media["demand"] == media["match"] == "insufficient" assert media["business_relevance"] == {"value": "unclear", "evidence": []} and media["next_step"] == "need_more_context" assert "media_uninterpreted" in media["missing_information"] @pytest.mark.parametrize("mutation", [ lambda item: item.update(role_evidence=["其他来源的服务"]), lambda item: item["expressions"][0].update(evidence=["其他来源的需求"]), lambda item: item["business_relevance"].update(evidence=["商家条件中的装修服务"]), lambda item: item["business_relevance"].update(evidence=[]), lambda item: item.update(role="confirmed_owner"), lambda item: item["expressions"].append(copy.deepcopy(item["expressions"][0])), lambda item: item["expressions"].append({"type": "unknown", "evidence": []}), lambda item: item.update(expressions=[{"type": "question", "evidence": ["我家准备装修"]}]), lambda item: item.update(demand="discussion"), lambda item: item.update(next_step="send_message"), lambda item: item.update(match="matched"), ]) def test_axes_reject_fabricated_cross_source_dimensions_and_model_actions(mutation): text = "我家准备装修" item = result(text) mutation(item) with pytest.raises(ModelCallError) as error: check(item, text) assert error.value.code == "MODEL_OUTPUT_INVALID" def test_axes_no_own_need_rejects_service_area_as_own_project_region(): text = "我在杭州提供装修服务" item = result(text, demand="discussion", role="suspected_service_provider", expressions=("service_offer",), region="杭州") with pytest.raises(ModelCallError): check(item, text) # A short question cannot support a service-provider role without an offer. item = result("哪里", demand="discussion", role="suspected_service_provider", expressions=("question",)) with pytest.raises(ModelCallError): check(item, "哪里") assert "设计师所在地、服务覆盖地" in SYSTEM_PROMPT and "role应unknown" in SYSTEM_PROMPT class Briefs: def __init__(self, owner): self.owner = owner self.records = {} self.reads = [] def add(self, key, **conditions): value = {**CONDITIONS, **conditions} self.records[key] = {"id": key, "digest": digest(value), "conditions": value} return key def get_confirmed(self, key, owner): self.reads.append((key, owner)) if owner != self.owner or key not in self.records: raise HTTPException(404, "不存在") return self.records[key] def test_axes_context_digest_reuse_old_records_and_sources_stay_immutable(analysis_client): client = analysis_client reading_id, owner = seed(client) calls = [] def handler(request): import json payload = json.loads(json.loads(request.content)["messages"][1]["content"]) calls.append(payload) assert set(payload) == {"business_context", "sources"} assert not any("private-" in encoded(value) for value in payload.values()) return wire(answer(payload)) service = attach(client, handler) briefs = service.business_briefs = Briefs(owner) first_id = briefs.add("brief-first") other_id = briefs.add("brief-other", region="杭州", region_scope="specified") store = service.store original_reading = store.one("SELECT data_json FROM reading_runs WHERE id=$1", (reading_id,)) old_id = identifier("analysis") old = {"status": "completed", "reading_digest": "old", "prompt_version": "renovation-demand-v2", "result_version": "demand-result-v1", "results": [{"demand": "discussion"}]} store.execute("INSERT INTO analysis_runs(id,owner_id,reading_run_id,data_json,created_at) VALUES($1,$2,$3,$4,$5)", (old_id, owner, reading_id, encoded(old), now())) original_old = service.get(old_id, owner) first = asyncio.run(service.create(reading_id, owner, first_id)) assert first["status"] == "completed" and first["prompt_version"] == PROMPT_VERSION and first["result_version"] == RESULT_VERSION assert first["context"]["kind"] == "confirmed_business_brief" and first["context"]["business_brief_digest"] == briefs.records[first_id]["digest"] assert asyncio.run(service.create(reading_id, owner, first_id)) == first and len(calls) == 1 second = asyncio.run(service.create(reading_id, owner, other_id)) assert second["id"] != first["id"] and len(calls) == 2 assert second["context"]["context_digest"] != first["context"]["context_digest"] assert second["results"][0]["match"] == "insufficient" and "region_missing" in second["results"][0]["match_reasons"] assert store.one("SELECT data_json FROM reading_runs WHERE id=$1", (reading_id,)) == original_reading assert service.get(old_id, owner) == original_old and "context" not in original_old assert service.get(first["id"], owner) == first with pytest.raises(HTTPException): asyncio.run(service.create(reading_id, owner, "draft-or-other-owner")) assert len(calls) == 2 def test_axes_implicit_reading_context_and_explicit_override(client): reading_id, owner = seed(client) service = attach(client, lambda _: wire({"results": []})) briefs = service.business_briefs = Briefs(owner) first = briefs.add("brief-first") other = briefs.add("brief-other", region_scope="unspecified") reading = service.reading.get(reading_id, owner) reading["business_brief_id"] = first assert service.context(reading, owner)["business_brief_id"] == first assert service.context(reading, owner, other)["business_brief_id"] == other legacy = service.context(service.reading.get(reading_id, owner), owner) assert legacy["kind"] == "discovery_conditions" and legacy["conditions"]["region_scope"] == "unspecified"