From 890ad5725006c8a290c20a41a4c40b23b5a87613 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 28 Sep 2026 04:07:55 +0800 Subject: [PATCH] feat(research): persist material-bound bilingual candidate explanations Co-Authored-By: Codex --- .../research_promotion_cycle.py | 133 +++++++++++- tests/test_research_promotion_resume.py | 203 ++++++++++++++++++ 2 files changed, 326 insertions(+), 10 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/research_promotion_cycle.py b/src/quant_platform_kit/strategy_lifecycle/research_promotion_cycle.py index 11079be5..bba32364 100644 --- a/src/quant_platform_kit/strategy_lifecycle/research_promotion_cycle.py +++ b/src/quant_platform_kit/strategy_lifecycle/research_promotion_cycle.py @@ -8,6 +8,7 @@ from __future__ import annotations import calendar +import copy import hashlib import json import math @@ -1321,6 +1322,14 @@ def _saved_proposal(raw: Mapping[str, Any]) -> OptimizationProposal: _MAX_NO_IMPROVEMENT_ROUNDS = 3 _MAX_IDLE_DAYS = 30 _SUMMARY_MAX_TEXT = 2000 +_SUMMARY_SEGMENT_MAX = 240 +_SUMMARY_OLD_KEYS = frozenset({"status", "provider", "model", "text"}) +_SUMMARY_NEW_KEYS = frozenset({"status", "provider", "model", "locales"}) +_SUMMARY_LOCALE_FIELDS = ("question", "basis", "limits", "suggestion") +_SUMMARY_PROVIDERS = frozenset({"codex", "cursor"}) +_HAN_RE = re.compile(r"[\u4e00-\u9fff]") +_LATIN_RE = re.compile(r"[A-Za-z]") +_DIGIT_RE = re.compile(r"[0-9\uff10-\uff19]") _RESEARCH_OWNER_FIELDS = frozenset({"repository", "issue_number", "watcher_issue_key"}) _REPOSITORY_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") _RESEARCH_OWNER_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{8,64}$") @@ -1538,6 +1547,114 @@ def _proposal_comparable(proposal: OptimizationProposal) -> bool: for value in current.cost_inputs.values()) +def _same_ticket_material(left: Any, right: Any) -> bool: + try: + return _canonical_ticket_value(left) == _canonical_ticket_value(right) + except (TypeError, ValueError, OverflowError): + return False + + +def _candidate_summary_binding( + ticket: ResearchPromotionTicket, + proposal: OptimizationProposal | None, + context: Mapping[str, Any], +) -> dict[str, Any] | None: + """Freeze candidate facts before a summary callback. None means do not call.""" + identity = context.get("identity") + if not isinstance(identity, Mapping): + return None + if (identity.get("strategy_profile") != ticket.strategy_profile + or identity.get("domain") != ticket.domain + or not _same_ticket_material(identity.get("proposed_params"), ticket.proposed_params)): + return None + if proposal is not None and ( + proposal.strategy_profile != ticket.strategy_profile + or proposal.domain != ticket.domain + or not _same_ticket_material(proposal.proposed_params, ticket.proposed_params)): + return None + raw = { + "ticket_id": ticket.ticket_id, + "strategy_profile": ticket.strategy_profile, + "domain": ticket.domain, + "proposed_params": dict(ticket.proposed_params), + "comparison": context.get("comparison"), + "shadow_evidence_kind": ticket.shadow_evidence_kind, + "shadow_passed": ticket.shadow_passed, + "notes": list(ticket.notes), + } + try: + frozen = json.loads(_canonical_ticket_value(raw)) + except (TypeError, ValueError, OverflowError): + return None + return frozen if isinstance(frozen, dict) else None + + +def _summary_segment(text: Any, *, chinese: bool) -> str | None: + if not isinstance(text, str) or not text.strip() or len(text) > _SUMMARY_SEGMENT_MAX: + return None + if _DIGIT_RE.search(text): + return None + has_han = _HAN_RE.search(text) is not None + if chinese: + return text if has_han else None + if has_han or _LATIN_RE.search(text) is None: + return None + return text + + +def _bilingual_summary_explanation( + value: Mapping[str, Any], binding: Mapping[str, Any], +) -> dict[str, Any] | None: + provider = value.get("provider") + model = value.get("model") + locales = value.get("locales") + if (not isinstance(provider, str) or provider not in _SUMMARY_PROVIDERS + or not isinstance(model, str) or not model.strip() + or not isinstance(locales, Mapping) or frozenset(locales) != {"zh-CN", "en"}): + return None + saved_locales: dict[str, dict[str, str]] = {} + for name, chinese in (("zh-CN", True), ("en", False)): + section = locales[name] + if not isinstance(section, Mapping) or frozenset(section) != frozenset(_SUMMARY_LOCALE_FIELDS): + return None + saved: dict[str, str] = {} + for field_name in _SUMMARY_LOCALE_FIELDS: + segment = _summary_segment(section.get(field_name), chinese=chinese) + if segment is None: + return None + saved[field_name] = segment + saved_locales[name] = saved + return { + "status": "available", + "provider": provider, + "model": model, + "scope": "candidate", + "locales": saved_locales, + "binding": json.loads(_canonical_ticket_value(binding)), + } + + +def _accepted_summary_explanation( + value: Any, binding: Mapping[str, Any] | None, +) -> dict[str, Any] | None: + if not isinstance(value, Mapping) or value.get("status") != "available": + return None + keys = frozenset(value) + if keys == _SUMMARY_NEW_KEYS: + if binding is None: + return None + return _bilingual_summary_explanation(value, binding) + if not keys <= _SUMMARY_OLD_KEYS or "text" not in value: + return None + text = value.get("text") + provider = value.get("provider") or "" + model = value.get("model") or "" + if (isinstance(text, str) and text.strip() and len(text) <= _SUMMARY_MAX_TEXT + and provider == "codex" and isinstance(model, str)): + return {"status": "available", "text": text.strip(), "provider": "codex", "model": model} + return None + + def _attach_research_summary( ticket: ResearchPromotionTicket, proposal: OptimizationProposal | None, @@ -1560,17 +1677,13 @@ def _attach_research_summary( ticket.research_summary = {**context, "ai_explanation": explanation} if persist is not None: persist() - if callable(summarize): + binding = _candidate_summary_binding(ticket, proposal, context) if callable(summarize) else None + if callable(summarize) and binding is not None: try: - value = summarize(context) - if isinstance(value, Mapping) and value.get("status") == "available": - text = value.get("text") - provider = value.get("provider") or "" - model = value.get("model") or "" - if (isinstance(text, str) and text.strip() and len(text) <= _SUMMARY_MAX_TEXT - and provider == "codex" and isinstance(model, str)): - explanation = {"status": "available", "text": text.strip(), - "provider": "codex", "model": model} + value = summarize(copy.deepcopy(context)) + accepted = _accepted_summary_explanation(value, binding) + if accepted is not None: + explanation = accepted except Exception: pass ticket.research_summary = {**context, "ai_explanation": explanation} diff --git a/tests/test_research_promotion_resume.py b/tests/test_research_promotion_resume.py index f9b739f0..9dbe04d5 100644 --- a/tests/test_research_promotion_resume.py +++ b/tests/test_research_promotion_resume.py @@ -838,3 +838,206 @@ def now(cls, tz=None): assert admission.call_count == 1 ai.assert_not_called() assert not list((tmp_path / "research_promotion_tickets").glob("*.json")) + + +def _summary_locale(*, chinese, **overrides): + fields = { + "question": "该候选与基线在固定窗口下是否接近？", + "basis": "比较来自同一窗口与同一成本条件。", + "limits": "这只是候选建议，尚未选定账户。", + "suggestion": "请人工审阅，不构成执行授权。", + } if chinese else { + "question": "Does this candidate stay close to the baseline?", + "basis": "The comparison uses one fixed window and one cost model.", + "limits": "This is candidate advice before any account is selected.", + "suggestion": "Ask a human to review it without execution authority.", + } + fields.update(overrides) + return fields + + +def _bilingual_summary(provider="codex", model="synthetic-summary-model", locales=None, **extra): + payload = { + "status": "available", + "provider": provider, + "model": model, + "locales": locales if locales is not None else { + "zh-CN": _summary_locale(chinese=True), + "en": _summary_locale(chinese=False), + }, + } + payload.update(extra) + return payload + + +def _candidate_cycle(tmp_path, summarize): + return invoke( + tmp_path, + optimize=Mock(return_value=_comparable_proposal(recommendation="research_candidate")), + summarize=summarize, + ) + + +def test_one_callback_saves_both_locales_and_reuses_without_recall(tmp_path): + summarize = Mock(return_value=_bilingual_summary()) + result = _candidate_cycle(tmp_path, summarize) + saved = cycle.load_research_promotion_ticket(result["ticket_path"]) + explanation = saved.research_summary["ai_explanation"] + context = summarize.call_args.args[0] + assert summarize.call_count == 1 + assert set(context) == {"identity", "strategy_description", "plugins", "comparison", "limitations"} + assert context["identity"]["proposed_params"] == {"a": 2} + assert explanation["status"] == "available" + assert explanation["provider"] == "codex" + assert explanation["model"] == "synthetic-summary-model" + assert explanation["scope"] == "candidate" + assert set(explanation["locales"]) == {"zh-CN", "en"} + assert set(explanation["locales"]["zh-CN"]) == {"question", "basis", "limits", "suggestion"} + assert explanation["binding"]["ticket_id"] == saved.ticket_id + assert explanation["binding"]["proposed_params"] == {"a": 2} + assert explanation["binding"]["comparison"] == saved.research_summary["comparison"] + assert explanation["binding"]["shadow_evidence_kind"] == saved.shadow_evidence_kind + assert explanation["binding"]["shadow_passed"] is True + assert explanation["binding"]["notes"] == list(saved.notes) + assert "selected_account" not in explanation + assert saved.state == cycle.ResearchPromotionState.AWAITING_HUMAN + assert saved.live_authority_granted is False + assert saved.human_decision == "" + _candidate_cycle(tmp_path, summarize) + assert summarize.call_count == 1 + reread = cycle.load_research_promotion_ticket(result["ticket_path"]) + assert reread.research_summary["ai_explanation"] == explanation + assert reread.live_authority_granted is False + assert reread.state == cycle.ResearchPromotionState.AWAITING_HUMAN + + +@pytest.mark.parametrize("provider", ["codex", "cursor"]) +def test_legal_single_provider_bilingual_summary_is_saved(tmp_path, provider): + summarize = Mock(return_value=_bilingual_summary(provider=provider)) + result = _candidate_cycle(tmp_path, summarize) + saved = cycle.load_research_promotion_ticket(result["ticket_path"]) + explanation = saved.research_summary["ai_explanation"] + assert summarize.call_count == 1 + assert explanation["provider"] == provider + assert explanation["scope"] == "candidate" + assert explanation["locales"]["en"]["question"].startswith("Does this candidate") + assert saved.live_authority_granted is False + assert saved.human_decision == "" + + +def test_saved_plain_text_summary_is_not_rewritten_as_bilingual(tmp_path): + old = Mock(return_value={"status": "available", "text": "候选与基线在固定窗口下接近。", + "provider": "codex", "model": "gpt-test"}) + result = _candidate_cycle(tmp_path, old) + fresh = Mock(return_value=_bilingual_summary(provider="cursor")) + _candidate_cycle(tmp_path, fresh) + saved = cycle.load_research_promotion_ticket(result["ticket_path"]) + explanation = saved.research_summary["ai_explanation"] + assert old.call_count == 1 + assert fresh.call_count == 0 + assert explanation == {"status": "available", "text": "候选与基线在固定窗口下接近。", + "provider": "codex", "model": "gpt-test"} + + +def test_summary_callback_cannot_mutate_persisted_binding_or_facts(tmp_path): + def mutate(context): + context["identity"]["strategy_profile"] = "hijacked" + context["identity"]["proposed_params"]["a"] = 9 + context["comparison"]["status"] = "hijacked" + context["limitations"].append("hijacked") + return _bilingual_summary() + + summarize = Mock(side_effect=mutate) + result = _candidate_cycle(tmp_path, summarize) + saved = cycle.load_research_promotion_ticket(result["ticket_path"]) + assert saved.strategy_profile == "demo_strategy" + assert saved.proposed_params == {"a": 2} + assert saved.research_summary["identity"]["proposed_params"] == {"a": 2} + assert saved.research_summary["comparison"]["status"] == "comparable" + assert saved.research_summary["limitations"] == [] + binding = saved.research_summary["ai_explanation"]["binding"] + assert binding["strategy_profile"] == "demo_strategy" + assert binding["proposed_params"] == {"a": 2} + assert binding["comparison"]["status"] == "comparable" + assert saved.live_authority_granted is False + + +def test_mismatched_proposal_does_not_call_summary_or_change_eligibility(tmp_path): + result = _candidate_cycle(tmp_path, None) + path = cycle.Path(result["ticket_path"]) + saved = cycle.load_research_promotion_ticket(path) + saved.research_summary = { + key: value for key, value in saved.research_summary.items() if key != "ai_explanation" + } + saved.proposed_params = {"a": 99} + cycle.save_research_promotion_ticket(saved, path) + summarize = Mock(return_value=_bilingual_summary()) + second = _candidate_cycle(tmp_path, summarize) + reread = cycle.load_research_promotion_ticket(path) + assert summarize.call_count == 0 + assert second["status"] == "awaiting_human" + assert reread.proposed_params == {"a": 99} + assert reread.research_summary["ai_explanation"]["status"] == "unavailable" + assert "locales" not in reread.research_summary["ai_explanation"] + assert reread.state == cycle.ResearchPromotionState.AWAITING_HUMAN + assert reread.live_authority_granted is False + assert reread.human_decision == "" + + +@pytest.mark.parametrize("payload", [ + _bilingual_summary(provider="other-model"), + _bilingual_summary(model=""), + _bilingual_summary(model=" "), + _bilingual_summary(locales={"zh-CN": _summary_locale(chinese=True)}), + _bilingual_summary(locales={ + "zh-CN": {key: value for key, value in _summary_locale(chinese=True).items() if key != "limits"}, + "en": _summary_locale(chinese=False), + }), + _bilingual_summary(binding={"ticket_id": "forged"}), + _bilingual_summary(locales={ + "zh-CN": {**_summary_locale(chinese=True), "account": "not-an-account"}, + "en": _summary_locale(chinese=False), + }), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True), + "en": _summary_locale(chinese=False, suggestion="See item 3"), + }), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True, suggestion="请看第３点"), + "en": _summary_locale(chinese=False), + }), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True, question=1), + "en": _summary_locale(chinese=False), + }), + _bilingual_summary(locales=["zh-CN", "en"]), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True, question="问" * 241), + "en": _summary_locale(chinese=False), + }), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True), + "en": _summary_locale(chinese=False, question="该候选 is close"), + }), + _bilingual_summary(locales={ + "zh-CN": _summary_locale(chinese=True, question="Is this close?"), + "en": _summary_locale(chinese=False), + }), + {**_bilingual_summary(), "text": "候选与基线在固定窗口下接近。"}, + _bilingual_summary(scope="account"), +]) +def test_invalid_bilingual_summary_stays_unavailable_and_is_not_retried(tmp_path, payload): + summarize = Mock(return_value=payload) + result = _candidate_cycle(tmp_path, summarize) + _candidate_cycle(tmp_path, summarize) + saved = cycle.load_research_promotion_ticket(result["ticket_path"]) + explanation = saved.research_summary["ai_explanation"] + assert summarize.call_count == 1 + assert explanation["status"] == "unavailable" + assert explanation["text"] == "暂无法生成 AI 简述。" + assert "locales" not in explanation + assert "binding" not in explanation + assert "scope" not in explanation + assert saved.state == cycle.ResearchPromotionState.AWAITING_HUMAN + assert saved.live_authority_granted is False + assert saved.human_decision == ""