Bound research context and audit source-grounded output corrections

backend/src/alphapulse/news_research/contracts.py

@@ -16,6 +16,7 @@ ASSIGN_SKILL = """你负责将一个完整新闻包归属到具体事件。外
 无论原文语言，生成的title、reason均用简体中文，主体名称沿用中文规范名；没有可靠中文名的专有名称、产品型号及通用缩写保留。quote逐字保留原语言，不翻译。keyword_canonicals必须逐字复用输入TAG，即使历史TAG是英文也不得在归属时翻译或重做索引。JSON键、event_id、new_key等标识及change_kind枚举值保持协议格式，不翻译。
 一个包可以关联多个事件：综述涉及多件独立事件时，各填一个归属，以对应连续原文quote说明相关部分，不能因主题相同而合并独立事件。
 事件以主体、具体事项和发生批次或统计期为边界；归属reason明确说明同一事实主线及原文依据。直接后果、回应、调查、更正、赔付属于原事件；独立决策或行动、新事故、新统计期另建事件，不能因为相关而强行合并。原文明示由某次数据发布引起的直接市场反应归入该次发布；独立政策决策另建。例：瑞士10月CPI环比、同比和受该次CPI影响的欧元兑瑞郎反应同属发布事件；瑞士11月CPI另建。币安同一次被盗及数月后追偿同属一事，第二次被盗另建。单次事故的资金追踪、冻结或追回即使由另一国家机关执行，也属于原事故直接处置，不能仅因执行主体或国家不同另建；只有原文明示独立案件范围（例如同一平台处理多起无关犯罪的执法）才另建并引用事故。关联不明时独立建事件并在reason说明缺口，不编造因果。
+已有事件尚未包含这次回应，不是另建事件的理由。例如同一次美国中期选举后，万斯解释败选原因、特朗普回应，应归入该次选举事件；正式宣布的新政策决定或独立行动另建事件。
 优先检查给定的长期事件候选；同一次事故、回应、调查、损失修订和处置可属于同一事件，即使横跨数月。仅主体相同不是同一事件。
 候选reporting_range是报道时间范围，evidence包含首条与近期成员原文线索，不等于事件发生时间；结合quote中的日期、统计期及具体事项区分每日数据、不同次事故等独立事件，不能仅因名称或TAG相同而归并，也不能仅因报道间隔长而拆开同一事件。
 每个target填写event_id或new_key且恰好一个非空、title、reason、quote、keyword_canonicals和change_kind、urgent。已有事件沿用ID；无合适候选则创建可独立识别的新事件名称，不编写事件总结。至少归属一个事件，单包事件也成立。title与quote只包含本事件事项，quote保留其主体、动作和对象；综述的其他事项不要混入。不能仅引用“下跌”“另有”等失去主体的片段；原文确实未具名时不补造。
@@ -31,7 +32,8 @@ summary为事件总结；facts为已知事实；reasoning为判断依据；opini
 importance_stars独立评估：1背景、2局部、3重要、4广泛、5系统性。risk_stars：0无具体负面迹象、1关注、2警惕、3重大损害、4严重广泛冲击、5系统性危机。证据不充分降低confidence，不机械提高风险。verification=insufficient时direction=unknown。
 claims聚焦原文明示的关键主张、实质冲突及会改变判断的误读；不凭空增加“已获官方认证”“已决定全国控制权”等主张再机械判为不足。报道某人作出预测与预测已经实现分开。先核对数字、单位、期间和口径，保留算术反证能力。summary概括结论，reasoning说明推理，uncertainty集中写缺口，counter_evidence写实质反证或说明未发现，不在各字段重复整段限制。official仅表示核心主张得到实际读取的官方正文支持，不因有一个官方链接就覆盖其他未核实主张。
 每次站内查询先明确要解决的关键缺口：是否会改变事件身份、核心事实、核验状态、评级或判断。普通行情可直接按reported提交，不强查原因或收盘状态；重大事故、矛盾数字和关键政策优先通过已采集的站内原文核验。不得外部搜索、读取网页或访问URL。不为补齐字段查询；没有有效站内线索时提交，并集中说明证据局限。可独立进行的站内查询合并在同轮调用，不为每项claims各查一轮。
-articles已有成员完整原文时直接复用正文和文章id，不再逐包调用read_package/read_news，也不把重复读取当独立核验；仅输入extractions而无全文时，可按需读取对应成员原文。工具返回already_provided即使用先前已提供正文及article_id，不重试同一成员，也不继续逐个读取其他已提供成员。先列出会改变判断的最少证据缺口，将无需依赖彼此结果的站内查询合并同一轮；拿到线索后才执行依赖它的读取。每轮检查是否有新增有效证据；没有可解决关键缺口的新具体线索时立即submit，未解决项写uncertainty，不为凑核验轮次或空字段继续查询，不跳过重大事实所需的有效取证。
+articles已有成员完整原文时直接复用正文和文章id，不再逐包调用read_package/read_news，也不把重复读取当独立核验；仅输入extractions而无全文时，可按需读取对应成员原文。工具返回context_limit表示该资料未完整交付，不得引用该资料，停止查询并依据已有证据提交，将缺口写入uncertainty。检索结果仅用于定位，不能替代原文。
+工具返回already_provided即使用先前已提供正文及article_id，不重试同一成员，也不继续逐个读取其他已提供成员。先列出会改变判断的最少证据缺口，将无需依赖彼此结果的站内查询合并同一轮；拿到线索后才执行依赖它的读取。每轮检查是否有新增有效证据；没有可解决关键缺口的新具体线索时立即submit，未解决项写uncertainty，不为凑核验轮次或空字段继续查询，不跳过重大事实所需的有效取证。
 claims逐项核验，只有可能改变结论的缺口才查询站内新闻包、事件或经济日历。不机械耗尽预算；同稿转载和多个事件引用同一原文不是独立确认。URL不是已读证据，不得联网核验。站内已采集的官方原文可支持official；证据不足时明确缺口并据实降低核验状态和confidence，不编造补证。恢复客户资产余额不等于追回被盗资金；金额估值、可追踪比例须保留各自时点与来源。
 时间线timeline_points按实际发展组织，转载可共用一个节点，不强制每包一节点；成员包进展的节点填写package_ids及各包对应原文evidence_refs；packages.id是包ID，packages.news_id是文章ID，二者不能混用。仅由工具补充材料支撑的背景节点package_ids填空数组，不能冒用成员包ID。发生、计划、报道时间和统计期分开；未知为null/unknown。time_text逐字取自某条quote，不用报道时间替代发生时间。
 日和月精度的occurred_at/planned_at统一以UTC零点存储日期标记（日取当日、月取首日），它不是实际发生瞬间，不按事件所在国家猜时区；明确的当地日期保留其原日期，不先换算UTC日期。分钟精度仅在原文明确时区或可由具名交易场所明确确定时填写，否则保留time_text并填null/unknown。只有报道日期不能推定发生日期。季度及年度保留period_text，不伪造月或日精度。

backend/src/alphapulse/news_research/pipeline.py

@@ -54,6 +54,49 @@ def validate_evidence(refs, articles):
         if source is None or not ref.quote.strip() or not any(ref.quote in (source.get(k) or '') for k in ('title', 'summary', 'content')):
             raise ValueError('模型证据并非对应文章原文引用')
 
+def align_quotes(value, articles):
+    """Recover unique original spans only when punctuation-adjacent whitespace differs."""
+    punctuation = set('，。！？；：、,.;:!?“”‘’"\'()（）[]【】')
+    def normalized(text):
+        chars, positions = [], []
+        for match in re.finditer(r'\S|\s+', text):
+            part = match.group()
+            if part.isspace() and ((match.start() and text[match.start()-1] in punctuation)
+                    or (match.end() < len(text) and text[match.end()] in punctuation)):
+                continue
+            chars.extend(part)
+            positions.extend(range(match.start(), match.end()))
+        return ''.join(chars), positions
+    changes = []
+    def visit(item, path):
+        if isinstance(item, list):
+            for i, child in enumerate(item): visit(child, f'{path}.{i}')
+        elif isinstance(item, dict):
+            if 'article_id' in item and 'quote' in item:
+                quote = item['quote']
+                source = articles.get(item['article_id'], {})
+                if not quote.strip() or any(quote in (source.get(k) or '') for k in ('title','summary','content')):
+                    return
+                needle, _ = normalized(quote)
+                matches = []
+                if needle:
+                    for field in ('title','summary','content'):
+                        original = source.get(field) or ''
+                        haystack, positions = normalized(original)
+                        start = haystack.find(needle)
+                        while start >= 0:
+                            matches.append(original[positions[start]:positions[start+len(needle)-1]+1])
+                            start = haystack.find(needle, start+1)
+                if len(matches) == 1 and len(matches[0]) <= 500:
+                    item['quote'] = matches[0]
+                    changes.append({'path':path+'.quote','original':quote,'replacement':matches[0],
+                                    'reason':'unique_punctuation_whitespace_alignment'})
+            else:
+                for key, child in item.items(): visit(child, f'{path}.{key}')
+    visit(value, '')
+    return changes
+
+
 def validate_event(result, articles):
     if result.excluded_article_ids:
         raise ValueError('事件包含不相关文章：'+str(result.excluded_article_ids))

backend/src/alphapulse/news_research/service.py

@@ -14,6 +14,7 @@ from alphapulse.news_research.pipeline import (
     validate_evidence,
     Evidence,
     validate_event,
+    align_quotes,
     extract_articles,
     compact_extractions,
 )
@@ -74,6 +75,22 @@ def validate_keywords(result, source):
             raise ValueError("关键词别名必须实际出现在本篇原文中")
 
 
+def clean_aliases(result, source):
+    """Retain every TAG; audit invalid aliases only when a valid alias remains."""
+    cleaned = result.model_copy(deep=True)
+    changes = []
+    for i, keyword in enumerate(cleaned.keywords):
+        valid = [alias for alias in keyword.aliases if alias.strip() and any(
+            alias in (source.get(field) or '') for field in ('title','summary','content'))]
+        if not valid:
+            raise ValueError('关键词没有任何原文有效别名')
+        if valid != keyword.aliases:
+            changes.append({'path':f'keywords.{i}.aliases','original':keyword.aliases,
+                            'replacement':valid,'reason':'aliases_absent_from_source'})
+            keyword.aliases = valid
+    return cleaned, changes
+
+
 def validate_assignment(result, source, candidate_ids, keywords):
     for target in result.targets:
         if target.event_id is not None and target.event_id not in candidate_ids:
@@ -462,7 +479,10 @@ async def _process_package(identifier, config, expected_stage):
                 IndexResult if stage == "index" else Assignment,
             )
             if stage == "index":
-                validate_keywords(output, source)
+                cleaned, changes = clean_aliases(output, source)
+                validate_keywords(cleaned, source)
+                output = cleaned
+                attempt.rules = {**attempt.rules, "output_adjustments": changes}
             else:
                 validate_assignment(output, source, {e.id for e in catalog}, payload["keywords"])
             async with publication_lock:
@@ -691,14 +711,15 @@ class Research(ResearchTools):
                             .limit(20)
                         )
                         result = [
-                            {"id": r.id, "title": r.title, "tags": r.tags} for r in rows
+                            {"id": r.id, "title": r.title, "tags": [t for t in r.tags
+                             if any(term.lower() in t.lower() for term in terms[:6])]} for r in rows
                         ]
                     else:
                         # Retain the phrase in addition to the bounded search terms.
                         condition, score = package_matching([query, *terms[:6]], vocabulary)
                         phrase, _ = package_matching([query], vocabulary)
                         rows = await session.execute(
-                            select(Package, NewsItem.content_hash)
+                            select(Package, NewsItem.content_hash, NewsItem.title)
                             .join(NewsItem)
                             .where(condition)
                             .order_by(
@@ -710,9 +731,13 @@ class Research(ResearchTools):
                             .limit(20)
                         )
                         result = [
-                            {"id": r.id, "news_id": r.news_id,
-                             "keywords": r.keywords if r.content_hash == source_hash else []}
-                            for r, source_hash in rows
+                            {"id": r.id, "news_id": r.news_id, "title": title,
+                             "keywords": [{"canonical": k["canonical"], "aliases": [a for a in k["aliases"]
+                                 if any(t.lower() in a.lower() for t in terms[:6])]}
+                                 for k in r.keywords if any(t.lower() in word.lower()
+                                     for t in terms[:6] for word in [k["canonical"], *k["aliases"]])]
+                                 if r.content_hash == source_hash else []}
+                            for r, source_hash, title in rows
                         ]
                 elif name == "read_package":
                     row = await session.get(Package, arguments["id"])
@@ -902,10 +927,13 @@ async def process_event(identifier, config, force=False):
             )
             if preparations:
                 used = {"preparation": preparations, "research": used}
-            validate_event(
-                output,
-                {**{a["id"]: a for a in payload["articles"]}, **researcher.sources},
-            )
+            sources = {**{a["id"]: a for a in payload["articles"]}, **researcher.sources}
+            raw_output = output.model_dump(mode="json")
+            changes = align_quotes(raw_output, sources)
+            if changes:
+                output = ResearchResult.model_validate(raw_output)
+            version.rules = {**version.rules, "output_adjustments": changes}
+            validate_event(output, sources)
             if output.id != identifier:
                 raise ValueError("返回事件ID不匹配")
             package_news = {p.id: n.id for p, m, n in members}

backend/src/alphapulse/news_research/tools.py

@@ -1,5 +1,6 @@
 """Read-only news and external research tools with bounded calls."""
 import json
+import copy
 from sqlalchemy import select, or_
 from pydantic import BaseModel, Field
 from alphapulse.ai import llm
@@ -47,18 +48,20 @@ class Research:
     def tool_specs():
         return [llm.ToolSpec(n, description, model.model_json_schema()) for n, (model, description) in TOOLS.items()]
 
-    def fits_messages(self, provider, system, messages, schema):
+    def fits_messages(self, provider, system, messages, schema, *, final=False):
         return fits_context(provider, system, {'messages': messages, 'tools': [
-            {'name': t.name, 'description': t.description, 'parameters': t.parameters} for t in self.tool_specs()]}, schema)
+            {'name': t.name, 'description': t.description, 'parameters': t.parameters} for t in ([] if final else self.tool_specs())]}, schema)
 
     def fits(self, provider, system, payload, schema):
-        return self.fits_messages(provider, system, [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}], schema)
+        reserved = copy.copy(provider)
+        reserved.context_window -= min(32768, max(0, (provider.context_window-provider.max_tokens)//5))
+        return self.fits_messages(reserved, system, [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}], schema)
 
     def restore(self, records):
         for record in records:
             name, args, result = record['tool'], record['arguments'], record['result']
             if name not in TOOLS: continue
-            if isinstance(result, dict) and 'error' in result: continue
+            if isinstance(result, dict) and ('error' in result or result.get('status') == 'context_limit'): continue
             # Membership and revisions can change between attempts; reread local objects.
             if name in ('read_event', 'read_package'): continue
             self.cache[name+json.dumps(args, sort_keys=True)] = result
@@ -121,12 +124,14 @@ class Research:
         messages = [{'role': 'user', 'text': json.dumps(payload, ensure_ascii=False)}]
         usages = self.usages = []
         from alphapulse.news_research.progress import step, current
+        final_only = False
         for round_no in range(9):
             if current.get() is not None: step('模型研判', round=round_no+1, model=provider.model, reasoning_effort=provider.reasoning_effort)
-            if not self.fits_messages(provider, system, messages, schema):
+            final_only |= round_no == 8
+            if not self.fits_messages(provider, system, messages, schema, final=final_only):
                 raise ValueError('研究上下文不足，请增大上下文配置；证据未截断')
             completion = None
-            async for item in model_stream(provider, system, messages, [*specs, submit] if round_no < 8 else [submit], provider.reasoning_effort):
+            async for item in model_stream(provider, system, messages, [submit] if final_only else [*specs, submit], provider.reasoning_effort):
                 if isinstance(item, llm.Completion): completion = item
             if completion is not None: usages.append(completion.usage)
             if completion is None or completion.truncated: raise ValueError('研究输出未完成或被截断')
@@ -135,14 +140,28 @@ class Research:
                 if len(calls) != 1: raise ValueError('提交结果不能同时执行查询')
                 result = await validate_submission(provider, system, messages, schema, completion, usages)
                 return result, {'rounds': usages, 'tools': self.records}
-            if round_no == 8 or not calls: raise ValueError('研究模型未提交结果')
+            if final_only or not calls: raise ValueError('研究模型未提交结果')
             assistant = {'role': 'assistant', 'text': completion.text, 'tool_calls': calls}
             if completion.thinking_blocks: assistant['thinking_blocks'] = completion.thinking_blocks
             messages.append(assistant)
-            for call in calls:
-                try: result = await self.tool(call['name'], call['arguments'])
+            for index, call in enumerate(calls):
+                snapshot = {name: copy.deepcopy(getattr(self, name)) for name in (
+                    'sources', 'provided_news', 'provided_packages', 'read_event_versions', 'read_event_ids')}
+                records_before = len(self.records)
+                try: result = await self.tool(call['name'], call['arguments']) if not final_only else {'status':'context_limit', 'instruction':'停止查询，依据已有证据提交并说明缺口。'}
                 except Exception as error:
                     result = {'error': str(error)}
                     self.records.append({'tool': call['name'], 'arguments': call['arguments'], 'result': result})
-                messages.append({'role': 'tool', 'tool_call_id': call['id'], 'result': json.dumps(result, ensure_ascii=False, default=str)})
+                response = {'role':'tool', 'tool_call_id':call['id'], 'result':json.dumps(result, ensure_ascii=False, default=str)}
+                remaining = [{'role':'tool', 'tool_call_id':c['id'], 'result':json.dumps({'status':'context_limit'})} for c in calls[index+1:]]
+                if not self.fits_messages(provider, system, [*messages, response, *remaining], schema, final=True):
+                    for name, value in snapshot.items(): setattr(self, name, value)
+                    result = {'status':'context_limit', 'instruction':'资料未完整交付，不得引用；依据已有证据提交并说明缺口。'}
+                    for record in self.records[records_before:]:
+                        record['undelivered_result'] = record['result']
+                        record['result'] = result
+                    response['result'] = json.dumps(result, ensure_ascii=False)
+                    final_only = True
+                messages.append(response)
+                if not self.fits_messages(provider, system, messages, schema): final_only = True
         raise ValueError('研究轮次已用尽')

backend/tests/test_news_pipeline.py

@@ -174,3 +174,75 @@ def test_oversized_scope_fails_before_model_call_with_article_identity(pipeline,
         asyncio.run(pipeline.extract_articles(SimpleNamespace(max_tokens=100, context_window=1500), '规则',
             [{'id': 9, 'title': '标题', 'content': '正文'}], scopes=[{'news_id': 9, 'scope_quote': '长' * 5000}]))
     complete.assert_not_awaited()
+
+
+def test_alias_cleanup_preserves_tags_valid_aliases_and_raw_output(trading):
+    from alphapulse.news_research.contracts import IndexResult
+    from alphapulse.news_research.service import clean_aliases, validate_keywords
+    raw = IndexResult(keywords=[dict(canonical='插件调用计费', aliases=['插件调用','插件调用收费'], quote='插件调用单独计费')])
+    source = {'id':1, 'content':'插件调用单独计费'}
+    output, changes = clean_aliases(raw, source)
+    validate_keywords(output, source)
+    assert output.keywords[0].aliases == ['插件调用']
+    assert raw.keywords[0].aliases == ['插件调用','插件调用收费']
+    assert changes[0]['original'] == raw.keywords[0].aliases
+    with pytest.raises(ValueError, match='没有任何原文有效别名'):
+        clean_aliases(raw, {'id':1,'content':'其他文字'})
+
+
+@pytest.mark.parametrize('source,quote,corrected', [
+    ('官员“感到担忧”。','官员“感到担忧” 。',True),
+    ('官员“感到担忧”。官员“感到担忧”。','官员“感到担忧” 。',False),
+    ('金额100美元。','金额1000美元。',False),
+    ('US D company.','USD company.',False),
+])
+def test_quote_alignment_only_recovers_unique_punctuation_whitespace(pipeline, source, quote, corrected):
+    value = {'claims':[{'evidence_refs':[{'article_id':-2,'quote':quote}]}]}
+    changes = pipeline.align_quotes(value, {-2:{'content':source}})
+    assert bool(changes) == corrected
+    if corrected:
+        pipeline.validate_evidence([pipeline.Evidence.model_validate(value['claims'][0]['evidence_refs'][0])], {-2:{'content':source}})
+        assert changes[0]['original'] == quote
+    else:
+        assert value['claims'][0]['evidence_refs'][0]['quote'] == quote
+
+
+def test_oversized_tool_response_is_not_delivered_or_registered(trading, monkeypatch):
+    from alphapulse.news_research.tools import Research
+    from alphapulse.news_research import model
+    from alphapulse.ai.llm import Completion
+    from pydantic import BaseModel
+    class Result(BaseModel):
+        facts: str
+    reader = Research()
+    calls = []
+    async def stream(provider, system, messages, tools, effort):
+        calls.append(messages.copy())
+        if len(calls) == 1:
+            yield Completion(tool_calls=[{'id':'1','name':'read_news','arguments':{'id':99}}],usage={'input_tokens':10})
+        else:
+            assert [t.name for t in tools] == ['submit']
+            assert 'context_limit' in messages[-1]['result']
+            assert 'x'*100 not in messages[-1]['result']
+            yield Completion(tool_calls=[{'id':'2','name':'submit','arguments':{'facts':'证据不足'}}],usage={'input_tokens':20})
+    async def tool(name, args):
+        result = {'article_id':-1,'content':'x'*20000}
+        reader.sources[-1] = result
+        reader.provided_news[99] = -1
+        reader.records.append({'tool':name,'arguments':args,'result':result})
+        return result
+    monkeypatch.setattr(reader, 'tool', tool)
+    monkeypatch.setattr(model.llm, 'stream', stream)
+    output, usage = asyncio.run(reader.run(SimpleNamespace(model='test',reasoning_effort='medium',max_tokens=100,context_window=8000), '规则', {}, Result))
+    assert output.facts == '证据不足' and len(usage['rounds']) == 2
+    assert not reader.sources and not reader.provided_news
+    assert reader.records[0]['result']['status'] == 'context_limit'
+    assert len(reader.records[0]['undelivered_result']['content']) == 20000
+
+
+def test_undelivered_source_is_not_restored_as_read(trading):
+    from alphapulse.news_research.tools import Research
+    reader = Research()
+    reader.restore([{'tool':'read_news','arguments':{'id':99},
+        'result':{'status':'context_limit'},'undelivered_result':{'article_id':-1,'content':'原文'}}])
+    assert not reader.sources and not reader.cache and not reader.provided_news

backend/tests/test_news_research_workspace.py

@@ -221,7 +221,7 @@ def test_retry_refreshes_local_search_with_current_vocabulary(
     session = AsyncMock()
     session.__aenter__.return_value = session
     session.execute.return_value = [
-        (N(id=2, news_id=12, content_hash="current", keywords=[]), "current")
+        (N(id=2, news_id=12, content_hash="current", keywords=[]), "current", "Fresh news")
     ]
     session.scalars.return_value = [N(id=2, title="Fresh event", tags=["binance"])]
     monkeypatch.setattr(service, "SessionLocal", lambda: session)