refactor(auto-fin): research news by topic before market open

plugins/auto-fin/README.md

@@ -2,8 +2,8 @@
 
 [中文](README_ZH.md)
 
-Auto Fin fetches a rolling window of CLS telegraph news (24 hours by default), selects items related to configured
-topics, searches ReMe for useful historical context, and writes one Chinese Markdown report with validated wikilinks.
+Auto Fin fetches a rolling window of CLS telegraph news (24 hours by default), groups items by configured
+topics, researches each topic against ReMe history, and writes one Chinese Markdown report with validated wikilinks.
 Current news and topic selection stay in runtime memory; only the final report becomes durable memory. This directory
 is an independent Python distribution. Its single `reme.plugins` entry point exposes a `plugin.yaml` containing the
 three Step backends and their Job configuration under `application_defaults`. Enable the installed plugin explicitly
@@ -60,7 +60,7 @@ reme start plugins='["auto-fin"]' \
 ```
 
 Custom application configs must provide `agent_wrapper.default`, a `file_store.default` with an enabled tag index, and
-the `search`, `read`, `list_tags`, `frontmatter_read`, and `frontmatter_update` Jobs used by Auto Fin and automatic
+the `search`, `list_tags`, `frontmatter_read`, and `frontmatter_update` Jobs used by Auto Fin and automatic
 tagging.
 
 ## Pipeline
@@ -70,11 +70,11 @@ CLS public telegraph endpoint (rolling 24 hours)
         ↓
 normalize and deduplicate in RuntimeContext
         ↓
-topic Agent selects real news IDs in bounded batches
+topic Agent returns topic-to-news-ID mappings in prompt-sized batches
         ↓
-research Agent uses search + read on historical memory
+one research Agent per topic examines its latest 20 articles and searches history up to three times
         ↓
-validate historical wikilinks in code
+combine topic results and validate historical wikilinks in code
         ↓
 daily/YYYY-MM-DD/auto_fin.md
         ↓
@@ -85,17 +85,17 @@ generate memory tags; the background file watcher refreshes indexes
 stops only after covering the exact preceding 24 hours. Requests are rate-limited and retried; malformed records and
 records outside the window are discarded.
 
-`auto_fin_topic_step` receives batches of current news and returns only related `news_id` values. Code ignores unknown
-IDs and deduplicates repeated IDs, then preserves the source-news order. If nothing is relevant, the job succeeds as a
+`auto_fin_topic_step` batches current news under a 100,000-character full-prompt limit and returns related `news_id`
+values for each topic. Code ignores unknown IDs and deduplicates repeated IDs, then preserves source-news order. One article may belong to multiple topics. If nothing is relevant, the job succeeds as a
 skip without writing a report.
 
-`auto_fin_merge_step` receives only selected current news. It exposes `search` and `read`, and keeps current CLS IDs,
+`auto_fin_merge_step` researches each nonempty topic with its latest 20 articles. It exposes only `search`, enforces a three-call search budget per topic, and keeps current CLS IDs,
 times, and titles as plain evidence. The prompt limits
 wikilinks to historical Markdown actually used by the Agent; the code-level boundary independently keeps only existing,
 workspace-relative Markdown targets. Missing, absolute, escaping, backslash, and self-referential targets are degraded
 to their readable aliases.
 
-Same-day reruns use the existing report as context and replace it with the revised result. The final write is atomic and
+Same-day reruns use up to the first 30,000 characters of the existing report as context and replace it with the revised result. The combined final write is atomic and
 refreshes the daily index. The workflow then runs `auto_tag_step` to update the generated report's memory-tag
 frontmatter; the normal background file watcher observes that source-file change and refreshes derived indexes. No
 JSONL, intermediate Markdown, or structured Agent output is written.
@@ -111,7 +111,7 @@ JSONL, intermediate Markdown, or structured Agent output is written.
 | `request_interval` |                   `10` | Minimum delay in seconds after every CLS request attempt; may be zero    |
 | `max_retries`      |                    `3` | Maximum attempts for each CLS page request; must be at least one         |
 
-The plugin cron Job starts with the application and runs daily at 18:00 in the application timezone.
+The plugin cron Job starts with the application and runs daily at 09:00 in the application timezone, which defaults to `Asia/Shanghai`. The rolling window uses timestamps and may cross calendar days. Report completion depends on news volume and model latency.
 
 ## Output
 

plugins/auto-fin/README_ZH.md

@@ -2,7 +2,7 @@
 
 [English](README.md)
 
-Auto Fin 自动拉取一个滚动时间窗口内的财联社电报（默认 24 小时），按配置 topics 筛选相关新闻，搜索 ReMe 中有回顾价值的历史材料，最后写入一份带校验
+Auto Fin 自动拉取一个滚动时间窗口内的财联社电报（默认 24 小时），按配置 topics 归类相关新闻，逐主题搜索 ReMe 中有回顾价值的历史材料，最后写入一份带校验
 wikilink 的中文 Markdown 报告。当前新闻和筛选结果只存在于本次运行内存中，只有最终报告成为持久记忆。本目录是一个独立 Python
 distribution：单个 `reme.plugins` entry point 暴露 `plugin.yaml`，其中声明三个 Step backend，并在
 `application_defaults` 下提供 Job 配置；通过 `plugins=["auto-fin"]` 显式启用这个已安装插件。
@@ -55,7 +55,7 @@ reme start plugins='["auto-fin"]' \
 ```
 
 自定义应用配置需要提供 `agent_wrapper.default`、启用 tag index 的 `file_store.default`，以及
-Auto Fin 和自动标签使用的 `search`、`read`、`list_tags`、`frontmatter_read` 和
+Auto Fin 和自动标签使用的 `search`、`list_tags`、`frontmatter_read` 和
 `frontmatter_update` Jobs。
 
 ## 流程
@@ -65,11 +65,11 @@ Auto Fin 和自动标签使用的 `search`、`read`、`list_tags`、`frontmatter
         ↓
 在 RuntimeContext 中规范化和去重
         ↓
-Topic Agent 分批选择真实 news_id
+Topic Agent 按提示词长度分批输出“主题 → news_id”
         ↓
-Research Agent 使用 search + read 检索历史记忆
+每个有新闻的主题由独立 Research Agent 研究最新 20 篇；每个 Agent 最多搜索 3 次
         ↓
-代码校验历史 wikilink
+合并各主题结果并校验历史 wikilink
         ↓
 daily/YYYY-MM-DD/auto_fin.md
         ↓
@@ -79,14 +79,14 @@ daily/YYYY-MM-DD/auto_fin.md
 `auto_fin_data_step` 使用财联社网页同源接口的签名和分页方式，从分析时刻开始向前翻页，直到完整覆盖严格的最近 24
 小时。请求带有限速和重试；损坏记录及窗口外记录会被丢弃。
 
-`auto_fin_topic_step` 分批接收当前新闻，只返回相关的 `news_id`。代码会忽略未知 ID、去除重复 ID，并保持源新闻顺序。如果没有相关新闻，Job
+`auto_fin_topic_step` 按完整提示词的 10 万字符上限分批接收当前新闻，返回每个主题的相关 `news_id`。代码会忽略未知 ID、去除重复 ID，并保持源新闻顺序；一条新闻可属于多个主题。如果没有相关新闻，Job
 会成功跳过，不写报告。
 
-`auto_fin_merge_step` 只接收筛选后的当前新闻，并向 Agent 开放 `search` 和 `read`。当前新闻以 CLS ID、时间和标题作为普通证据。
+`auto_fin_merge_step` 依次研究每个有新闻的主题，只送入该主题最新的 20 篇新闻，并向 Agent 仅开放 `search`；代码按主题限制最多 3 次搜索。当前新闻以 CLS ID、时间和标题作为普通证据。
 Prompt 要求 Agent 只链接实际使用过的历史 Markdown；代码边界则独立保证只保留真实存在、相对
 workspace 的 Markdown 目标。不存在、绝对路径、越界、带反斜杠和自引用的目标都会降级为可读 alias。
 
-同日重跑会参考当天已有报告并覆盖为修订结果。最终写入使用原子替换并刷新当天索引，随后通过 `auto_tag_step`
+同日重跑会参考当天已有报告（最多前 3 万字符）并覆盖为修订结果。各主题结果合并后原子写入并刷新当天索引，随后通过 `auto_tag_step`
 更新报告的记忆标签 frontmatter；常规后台文件 watcher 会观察该源文件变化并刷新派生索引。流程不会写入 JSONL、
 中间 Markdown 或 Agent 结构化输出。
 
@@ -101,7 +101,7 @@ workspace 的 Markdown 目标。不存在、绝对路径、越界、带反斜杠
 | `request_interval` |                   `10` | 每次财联社请求尝试后的最小等待秒数，可设为 0 |
 | `max_retries`      |                    `3` | 每页财联社请求的最大尝试次数，至少为 1       |
 
-插件的 cron Job 随应用启动，并按应用配置的时区在每天 18:00 运行。
+插件的 cron Job 随应用启动，并按应用配置的时区在每天 09:00 运行，默认时区为 `Asia/Shanghai`。滚动窗口按时间戳计算，允许跨自然日；09:00 是启动时间，报告完成时间取决于新闻量与模型耗时。
 
 ## 产物
 

plugins/auto-fin/src/reme_auto_fin/base.py

@@ -77,6 +77,7 @@ class AutoFinStep(BaseStep):
         model: type[BaseModel],
         job_tools: list[str] | None = None,
         injected_job_kwargs: dict[str, Any] | None = None,
+        tool_context_id: str | None = None,
         **values: str,
     ) -> BaseModel:
         if self.agent_wrapper is None:
@@ -87,11 +88,18 @@ class AutoFinStep(BaseStep):
             f"[{self.name}] agent input prompt={prompt_name} schema={model.__name__} "
             f"query={self._preview(prompt, AGENT_INPUT_LOG_LIMIT)}",
         )
-        kwargs: dict[str, Any] = {"output_schema": model}
+        kwargs: dict[str, Any] = {
+            "output_schema": model,
+            "builtin_tools": [],
+            "use_builtin_tools": False,
+            "skills": [],
+        }
         if job_tools:
             kwargs["job_tools"] = job_tools
         if injected_job_kwargs:
             kwargs["injected_job_kwargs"] = injected_job_kwargs
+        if tool_context_id:
+            kwargs["tool_context_id"] = tool_context_id
         result = await self.agent_wrapper.reply(prompt, **kwargs)
         if not isinstance(result, dict) or result.get("structured_output") is None:
             raise ValueError(f"Auto Fin Agent returned no structured output: {self._preview(result)}")

plugins/auto-fin/src/reme_auto_fin/data.py

@@ -72,6 +72,11 @@ class AutoFinDataStep(AutoFinStep):
         interval = max(0.0, float(self._value("request_interval", 10)))
         max_retries = max(1, int(self._value("max_retries", 3)))
         records: dict[str, dict[str, str]] = {}
+        page_count = 0
+        self.logger.info(
+            f"[{self.name}] fetching CLS news window_start={cutoff.isoformat()} "
+            f"window_end={decision_at.isoformat()} request_interval={interval}s",
+        )
         async with httpx.AsyncClient(headers=HEADERS, timeout=httpx.Timeout(20, connect=5)) as client:
             while True:
                 for attempt in range(max_retries):
@@ -81,12 +86,17 @@ class AutoFinDataStep(AutoFinStep):
                     except (httpx.HTTPError, ValueError, RuntimeError) as exc:
                         if attempt + 1 == max_retries:
                             raise RuntimeError(f"CLS request failed after {max_retries} attempts: {exc}") from exc
+                        self.logger.warning(
+                            f"[{self.name}] CLS page request failed cursor={cursor} "
+                            f"attempt={attempt + 1}/{max_retries}: {exc}",
+                        )
                         await asyncio.sleep(2**attempt)
                     finally:
                         if interval:
                             await asyncio.sleep(interval)
                 if not rows:
                     raise RuntimeError("CLS API returned no news before the 24-hour window was covered")
+                page_count += 1
                 timestamps = [int(row["ctime"]) for row in rows if str(row.get("ctime", "")).isdigit()]
                 if not timestamps:
                     raise RuntimeError("CLS API page contained no valid timestamps")
@@ -97,9 +107,14 @@ class AutoFinDataStep(AutoFinStep):
                     normalized = self._normalize(row, cutoff, decision_at)
                     if normalized is not None:
                         records.setdefault(normalized["news_id"], normalized)
+                self.logger.info(
+                    f"[{self.name}] CLS page={page_count} rows={len(rows)} oldest={oldest} "
+                    f"unique_in_window={len(records)}",
+                )
                 if oldest <= int(cutoff.timestamp()):
                     break
                 cursor = oldest
+        self.logger.info(f"[{self.name}] completed CLS fetch pages={page_count} news={len(records)}")
         return sorted(records.values(), key=lambda row: (row["event_time"], row["news_id"]))
 
     @staticmethod

plugins/auto-fin/src/reme_auto_fin/merge.py

@@ -7,6 +7,7 @@ import re
 from datetime import date, timedelta
 from pathlib import Path
 from types import SimpleNamespace
+from uuid import uuid4
 
 from reme.steps.file_io import refresh_day_index
 
@@ -29,7 +30,7 @@ class AutoFinMergeStep(AutoFinStep):
         """Return today's existing report so intra-day reruns refine it, not replace it."""
         path = self._report_path(run_date)
         if path.is_file():
-            return path.read_text(encoding="utf-8")
+            return path.read_text(encoding="utf-8")[:30_000]
         return "今日暂无更早时段的推荐，本次为当日首次生成。"
 
     def _normalize_hybrid_wikilinks(self, body: str) -> str:
@@ -113,27 +114,61 @@ class AutoFinMergeStep(AutoFinStep):
             "min_score": 0.0,
             "start_date": None,
             "end_date": (run_date - timedelta(days=1)).isoformat(),
+            "max_search_calls": 3,
         }
-        output = await self._reply(
-            "merge_user",
-            AutoFinReportOutput,
-            job_tools=list(self.kwargs.get("job_tools") or []),
-            injected_job_kwargs=historical_search,
-            decision_at=str(self._required("auto_fin_decision_at")),
-            window_start=str(self._required("auto_fin_window_start")),
-            topics=json.dumps(self._required("auto_fin_topics"), ensure_ascii=False),
-            news=json.dumps(self._required("auto_fin_selected_news"), ensure_ascii=False),
-            current_report=self._current_report(run_date),
-        )
-        output = self._normalize(output)
-        output = output.model_copy(update={"body": self._normalize_hybrid_wikilinks(output.body)})
-        body, source_paths = self._validate_wikilinks(output.body, run_date)
-        output = output.model_copy(update={"body": body})
-        markdown = f"# {output.title}\n\n> {output.description}\n\n{output.body}\n\n"
+        by_topic = self._required("auto_fin_news_by_topic")
+        sections: list[str] = []
+        source_paths: list[str] = []
+        researched_topics: list[str] = []
+        earlier_report = self._current_report(run_date)
+        for topic in self._required("auto_fin_topics"):
+            related = by_topic[topic]
+            if not related:
+                self.logger.info(f"[{self.name}] skipping topic={topic} reason=no_related_news")
+                continue
+            recent = sorted(related, key=lambda row: (row["event_time"], row["news_id"]), reverse=True)[:20]
+            tool_context_id = f"auto_fin:{uuid4().hex}"
+            self.logger.info(
+                f"[{self.name}] researching topic={topic} related_news={len(related)} "
+                f"sent_news={len(recent)} omitted_news={len(related) - len(recent)} search_limit=3",
+            )
+            try:
+                output = await self._reply(
+                    "merge_user",
+                    AutoFinReportOutput,
+                    job_tools=["search"],
+                    injected_job_kwargs=historical_search,
+                    tool_context_id=tool_context_id,
+                    decision_at=str(self._required("auto_fin_decision_at")),
+                    window_start=str(self._required("auto_fin_window_start")),
+                    topic=topic,
+                    news=json.dumps(recent, ensure_ascii=False),
+                    omitted_news_count=str(len(related) - len(recent)),
+                    current_report=earlier_report,
+                )
+            finally:
+                if self.app_context is not None:
+                    self.app_context.metadata.get("__search_call_budgets", {}).pop(tool_context_id, None)
+            output = self._normalize(output)
+            body, paths = self._validate_wikilinks(self._normalize_hybrid_wikilinks(output.body), run_date)
+            source_paths.extend(path for path in paths if path not in source_paths)
+            sections.append(f"## {topic}：{output.title}\n\n> {output.description}\n\n{body}")
+            researched_topics.append(topic)
+            self.logger.info(
+                f"[{self.name}] researched topic={topic} body_chars={len(body)} valid_sources={len(paths)}",
+            )
+        title = f"主题新闻观察（{run_date}）"
+        description = f"截至 {self._required('auto_fin_decision_at')}，关注 {', '.join(researched_topics)}。"
+        body = "\n\n".join(sections)
+        markdown = f"# {title}\n\n> {description}\n\n{body}\n\n"
         markdown += "> 未接入可靠行情数据；本文只提供新闻研究和回顾线索，不提供收益、目标价或买卖建议。\n"
         report = self._report_path(run_date)
         change = "modified" if report.is_file() else "added"
         _write(report, markdown)
+        self.logger.info(
+            f"[{self.name}] wrote report path={report.relative_to(self.workspace_path)} "
+            f"change={change} topics={len(researched_topics)} sources={len(source_paths)} chars={len(markdown)}",
+        )
         await refresh_day_index(
             SimpleNamespace(workspace_path=self.workspace_path),
             str(run_date),
@@ -143,7 +178,7 @@ class AutoFinMergeStep(AutoFinStep):
         self.context["changes"] = [{"change": change, "path": relative}]
         self.context["markdown_path"] = relative
         self.context["auto_fin_digest_path"] = relative
-        self.context.response.answer = output.body
+        self.context.response.answer = body
         self.context.response.metadata.update(
             {
                 "markdown_path": relative,

plugins/auto-fin/src/reme_auto_fin/merge.yaml

@@ -1,20 +1,23 @@
 merge_user: |
-  你是主题新闻研究 Agent。当前新闻已经按 topics 做过语义筛选。你可以使用 `search` 搜索历史记忆，
-  并使用 `read` 阅读可能相关的完整 Markdown。不得使用外部搜索，不得虚构行情、收益、价格或未提供的数据。
-
+  ## 输入材料
   研究窗口：{window_start} 至 {decision_at}
-  topics：{topics}
-  当前新闻：{news}
-  今天早些时段的报告（如有，请保留仍成立的判断，只修订变化部分）：
+  当前主题：{topic}
+  该主题最新的新闻（最多 20 篇，另有 {omitted_news_count} 篇未送入本次研究）：
+  {news}
+  今天早些时段的报告：
   {current_report}
 
-  围绕 topics 和最重要的当前事件检索历史记忆；`search` 与 `read` 合计最多调用 3 次。
-  优先调用 `search`，仅对明显相关的结果调用 `read`；用完调用额度后直接完成报告，不再调用工具。
+  ## 任务指令
+  你是主题新闻研究 Agent。当前新闻已按主题筛选。你只能使用 `search` 搜索历史记忆，最多调用 3 次。
+  不得使用外部搜索，不得虚构行情、收益、价格或未提供的数据。搜索片段不足以支持的细节应标明待核实。
+  复核今天早些时段报告中与当前主题有关的判断。
+  围绕当前主题和重要事件检索历史记忆；用完搜索额度后直接完成报告。
   说明历史事件与当前事件的相同点、关键差异，以及旧判断是否仍适用。
   给出值得回顾的新闻、应继续观察的信息，以及哪些条件会强化或推翻判断，但不要给出投资建议。
 
   当前新闻不是 workspace 文件：在正文中保留其 CLS news_id、发布时间和标题，不要为它虚构 wikilink。
-  对实际搜索、阅读并支持判断的历史 Markdown，把完整 workspace-relative path 作为 contextual wikilink 直接写进
-  相关句子，只写一次。不得写裸链接行。代码会把不存在、越界或指向当前报告的链接降级为普通文本。
+  对搜索结果中实际支持判断的历史 Markdown，把完整 workspace-relative path 作为 contextual wikilink 写进
+  相关句子，例如：这与[[daily/2026-08-01/auto_fin.md|此前的供给判断]]相似，但本次政策范围不同。
+  不得写裸链接行或虚构路径。代码会把不存在、越界或指向当前报告的链接降级为普通文本。
 
   按结构化输出契约返回 title、description 和完整中文 Markdown body。

plugins/auto-fin/src/reme_auto_fin/plugin.yaml

@@ -42,10 +42,10 @@ application_defaults:
         - backend: auto_fin_data_step
         - backend: auto_fin_topic_step
         - backend: auto_fin_merge_step
-          job_tools: [search, read]
+          job_tools: [search]
         - backend: auto_tag_step
 
     auto_fin_cron:
       backend: cron
-      cron: "0 18 * * *"
+      cron: "0 9 * * *"
       steps: *auto_fin_steps

plugins/auto-fin/src/reme_auto_fin/topic.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 import json
 import re
 from time import perf_counter
@@ -12,22 +13,26 @@ from .base import AGENT_INPUT_LOG_LIMIT, AGENT_OUTPUT_LOG_LIMIT, AutoFinStep
 class AutoFinTopicStep(AutoFinStep):
     """Filter current news in bounded Agent batches without writing files."""
 
+    PROMPT_CHAR_LIMIT = 100_000
+
     @staticmethod
-    def _parse_news_ids(value: object) -> list[str]:
+    def _parse_news_ids(value: object, topics: list[str]) -> dict[str, list[str]]:
         if not isinstance(value, str):
             raise ValueError("Auto Fin topic Agent returned no text")
         match = re.search(r"```json\s*(.*?)```", value, re.IGNORECASE | re.DOTALL)
         if match is None:
             raise ValueError("Auto Fin topic Agent returned no JSON code block")
         try:
-            ids = json.loads(match.group(1).strip())
+            mapping = json.loads(match.group(1).strip())
         except json.JSONDecodeError as exc:
             raise ValueError("Auto Fin topic Agent returned invalid JSON") from exc
-        if not isinstance(ids, list) or any(not isinstance(item, str) for item in ids):
-            raise ValueError("Auto Fin topic Agent must return a JSON array of strings")
-        return ids
+        if not isinstance(mapping, dict) or set(mapping) != set(topics):
+            raise ValueError("Auto Fin topic Agent must return exactly the configured topic keys")
+        if any(not isinstance(ids, list) or any(not isinstance(item, str) for item in ids) for ids in mapping.values()):
+            raise ValueError("Auto Fin topic Agent must return string news ID arrays")
+        return mapping
 
-    async def _select_news_ids(self, prompt: str) -> list[str]:
+    async def _select_news_ids(self, prompt: str, topics: list[str]) -> dict[str, list[str]]:
         """Request and validate plain-text topic IDs, retrying one malformed reply."""
         if self.agent_wrapper is None:
             raise RuntimeError("Auto Fin analysis requires an agent_wrapper")
@@ -38,7 +43,7 @@ class AutoFinTopicStep(AutoFinStep):
             started_at = perf_counter()
             result = await self.agent_wrapper.reply(prompt)
             try:
-                ids = self._parse_news_ids(result.get("result") if isinstance(result, dict) else None)
+                ids = self._parse_news_ids(result.get("result") if isinstance(result, dict) else None, topics)
             except ValueError as exc:
                 if attempt:
                     raise ValueError(f"Auto Fin topic Agent returned invalid news IDs: {exc}") from exc
@@ -51,6 +56,31 @@ class AutoFinTopicStep(AutoFinStep):
             return ids
         raise RuntimeError("Auto Fin topic Agent produced no response")
 
+    def _prompt(self, news: list[dict], topics: list[str], window_hours: str) -> str:
+        return self.prompt_format(
+            "topic_user",
+            topics=json.dumps(topics, ensure_ascii=False),
+            news=json.dumps(news, ensure_ascii=False),
+            window_hours=window_hours,
+            output_example=json.dumps({topic: [] for topic in topics}, ensure_ascii=False),
+        )
+
+    def _batches(self, news: list[dict], topics: list[str], window_hours: str) -> Iterator[list[dict]]:
+        batch: list[dict] = []
+        for row in news:
+            item = {**row, "title": str(row.get("title") or "")[:300], "content": str(row.get("content") or "")[:1000]}
+            if len(self._prompt([*batch, item], topics, window_hours)) > self.PROMPT_CHAR_LIMIT:
+                if not batch:
+                    raise ValueError("Auto Fin news item exceeds the topic Agent prompt limit")
+                yield batch
+                batch = [item]
+                if len(self._prompt(batch, topics, window_hours)) > self.PROMPT_CHAR_LIMIT:
+                    raise ValueError("Auto Fin news item exceeds the topic Agent prompt limit")
+            else:
+                batch.append(item)
+        if batch:
+            yield batch
+
     async def execute(self):
         """Select relevant news from each batch for the current invocation."""
         assert self.context is not None
@@ -58,22 +88,34 @@ class AutoFinTopicStep(AutoFinStep):
         topics = list(self._required("auto_fin_topics"))
         window_hours = float(self._value("auto_fin_window_hours", 24))
         formatted_hours = f"{window_hours:g}"
-        batch_size = max(1, int(self._value("topic_batch_size", 50)))
-        selected: set[str] = set()
-        for start in range(0, len(news), batch_size):
-            batch = [
-                {**row, "content": str(row.get("content") or "")[:1000]} for row in news[start : start + batch_size]
-            ]
-            prompt = self.prompt_format(
-                "topic_user",
-                topics=json.dumps(topics, ensure_ascii=False),
-                news=json.dumps(batch, ensure_ascii=False),
-                window_hours=formatted_hours,
+        selected: dict[str, set[str]] = {topic: set() for topic in topics}
+        batch_count = 0
+        for batch in self._batches(news, topics, formatted_hours):
+            batch_count += 1
+            valid_ids = {row["news_id"] for row in batch}
+            prompt = self._prompt(batch, topics, formatted_hours)
+            self.logger.info(
+                f"[{self.name}] filtering batch={batch_count} news={len(batch)} prompt_chars={len(prompt)} "
+                f"first_id={batch[0]['news_id']} last_id={batch[-1]['news_id']}",
             )
-            selected.update(await self._select_news_ids(prompt))
-        relevant = [row for row in news if row["news_id"] in selected]
+            for topic, ids in (await self._select_news_ids(prompt, topics)).items():
+                unknown = len(set(ids) - valid_ids)
+                if unknown:
+                    self.logger.warning(
+                        f"[{self.name}] ignored unknown IDs batch={batch_count} topic={topic} count={unknown}",
+                    )
+                selected[topic].update(valid_ids.intersection(ids))
+        by_topic = {topic: [row for row in news if row["news_id"] in selected[topic]] for topic in topics}
+        relevant_ids = set().union(*selected.values()) if selected else set()
+        relevant = [row for row in news if row["news_id"] in relevant_ids]
+        self.context["auto_fin_news_by_topic"] = by_topic
         self.context["auto_fin_selected_news"] = relevant
         self.context.response.metadata["relevant_news_count"] = len(relevant)
+        self.context.response.metadata["topic_batch_count"] = batch_count
+        self.logger.info(
+            f"[{self.name}] topic filtering complete batches={batch_count} selected_unique={len(relevant)} "
+            f"per_topic={{{', '.join(f'{topic!r}: {len(rows)}' for topic, rows in by_topic.items())}}}",
+        )
         if not relevant:
             reason = f"最近{formatted_hours}小时没有与 {', '.join(topics)} 相关的财联社新闻。"
             self.context["auto_fin_skipped"] = True

plugins/auto-fin/src/reme_auto_fin/topic.yaml

@@ -1,14 +1,14 @@
 topic_user: |
-  从下面最近{window_hours}小时的财联社新闻中，选择与至少一个 topics 存在真实、可解释关系的新闻。
-  只返回输入中真实存在的 news_id；仅出现关键词但没有实质关系的新闻不要选择。
-  在 ```json 代码块中返回 JSON 字符串数组。没有相关新闻时返回 []。
-  输出格式示例：
-  ```json
-  [
-    "123",
-    "456"
-  ]
-  ```
+  ## 输入材料
+  时间范围：最近 {window_hours} 小时
+  主题：{topics}
+  财联社新闻：
+  {news}
 
-  topics：{topics}
-  新闻：{news}
+  ## 任务指令
+  按主题归类新闻。只选择与主题存在真实、可解释关系的新闻；仅出现关键词而无实质关系的不要选择。
+  同一新闻可以属于多个主题。只返回材料中真实存在的 news_id，每个输入主题都必须出现，没有相关新闻时使用空数组。
+  在 ```json 代码块中只返回 JSON 对象，不要补充解释。使用以下键结构，并填入真实 news_id：
+  ```json
+  {output_example}
+  ```

plugins/auto-fin/tests/test_auto_fin.py

@@ -100,20 +100,20 @@ async def test_data_step_uses_configurable_window_hours(tmp_path: Path, monkeypa
 
 
 class _TopicAgent(BaseAgentWrapper):
-    def __init__(self, news_ids: list[str], **kwargs):
+    def __init__(self, selected: dict[str, list[str]], **kwargs):
         super().__init__(**kwargs)
-        self.news_ids = news_ids
+        self.selected = selected
         self.calls = []
 
     async def reply(self, inputs, **kwargs):
         self.calls.append((str(inputs), kwargs))
-        return {"result": f"筛选结果：\n```json\n{json.dumps(self.news_ids)}\n```\n以上是相关 ID。"}
+        return {"result": f"筛选结果：\n```json\n{json.dumps(self.selected)}\n```\n以上是相关 ID。"}
 
 
 @pytest.mark.asyncio
 async def test_topic_step_keeps_real_ids_in_memory_only(tmp_path: Path):
     app_context = ApplicationContext(workspace_dir=str(tmp_path), timezone="Asia/Shanghai")
-    agent = _TopicAgent(["2", "missing", "2"], app_context=app_context)
+    agent = _TopicAgent({"黄金": ["2", "missing", "2"]}, app_context=app_context)
     context = RuntimeContext(
         auto_fin_news=[
             {
@@ -135,8 +135,10 @@ async def test_topic_step_keeps_real_ids_in_memory_only(tmp_path: Path):
     response = await AutoFinTopicStep(app_context=app_context, agent_wrapper=agent)(context)
 
     assert [row["news_id"] for row in context["auto_fin_selected_news"]] == ["2"]
+    assert [row["news_id"] for row in context["auto_fin_news_by_topic"]["黄金"]] == ["2"]
     assert agent.calls[0][1] == {}
-    assert '```json\n[\n"123",\n"456"\n]\n```' in agent.calls[0][0]
+    assert '```json\n{"黄金": []}' in agent.calls[0][0]
+    assert agent.calls[0][0].index("## 输入材料") < agent.calls[0][0].index("## 任务指令")
     assert response.metadata["relevant_news_count"] == 1
     assert not list(tmp_path.rglob("*.*"))
 
@@ -144,7 +146,7 @@ async def test_topic_step_keeps_real_ids_in_memory_only(tmp_path: Path):
 @pytest.mark.asyncio
 async def test_topic_step_marks_empty_selection_as_successful_skip(tmp_path: Path):
     app_context = ApplicationContext(workspace_dir=str(tmp_path), timezone="Asia/Shanghai")
-    agent = _TopicAgent([], app_context=app_context)
+    agent = _TopicAgent({"黄金": []}, app_context=app_context)
     context = RuntimeContext(
         auto_fin_news=[
             {
@@ -166,21 +168,21 @@ async def test_topic_step_marks_empty_selection_as_successful_skip(tmp_path: Pat
     assert context["auto_fin_skipped"] is True
     assert response.metadata["skipped"] is True
     assert response.answer == "最近12小时没有与 黄金 相关的财联社新闻。"
-    assert "最近12小时" in agent.calls[0][0]
+    assert "最近 12 小时" in agent.calls[0][0]
     assert not list(tmp_path.rglob("*.md"))
 
 
 @pytest.mark.asyncio
 async def test_topic_step_retries_invalid_json_once(tmp_path: Path):
     class RetryAgent(_TopicAgent):
-        """Return one malformed response before a fenced JSON array."""
+        """Return one malformed response before a fenced topic mapping."""
 
         async def reply(self, inputs, **kwargs):
             self.calls.append((str(inputs), kwargs))
-            return {"result": ('{"new_ids": "[\\"1\\"]"}' if len(self.calls) == 1 else '```json\n["1"]\n```')}
+            return {"result": ('{"new_ids": "[\\"1\\"]"}' if len(self.calls) == 1 else '```json\n{"黄金": ["1"]}\n```')}
 
     app_context = ApplicationContext(workspace_dir=str(tmp_path), timezone="Asia/Shanghai")
-    agent = RetryAgent([], app_context=app_context)
+    agent = RetryAgent({"黄金": []}, app_context=app_context)
     context = RuntimeContext(
         auto_fin_news=[
             {
@@ -201,11 +203,39 @@ async def test_topic_step_retries_invalid_json_once(tmp_path: Path):
 
 @pytest.mark.parametrize(
     "value",
-    ['{"new_ids": ["1"]}', "```json\n[1]\n```", "not json", ""],
+    ['{"new_ids": ["1"]}', "```json\n[1]\n```", '```json\n{"黄金": [1]}\n```', "not json", ""],
 )
 def test_topic_step_rejects_non_array_or_non_string_ids(value: str):
     with pytest.raises(ValueError):
-        AutoFinTopicStep._parse_news_ids(value)
+        AutoFinTopicStep._parse_news_ids(value, ["黄金"])
+
+
+@pytest.mark.asyncio
+async def test_topic_step_batches_by_prompt_length_and_merges_topics(tmp_path: Path):
+    app_context = ApplicationContext(workspace_dir=str(tmp_path))
+    agent = _TopicAgent({"黄金": ["1", "2", "2"], "机器人": ["2"]}, app_context=app_context)
+    news = [
+        {
+            "news_id": str(index),
+            "event_time": f"2026-08-10T0{index}:00:00+08:00",
+            "title": "新闻",
+            "content": "正文" * 80,
+        }
+        for index in (1, 2, 3)
+    ]
+    step = AutoFinTopicStep(app_context=app_context, agent_wrapper=agent)
+    one_item_length = len(step._prompt([{**news[0], "content": news[0]["content"][:1000]}], ["黄金", "机器人"], "24"))
+    step.PROMPT_CHAR_LIMIT = one_item_length + 5
+    context = RuntimeContext(auto_fin_news=news, auto_fin_topics=["黄金", "机器人"])
+
+    response = await step(context)
+
+    assert len(agent.calls) == 3
+    assert all(len(prompt) <= step.PROMPT_CHAR_LIMIT for prompt, _ in agent.calls)
+    assert [row["news_id"] for row in context["auto_fin_news_by_topic"]["黄金"]] == ["1", "2"]
+    assert [row["news_id"] for row in context["auto_fin_news_by_topic"]["机器人"]] == ["2"]
+    assert [row["news_id"] for row in context["auto_fin_selected_news"]] == ["1", "2"]
+    assert response.metadata["topic_batch_count"] == 3
 
 
 class _ResearchAgent(BaseAgentWrapper):
@@ -244,6 +274,16 @@ async def test_merge_writes_only_final_report_and_validates_historical_links(
         auto_fin_decision_at="2026-08-10T09:30:00+08:00",
         auto_fin_window_start="2026-08-09T09:30:00+08:00",
         auto_fin_topics=["黄金"],
+        auto_fin_news_by_topic={
+            "黄金": [
+                {
+                    "news_id": "1",
+                    "event_time": "2026-08-10T09:00:00+08:00",
+                    "title": "黄金上涨",
+                    "content": "避险需求增强",
+                },
+            ],
+        },
         auto_fin_selected_news=[
             {
                 "news_id": "1",
@@ -257,22 +297,23 @@ async def test_merge_writes_only_final_report_and_validates_historical_links(
     response = await AutoFinMergeStep(
         app_context=app_context,
         agent_wrapper=agent,
-        job_tools=["search", "read"],
+        job_tools=["search"],
     )(context)
 
     prompt, kwargs = agent.calls[0]
     assert "end_date" not in prompt
-    assert "调用 `search`" in prompt
-    assert "调用 `read`" in prompt
-    assert kwargs == {
-        "output_schema": AutoFinReportOutput,
-        "job_tools": ["search", "read"],
-        "injected_job_kwargs": {
-            "limit": 5,
-            "min_score": 0.0,
-            "start_date": None,
-            "end_date": "2026-08-09",
-        },
+    assert "`search`" in prompt
+    assert "`read`" not in prompt
+    assert prompt.index("## 输入材料") < prompt.index("## 任务指令")
+    assert kwargs["output_schema"] == AutoFinReportOutput
+    assert kwargs["job_tools"] == ["search"]
+    assert kwargs["tool_context_id"].startswith("auto_fin:")
+    assert kwargs["injected_job_kwargs"] == {
+        "limit": 5,
+        "min_score": 0.0,
+        "start_date": None,
+        "end_date": "2026-08-09",
+        "max_search_calls": 3,
     }
     report = (tmp_path / "daily" / "2026-08-10" / "auto_fin.md").read_text(encoding="utf-8")
     assert "[[daily/2026-08-01/auto_fin.md|历史黄金观察]]" in report
@@ -285,6 +326,40 @@ async def test_merge_writes_only_final_report_and_validates_historical_links(
     assert response.metadata["source_paths"] == ["daily/2026-08-01/auto_fin.md"]
 
 
+@pytest.mark.asyncio
+async def test_merge_researches_each_topic_with_latest_twenty_news(tmp_path: Path):
+    app_context = ApplicationContext(workspace_dir=str(tmp_path))
+    agent = _ResearchAgent(app_context=app_context)
+    gold = [
+        {"news_id": str(index), "event_time": f"2026-08-10T09:{index:02}:00+08:00", "title": "黄金", "content": "正文"}
+        for index in range(25)
+    ]
+    robot = [{"news_id": "30", "event_time": "2026-08-10T08:00:00+08:00", "title": "机器人", "content": "正文"}]
+    context = RuntimeContext(
+        auto_fin_date="2026-08-10",
+        auto_fin_decision_at="2026-08-10T09:30:00+08:00",
+        auto_fin_window_start="2026-08-09T09:30:00+08:00",
+        auto_fin_topics=["黄金", "机器人", "半导体"],
+        auto_fin_news_by_topic={"黄金": gold, "机器人": robot, "半导体": []},
+        auto_fin_selected_news=[*gold, *robot],
+    )
+
+    response = await AutoFinMergeStep(app_context=app_context, agent_wrapper=agent)(context)
+
+    assert len(agent.calls) == 2
+    gold_prompt, gold_kwargs = agent.calls[0]
+    robot_prompt, robot_kwargs = agent.calls[1]
+    assert '"news_id": "24"' in gold_prompt
+    assert '"news_id": "4"' not in gold_prompt
+    assert "另有 5 篇" in gold_prompt
+    assert "当前主题：机器人" in robot_prompt
+    assert gold_kwargs["tool_context_id"] != robot_kwargs["tool_context_id"]
+    assert response.answer.count("## ") >= 2
+    assert response.metadata["selected_news_count"] == 26
+    report = (tmp_path / "daily" / "2026-08-10" / "auto_fin.md").read_text(encoding="utf-8")
+    assert "## 黄金：" in report and "## 机器人：" in report
+
+
 def test_hybrid_wikilink_normalization_is_conservative_and_failure_safe(tmp_path: Path, monkeypatch):
     import reme_auto_fin.merge as merge_module
 
@@ -327,9 +402,9 @@ def test_plugin_config_has_default_topics_and_no_intermediate_index_step():
         "auto_fin_merge_step",
         "auto_tag_step",
     ]
-    assert job["steps"][2]["job_tools"] == ["search", "read"]
+    assert job["steps"][2]["job_tools"] == ["search"]
     assert job["steps"][3] == {"backend": "auto_tag_step"}
-    assert jobs["auto_fin_cron"]["cron"] == "0 18 * * *"
+    assert jobs["auto_fin_cron"]["cron"] == "0 9 * * *"
     assert jobs["auto_fin_cron"]["steps"] == job["steps"]
     assert (
         not {

reme/steps/index/search.py

@@ -225,6 +225,7 @@ class SearchStep(BaseStep):
         expand_links_enabled: bool = bool(self.kwargs.get("expand_links", True))
         max_links_per_direction: int = int(self.kwargs.get("max_links_per_direction", 10))
         tool_context_id: str = (self.context.get("tool_context_id", "") or "").strip()
+        max_search_calls = self.context.get("max_search_calls")
         strict_date_filter: bool = bool(
             self.context.get("strict_date_filter") or self.kwargs.get("strict_date_filter", False),
         )
@@ -235,6 +236,27 @@ class SearchStep(BaseStep):
             return self.context.response
         assert limit > 0, f"limit must be positive, got {limit}"
 
+        if max_search_calls is not None:
+            if not tool_context_id or self.app_context is None:
+                raise ValueError("max_search_calls requires an application and tool_context_id")
+            maximum = int(max_search_calls)
+            if maximum < 1:
+                raise ValueError("max_search_calls must be positive")
+            budgets = self.app_context.metadata.setdefault("__search_call_budgets", {})
+            budget = budgets.setdefault(tool_context_id, {"count": 0, "lock": asyncio.Lock()})
+            async with budget["lock"]:
+                if budget["count"] >= maximum:
+                    self.logger.warning(
+                        f"[{self.name}] search budget exhausted context={tool_context_id} limit={maximum}",
+                    )
+                    self.context.response.success = False
+                    self.context.response.answer = f"Error: search call limit of {maximum} reached"
+                    return self.context.response
+                budget["count"] += 1
+                self.logger.info(
+                    f"[{self.name}] search budget context={tool_context_id} " f"call={budget['count']}/{maximum}",
+                )
+
         candidates = min(_MAX_CANDIDATES, max(1, int(limit * candidate_multiplier)))
         search_filter: dict = dict(self.context.get("search_filter", {}) or {})
         raw_tags = self.context.get("tags", []) or []

tests/unit/test_cookbook_config.py

@@ -196,7 +196,7 @@ def test_cookbook_overrides_merge_with_pure_plugin_defaults(monkeypatch):
     assert config.jobs["auto_fin"].backend == "base"
     assert config.jobs["auto_fin"].parameters["properties"]["topics"]["default"] == "黄金,机器人,半导体"
     assert config.jobs["auto_fin_cron"].backend == "cron"
-    assert config.jobs["auto_fin_cron"].model_extra["cron"] == "0 18 * * *"
+    assert config.jobs["auto_fin_cron"].model_extra["cron"] == "0 9 * * *"
     assert config.jobs["daily_paper"].backend == "base"
     assert config.jobs["daily_paper_cron"].backend == "cron"
     assert config.jobs["daily_paper_cron"].model_extra["cron"] == "0 8 * * *"

tests/unit/test_search_step.py

@@ -93,6 +93,32 @@ class FakeSearchStore(BaseFileStore):
         return self.keyword_results[:limit]
 
 
+def test_search_call_budget_is_scoped_to_tool_context(tmp_path):
+    """An injected budget rejects a fourth call before touching the store."""
+    store = FakeSearchStore()
+    app_context = ApplicationContext(workspace_dir=str(tmp_path))
+
+    async def run():
+        for _ in range(3):
+            response = await SearchStep(app_context=app_context, file_store=store, expand_links=False)(
+                RuntimeContext(query="gold", tool_context_id="topic-a", max_search_calls=3),
+            )
+            assert response.success
+        calls = len(store.calls)
+        rejected = await SearchStep(app_context=app_context, file_store=store, expand_links=False)(
+            RuntimeContext(query="gold", tool_context_id="topic-a", max_search_calls=3),
+        )
+        assert not rejected.success
+        assert "limit of 3" in rejected.answer
+        assert len(store.calls) == calls
+        other = await SearchStep(app_context=app_context, file_store=store, expand_links=False)(
+            RuntimeContext(query="gold", tool_context_id="topic-b", max_search_calls=3),
+        )
+        assert other.success
+
+    asyncio.run(run())
+
+
 class TaggedFakeSearchStore(FakeSearchStore):
     """Fake store with a tag index and ordinary file-store filtering."""