Fix research context limits and preserve output correction audits

backend/src/alphapulse/news_research/model.py

@@ -5,6 +5,9 @@ from alphapulse.ai import llm
 from alphapulse.models import AiProvider
 
 
+MAX_CONTEXT_WINDOW = 250_000
+
+
 class OutputTruncated(ValueError):
     """A read-only model call may be repeated with a smaller input batch."""
 
@@ -87,7 +90,7 @@ async def complete(provider: AiProvider, system: str, payload: dict, schema: typ
 def fits_context(provider, system, payload, schema):
     # UTF-8 bytes bound source tokens conservatively, including the tool schema.
     text = system + json.dumps(payload, ensure_ascii=False) + json.dumps(schema.model_json_schema(), ensure_ascii=False)
-    return len(text.encode('utf-8')) + provider.max_tokens <= provider.context_window
+    return len(text.encode('utf-8')) + provider.max_tokens <= min(provider.context_window, MAX_CONTEXT_WINDOW)
 
 async def provider_for(session, settings):
     provider = await session.get(AiProvider, settings.provider_id) if settings.provider_id else None

backend/src/alphapulse/news_research/service.py

@@ -459,6 +459,7 @@ async def _process_package(identifier, config, expected_stage):
         attempt_id = attempt.id
         output = None
         used = None
+        output_audit = {}
         try:
             vocabulary = await vocab.snapshot(session)
             if stage == "assign":
@@ -480,9 +481,12 @@ async def _process_package(identifier, config, expected_stage):
             )
             if stage == "index":
                 cleaned, changes = clean_aliases(output, source)
+                output_audit = {"output_adjustments": changes}
+                if changes:
+                    output_audit["raw_output"] = output.model_dump(mode="json")
+                attempt.rules = {**attempt.rules, **output_audit}
                 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:
@@ -586,6 +590,7 @@ async def _process_package(identifier, config, expected_stage):
         except Exception as error:
             await session.rollback()
             attempt = await session.get(Attempt, attempt_id)
+            attempt.rules = {**attempt.rules, **output_audit}
             attempt.state = "failed"
             attempt.error = str(error)[:2000]
             attempt.output = output.model_dump(mode="json") if output else None
@@ -859,6 +864,7 @@ async def process_event(identifier, config, force=False):
         researcher = Research(event={"id": identifier, "revision": revision})
         output = None
         used = None
+        output_audit = {}
         preparations = []
         try:
             attempts = {a.package_id: a for a in await session.scalars(
@@ -930,9 +936,11 @@ async def process_event(identifier, config, force=False):
             sources = {**{a["id"]: a for a in payload["articles"]}, **researcher.sources}
             raw_output = output.model_dump(mode="json")
             changes = align_quotes(raw_output, sources)
+            output_audit = {"output_adjustments": changes}
             if changes:
+                output_audit["raw_output"] = output.model_dump(mode="json")
                 output = ResearchResult.model_validate(raw_output)
-            version.rules = {**version.rules, "output_adjustments": changes}
+            version.rules = {**version.rules, **output_audit}
             validate_event(output, sources)
             if output.id != identifier:
                 raise ValueError("返回事件ID不匹配")
@@ -985,6 +993,7 @@ async def process_event(identifier, config, force=False):
         except Exception as error:
             await session.rollback()
             version = await session.get(Version, version_id)
+            version.rules = {**version.rules, **output_audit}
             version.state = "failed"
             version.error = str(error)[:2000]
             version.evidence = researcher.records

backend/src/alphapulse/news_research/tools.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field
 from alphapulse.ai import llm
 from alphapulse.core.db import SessionLocal
 from alphapulse.models import NewsItem
-from alphapulse.news_research.model import fits_context, validate_submission, model_stream
+from alphapulse.news_research.model import MAX_CONTEXT_WINDOW, fits_context, validate_submission, model_stream
 
 
 class Search(BaseModel):
@@ -54,7 +54,8 @@ class Research:
 
     def fits(self, provider, system, payload, schema):
         reserved = copy.copy(provider)
-        reserved.context_window -= min(32768, max(0, (provider.context_window-provider.max_tokens)//5))
+        reserved.context_window = min(provider.context_window, MAX_CONTEXT_WINDOW)
+        reserved.context_window -= min(32768, max(0, (reserved.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):
@@ -125,11 +126,13 @@ class Research:
         usages = self.usages = []
         from alphapulse.news_research.progress import step, current
         final_only = False
+        limited = {'status':'context_limit', 'instruction':'资料未完整交付，不得引用；依据已有证据提交并说明缺口。'}
+        limited_result = json.dumps(limited, ensure_ascii=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)
             final_only |= round_no == 8
             if not self.fits_messages(provider, system, messages, schema, final=final_only):
-                raise ValueError('研究上下文不足，请增大上下文配置；证据未截断')
+                raise ValueError('研究上下文不足；证据未截断或继续发送')
             completion = None
             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
@@ -148,20 +151,22 @@ class Research:
                 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':'停止查询，依据已有证据提交并说明缺口。'}
+                try: result = await self.tool(call['name'], call['arguments']) if not final_only else limited
                 except Exception as error:
                     result = {'error': str(error)}
                     self.records.append({'tool': call['name'], 'arguments': call['arguments'], 'result': result})
                 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:]]
+                remaining = [{'role':'tool', 'tool_call_id':c['id'], 'result':limited_result} 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':'资料未完整交付，不得引用；依据已有证据提交并说明缺口。'}
+                    result = limited
                     for record in self.records[records_before:]:
                         record['undelivered_result'] = record['result']
                         record['result'] = result
-                    response['result'] = json.dumps(result, ensure_ascii=False)
+                    response['result'] = limited_result
                     final_only = True
+                    if not self.fits_messages(provider, system, [*messages, response, *remaining], schema, final=True):
+                        raise ValueError('研究上下文不足；证据未截断或继续发送')
                 messages.append(response)
                 if not self.fits_messages(provider, system, messages, schema): final_only = True
         raise ValueError('研究轮次已用尽')

backend/tests/test_news_pipeline.py

@@ -246,3 +246,48 @@ def test_undelivered_source_is_not_restored_as_read(trading):
     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
+
+
+@pytest.mark.parametrize('context_window,body_length', [(8000, 7310), (250000, 249310)])
+def test_parallel_tool_responses_leave_room_for_final_submission(trading, monkeypatch, context_window, body_length):
+    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()
+    requests = []
+    tool_calls = [{'id': str(i), 'name': 'read_news', 'arguments': {'id': i}} for i in (1, 2)]
+
+    async def stream(provider, system, messages, tools, effort):
+        requests.append(messages.copy())
+        if len(requests) == 1:
+            yield Completion(tool_calls=tool_calls, usage={'input_tokens': 10})
+        else:
+            assert [t.name for t in tools] == ['submit']
+            assert reader.fits_messages(provider, system, messages, Result, final=True)
+            responses = [m for m in messages if m['role'] == 'tool']
+            assert [r['tool_call_id'] for r in responses] == ['1', '2']
+            assert all('context_limit' in r['result'] for r in responses)
+            yield Completion(tool_calls=[{'id': '3', 'name': 'submit', 'arguments': {'facts': '依据已有资料提交，补证不足。'}}],
+                             usage={'input_tokens': 20})
+
+    async def tool(name, args):
+        assert args['id'] == 1
+        result = {'content': 'x' * body_length}
+        reader.sources[-1] = result
+        reader.provided_news[1] = -1
+        reader.records.append({'tool': name, 'arguments': args, 'result': result})
+        return result
+
+    monkeypatch.setattr(reader, 'tool', tool)
+    monkeypatch.setattr(model.llm, 'stream', stream)
+    provider = SimpleNamespace(model='test', reasoning_effort='medium', max_tokens=100, context_window=context_window)
+    output, usage = asyncio.run(reader.run(provider, '规则', {}, Result))
+    assert output.facts and len(requests) == len(usage['rounds']) == 2
+    assert not reader.sources and not reader.provided_news
+    assert len(reader.records) == 1
+    assert reader.records[0]['undelivered_result']['content'] == 'x' * body_length

backend/tests/test_news_research_workspace.py

@@ -685,6 +685,132 @@ def test_long_event_compaction_uses_final_research_context(trading, monkeypatch,
     assert reader.fits(provider, skill, compressed, ResearchResult) != needs_compaction
 
 
+@pytest.mark.parametrize("failure_at", [None, "evidence", "timeline", "publication"])
+def test_quote_adjustment_audit_survives_validation_and_publication_failures(trading, monkeypatch, failure_at):
+    import asyncio
+    import copy
+    from unittest.mock import AsyncMock
+    from alphapulse.news_research import service
+    from alphapulse.news_research.contracts import ResearchResult
+    from alphapulse.news_research.models import Version
+
+    quote, original_quote = "官员“感到担忧”。", "官员“感到担忧” 。"
+    event = N(id=1, revision=1, title="官员回应", state="pending", result=None,
+              due_at=datetime.now(UTC) - timedelta(seconds=1))
+    news = N(id=101, title="官员回应", summary="", content_text=quote, content_hash="hash",
+             url="https://example.invalid/news", published_at=datetime.now(UTC))
+    package = N(id=7, news_id=101, state="done", content_hash="hash")
+    member = N(event_id=1, package_id=7, evidence=quote, reason="同一回应")
+    assignment = N(package_id=7, content_hash="hash", input={"keywords": [
+        {"canonical": "官员", "aliases": ["官员"], "quote": quote}]},
+        rules={"membership_targets": {"1": {"keyword_canonicals": ["官员"]}}})
+    output = ResearchResult(
+        id=1, title="官员回应", summary="官员表达担忧。", category="政策", entities=[],
+        risk_stars=1, importance_stars=2, direction="unknown", verification="reported", confidence="low",
+        reasoning="依据报道。", uncertainty="缺少独立佐证。", facts="官员表达担忧。", timeline="官员回应。",
+        asset_impacts="未知。", opinions="持续观察。", counter_evidence="未发现。", watchpoints="后续回应。",
+        claims=[{"statement": "官员表达担忧。", "verdict": "supported", "evidence_refs": [
+            {"article_id": 101, "quote": "不存在的引文" if failure_at == "evidence" else quote}]}],
+        evidence_refs=[{"article_id": 101, "quote": original_quote}],
+        timeline_points=[{"occurred_at": None, "time_precision": "unknown", "time_text": "",
+            "reported_at": None, "description": "官员表达担忧。", "verification": "supported",
+            "package_ids": [999 if failure_at == "timeline" else 7],
+            "evidence_refs": [{"article_id": 101, "quote": quote}]}], references=[])
+    original = output.model_dump(mode="json")
+    versions, saved_rules = [], {}
+
+    class Session:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        def add(self, row):
+            row.id = 1
+            versions.append(row)
+        async def commit(self):
+            nonlocal saved_rules
+            if versions[0].state == "done" and failure_at == "publication":
+                raise RuntimeError("发布写入失败")
+            saved_rules = copy.deepcopy(versions[0].rules)
+        async def rollback(self):
+            # Rollback discards uncommitted JSON changes, as the real ORM does.
+            versions[0].rules = copy.deepcopy(saved_rules)
+
+    session = Session()
+    session.get = AsyncMock(side_effect=lambda model, *args, **kwargs: versions[0] if model is Version else event)
+    session.execute = AsyncMock(side_effect=[[(package, member, news)], N(all=lambda: [(101, "hash")])])
+    session.scalars = AsyncMock(return_value=[assignment])
+    session.scalar = AsyncMock(return_value=None)
+    monkeypatch.setattr(service, "SessionLocal", lambda: session)
+    monkeypatch.setattr(service, "model_config", AsyncMock(return_value=(
+        N(max_tokens=32000, context_window=250000), "研究规则", {"system_prompt": "研究规则"})))
+    monkeypatch.setattr(service.vocab, "snapshot", AsyncMock(return_value={}))
+    monkeypatch.setattr(service.Research, "run", AsyncMock(return_value=(output, {"input_tokens": 10})))
+
+    asyncio.run(service.process_event(1, {}))
+    version = versions[0]
+    assert version.state == ("failed" if failure_at else "done"), version.error
+    if failure_at:
+        assert {"evidence": "原文引用", "timeline": "未知新闻包", "publication": "发布写入失败"}[failure_at] in version.error
+    assert version.output["evidence_refs"][0]["quote"] == quote
+    assert version.rules["raw_output"] == original
+    assert version.rules["output_adjustments"] == [{
+        "path": ".evidence_refs.0.quote", "original": original_quote, "replacement": quote,
+        "reason": "unique_punctuation_whitespace_alignment"}]
+    assert version.rules["system_prompt"] == "研究规则"
+    assert news.content_text == quote and output.model_dump(mode="json") == original
+
+
+@pytest.mark.parametrize("failure_at", [None, "validation", "publication"])
+def test_alias_adjustment_audit_survives_index_failures(trading, monkeypatch, failure_at):
+    import asyncio
+    import copy
+    from unittest.mock import AsyncMock
+    from alphapulse.news_research import service
+    from alphapulse.news_research.contracts import IndexResult
+    from alphapulse.news_research.models import Attempt, Package
+
+    news = N(id=101, title="", summary="", content_text="插件调用单独计费", content_hash="hash",
+             url="https://example.invalid/news", published_at=datetime.now(UTC))
+    package = N(id=7, news_id=101, state="index_pending", content_hash="hash")
+    output = IndexResult(keywords=[{"canonical": "插件计费", "aliases": ["插件调用", "插件调用收费"],
+                                   "quote": "不存在的引文" if failure_at == "validation" else "插件调用单独计费"}])
+    original = output.model_dump(mode="json")
+    attempts, saved_rules = [], {}
+
+    class Session:
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        def add(self, row):
+            row.id = 1
+            attempts.append(row)
+        async def commit(self):
+            nonlocal saved_rules
+            if attempts[0].state == "done" and failure_at == "publication":
+                raise RuntimeError("发布写入失败")
+            saved_rules = copy.deepcopy(attempts[0].rules)
+        async def rollback(self):
+            attempts[0].rules = copy.deepcopy(saved_rules)
+
+    session = Session()
+    session.get = AsyncMock(side_effect=lambda model, *args, **kwargs:
+                            attempts[0] if model is Attempt else package if model is Package else news)
+    session.execute = AsyncMock()
+    monkeypatch.setattr(service, "SessionLocal", lambda: session)
+    monkeypatch.setattr(service.vocab, "snapshot", AsyncMock(return_value={"entries": []}))
+    monkeypatch.setattr(service, "model_config", AsyncMock(return_value=(None, "索引规则", {})))
+    monkeypatch.setattr(service, "complete", AsyncMock(return_value=(output, {"input_tokens": 10})))
+
+    asyncio.run(service._process_package(7, {}, "index"))
+    attempt = attempts[0]
+    assert attempt.state == ("failed" if failure_at else "done"), attempt.error
+    if failure_at:
+        assert {"validation": "原文引用", "publication": "发布写入失败"}[failure_at] in attempt.error
+    assert attempt.rules["raw_output"] == original
+    assert attempt.rules["output_adjustments"] == [{
+        "path": "keywords.0.aliases", "original": ["插件调用", "插件调用收费"],
+        "replacement": ["插件调用"], "reason": "aliases_absent_from_source"}]
+    assert news.content_text == "插件调用单独计费" and output.model_dump(mode="json") == original
+
+
 @pytest.mark.parametrize("task", ["index", "assign", "research"])
 def test_model_config_sends_only_selected_task_and_records_exact_prompt(trading, monkeypatch, task):
     import asyncio

backend/tests/test_news_risk.py

@@ -36,3 +36,38 @@ def test_context_budget_includes_unicode_and_tool_schema(risk):
     from types import SimpleNamespace
     provider = SimpleNamespace(max_tokens=256, context_window=3500)
     assert not risk.fits_context(provider, '只使用原文证据', {'content': '风'*1000}, IndexResult)
+
+
+@pytest.mark.parametrize('context_window', [250000, 1000000])
+def test_news_context_is_capped_at_250k_including_output(risk, monkeypatch, context_window):
+    from pydantic import BaseModel
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    class Result(BaseModel):
+        facts: str
+
+    provider = SimpleNamespace(max_tokens=32000, context_window=context_window)
+    assert risk.fits_context(provider, '', {'content': 'x' * 200000}, Result)
+    stream = AsyncMock()
+    monkeypatch.setattr(risk.llm, 'stream', stream)
+    with pytest.raises(ValueError, match='超出模型上下文'):
+        asyncio.run(risk.complete(provider, '', {'content': 'x' * 220000}, Result))
+    stream.assert_not_called()
+
+
+def test_research_reserves_tool_room_within_250k_cap(risk):
+    from alphapulse.news_research.tools import Research
+    from pydantic import BaseModel
+    from types import SimpleNamespace
+
+    class Result(BaseModel):
+        facts: str
+
+    reader = Research()
+    payload = {'content': 'x' * 190000}
+    for context_window in (250000, 1000000):
+        provider = SimpleNamespace(max_tokens=32000, context_window=context_window)
+        assert risk.fits_context(provider, '', payload, Result)
+        assert not reader.fits(provider, '', payload, Result)
+        assert provider.context_window == context_window