fix: stabilize API routing and simplify delivery reports

README.md

@@ -20,6 +20,7 @@ capabilities/am516-delivery-prediction/
 - Generate internal similar-model candidates.
 - Query the AR836 delivery prediction API through the controlled local script.
 - Produce an internal delivery candidate report for Feishu.
+- Resolve code-only shortage names from the Git-synchronized material-name dictionary without a live Kingdee request during replies.
 - Append a non-secret candidate report only to `capabilities/am516-delivery-prediction/records/delivery_prediction_records.md`.
 
 ## What This Bot Must Not Do
@@ -81,6 +82,8 @@ Minimum handoff material:
 
 ## Current Boundary
 
+The material-name dictionary at `capabilities/am516-delivery-prediction/data/material_names.json` is approved by AR51's direct local request for Git synchronization. It includes only material codes, names and source metadata; full material-master exports, BOMs and credentials remain excluded. Refresh instructions are in `capabilities/am516-delivery-prediction/docs/material_name_cache.md`.
+
 This is a pilot package. It is not a formal company AM, official system, customer commitment tool, or data security approval.
 
 AR51 owns AM516 rules, functional design, implementation, validation, iteration, network access, account permissions, API-key use, and outbound-transfer approval. TH8 is the supporting IT department but owns only Git repository and synchronization-mechanism construction in this project. Outbound API access remains disabled until AR51 creates a trusted approval record explicitly covering `api.zeroerr-agent.com`, the AM516 use case, and API-key use.

capabilities/am516-delivery-prediction/AGENTS.md

@@ -97,6 +97,17 @@ Include:
 - human review item;
 - fixed disclaimer.
 
+### Material Name Cache
+
+AR51's direct local request on 2026-10-09 authorizes a Git-synchronized dictionary of all Kingdee material codes and names for shortage-name lookup. The fixed file is `data/material_names.json`; it contains only `FNumber` / `FName` mappings and non-secret source/count/time/hash metadata.
+
+- When a returned shortage has no name, resolve its `缺料物料` code from this local file. Cache contents do not determine shortages or delivery dates.
+- Feishu processing may read this file, but must not write or refresh it or call Kingdee for name lookup.
+- AR51 may maintain it outside Feishu using `scripts/refresh_material_names.py`, either importing a verified full `BD_MATERIAL` reference cache or invoking the bundled restricted `shared/runtime/kingdee_materials.py` / existing approved read-only `kingdee_query.py` for `FNumber,FName` only, with no material filter.
+- AR51's direct local request authorizes the native background refresh: check hourly, read ERP when 48 hours have elapsed since the last successful refresh, add new names and update changed names. State and locks live in Git-ignored `logs/`; credentials stay in each computer's environment, Keychain or ignored `configs/material_cache_erp.local`. Feishu processing never triggers this updater.
+- The dictionary is intended to be tracked in this AM516 Git repository. Keep credentials, full material-master exports, BOMs, specifications, quantities, prices and other raw ERP fields out of it.
+- Empty, invalid, partial, or conflicting refresh results must preserve the previous file. Missing names must remain explicit unresolved items.
+
 ## 8. Fixed Disclaimer
 
 ```text

capabilities/am516-delivery-prediction/README.md

@@ -34,6 +34,11 @@ model + quantity
 | `skills/joint-module-model-recommendation/SKILL.md` | Model recommendation rules |
 | `skills/joint-module-leadtime-report/SKILL.md` | Lead-time report workflow |
 | `scripts/delivery_prediction_query_template.py` | Controlled API query script |
+| `scripts/refresh_material_names.py` | Local-only maintenance: export the full material-code/name dictionary |
+| `../../shared/runtime/install_material_name_refresh.py` | Install the native macOS background 48-hour refresh service |
+| `../../shared/runtime/kingdee_materials.py` | Portable ERP client restricted to material codes and names |
+| `data/material_names.json` | Git-synchronized material-code/name lookup; no ERP call during replies |
+| `docs/material_name_cache.md` | Cache source, refresh and cross-computer synchronization guide |
 | `rules/eRob关节模组相似型号推荐规则_v3.1.md` | Local packaged model recommendation rule source |
 | `records/delivery_prediction_records.md` | Local append-only candidate report record |
 | `docs/deployment_checklist.md` | Deployment and environment checks |

capabilities/am516-delivery-prediction/docs/acceptance_samples.md

@@ -14,8 +14,11 @@ Expected:
 
 - model parses successfully;
 - quantity is accepted;
+- original demand 10 remains visible and each model is queried exactly once with quantity 10; stock accounting is handled by the API, with no local inventory deduction, yield adjustment or batch rounding;
 - the bot immediately adds the Feishu `OnIt` (typing) reaction to the original message after the valid trigger is accepted and before the API query starts;
 - the controlled script uses the macOS system trust store and keeps TLS verification enabled;
+- the controlled script defaults to a direct HTTPS connection to `api.zeroerr-agent.com`, ignoring system/environment proxies for this API only; it does not change Feishu or other applications' proxy settings;
+- an explicit `ZEROERR_AGENT_PROXY_MODE=system` uses the configured proxies; an invalid mode or a non-approved direct target stops before the request, and a failed request does not trigger an automatic proxy fallback or retry;
 - no temporary TLS-bypass parameter is required for the normal happy path;
 - API script runs;
 - output does not expose API key;
@@ -24,10 +27,14 @@ Expected:
 - output omits the `候选明细` heading and does not repeat a human-review line under each candidate;
 - model differences use the compact `字段 原值→新值` format and join multiple changes with `；`;
 - Feishu output uses rich-text `post`; the report title, overall-judgment heading, numbered candidate headings, and unavailable-candidate heading are bold;
+- in initial and continuation replies, the `X个工作日` and final delivery date values are bold in both `最快候选交期` and each model's `AM516预测交期`, using native Feishu rich-text styles without displaying Markdown asterisks;
 - the judgment level and fixed disclaimer provide the global human-review boundary;
 - inventory output is limited to summary-level available quantity and coverage; non-reducer shortage output never exposes material codes or raw ERP details;
-- each usable model includes `缺料物料明细` listing only names from the API shortage records (`物料名称`, with `缺料物料名称` as the fallback), preserving order and deduplicating names; the same behavior applies to continuation replies;
-- missing names are reported explicitly without substituting codes or inferring names; an empty shortage list says `API未返回缺料记录`, and missing shortage data or invalid BOM keeps the unavailable-data reminder;
+- each usable model includes `缺料物料明细` listing only shortage names, preserving order and deduplicating names; use the API's `物料名称` / `缺料物料名称` first, then resolve `缺料物料` through `data/material_names.json`; the same behavior applies to continuation replies;
+- code-only shortage records with cached names display those names without calling Kingdee; unresolved codes and missing/invalid caches are reported explicitly without displaying codes or inferring names; an empty shortage list says `API未返回缺料记录`, and missing shortage data or invalid BOM keeps the unavailable-data reminder;
+- the cache is Git-trackable, contains only code/name mappings and source metadata, and works in another checkout without the source computer's paths or ERP credentials; refreshes with empty, invalid or conflicting data preserve the previous cache;
+- the native background refresher queries ERP only when at least 48 hours have elapsed since the last successful refresh; unchanged dictionaries do not rewrite the Git cache, added/renamed names are merged, and failed refreshes preserve the cache and last-success state;
+- running the updater on another computer uses that computer's local read-only ERP credentials, never Git-synchronized credentials; the bot sees valid cache changes on its next lookup without a restart;
 - reducer shortages are the sole exception: when `缺料物料` starts with `20.30.`, output only the reducer code, inventory, inspection, in-transit, existing-demand quantities, and in-transit coverage conclusion in the Skill-defined order;
 - shortage items outside the `20.30.` reducer prefix never expose their material codes or quantity details;
 - result is marked as internal candidate;
@@ -80,6 +87,7 @@ Expected:
 
 - the bot immediately adds the Feishu `OnIt` (typing) reaction to the original message after the continuation and active-context gates pass;
 - the bot reuses the active model and quantity only for that same sender and chat;
+- continuation retains original demand 10 and computes each new candidate from its own current stock; with zero stock production is 15. It never compounds an earlier adjustment or reuses baseline stock;
 - the bot queries candidate positions 5-8 in the unchanged rule order;
 - the original model and candidate positions 1-4 are not queried or displayed again;
 - unavailable candidates still consume their original positions and are summarized under `不可用候选`;
@@ -239,7 +247,7 @@ python3 capabilities/am516-delivery-prediction/scripts/delivery_prediction_query
 Expected:
 
 - the script creates an unverified SSL context only because the flag is explicitly present;
-- the context is passed to the single controlled `urlopen` call;
+- the context is passed to the request-local HTTPS handler used by the single controlled `opener.open` call;
 - the result does not expose the API Key;
 - a successful response remains an internal candidate and requires human review;
 - the temporary result does not remove the long-term certificate-repair requirement.
@@ -263,3 +271,12 @@ Expected:
 ```text
 以上为内部候选方案，不构成客户正式交期承诺；需 AR516 或授权责任人结合库存、生产、采购、客户优先级和最新系统数据人工复核后使用。
 ```
+
+## Original-Demand API Quantity Acceptance
+
+- Regression: `eRob70F50I-FHM-18ET[V4] 12台`, with API stock 10. Send quantity 12 once per model; show stock partly covering demand with a gap of 2. Preserve the API workdays/date and shortages. Do not show the old false-stock-branch error or calculate production 5 locally.
+- D=8/S=5: send 8; D=24/S=6: send 24; D=24/S=0: send 24; D=24/S>=24: still send 24 once and preserve the API response.
+- Different candidates display their own stock, but all use unchanged original demand. Initial and continuation batches query each current-batch model once.
+- Missing/invalid stock is unknown, never zero; valid API workdays/date remain usable with an explicit inventory-review item. Missing/invalid workdays/date or failed requests retain existing error handling.
+- Initial and continuation Feishu reports show demand quantity, lead time and inventory coverage, but omit `数量传递规则`, `库存覆盖口径` and per-model `API查询数量`. Internal record metadata still contains actual API quantities and quantity rules. No local net-demand or production-quantity fields are shown. Historical records and existing candidate-table columns remain unchanged.
+- These checks use mocked responses locally. Runtime acceptance must verify actual API arguments and compare the displayed workdays/date with the same returned response; mocked dates do not constitute delivery predictions.

capabilities/am516-delivery-prediction/docs/material_name_cache.md

@@ -0,0 +1,91 @@
+# 缺料物料名称本地缓存
+
+AR51 于 2026-10-09 在本地 Codex 请求将金蝶 ERP 全量物料编码和名称缓存到可同步的 Git 文件，用于在 AR836 交期接口只返回缺料编码时补充名称。本文件记录该实现的范围和维护方式。
+
+## 文件和数据范围
+
+固定文件：`capabilities/am516-delivery-prediction/data/material_names.json`。
+
+- 来源为 `BD_MATERIAL`，不按物料类型、审核状态、禁用状态或组织筛选。
+- 仅保存编码 `FNumber` 和名称 `FName`，以及快照时间、来源行数、去重后编码数和内容哈希。
+- 多组织的相同编码、相同名称合并成一条；同一编码出现不同名称时停止刷新，保留原缓存。
+- 文件位于项目内，纳入 Git 同步。凭证、完整物料主数据、规格、BOM、数量和价格不进入这个文件。
+- `source.snapshot_at` 是 ERP 查询或原参考缓存的时点；`generated_at` 是生成本文件的时间，不能把二者混为最新 ERP 数据时间。
+
+## 回复逻辑
+
+1. 仍由 AR836 接口提供缺料记录，不根据名称缓存判断缺料。
+2. 优先使用缺料记录里的名称；名称缺失时，根据该条记录的 `缺料物料` 编码查本地缓存。
+3. 首次需要时加载并校验文件，后续在内存中查询；检测到本地文件变化时重新加载。回复过程不访问金蝶、不刷新缓存，后台新增名称在下一次查询时生效。
+4. 回复只列缺料物料名称并去重，原有减速器展开规则继续有效。
+5. 编码未命中，或文件缺失、损坏时，仍返回可用的其他名称，并提示未找到名称的记录数及维护原因，不编造名称。
+
+## 在 AR51 的维护电脑刷新
+
+以下命令均在 AM516 项目根目录运行。刷新脚本只能由本地维护执行，不能由飞书消息触发。
+
+使用已配置只读金蝶凭证的现有客户端，全量分页查询两个字段：
+
+```bash
+PYTHONDONTWRITEBYTECODE=1 python3 capabilities/am516-delivery-prediction/scripts/refresh_material_names.py \
+  --kingdee-client /path/to/TH51_Knowledge_System/91_Scripts/kingdee_openapi/kingdee_query.py
+```
+
+其中路径替换为维护电脑上实际的已批准客户端路径。凭证仍由该客户端通过其已有的环境变量、钥匙串或本地凭证文件读取，不复制到 AM516 仓库。
+
+也可以从已经核验的全量物料参考缓存导出，无需联网：
+
+```bash
+python3 capabilities/am516-delivery-prediction/scripts/refresh_material_names.py \
+  --reference-json /path/to/material_master.json
+```
+
+导出会校验来源对象、全量未筛选范围、计数和内容哈希，并保留原快照时间。空结果、异常分页、无效记录或名称冲突不会覆盖旧文件。成功后输出行数、编码数和快照时间，不输出全量物料内容或凭证。
+
+## 同步到另一台电脑
+
+维护电脑提交代码、规则和 `data/material_names.json`，推送到 AM516 Git 仓库。另一台电脑在该项目拉取对应提交后，重启原有飞书机器人进程，使新代码和缓存生效。
+
+仅使用名称缓存回复时不需要 ERP 凭证；运行后台自动刷新时，需要在执行电脑配置自己的只读 ERP 凭证。新增物料或改名后，缓存文件变化会自动生效；修改机器人代码后仍需重启机器人。
+
+## 每两天后台自动刷新
+
+使用 macOS `launchd` 用户后台服务 `com.am516.material-name-refresh`，安装器根据执行电脑的实际路径生成配置。每小时启动一次轻量检查；距离上次成功查询满 48 小时后，才全量查询 ERP 的 `FNumber,FName`。首次安装立即查询一次；失败时保留缓存和上次成功时间，下次小时检查重试。电脑休眠或关机时不能保证精确到点，恢复运行后补查。
+
+自动刷新会加入新增编码、更新已有名称，并保留本轮未返回的旧编码。没有名称变化时不重写 Git 中的缓存，仅更新 Git 忽略的 `logs/material_names_refresh_state.json`；有变化时先完成分页、验证名称无冲突，再原子替换文件。进程锁避免并发刷新，日志只记录状态、数量、时间和安全错误摘要。
+
+### 另一台 Mac 的安装步骤
+
+1. 拉取包含缓存和脚本的提交，进入 AM516 项目根目录。
+2. 随项目提供的 `shared/runtime/kingdee_materials.py` 只允许查询物料编码和名称，不依赖 TH51 仓库。它读取本机环境变量、钥匙串中的 `TH51_KINGDEE_ACCT_ID` / `TH51_KINGDEE_USERNAME` / `TH51_KINGDEE_PASSWORD`，或本机配置文件 `configs/material_cache_erp.local`。
+3. 如果本机未配置只读凭证，从 `configs/material_cache_erp.local.example` 复制为 `configs/material_cache_erp.local`，在本机编辑账套、用户名和密码，并将权限设为 `600`。该文件匹配 `configs/*.local`，不会进入 Git；不要把实际凭证写入命令、日志或提交。
+4. 运行安装命令：
+
+```bash
+python3 shared/runtime/install_material_name_refresh.py install
+```
+
+安装后的路径随电脑生成，首次立即检查，以后每小时检查是否已满 48 小时。用户服务在该 macOS 用户登录期间运行；本机需允许该后台 Python 进程访问项目目录和使用已批准的网络。
+
+维护电脑也可复用已经配置好的现有只读客户端：
+
+```bash
+python3 shared/runtime/install_material_name_refresh.py install \
+  --kingdee-client /path/to/TH51_Knowledge_System/91_Scripts/kingdee_openapi/kingdee_query.py
+```
+
+查看、预览和卸载：
+
+```bash
+python3 shared/runtime/install_material_name_refresh.py status
+python3 shared/runtime/install_material_name_refresh.py preview
+python3 shared/runtime/install_material_name_refresh.py uninstall
+```
+
+手动执行一次到期检查（不会重复查询未到期的数据）：
+
+```bash
+python3 capabilities/am516-delivery-prediction/scripts/refresh_material_names.py --due
+```
+
+日志位于执行用户的 `~/Library/Logs/AM516_Feishu_Bot/material_name_refresh.log` 和 `material_name_refresh.error.log`。后台服务只更新本机缓存和状态；Git 提交、推送和其他电脑的拉取由既有同步流程执行，不在定时脚本内处理。

capabilities/am516-delivery-prediction/docs/production_quantity.md

@@ -0,0 +1,43 @@
+# 交期预测数量传递规则
+
+## 计算责任
+
+AR51 于 2026-10-15 检查上游 API 的交期预测逻辑后确认：接口已核算可用库存；并明确取消机器人本地的扣库存、95% 合格率折算和向上取 5 台倍数规则。此确认是本次修正依据，不代表已经实时验证服务端或运行电脑。
+
+```text
+D = 用户原始需求数量（正整数）
+API quantity = D
+```
+
+每个型号只通过受控脚本查询一次，原型号和各候选型号都传原始需求 D。机器人保留型号推荐、结果整理和记录能力；库存扣减及交期测算由上游 API 处理。
+
+## 示例与输出
+
+| 原始需求 | API 返回可用库存 | API 查询数量 | 库存展示 |
+|---:|---:|---:|---|
+| 12 | 10 | 12 | 部分覆盖，仍需补齐 2 台 |
+| 8 | 5 | 8 | 部分覆盖，仍需补齐 3 台 |
+| 24 | 0 | 24 | 需生产/采购 |
+| 24 | 24 | 24 | 完全覆盖 |
+| 24 | 30 | 24 | 完全覆盖 |
+
+库存覆盖按原始需求与该型号 API 返回的库存比较，仅用于展示。展示差额不回传 API，不作为机器人计算的排产数量。每个候选展示自己的库存，不能沿用原型号或其他候选的库存。
+
+库存缺失或无效时显示待复核，不视作 0；如果 API 已返回有效工作日交期和最终交货日期，机器人保留该交期并标明库存覆盖无法判断。缺失有效交期字段或请求失败时继续执行原有异常处理。
+
+飞书报告展示需求数量、API 返回交期和库存覆盖情况。按 AR51 于 2026-10-16 的显示要求，首次及续查回复均不显示“数量传递规则”“库存覆盖口径”和各型号下的“API查询数量”，也不输出本地净需求或排产测算数量。内部记录仍保存数量规则、原始需求及每个型号实际 API 查询数量，历史记录不改写。
+
+## 已取消的旧逻辑
+
+- 不再先读取库存、在本地扣减后发起第二次生产数量查询。
+- 不再按 95% 合格率放大需求或取 5 台倍数。
+- 不再根据“库存不足原需求、但足以覆盖本地排产量”拦截结果。
+- 不再要求接口为这个本地拦截增加“原始需求、排产数量”双参数。
+
+## 运行电脑同步验收
+
+运行电脑须拉取包含桥接代码、项目内 Skill、测试及说明的完整版本并重启交期回复机器人。本次无需新增依赖或配置。
+
+先执行 `python3 -m unittest discover -s tests -q`。随后在既有 API 审批及功能测试授权范围内验证 `eRob70F50I-FHM-18ET[V4] 12台`，核对实际发送的 quantity 为 12，机器人显示的工作日、最终日期及缺料摘要与同一 API 响应一致。API 库存以验证时实时返回值为准，不能假定仍为 10 台。
+
+分别报告本机验证、远端同步和运行电脑验证；本地模拟测试通过不代表服务端或运行电脑已验证，推送成功不代表运行端已生效。

capabilities/am516-delivery-prediction/docs/runbook.md

@@ -9,7 +9,7 @@
 5. If that domain-and-permission record is absent, stop before any outbound request; a general functional-test approval is insufficient by itself.
 6. Use model recommendation rules to generate the complete ordered internal candidate pool.
 7. Immediately add the Feishu `OnIt` (typing) reaction to the original message to acknowledge the valid request.
-8. Run the controlled script for the original model baseline and only the first four candidate models.
+8. For the baseline and first four candidate models, follow [original-demand quantity rules](production_quantity.md): call the controlled script once per model with the unchanged original demand. The upstream API accounts for stock. Do not deduct stock, apply local yield/batch adjustments, or query again with a local production quantity. Display each model's API inventory coverage against original demand only.
 9. Read `data.交期汇总.AM516工作日交期` and `data.交期汇总.AM516最终交货日期` directly. Require a nonnegative whole-number workday count and a valid `YYYY-MM-DD` date; missing or invalid fields make the result unavailable. Do not convert natural days or recalculate dates locally.
 10. Summarize the current batch with the API workdays and final delivery date, for example `AM516预测交期：8个工作日，最终交货日期：2026-10-08`; state how many candidates remain and invite the same sender to ask naturally for more when candidates remain. Newly appended records use the same form; historical records are unchanged.
 11. Append a non-secret record.
@@ -39,6 +39,28 @@ python3 capabilities/am516-delivery-prediction/scripts/delivery_prediction_query
 
 Do not run this check until AR51 confirms the functional test and creates a separate trusted approval record explicitly covering `api.zeroerr-agent.com`, the AM516 use case, and API-key use.
 
+## API Network Route
+
+The controlled query script defaults to `ZEROERR_AGENT_PROXY_MODE=direct` for
+`https://api.zeroerr-agent.com`. It uses a request-local opener with an empty
+proxy map, so this API does not inherit macOS or environment proxy settings.
+Feishu, Codex and other applications retain their existing network settings.
+The bridge launches the script for each query, so script updates take effect on
+the next query without restarting the bridge or clearing continuation context.
+
+On 2026-10-16, the proxy log for the 08:51 failed query showed a default-proxy
+connection timeout to this API. AR51-approved comparison queries with the fixed
+sample both succeeded later: system proxy 20.04 seconds, direct 4.79 seconds.
+These are individual observations, not an availability or latency guarantee.
+
+If another approved runtime requires a company proxy, configure
+`ZEROERR_AGENT_PROXY_MODE=system` in that runtime's environment (or the bridge's
+local `.env`, followed by a bridge restart). This selects the configured proxy
+route explicitly. It does not authorize a new destination. Invalid modes and
+direct requests outside the approved HTTPS hostname stop before the request.
+There is no automatic proxy fallback or extra API retry, and TLS certificate
+verification remains enabled.
+
 ## Normal TLS Trust Configuration
 
 The controlled query script injects `truststore` before making the request. Normal queries therefore keep TLS certificate verification enabled while using the macOS system trust store, including an AR51-approved company proxy root certificate installed and trusted in the system keychain.

capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py

@@ -17,7 +17,9 @@ truststore.inject_into_ssl()
 
 DEFAULT_BASE_URL = os.getenv("ZEROERR_AGENT_BASE_URL", "https://api.zeroerr-agent.com").rstrip("/")
 DEFAULT_ENDPOINT = "/api/SaleAgent/DeliveryPrediction/WithTransit"
-APPROVED_SKIP_TLS_HOSTNAME = "api.zeroerr-agent.com"
+APPROVED_API_HOSTNAME = "api.zeroerr-agent.com"
+APPROVED_SKIP_TLS_HOSTNAME = APPROVED_API_HOSTNAME
+PROXY_MODE_ENV = "ZEROERR_AGENT_PROXY_MODE"
 API_KEY_ENV = "ZEROERR_AGENT_API_KEY"
 PROJECT_ROOT = Path(__file__).resolve().parents[3]
 LOCAL_API_KEY_FILE = PROJECT_ROOT / "configs" / "delivery_api_key.local"
@@ -66,6 +68,23 @@ def get_api_key() -> str:
         return ""
 
 
+def build_api_opener(url: str, *, skip_tls_verify: bool = False) -> urllib.request.OpenerDirector:
+    """Keep the approved API independent of the desktop's changing proxy route."""
+    mode = os.getenv(PROXY_MODE_ENV, "direct").strip().lower()
+    if mode not in {"direct", "system"}:
+        raise ValueError("invalid API proxy mode: expected direct or system")
+    target = urllib.parse.urlsplit(url)
+    if mode == "direct" and (
+        target.scheme != "https" or target.hostname != APPROVED_API_HOSTNAME
+    ):
+        raise ValueError("direct API connection requires the approved HTTPS hostname")
+    # An explicit empty proxy map disables both environment and macOS proxies
+    # for this opener only. Never alter the process-wide urllib opener or env.
+    proxy = urllib.request.ProxyHandler({}) if mode == "direct" else urllib.request.ProxyHandler()
+    context = ssl._create_unverified_context() if skip_tls_verify else ssl.create_default_context()
+    return urllib.request.build_opener(proxy, urllib.request.HTTPSHandler(context=context))
+
+
 def main() -> int:
     args = parse_args()
     product = args.product.strip()
@@ -110,9 +129,13 @@ def main() -> int:
     }
 
     request = urllib.request.Request(url, headers=headers, method="GET")
-    context = ssl._create_unverified_context() if args.skip_tls_verify else None
     try:
-        with urllib.request.urlopen(request, timeout=30, context=context) as response:
+        opener = build_api_opener(url, skip_tls_verify=args.skip_tls_verify)
+    except ValueError as exc:
+        emit_error(str(exc))
+        return 2
+    try:
+        with opener.open(request, timeout=30) as response:
             raw_body = response.read().decode("utf-8")
     except urllib.error.HTTPError as exc:
         body = exc.read().decode("utf-8", errors="replace")

capabilities/am516-delivery-prediction/scripts/refresh_material_names.py

@@ -0,0 +1,210 @@
+"""Refresh the Git-synchronized code/name cache outside the Feishu runtime."""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[3]
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from shared.runtime.material_name_cache import (
+    MATERIAL_NAME_CACHE_FILE,
+    MaterialNameCacheError,
+    load_material_names,
+    material_names_from_records,
+    material_names_hash,
+)
+
+BUNDLED_CLIENT = PROJECT_ROOT / "shared/runtime/kingdee_materials.py"
+DEFAULT_STATE = PROJECT_ROOT / "logs/material_names_refresh_state.json"
+
+def read_reference(path: Path) -> tuple[list[dict[str, Any]], str]:
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict) or payload.get("form_id") != "BD_MATERIAL" or payload.get("filter") != "":
+        raise MaterialNameCacheError("来源必须为未筛选的 BD_MATERIAL 全量参考缓存。")
+    rows = payload.get("records")
+    if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows):
+        raise MaterialNameCacheError("来源缓存记录结构无效。")
+    digest = hashlib.sha256(
+        json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    if payload.get("record_count") != len(rows) or payload.get("content_sha256") != digest:
+        raise MaterialNameCacheError("来源缓存记录计数或完整性校验失败。")
+    snapshot_at = payload.get("generated_at")
+    if not isinstance(snapshot_at, str):
+        raise MaterialNameCacheError("来源缓存缺少快照时间。")
+    datetime.fromisoformat(snapshot_at)
+    return rows, snapshot_at
+
+
+def query_materials(client_path: Path) -> tuple[list[dict[str, Any]], str]:
+    """Use the existing approved read-only Kingdee client, querying only two fields."""
+    spec = importlib.util.spec_from_file_location("am516_kingdee_query", client_path)
+    if spec is None or spec.loader is None:
+        raise MaterialNameCacheError("无法加载金蝶只读查询器。")
+    client = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(client)
+    # Resolve the client's existing local credential fallback without copying it.
+    client.LOCAL_CREDENTIALS_PATH = client_path.resolve().parents[2] / client.LOCAL_CREDENTIALS_PATH
+    snapshot_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
+    opener = client.build_opener()
+    client.login(opener)
+    rows: list[dict[str, Any]] = []
+    seen_pages: set[str] = set()
+    start_row = 0
+    while True:
+        args = SimpleNamespace(
+            command="bill-query", no_audit_log=True, form_id="BD_MATERIAL",
+            fields="FNumber,FName", filter="", order="FNumber ASC,FMaterialId ASC",
+            limit=2000, start_row=start_row,
+        )
+        page = client.execute_list_query(opener, args)
+        if not isinstance(page, list) or any(not isinstance(row, list) or len(row) != 2 for row in page):
+            raise MaterialNameCacheError("金蝶返回的物料分页结构无效，保留原缓存。")
+        if not page:
+            break
+        page_hash = hashlib.sha256(json.dumps(page, ensure_ascii=False).encode()).hexdigest()
+        if page_hash in seen_pages:
+            raise MaterialNameCacheError("金蝶分页重复，无法确认全量完成，保留原缓存。")
+        seen_pages.add(page_hash)
+        rows.extend({"FNumber": row[0], "FName": row[1]} for row in page)
+        start_row += len(page)
+        if len(page) < args.limit:
+            break
+        print(f"已读取物料主数据：{len(rows)}行", flush=True)
+    return rows, snapshot_at
+
+
+def write_json_atomic(output: Path, payload: dict[str, Any]) -> None:
+    output.parent.mkdir(parents=True, exist_ok=True)
+    temp_path = None
+    try:
+        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as handle:
+            temp_path = Path(handle.name)
+            json.dump(payload, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp_path, output)
+    finally:
+        if temp_path is not None:
+            temp_path.unlink(missing_ok=True)
+
+
+def write_cache(
+    records: list[dict[str, Any]], snapshot_at: str, output: Path, *, merge_existing: bool = False,
+) -> dict[str, Any]:
+    materials = material_names_from_records(records)
+    retained_count = 0
+    if merge_existing and output.exists():
+        previous = load_material_names(output)
+        retained_count = len(previous.keys() - materials.keys())
+        materials = dict(sorted({**previous, **materials}.items()))
+        if materials == previous:
+            return json.loads(output.read_text(encoding="utf-8"))
+    payload = {
+        "schema_version": 1,
+        "source": {
+            "system": "Kingdee K3 Cloud", "form_id": "BD_MATERIAL",
+            "field_keys": ["FNumber", "FName"], "filter": "",
+            "snapshot_at": snapshot_at, "record_count": len(records),
+        },
+        "generated_at": datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds"),
+        "material_count": len(materials),
+        "content_sha256": material_names_hash(materials),
+        "materials": materials,
+    }
+    if retained_count:
+        payload["source"]["retained_missing_count"] = retained_count
+    write_json_atomic(output, payload)
+    return payload
+
+
+def refresh_once(
+    *, output: Path, state_file: Path, due_only: bool, interval_hours: int,
+    client_path: Path, reference_json: Path | None = None, current_time: datetime | None = None,
+) -> dict[str, Any]:
+    now = current_time or datetime.now(timezone(timedelta(hours=8)))
+    if due_only and state_file.exists():
+        try:
+            state = json.loads(state_file.read_text(encoding="utf-8"))
+            last = datetime.fromisoformat(state["last_success_at"])
+            if last.tzinfo is None:
+                raise ValueError
+            next_due = last + timedelta(hours=interval_hours)
+        except (OSError, ValueError, KeyError, TypeError):
+            next_due = now  # Rebuild invalid local state only after a successful refresh.
+        if now < next_due:
+            return {"ok": True, "status": "NOT_DUE", "next_due_at": next_due.isoformat()}
+    previous = load_material_names(output) if output.exists() else {}
+    records, snapshot_at = read_reference(reference_json) if reference_json else query_materials(client_path)
+    incoming = material_names_from_records(records)
+    added_count = len(incoming.keys() - previous.keys())
+    renamed_count = sum(previous[code] != name for code, name in incoming.items() if code in previous)
+    payload = write_cache(records, snapshot_at, output, merge_existing=due_only)
+    completed_at = current_time or datetime.now(timezone(timedelta(hours=8)))
+    result = {
+        "ok": True, "status": "UPDATED" if added_count or renamed_count else "UNCHANGED",
+        "source_record_count": len(records), "material_count": payload["material_count"],
+        "added_count": added_count, "renamed_count": renamed_count,
+        "retained_missing_count": len(previous.keys() - incoming.keys()) if due_only else 0,
+        "snapshot_at": snapshot_at, "output": str(output),
+    }
+    if due_only:
+        write_json_atomic(state_file, {
+            "last_success_at": completed_at.isoformat(timespec="seconds"),
+            "last_result": result,
+        })
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    source = parser.add_mutually_exclusive_group()
+    source.add_argument("--reference-json", type=Path, help="Existing verified full material-master cache")
+    source.add_argument("--kingdee-client", type=Path, help="Approved kingdee_query.py for a live read-only refresh")
+    parser.add_argument("--output", type=Path, default=MATERIAL_NAME_CACHE_FILE)
+    parser.add_argument("--due", action="store_true", help="Query only after the successful refresh interval")
+    parser.add_argument("--interval-hours", type=int, default=48)
+    parser.add_argument("--state-file", type=Path, default=DEFAULT_STATE)
+    args = parser.parse_args()
+    if args.interval_hours <= 0 or (args.due and args.reference_json):
+        parser.error("--interval-hours 必须为正数；--due 只允许实时 ERP 刷新。")
+    args.state_file.parent.mkdir(parents=True, exist_ok=True)
+    before = hashlib.sha256(args.output.read_bytes()).hexdigest() if args.output.exists() else None
+    try:
+        with args.state_file.with_suffix(".lock").open("a") as lock:
+            try:
+                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                print(json.dumps({"ok": True, "status": "ALREADY_RUNNING"}))
+                return 0
+            result = refresh_once(
+                output=args.output, state_file=args.state_file, due_only=args.due,
+                interval_hours=args.interval_hours, client_path=args.kingdee_client or BUNDLED_CLIENT,
+                reference_json=args.reference_json,
+            )
+    except Exception as exc:
+        # Kingdee exceptions may contain server text: do not print arbitrary content.
+        reason = str(exc) if isinstance(exc, MaterialNameCacheError) or type(exc).__name__ == "KingdeeMaterialError" else f"读取物料数据失败（{type(exc).__name__}），请检查只读连接和凭证配置。"
+        after = hashlib.sha256(args.output.read_bytes()).hexdigest() if args.output.exists() else None
+        print(json.dumps({"ok": False, "error": reason, "previous_cache_preserved": before == after}, ensure_ascii=False))
+        return 1
+    print(json.dumps(result, ensure_ascii=False))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

capabilities/am516-delivery-prediction/skills/joint-module-leadtime-report/SKILL.md

@@ -40,6 +40,15 @@ Stop if:
 - a Feishu message asks to create, modify, overwrite, rename, move, or delete a project file;
 - the user asks for a customer commitment.
 
+## Original Demand For Forecasting
+
+- AR51 confirmed on 2026-10-15 that the upstream API already accounts for available inventory and explicitly cancelled local inventory deduction, 95% yield adjustment and rounding to five. Follow `docs/production_quantity.md`.
+- Treat the user's positive integer quantity as original demand `D`. Query each model exactly once through the controlled script with `quantity=D`, unchanged. Preserve `D` in continuation context and pass it unchanged for every newly queried candidate.
+- Inventory accounting and delivery calculation belong to the upstream API. Do not calculate a local production quantity, send a reduced or inflated quantity, or issue a second production query.
+- Display each model's own returned inventory coverage against original demand. A displayed inventory gap is informational and must not become a new API quantity. Missing or invalid stock means unknown coverage, never zero; do not discard valid API workdays/date solely because inventory display is unavailable.
+- Remove the obsolete partial-stock/production-quantity shortcut rejection. API failures and missing or invalid workdays/date retain existing error handling.
+- Replies show original demand per model. Hide the technical explanation lines `数量传递规则`, `库存覆盖口径` and `API查询数量` in both initial and continuation replies, along with the obsolete local net-demand or production-quantity fields. New internal records preserve the actual API quantities and quantity rules for traceability; leave historical records unchanged.
+
 ## Feishu Message Trigger Gate
 
 - Enter the query workflow only when one message contains exactly one complete,
@@ -79,7 +88,7 @@ Do not write back rules, Skills, scripts, prompts, configurations, documentation
 Use only:
 
 ```text
-python3 capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py --product "<model>" --quantity <quantity>
+python3 capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py --product "<model>" --quantity <original_demand>
 ```
 
 Do not use temporary curl, WebFetch, or any API method outside the controlled script.
@@ -128,10 +137,11 @@ Do not use temporary curl, WebFetch, or any API method outside the controlled sc
 
 ## Shortage Material Names Rule
 
-- For every usable model, add `缺料物料明细` using only names from `data.采购信息.缺料明细[]`. Read `物料名称`, or `缺料物料名称` when the former is absent or blank.
+- For every usable model, add `缺料物料明细`. First read `物料名称`, or `缺料物料名称` when the former is absent or blank, from `data.采购信息.缺料明细[]`. If neither name is usable, look up that shortage record's `缺料物料` code in the Git-synchronized `data/material_names.json` cache.
 - Preserve the API order, omit duplicate names, and display names separated by `；`. Include both reducer and non-reducer shortage names.
-- Do not substitute material codes, infer names from code prefixes, or list materials from the complete BOM. This added detail contains no codes, quantities, or specifications.
-- If some shortage records lack a nonblank string name, retain the available names and state how many records did not return a name. If all names are missing, state `N项缺料记录未返回物料名称，需人工复核`.
+- Do not display material codes in the names list, infer names from code prefixes, or list materials from the complete BOM. The lookup only enriches API-returned shortages; it never identifies shortages itself. This added detail contains no codes, quantities, or specifications.
+- Reuse the local dictionary in memory and reload it when the file's timestamp, size or inode changes. Never call Kingdee or refresh/write the cache while replying to Feishu messages. Background cache updates take effect on the next lookup without restarting the bot.
+- If a name is absent from both the API and local cache, retain the available names and state how many records could not be resolved. Report missing or invalid cache files explicitly. Ask for cache maintenance outside Feishu; do not invent names or treat unresolved records as no shortage.
 - If the shortage list is empty, state `API未返回缺料记录`; missing shortage data or an invalid BOM retains the existing unavailable-data reminder. Do not infer that missing data means no shortage.
 
 ## Reducer Shortage Detail Rule
@@ -182,7 +192,7 @@ Use this field order for a successful Feishu report:
 - 主导因素：{main factor}
 
 1. {model}（{candidate type and reducer brand}）
-- 数量：{quantity}台
+- 需求数量：{input quantity}台
 - AM516预测交期：{API workdays}个工作日，最终交货日期：{API final delivery date}
 - 可用库存/覆盖情况：{summary-level inventory and coverage only}
 - 主导因素：{main factor}
@@ -206,6 +216,12 @@ Use Feishu rich-text `post` output. Bold only:
 - each numbered candidate heading;
 - `不可用候选：`.
 
+Also bold the `X个工作日` value and the final delivery date (`YYYY-MM-DD`)
+in `最快候选交期` and each model's `AM516预测交期`, for both initial and
+continuation replies. Use native Feishu text-element `style: ["bold"]`, keeping
+the field labels and separating punctuation in normal weight. Dates in the
+report title or data timestamp are not included in this emphasis.
+
 Do not repeat a separate human-review line under each candidate. The global judgment level and fixed disclaimer carry the human-review boundary.
 
 If the current candidate-batch report exceeds the Feishu rich-text size limit, split it into multiple ordered `post` replies without dropping any candidate queried in that batch. Use the original report title on the first part and append `（续）` to continuation titles.

configs/env.example

@@ -3,6 +3,10 @@
 
 ZEROERR_AGENT_API_KEY=replace_with_approved_key
 ZEROERR_AGENT_BASE_URL=https://api.zeroerr-agent.com
+# Only this controlled API script uses the setting; Feishu and desktop apps keep their own routes.
+# direct (default): approved HTTPS API host without system/environment proxies.
+# system: use the runtime's configured proxies if an approved network requires them.
+ZEROERR_AGENT_PROXY_MODE=direct
 AM516_RECORD_FILE=capabilities/am516-delivery-prediction/records/delivery_prediction_records.md
 
 # Feishu custom-app credentials. Keep real values in the local .env file only.

configs/material_cache_erp.local.example

@@ -0,0 +1,5 @@
+{
+  "acct_id": "",
+  "username": "",
+  "password": ""
+}

shared/runtime/README.md

@@ -60,7 +60,13 @@ continuation wording is flexible: `还有吗`, `有没有别的`, `再推荐几
 bridge still requires both continuation intent and valid context, so unrelated
 phrases such as `more interesting` or `继续查询英语例句` do not trigger the old query.
 
-The bridge invokes only the existing controlled delivery-prediction script. It
+The bridge invokes only the existing controlled delivery-prediction script.
+That script defaults to direct HTTPS for `api.zeroerr-agent.com` without
+inheriting desktop proxy settings, while keeping certificate verification on.
+`ZEROERR_AGENT_PROXY_MODE=system` explicitly restores the configured proxy route
+for approved networks that require it. This setting affects only the API script,
+not the Feishu connection. See the capability runbook for route diagnostics.
+It
 generates the full Cartesian
 set of candidates allowed by the packaged Skill: brake upgrades, encoder
 upgrades, sensor upgrades, low-temperature grease upgrades, and DZ/LF reducer

shared/runtime/feishu_bot_bridge.py

@@ -22,6 +22,11 @@ truststore.inject_into_ssl()
 
 import lark_oapi as lark
 
+if __package__:
+    from .material_name_cache import MaterialNameCacheError, load_material_names
+else:
+    from material_name_cache import MaterialNameCacheError, load_material_names
+
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 QUERY_SCRIPT = (
@@ -175,6 +180,12 @@ class CandidateUnavailableError(RuntimeError):
     pass
 
 
+def validate_demand_quantity(quantity: int) -> None:
+    """Reject invalid demand before calling the controlled API."""
+    if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0:
+        raise InputError("需求数量必须为正整数。")
+
+
 _candidate_continuations: dict[str, CandidateContinuation] = {}
 _candidate_continuation_lock = threading.Lock()
 
@@ -599,6 +610,8 @@ def shortage_material_names_line(shortages: list[Any]) -> str:
     """Show only names from shortage records, never codes or other raw fields."""
     names: list[str] = []
     missing_names = 0
+    cached_names: dict[str, str] | None = None
+    cache_error: str | None = None
     for item in shortages:
         name = None
         if isinstance(item, dict):
@@ -607,6 +620,16 @@ def shortage_material_names_line(shortages: list[Any]) -> str:
                 if isinstance(value, str) and value.strip():
                     name = " ".join(value.split())
                     break
+            if name is None:
+                code = item.get("缺料物料")
+                if isinstance(code, str) and code.strip():
+                    if cached_names is None and cache_error is None:
+                        try:
+                            cached_names = load_material_names()
+                        except MaterialNameCacheError as exc:
+                            cache_error = str(exc)
+                    if cached_names is not None:
+                        name = cached_names.get(code.strip())
         if name is None:
             missing_names += 1
         elif name not in names:
@@ -614,7 +637,12 @@ def shortage_material_names_line(shortages: list[Any]) -> str:
 
     text = "；".join(names)
     if missing_names:
-        missing_text = f"{missing_names}项缺料记录未返回物料名称，需人工复核"
+        missing_text = f"{missing_names}项缺料记录未找到物料名称"
+        missing_text += (
+            f"（{cache_error}），需补齐缓存后复核"
+            if cache_error is not None
+            else "，请核对物料编码并更新本地名称缓存后复核"
+        )
         text = f"{text}（另有{missing_text}）" if text else missing_text
     return f"- 缺料物料明细：{text}"
 
@@ -683,7 +711,14 @@ def lead_time_details(
         if isinstance(inventory_info, dict)
         else None
     )
-    if isinstance(available_inventory, (int, float)):
+    # Inventory is display-only; invalid values do not invalidate API dates.
+    if (
+        isinstance(available_inventory, bool)
+        or not isinstance(available_inventory, (int, float))
+        or (isinstance(available_inventory, float) and not available_inventory.is_integer())
+    ):
+        available_inventory = None
+    if available_inventory is not None:
         if available_inventory >= requested_quantity:
             inventory_text = f"整机可用库存{available_inventory:g}台（完全覆盖）"
             full_coverage: bool | None = True
@@ -801,7 +836,7 @@ def candidate_report_lines(candidate: dict[str, Any], quantity: int) -> list[str
     details = candidate["details"]
     return [
         f"{candidate['index']}. {candidate['model']}（{candidate['type']}）",
-        f"- 数量：{quantity}台",
+        f"- 需求数量：{quantity}台",
         f"- {details['prediction']}",
         f"- 可用库存/覆盖情况：{details['inventory']}",
         f"- 主导因素：{details['main_factor']}",
@@ -872,6 +907,13 @@ def render_record(
         "",
         f"**输入型号**：{request.model}",
         f"**输入数量**：{request.quantity}台",
+        "**数量传递规则**：原始需求原样传入API；库存及交期测算由API处理。",
+        "**库存覆盖口径**：按原始需求数量展示API返回的库存覆盖情况。",
+        "**各型号API查询数量**：",
+        *[
+            f"- {item['model']}：最终API查询数量{item['details']['api_quantity']}台"
+            for item in candidates
+        ],
         f"**生成时间**：{generation_time:%Y-%m-%d %H:%M:%S %z}",
         f"**规则源实际路径**：`{RULE_SOURCE_DISPLAY}`",
         f"**接口依据**：`{API_SOURCE_DISPLAY}`",
@@ -944,9 +986,19 @@ def write_record(
         file.write(record)
 
 
+def query_model_for_demand(request: RequestInput) -> dict[str, Any]:
+    """Pass original demand once; the upstream API accounts for inventory."""
+    validate_demand_quantity(request.quantity)
+    payload = run_controlled_query(request)
+    details = lead_time_details(payload, request.quantity)
+    details['api_quantity'] = request.quantity
+    return details
+
+
 def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
     if candidate_offset < 0:
         raise InputError("候选批次位置无效，请重新发起型号和数量查询。")
+    validate_demand_quantity(request.quantity)
 
     all_candidate_specs = generate_candidate_specs(request.model)
     candidate_specs = all_candidate_specs[
@@ -960,18 +1012,14 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
     baseline: dict[str, Any] | None = None
     if is_initial_batch:
         try:
-            baseline_payload = run_controlled_query(request)
+            baseline = query_model_for_demand(request)
         except CandidateUnavailableError as exc:
             raise RuntimeError(f"原型号未生成候选结果。\n原因：{exc}") from exc
-        baseline = lead_time_details(
-            baseline_payload,
-            request.quantity,
-        )
         if not baseline["has_lead_time"]:
             raise RuntimeError(
-                "原型号接口未返回有效工作日交期和最终交货日期，未生成候选结果。\n"
+                "原型号未生成可用交期结果，未生成候选结果。\n"
                 f"原因：{baseline['unavailable_reason']}\n"
-                "处理建议：请接口维护人员核对上述字段及服务端计算结果后重试。"
+                "处理建议：请接口维护人员核对上述原因、库存口径及服务端计算结果后重试。"
             )
         candidates.append(
             {
@@ -990,15 +1038,14 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
 
     def query_candidate(spec: dict[str, str]) -> tuple[dict[str, str], dict[str, Any] | None]:
         try:
-            payload = run_controlled_query(RequestInput(spec["model"], request.quantity))
+            details = query_model_for_demand(RequestInput(spec["model"], request.quantity))
         except CandidateUnavailableError as exc:
             return {**spec, "unavailable_reason": str(exc)}, None
-        details = lead_time_details(
-            payload,
-            request.quantity,
-        )
         if not details["has_lead_time"]:
-            return {**spec, "unavailable_reason": details["unavailable_reason"]}, None
+            return {
+                **spec,
+                "unavailable_reason": details["unavailable_reason"],
+            }, None
         return spec, details
 
     if candidate_specs:
@@ -1107,6 +1154,19 @@ def message_text(data: lark.im.v1.P2ImMessageReceiveV1) -> str | None:
 
 
 def post_paragraph(line: str) -> list[dict[str, Any]]:
+    lead_time = re.fullmatch(
+        r"(- (?:最快候选交期|AM516预测交期)：)(\d+个工作日)"
+        r"(，最终交货日期：)(\d{4}-\d{2}-\d{2})",
+        line,
+    )
+    if lead_time:
+        label, workdays, date_label, delivery_date = lead_time.groups()
+        return [
+            {"tag": "text", "text": label},
+            {"tag": "text", "text": workdays, "style": ["bold"]},
+            {"tag": "text", "text": date_label},
+            {"tag": "text", "text": delivery_date, "style": ["bold"]},
+        ]
     element: dict[str, Any] = {"tag": "text", "text": line or " "}
     if (
         line in {"总体判断：", "本批次判断：", "不可用候选："}

shared/runtime/install_material_name_refresh.py

@@ -0,0 +1,67 @@
+"""Install the native macOS background material-name refresher."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import plistlib
+import subprocess
+import sys
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+LABEL = "com.am516.material-name-refresh"
+
+
+def launch_agent(project: Path, python: Path, log_dir: Path, client: Path | None = None) -> dict:
+    command = [str(python), str(project / "capabilities/am516-delivery-prediction/scripts/refresh_material_names.py"), "--due", "--interval-hours", "48"]
+    if client is not None:
+        command.extend(["--kingdee-client", str(client)])
+    return {
+        "Label": LABEL, "ProgramArguments": command, "WorkingDirectory": str(project),
+        "RunAtLoad": True, "StartInterval": 3600, "ProcessType": "Background",
+        "EnvironmentVariables": {"PYTHONDONTWRITEBYTECODE": "1", "PYTHONUNBUFFERED": "1"},
+        "StandardOutPath": str(log_dir / "material_name_refresh.log"),
+        "StandardErrorPath": str(log_dir / "material_name_refresh.error.log"),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=("install", "status", "uninstall", "preview"))
+    parser.add_argument("--python", type=Path, default=Path(sys.executable))
+    parser.add_argument("--kingdee-client", type=Path)
+    args = parser.parse_args()
+    if args.kingdee_client is not None:
+        args.kingdee_client = args.kingdee_client.resolve()
+    destination = Path.home() / "Library/LaunchAgents" / f"{LABEL}.plist"
+    logs = Path.home() / "Library/Logs/AM516_Feishu_Bot"
+    domain = f"gui/{os.getuid()}"
+    service = f"{domain}/{LABEL}"
+    if args.action == "preview":
+        print(plistlib.dumps(launch_agent(PROJECT_ROOT, args.python.resolve(), logs, args.kingdee_client)).decode())
+        return 0
+    if args.action == "status":
+        return subprocess.run(["launchctl", "print", service]).returncode
+    if args.action in ("install", "uninstall"):
+        exists = subprocess.run(["launchctl", "print", service], capture_output=True).returncode == 0
+        if exists:
+            subprocess.run(["launchctl", "bootout", service], check=True)
+    if args.action == "uninstall":
+        destination.unlink(missing_ok=True)
+        print("已移除物料名称后台刷新服务。")
+        return 0
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    logs.mkdir(parents=True, exist_ok=True)
+    (PROJECT_ROOT / "logs").mkdir(exist_ok=True)
+    with destination.open("wb") as handle:
+        plistlib.dump(launch_agent(PROJECT_ROOT, args.python.resolve(), logs, args.kingdee_client), handle)
+    subprocess.run(["launchctl", "bootstrap", domain, str(destination)], check=True)
+    print("已安装物料名称后台刷新：每小时检查，成功刷新后间隔48小时再次查询 ERP。")
+    print(f"服务：{service}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

shared/runtime/kingdee_materials.py

@@ -0,0 +1,110 @@
+"""Standalone Kingdee reader restricted to material codes and names."""
+
+from __future__ import annotations
+
+import getpass
+import http.cookiejar
+import json
+import os
+import subprocess
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+BASE_URL = "https://kingdeeapi.zeroerr.com"
+LOCAL_CREDENTIALS_PATH = Path(__file__).resolve().parents[2] / "configs/material_cache_erp.local"
+LOGIN_PATH = "/Kingdee.BOS.WebApi.ServicesStub.AuthService.ValidateUser.common.kdsvc"
+QUERY_PATH = "/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.ExecuteBillQuery.common.kdsvc"
+
+
+class KingdeeMaterialError(RuntimeError):
+    pass
+
+
+def credentials() -> dict[str, str]:
+    result = {key: os.environ.get(env, "") for key, env in (
+        ("acct_id", "KINGDEE_ACCT_ID"), ("username", "KINGDEE_USERNAME"), ("password", "KINGDEE_PASSWORD"),
+    )}
+    if any(result.values()):
+        if not all(result.values()):
+            raise KingdeeMaterialError("金蝶运行环境凭证配置不完整。")
+        return result
+    if LOCAL_CREDENTIALS_PATH.exists():
+        try:
+            data = json.loads(LOCAL_CREDENTIALS_PATH.read_text(encoding="utf-8"))
+            result = {key: data.get(key) for key in result}
+            if not all(isinstance(value, str) and value for value in result.values()):
+                raise ValueError
+            return result
+        except (OSError, ValueError, AttributeError) as exc:
+            raise KingdeeMaterialError("本机金蝶凭证文件配置无效。") from exc
+    account = getpass.getuser()
+    for key, service in (
+        ("acct_id", "TH51_KINGDEE_ACCT_ID"), ("username", "TH51_KINGDEE_USERNAME"), ("password", "TH51_KINGDEE_PASSWORD"),
+    ):
+        try:
+            completed = subprocess.run(
+                ["security", "find-generic-password", "-a", account, "-s", service, "-w"],
+                capture_output=True, text=True, timeout=15,
+            )
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise KingdeeMaterialError("本机金蝶只读凭证不可用。") from exc
+        if completed.returncode:
+            raise KingdeeMaterialError("本机未配置金蝶只读凭证。")
+        result[key] = completed.stdout.rstrip("\r\n")
+    # macOS may return non-ASCII Keychain text as UTF-8 hexadecimal bytes.
+    for key, value in result.items():
+        try:
+            decoded = bytes.fromhex(value).decode("utf-8")
+        except (ValueError, UnicodeError):
+            continue
+        if any(ord(char) > 127 for char in decoded):
+            result[key] = decoded
+    return result
+
+
+def build_opener() -> urllib.request.OpenerDirector:
+    return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
+
+
+def post(opener: Any, path: str, payload: Any) -> Any:
+    request = urllib.request.Request(
+        BASE_URL + path, data=json.dumps(payload, ensure_ascii=True).encode("utf-8"),
+        headers={"Content-Type": "application/json; charset=utf-8"}, method="POST",
+    )
+    try:
+        with opener.open(request, timeout=30) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        raise KingdeeMaterialError(f"金蝶只读接口返回 HTTP {exc.code}。") from exc
+    except (urllib.error.URLError, OSError) as exc:
+        raise KingdeeMaterialError("金蝶只读接口网络或 TLS 连接失败。") from exc
+    except (ValueError, UnicodeError) as exc:
+        raise KingdeeMaterialError("金蝶只读接口响应格式无效。") from exc
+
+
+def login(opener: Any) -> None:
+    values = credentials()
+    result = post(opener, LOGIN_PATH, {
+        "acctID": values["acct_id"], "username": values["username"],
+        "password": values["password"], "lcid": 2052,
+    })
+    if not isinstance(result, dict) or not (result.get("IsSuccess") is True or result.get("LoginResultType") == 1):
+        raise KingdeeMaterialError("金蝶登录失败，请检查本机只读账号和凭证。")
+
+
+def execute_list_query(opener: Any, args: Any) -> list[Any]:
+    if args.form_id != "BD_MATERIAL" or args.fields != "FNumber,FName" or args.filter:
+        raise KingdeeMaterialError("物料名称客户端只允许未筛选的 BD_MATERIAL 编码和名称查询。")
+    parameters = {
+        "FormId": "BD_MATERIAL", "FieldKeys": "FNumber,FName", "FilterString": "",
+        "OrderString": "FNumber ASC,FMaterialId ASC", "TopRowCount": 0,
+        "StartRow": args.start_row, "Limit": args.limit,
+    }
+    for payload in (parameters, {"data": parameters}):
+        result = post(opener, QUERY_PATH, payload)
+        if isinstance(result, list) and all(isinstance(row, list) and len(row) == 2 for row in result):
+            return result
+    raise KingdeeMaterialError("金蝶物料名称查询未返回有效数据，请核对只读查询权限。")

shared/runtime/material_name_cache.py

@@ -0,0 +1,77 @@
+"""Portable, read-only material-code/name lookup for delivery replies."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+
+MATERIAL_NAME_CACHE_FILE = (
+    Path(__file__).resolve().parents[2]
+    / "capabilities/am516-delivery-prediction/data/material_names.json"
+)
+
+
+class MaterialNameCacheError(ValueError):
+    """A cache error that contains no credentials or raw ERP records."""
+
+
+def material_names_hash(materials: dict[str, str]) -> str:
+    content = json.dumps(materials, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(content.encode("utf-8")).hexdigest()
+
+
+def material_names_from_records(records: list[dict[str, Any]]) -> dict[str, str]:
+    if not records:
+        raise MaterialNameCacheError("物料查询结果为空，保留原缓存。")
+    materials: dict[str, str] = {}
+    for row in records:
+        code, name = row.get("FNumber"), row.get("FName")
+        if not isinstance(code, str) or not code.strip() or not isinstance(name, str) or not name.strip():
+            raise MaterialNameCacheError("物料记录缺少有效编码或名称，保留原缓存。")
+        code, name = code.strip(), " ".join(name.split())
+        if code in materials and materials[code] != name:
+            raise MaterialNameCacheError("同一物料编码存在不同名称，需核对后再刷新缓存。")
+        materials[code] = name
+    return dict(sorted(materials.items()))
+
+
+def load_material_names(path: Path = MATERIAL_NAME_CACHE_FILE) -> dict[str, str]:
+    """Reuse memory until the local file changes; never access ERP."""
+    try:
+        stat = path.stat()
+    except FileNotFoundError as exc:
+        raise MaterialNameCacheError("本地物料名称缓存文件缺失") from exc
+    except OSError as exc:
+        raise MaterialNameCacheError("本地物料名称缓存文件不可读取") from exc
+    return _load_material_names(path, (stat.st_mtime_ns, stat.st_size, stat.st_ino))
+
+
+@lru_cache(maxsize=4)
+def _load_material_names(path: Path, version: tuple[int, int, int]) -> dict[str, str]:
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise MaterialNameCacheError("本地物料名称缓存文件缺失") from exc
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise MaterialNameCacheError("本地物料名称缓存文件不可读取或格式错误") from exc
+    if not isinstance(payload, dict) or payload.get("schema_version") != 1:
+        raise MaterialNameCacheError("本地物料名称缓存版本或结构无效")
+    materials = payload.get("materials")
+    if not isinstance(materials, dict) or not materials:
+        raise MaterialNameCacheError("本地物料名称缓存为空或结构无效")
+    if any(
+        not isinstance(code, str) or not code.strip()
+        or not isinstance(name, str) or not name.strip()
+        for code, name in materials.items()
+    ):
+        raise MaterialNameCacheError("本地物料名称缓存包含无效编码或名称")
+    if payload.get("material_count") != len(materials) or payload.get("content_sha256") != material_names_hash(materials):
+        raise MaterialNameCacheError("本地物料名称缓存计数或完整性校验失败")
+    return materials
+
+
+load_material_names.cache_clear = _load_material_names.cache_clear

tests/test_delivery_prediction_network.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import contextlib
+import http.client
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import ssl
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+import urllib.parse
+import urllib.request
+
+
+SCRIPT = (
+    Path(__file__).resolve().parents[1]
+    / "capabilities/am516-delivery-prediction/scripts/delivery_prediction_query_template.py"
+)
+spec = importlib.util.spec_from_file_location("delivery_query", SCRIPT)
+query = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(query)
+
+API_URL = "https://api.zeroerr-agent.com/api/SaleAgent/DeliveryPrediction/WithTransit"
+
+
+class DeliveryNetworkTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.env = mock.patch.dict(os.environ, {}, clear=True)
+        self.env.start()
+        self.addCleanup(self.env.stop)
+        self.proxies = mock.patch.object(
+            urllib.request, "getproxies", return_value={"https": "http://127.0.0.1:7897"}
+        )
+        self.getproxies = self.proxies.start()
+        self.addCleanup(self.proxies.stop)
+        connection_patch = mock.patch.object(http.client, "HTTPSConnection")
+        self.connection = connection_patch.start()
+        self.addCleanup(connection_patch.stop)
+        response = self.connection.return_value.getresponse.return_value
+        response.status = response.code = 200
+        response.reason = "OK"
+        response.headers = {}
+        response.read.return_value = b'{"success": true, "data": {}}'
+        response.__enter__.return_value = response
+
+    def test_default_direct_ignores_system_and_environment_proxy(self) -> None:
+        os.environ["https_proxy"] = "http://127.0.0.1:9999"
+        before = dict(os.environ)
+        global_opener = urllib.request._opener
+
+        query.build_api_opener(API_URL).open(API_URL, timeout=30)
+
+        self.assertEqual(self.connection.call_args.args[0], "api.zeroerr-agent.com")
+        self.connection.return_value.set_tunnel.assert_not_called()
+        self.getproxies.assert_not_called()
+        self.assertEqual(dict(os.environ), before)
+        self.assertIs(urllib.request._opener, global_opener)
+        context = self.connection.call_args.kwargs["context"]
+        self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
+        self.assertTrue(context.check_hostname)
+
+    def test_system_mode_explicitly_uses_configured_proxy(self) -> None:
+        os.environ[query.PROXY_MODE_ENV] = "system"
+        with mock.patch.object(urllib.request, "proxy_bypass", return_value=False):
+            query.build_api_opener(API_URL).open(API_URL, timeout=30)
+
+        self.assertEqual(self.connection.call_args.args[0], "127.0.0.1:7897")
+        self.assertEqual(
+            self.connection.return_value.set_tunnel.call_args.args[0], "api.zeroerr-agent.com"
+        )
+        self.assertEqual(self.connection.call_args.kwargs["context"].verify_mode, ssl.CERT_REQUIRED)
+
+    def test_invalid_proxy_mode_stops_before_network_and_does_not_echo_value(self) -> None:
+        os.environ[query.PROXY_MODE_ENV] = "SECRET_MARKER"
+        with self.assertRaises(ValueError) as caught:
+            query.build_api_opener(API_URL)
+        self.assertNotIn("SECRET_MARKER", str(caught.exception))
+        self.connection.assert_not_called()
+
+    def test_direct_route_is_limited_to_approved_https_host(self) -> None:
+        for url in ("https://example.invalid/", "http://api.zeroerr-agent.com/",
+                    "https://api.zeroerr-agent.com.example.invalid/"):
+            with self.subTest(url=url), self.assertRaises(ValueError):
+                query.build_api_opener(url)
+        self.connection.assert_not_called()
+
+    def run_main(self, *, skip_tls_verify: bool = False) -> tuple[int, dict]:
+        output = io.StringIO()
+        with (
+            mock.patch.object(query, "parse_args", return_value=SimpleNamespace(
+                product="eRob110H160I-FHM-18ET[V6]", quantity=10,
+                skip_tls_verify=skip_tls_verify,
+            )),
+            mock.patch.object(query, "get_api_key", return_value="test-key-not-a-real-secret"),
+            mock.patch.object(query, "DEFAULT_BASE_URL", "https://api.zeroerr-agent.com"),
+            contextlib.redirect_stdout(output),
+        ):
+            code = query.main()
+        return code, json.loads(output.getvalue())
+
+    def test_controlled_script_sends_original_quantity_once_over_verified_direct_route(self) -> None:
+        code, payload = self.run_main()
+        self.assertEqual(code, 0)
+        self.assertTrue(payload["success"])
+        self.assertEqual(self.connection.call_args.args[0], "api.zeroerr-agent.com")
+        request = self.connection.return_value.request
+        request.assert_called_once()
+        method, path = request.call_args.args[:2]
+        self.assertEqual(method, "GET")
+        self.assertEqual(urllib.parse.parse_qs(urllib.parse.urlsplit(path).query), {
+            "product_specification": ["eRob110H160I-FHM-18ET[V6]"], "quantity": ["10"],
+        })
+
+    def test_certificate_failure_does_not_retry_or_fall_back_to_proxy(self) -> None:
+        self.connection.return_value.request.side_effect = ssl.SSLCertVerificationError(
+            1, "CERTIFICATE_VERIFY_FAILED"
+        )
+        code, payload = self.run_main()
+        self.assertEqual(code, 1)
+        self.assertEqual(payload["error"], "TLS certificate verification failed")
+        self.connection.assert_called_once()
+        self.connection.return_value.set_tunnel.assert_not_called()
+        self.assertEqual(self.connection.call_args.kwargs["context"].verify_mode, ssl.CERT_REQUIRED)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_feishu_bot_bridge.py

@@ -404,6 +404,128 @@ class TriggerGateTests(unittest.TestCase):
         self.assertNotIn(key, bridge._candidate_continuations)
 
 
+class OriginalDemandQueryTests(unittest.TestCase):
+    def test_invalid_demand_stops_before_api_or_recording(self) -> None:
+        for value in (0, -1, True, 5.0, "5"):
+            with (
+                self.subTest(value=value),
+                mock.patch.object(bridge, "run_controlled_query") as query,
+                mock.patch.object(bridge, "write_record") as record,
+            ):
+                with self.assertRaises(bridge.InputError):
+                    bridge.build_report(bridge.RequestInput("eRob70F50I-FHM-18ET[V4]", value))
+            query.assert_not_called()
+            record.assert_not_called()
+
+    def test_partial_full_and_zero_stock_do_not_change_api_quantity(self) -> None:
+        for demand, stock in ((12, 10), (8, 5), (24, 6), (24, 0), (24, 24), (24, 30), (24, -2), (5, 1)):
+            payload = successful_payload()
+            payload['data']['库存信息']['可用库存(扣除了待产数量)'] = stock
+            request = bridge.RequestInput('eRob70F50I-FHM-18ET[V4]', demand)
+            with (
+                self.subTest(demand=demand, stock=stock),
+                mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query,
+            ):
+                details = bridge.query_model_for_demand(request)
+            query.assert_called_once_with(request)
+            self.assertTrue(details['has_lead_time'])
+            self.assertEqual(details['api_quantity'], demand)
+
+    def test_screenshot_regression_passes_twelve_to_controlled_script(self) -> None:
+        request = bridge.RequestInput('eRob70F50I-FHM-18ET[V4]', 12)
+        payload = successful_payload()
+        payload['data']['库存信息']['可用库存(扣除了待产数量)'] = 10
+        payload['data']['交期汇总'].update(AM516工作日交期=8, AM516最终交货日期='2026-10-27', AM516主导因素='生产')
+        completed = SimpleNamespace(stdout=bridge.json.dumps(payload), stderr='', returncode=0)
+        with (
+            mock.patch.object(bridge.subprocess, 'run', return_value=completed) as run,
+            mock.patch.object(bridge, 'write_record') as record,
+        ):
+            report = bridge.build_report(request)
+        self.assertEqual(run.call_count, 5)
+        for call in run.call_args_list:
+            command = call.args[0]
+            self.assertEqual(command[command.index('--quantity') + 1], '12')
+        self.assertIn('整机可用库存10台（部分覆盖，仍需补齐2台）', report)
+        self.assertIn('8个工作日，最终交货日期：2026-10-27', report)
+        self.assertNotIn('误走现货分支', report)
+        self.assertNotIn('排产测算数量', report)
+        recorded = bridge.render_record(*record.call_args.args)
+        self.assertIn('**输入数量**：12台', recorded)
+        self.assertEqual(recorded.count('最终API查询数量12台'), 5)
+        self.assertNotIn('净需求', recorded)
+
+    def test_initial_and_continuation_preserve_original_demand(self) -> None:
+        request = bridge.RequestInput('eRob70H50I-FS-18CN[V5]', 24)
+        for offset, model_count in ((0, 5), (bridge.CANDIDATE_BATCH_SIZE, 4)):
+            with (
+                self.subTest(offset=offset),
+                mock.patch.object(bridge, 'run_controlled_query', return_value=successful_payload()) as query,
+                mock.patch.object(bridge, 'write_record'),
+            ):
+                bridge.build_report(request, candidate_offset=offset)
+            self.assertEqual(query.call_count, model_count)
+            self.assertEqual(len({call.args[0].model for call in query.call_args_list}), model_count)
+            self.assertTrue(all(call.args[0].quantity == 24 for call in query.call_args_list))
+
+    def test_each_candidate_displays_its_own_stock(self) -> None:
+        request = bridge.RequestInput('eRob70H50I-FS-18CN[V5]', 24)
+        models = [request.model, *[s['model'] for s in bridge.generate_candidate_specs(request.model)[:4]]]
+        stocks = dict(zip(models, [6, 0, 24, 20, 30]))
+        def query(r):
+            payload = successful_payload()
+            payload['data']['库存信息']['可用库存(扣除了待产数量)'] = stocks[r.model]
+            return payload
+        with (
+            mock.patch.object(bridge, 'run_controlled_query', side_effect=query) as run,
+            mock.patch.object(bridge, 'write_record') as record,
+        ):
+            bridge.build_report(request)
+        self.assertEqual(run.call_count, len(models))
+        for candidate in record.call_args.args[1]:
+            self.assertEqual(candidate['details']['available_inventory'], stocks[candidate['model']])
+            self.assertEqual(candidate['details']['api_quantity'], 24)
+            self.assertEqual(candidate['details']['full_coverage'], stocks[candidate['model']] >= 24)
+
+    def test_missing_or_invalid_stock_is_unknown_without_overriding_api_forecast(self) -> None:
+        for stock in (None, True, '6', 1.5, float('nan'), float('inf')):
+            payload = successful_payload()
+            payload['data']['库存信息']['可用库存(扣除了待产数量)'] = stock
+            with (
+                self.subTest(stock=stock),
+                mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query,
+            ):
+                details = bridge.query_model_for_demand(bridge.RequestInput('model', 24))
+            self.assertTrue(details['has_lead_time'])
+            self.assertIsNone(details['available_inventory'])
+            self.assertIsNone(details['full_coverage'])
+            self.assertIn('需人工复核', details['inventory'])
+            query.assert_called_once()
+
+    def test_forecast_preserves_api_date_and_materials(self) -> None:
+        payload = successful_payload()
+        payload['data']['库存信息']['可用库存(扣除了待产数量)'] = 10
+        payload['data']['交期汇总'].update(AM516工作日交期=8, AM516最终交货日期='2026-10-27')
+        payload['data']['采购信息']['缺料明细'] = [{'物料名称': '测算物料'}]
+        with mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query:
+            details = bridge.query_model_for_demand(bridge.RequestInput('model', 12))
+        query.assert_called_once()
+        self.assertEqual(details['prediction_value'], '8个工作日，最终交货日期：2026-10-27')
+        self.assertIn('- 缺料物料明细：测算物料', details['evidence_lines'])
+
+    def test_missing_api_date_still_stops_before_candidates_and_record(self) -> None:
+        payload = successful_payload()
+        del payload['data']['交期汇总']['AM516最终交货日期']
+        with (
+            mock.patch.object(bridge, 'run_controlled_query', return_value=payload) as query,
+            mock.patch.object(bridge, 'write_record') as record,
+        ):
+            with self.assertRaisesRegex(RuntimeError, '原型号未生成可用交期结果'):
+                bridge.build_report(bridge.RequestInput('eRob70F50I-FHM-18ET[V4]', 12))
+        query.assert_called_once()
+        record.assert_not_called()
+
+
 class CandidateBatchTests(unittest.TestCase):
     def setUp(self) -> None:
         self.request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
@@ -569,6 +691,58 @@ class UnavailableReasonTests(unittest.TestCase):
 
 
 class ShortageMaterialNamesTests(unittest.TestCase):
+    def setUp(self) -> None:
+        patcher = mock.patch.object(bridge, "load_material_names", return_value={})
+        self.load_names = patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def test_code_only_shortages_use_local_names_and_preserve_api_name_priority(self) -> None:
+        self.load_names.return_value = {"20.40.00001": "编码器", "20.50.00002": "轴承"}
+        lines = bridge.shortage_material_names_line([
+            {"缺料物料": "20.40.00001"},
+            {"缺料物料": " 20.50.00002 "},
+            {"缺料物料": "20.40.00001"},
+            {"缺料物料": "20.50.00002", "物料名称": "API更新后的轴承名称"},
+        ])
+        self.assertEqual(lines, "- 缺料物料明细：编码器；轴承；API更新后的轴承名称")
+        self.load_names.assert_called_once_with()
+        self.assertNotIn("20.40", lines)
+        self.assertNotIn("20.50", lines)
+
+    def test_missing_or_invalid_cache_keeps_available_names_and_explains_failure(self) -> None:
+        for reason in ("本地物料名称缓存文件缺失", "本地物料名称缓存计数或完整性校验失败"):
+            with self.subTest(reason=reason):
+                self.load_names.side_effect = bridge.MaterialNameCacheError(reason)
+                line = bridge.shortage_material_names_line([
+                    {"物料名称": "编码器"},
+                    {"缺料物料": "20.50.00002"},
+                    {"缺料物料": "20.50.00003"},
+                ])
+                self.assertIn("编码器", line)
+                self.assertIn("2项缺料记录未找到物料名称", line)
+                self.assertIn(reason, line)
+                self.assertNotIn("20.50.", line)
+
+    def test_initial_and_continuation_code_only_shortages_resolve_offline(self) -> None:
+        request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
+        self.load_names.return_value = {"20.40.00001": "编码器"}
+
+        def query(query_request: bridge.RequestInput) -> dict[str, object]:
+            payload = successful_payload()
+            payload["data"]["采购信息"]["缺料明细"] = [{"缺料物料": "20.40.00001"}]
+            return payload
+
+        for offset in (0, bridge.CANDIDATE_BATCH_SIZE):
+            with (
+                self.subTest(offset=offset),
+                mock.patch.object(bridge, "run_controlled_query", side_effect=query) as run_query,
+                mock.patch.object(bridge, "write_record"),
+            ):
+                report = bridge.build_report(request, candidate_offset=offset)
+            self.assertEqual(report.count("- 缺料物料明细：编码器"), run_query.call_count)
+            self.assertNotIn("未找到物料名称", report)
+            self.assertNotIn("20.40.00001", report)
+
     def test_all_shortage_names_are_shown_without_non_reducer_raw_fields(self) -> None:
         payload = successful_payload()
         payload["data"]["采购信息"]["缺料明细"] = [
@@ -601,7 +775,7 @@ class ShortageMaterialNamesTests(unittest.TestCase):
 
                 lines = bridge.lead_time_details(payload, requested_quantity=10)["evidence_lines"]
 
-                expected = "4项缺料记录未返回物料名称，需人工复核"
+                expected = "4项缺料记录未找到物料名称，请核对物料编码并更新本地名称缓存后复核"
                 if valid_name:
                     expected = f"轴承（另有{expected}）"
                 self.assertEqual(lines, [f"- 缺料物料明细：{expected}"])
@@ -759,7 +933,7 @@ class ApiWorkdayOutputTests(unittest.TestCase):
             mock.patch.object(bridge, "run_controlled_query", return_value=payload),
             mock.patch.object(bridge, "write_record") as write_record,
         ):
-            with self.assertRaisesRegex(RuntimeError, "工作日交期和最终交货日期"):
+            with self.assertRaisesRegex(RuntimeError, "缺少“AM516工作日交期”字段"):
                 bridge.build_report(request)
         write_record.assert_not_called()
 

tests/test_material_name_cache.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from datetime import datetime, timedelta
+from unittest import mock
+
+from shared.runtime import material_name_cache as cache
+from shared.runtime import install_material_name_refresh as installer
+from shared.runtime import kingdee_materials as kingdee
+
+
+script_path = Path(__file__).resolve().parents[1] / "capabilities/am516-delivery-prediction/scripts/refresh_material_names.py"
+spec = importlib.util.spec_from_file_location("refresh_material_names", script_path)
+refresh = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(refresh)
+
+
+class MaterialNameCacheTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.path = Path(self.temp.name) / "another-checkout/data/material_names.json"
+        cache.load_material_names.cache_clear()
+        self.addCleanup(cache.load_material_names.cache_clear)
+
+    def test_generated_cache_is_portable_and_contains_only_name_lookup_data(self) -> None:
+        rows = [
+            {"FNumber": "20.40.00001", "FName": "编码器", "FMaterialId": 123, "FSpecification": "PRIVATE_MARKER"},
+            {"FNumber": "20.40.00001", "FName": "编码器", "FMaterialId": 456},
+            {"FNumber": "20.50.00002", "FName": "轴承"},
+        ]
+        payload = refresh.write_cache(rows, "2026-10-09T10:13:22+08:00", self.path)
+        self.assertEqual(cache.load_material_names(self.path), {"20.40.00001": "编码器", "20.50.00002": "轴承"})
+        self.assertEqual(payload["source"]["record_count"], 3)
+        self.assertEqual(payload["material_count"], 2)
+        content = self.path.read_text()
+        for unwanted in ("FMaterialId", "FSpecification", "PRIVATE_MARKER", "/Users/"):
+            self.assertNotIn(unwanted, content)
+
+    def test_empty_invalid_or_conflicting_refresh_preserves_existing_cache(self) -> None:
+        refresh.write_cache([{"FNumber": "A", "FName": "原物料"}], "2026-10-09T10:00:00+08:00", self.path)
+        previous = self.path.read_bytes()
+        for rows in (
+            [], [{"FNumber": "A", "FName": ""}],
+            [{"FNumber": "A", "FName": "名称1"}, {"FNumber": "A", "FName": "名称2"}],
+        ):
+            with self.subTest(rows=rows):
+                with self.assertRaises(cache.MaterialNameCacheError):
+                    refresh.write_cache(rows, "2026-10-09T11:00:00+08:00", self.path)
+                self.assertEqual(self.path.read_bytes(), previous)
+
+    def test_cache_updates_are_seen_without_restart_and_integrity_is_checked(self) -> None:
+        refresh.write_cache([{"FNumber": "A", "FName": "编码器"}], "2026-10-09T10:00:00+08:00", self.path)
+        self.assertEqual(cache.load_material_names(self.path)["A"], "编码器")
+        refresh.write_cache([{"FNumber": "A", "FName": "新编码器"}, {"FNumber": "B", "FName": "轴承"}], "2026-10-11T10:00:00+08:00", self.path)
+        self.assertEqual(cache.load_material_names(self.path), {"A": "新编码器", "B": "轴承"})
+        payload = json.loads(self.path.read_text())
+        payload["materials"]["A"] = "篡改名称"
+        self.path.write_text(json.dumps(payload))
+        with self.assertRaisesRegex(cache.MaterialNameCacheError, "完整性校验失败"):
+            cache.load_material_names(self.path)
+
+    def test_reference_export_rejects_filtered_or_incomplete_source(self) -> None:
+        rows = [{"FNumber": "A", "FName": "编码器"}]
+        payload = {
+            "form_id": "BD_MATERIAL", "filter": "", "records": rows,
+            "record_count": len(rows), "generated_at": "2026-10-09T10:00:00+08:00",
+            "content_sha256": hashlib.sha256(json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest(),
+        }
+        source = Path(self.temp.name) / "source.json"
+        source.write_text(json.dumps(payload))
+        self.assertEqual(refresh.read_reference(source)[0], rows)
+        for field, value in (("filter", "FForbidStatus='A'"), ("record_count", 2), ("content_sha256", "wrong")):
+            with self.subTest(field=field):
+                source.write_text(json.dumps({**payload, field: value}))
+                with self.assertRaises(cache.MaterialNameCacheError):
+                    refresh.read_reference(source)
+
+    def test_live_refresh_pages_two_fields_only_and_rejects_repeated_pages(self) -> None:
+        full_page = [[f"code-{i:04}", "编码器"] for i in range(2000)]
+        query = mock.Mock(side_effect=[full_page, [["last-code", "轴承"]]])
+        client = SimpleNamespace(
+            LOCAL_CREDENTIALS_PATH=Path("credentials.local.json"),
+            build_opener=mock.Mock(return_value=object()), login=mock.Mock(),
+            execute_list_query=query,
+        )
+        fake_spec = SimpleNamespace(loader=SimpleNamespace(exec_module=mock.Mock()))
+        with (
+            mock.patch.object(refresh.importlib.util, "spec_from_file_location", return_value=fake_spec),
+            mock.patch.object(refresh.importlib.util, "module_from_spec", return_value=client),
+            mock.patch("builtins.print"),
+        ):
+            rows, _ = refresh.query_materials(Path("/approved/source/91_Scripts/kingdee_openapi/kingdee_query.py"))
+            self.assertEqual(len(rows), 2001)
+            self.assertEqual([call.args[1].start_row for call in query.call_args_list], [0, 2000])
+            for call in query.call_args_list:
+                args = call.args[1]
+                self.assertEqual(args.fields, "FNumber,FName")
+                self.assertEqual(args.filter, "")
+                self.assertTrue(args.no_audit_log)
+            query.side_effect = [full_page, full_page]
+            with self.assertRaisesRegex(cache.MaterialNameCacheError, "分页重复"):
+                refresh.query_materials(Path("/approved/source/91_Scripts/kingdee_openapi/kingdee_query.py"))
+
+    def test_due_refresh_queries_after_48_hours_and_merges_without_deleting_names(self) -> None:
+        start = datetime.fromisoformat("2026-10-09T10:00:00+08:00")
+        state = Path(self.temp.name) / "state.json"
+        rows = [{"FNumber": "A", "FName": "编码器"}, {"FNumber": "B", "FName": "轴承"}]
+        args = dict(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py"))
+        with mock.patch.object(refresh, "query_materials", return_value=(rows, start.isoformat())) as query:
+            first = refresh.refresh_once(**args, current_time=start)
+            self.assertEqual(first["added_count"], 2)
+            result = refresh.refresh_once(**args, current_time=start + timedelta(hours=47))
+            self.assertEqual(result["status"], "NOT_DUE")
+            query.assert_called_once()
+            query.return_value = ([{"FNumber": "A", "FName": "新编码器"}, {"FNumber": "C", "FName": "减速器"}], start.isoformat())
+            result = refresh.refresh_once(**args, current_time=start + timedelta(hours=48))
+            self.assertEqual(result["added_count"], 1)
+            self.assertEqual(result["renamed_count"], 1)
+            self.assertEqual(result["retained_missing_count"], 1)
+            self.assertEqual(cache.load_material_names(self.path), {"A": "新编码器", "B": "轴承", "C": "减速器"})
+
+    def test_unchanged_refresh_does_not_rewrite_git_cache_but_advances_due_state(self) -> None:
+        start = datetime.fromisoformat("2026-10-09T10:00:00+08:00")
+        state = Path(self.temp.name) / "state.json"
+        rows = [{"FNumber": "A", "FName": "编码器"}]
+        refresh.write_cache(rows, start.isoformat(), self.path)
+        before = self.path.read_bytes()
+        with mock.patch.object(refresh, "query_materials", return_value=(rows, start.isoformat())):
+            result = refresh.refresh_once(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py"), current_time=start)
+        self.assertEqual(result["status"], "UNCHANGED")
+        self.assertEqual(self.path.read_bytes(), before)
+        self.assertEqual(json.loads(state.read_text())["last_success_at"], start.isoformat())
+
+    def test_failed_due_refresh_preserves_cache_and_last_success(self) -> None:
+        start = datetime.fromisoformat("2026-10-09T10:00:00+08:00")
+        state = Path(self.temp.name) / "state.json"
+        refresh.write_cache([{"FNumber": "A", "FName": "编码器"}], start.isoformat(), self.path)
+        state.write_text(json.dumps({"last_success_at": start.isoformat()}))
+        previous = self.path.read_bytes(), state.read_bytes()
+        with mock.patch.object(refresh, "query_materials", side_effect=cache.MaterialNameCacheError("网络失败")):
+            with self.assertRaises(cache.MaterialNameCacheError):
+                refresh.refresh_once(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py"), current_time=start + timedelta(hours=48))
+        self.assertEqual((self.path.read_bytes(), state.read_bytes()), previous)
+
+    def test_launch_agent_uses_host_paths_due_gate_and_no_credentials(self) -> None:
+        job = installer.launch_agent(Path("/another-machine/AM516"), Path("/usr/bin/python3"), Path("/user/logs"))
+        self.assertEqual(job["StartInterval"], 3600)
+        self.assertTrue(job["RunAtLoad"])
+        self.assertIn("--due", job["ProgramArguments"])
+        self.assertIn("48", job["ProgramArguments"])
+        self.assertIn("/another-machine/AM516", job["WorkingDirectory"])
+        self.assertNotIn("KINGDEE_PASSWORD", job["EnvironmentVariables"])
+        self.assertNotIn("codex", " ".join(job["ProgramArguments"]).lower())
+
+    def test_bundled_kingdee_client_rejects_other_forms_or_fields_before_network(self) -> None:
+        for form, fields, filter_string in (("STK_INVENTORY", "FNumber,FName", ""), ("BD_MATERIAL", "FNumber,FName,FSpecification", ""), ("BD_MATERIAL", "FNumber,FName", "FNumber='A'")):
+            with self.subTest(form=form, fields=fields), mock.patch.object(kingdee, "post") as request:
+                with self.assertRaises(kingdee.KingdeeMaterialError):
+                    kingdee.execute_list_query(object(), SimpleNamespace(form_id=form, fields=fields, filter=filter_string))
+                request.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()