business-assistant/src/step3_question_answer/tests/test_synthesis_reuse_threshold.py

from copy import deepcopy
import json
import unittest

from src.step3_question_answer.synthesis_input import (
    compact_payload, expand_text_entry, text_reuse_prompt, text_reuse_minimum, TEXT_REUSE_RULE,
)
from src.step3_question_answer.tests.test_synthesis_text_reuse import entry
from src.step4_api.logging import summary_diagnostics


class SynthesisReuseThresholdTests(unittest.TestCase):
    def test_threshold_includes_instruction_cost_and_scales_with_payload(self):
        self.assertEqual(text_reuse_minimum(1000), 1024)
        self.assertEqual(text_reuse_minimum(40000), 2000)
        self.assertEqual(text_reuse_minimum(40001), 2001)
        # The observed request saved 773 JSON bytes but paid 506 rule bytes.
        # This is a general size gate; no session ID or business keyword routes it.
        self.assertLess(773 - len(TEXT_REUSE_RULE.encode()), text_reuse_minimum(40000))

    def test_low_gain_keeps_plain_input_and_reports_only_numeric_skip_metadata(self):
        payload = {'evidence': [entry()]}
        before = deepcopy(payload)
        diagnostics = {}
        result, _ = compact_payload(payload, diagnostics)
        self.assertEqual(result, before)
        self.assertEqual(payload, before)
        self.assertEqual(text_reuse_prompt('任务', result), '任务')
        self.assertEqual(diagnostics['text_reuse_applied'], 0)
        self.assertLess(diagnostics['text_reuse_candidate_net_bytes'], diagnostics['text_reuse_minimum_net_bytes'])
        safe = summary_diagnostics({**diagnostics, 'text_pool': {'t0': '合成正文'}})
        self.assertEqual(safe, diagnostics)

    def test_large_unique_context_blocks_low_relative_gain_without_dropping_it(self):
        payload = {'evidence': [entry(i) for i in range(3)],
                   'conversation_context': {'note': ''.join(f'唯一背景{i}。' for i in range(5000))}}
        before = deepcopy(payload)
        result, _ = compact_payload(payload)
        self.assertEqual(result, before)
        self.assertNotIn('text_pool', str(result.keys()))

    def test_meaningful_gain_still_encodes_and_roundtrips(self):
        payload = {'cards': [entry(i) for i in range(3)]}
        diagnostics = {}
        result, _ = compact_payload(payload, diagnostics)
        self.assertEqual(diagnostics['text_reuse_applied'], 1)
        self.assertGreaterEqual(diagnostics['text_reuse_candidate_net_bytes'], diagnostics['text_reuse_minimum_net_bytes'])
        self.assertEqual([expand_text_entry(e) for e in result['cards']], payload['cards'])
        before = len(json.dumps(payload, ensure_ascii=False).encode())
        after = len(json.dumps(result, ensure_ascii=False).encode()) + len(TEXT_REUSE_RULE.encode())
        self.assertLess(after, before)