feat: account for defects and five-unit batches in delivery forecasts

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,6 +14,7 @@ Expected:
 
 - model parses successfully;
 - quantity is accepted;
+- the original demand of 10 remains visible in the reply and record, while the baseline and every recommended candidate call the API with 15 after dividing by 95% yield and rounding to a batch of five;
 - 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;
 - no temporary TLS-bypass parameter is required for the normal happy path;
@@ -26,8 +27,11 @@ Expected:
 - Feishu output uses rich-text `post`; the report title, overall-judgment heading, numbered candidate headings, and unavailable-candidate heading are bold;
 - 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 +84,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;
+- the continuation retains original demand 10 and again uses forecast quantity 15; it never compounds the adjustment to 20;
 - 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 `不可用候选`;

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,36 @@
+# 交期预测数量口径
+
+用户输入为原始需求数量。交期回复机器人在查询原型号、候选型号及后续批次前，统一按以下规则计算 API 输入数量：
+
+```text
+排产测算数量 = 5 × ceil(原始需求数量 ÷ 0.95 ÷ 5)
+```
+
+固定不良率为 5%，对应良率 95%。先除以良率，再向上取到 5 的倍数。代码使用整数运算，避免恰好落在倍数边界时因浮点误差多加一个批次。
+
+| 原始需求 | 除以95%良率 | 排产测算数量 / API quantity |
+|---:|---:|---:|
+| 5 | 5.263… | 10 |
+| 10 | 10.526… | 15 |
+| 19 | 20 | 20 |
+| 20 | 21.052… | 25 |
+| 25 | 26.315… | 30 |
+| 95 | 100 | 100 |
+
+回复示例中的数量部分：
+
+```text
+输入需求：eRob70H120I-BHM-18CT[V5]，5台
+排产测算数量：10台（不良率5%，按5台的倍数向上取整）
+库存覆盖口径：按排产测算数量判断
+
+1. eRob70H120I-BHM-18CT[V5]（原型号，DZ）
+- 需求数量：5台
+- 排产测算数量：10台
+```
+
+实际交期继续读取 API 返回的工作日交期和最终交货日期。库存覆盖也以本次 API 使用的排产测算数量判断，例如原始需求 5、API 数量 10、库存 7 时，显示部分覆盖，仍需补齐 3 台。
+
+对话上下文和本地记录保留原始需求；记录元数据另列排产测算数量和计算规则。继续推荐仍从原始需求重新计算一次，不把 10 台再次放大为 15 台。低层受控脚本的 `--quantity` 参数就是最终发送给 API 的数量，不在脚本内重复换算。
+
+本机通过相应测试后，运行电脑需要同步这次代码和规则变更，并重启交期回复机器人；另一个后台物料名称刷新服务不因数量规则变化而需要重新安装。

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.
 
+## Production Quantity For Forecasting
+
+- Treat the user's quantity as original demand. Before querying any model, divide by the fixed 95% yield (5% defect rate), then round upward to a multiple of five: `production_quantity = 5 × ceil(original_demand / 0.95 / 5)`.
+- For example, demand 5 gives `5 / 0.95 = 5.263...`, then a forecast quantity of 10. Demand 10 gives 15; demand 19 gives exactly 20; demand 25 gives 30; demand 95 gives exactly 100.
+- Use exact integer arithmetic to avoid rounding an exact batch boundary upward due to floating-point error. Do not replace division by yield with multiplication by 1.05, and do not round to five before applying the yield.
+- Send this computed quantity as the controlled script's `--quantity` / API `quantity` for the baseline, all candidates and continuation batches. Calculate from the original demand each time; never adjust an already adjusted quantity again. The low-level query script passes the supplied API quantity unchanged.
+- Preserve original demand in the conversation context, reply and record. Display both `需求数量` and `排产测算数量`; state the fixed defect rate and batching rule. Newly appended records include the forecast quantity as metadata without rewriting historical records or changing the candidate table columns.
+- Judge inventory coverage against the same production quantity used by the API and label this quantity scope in the reply.
+
 ## 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 <production_quantity>
 ```
 
 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
@@ -169,6 +179,8 @@ Use this field order for a successful Feishu report:
 关节模组交期预估报告 | {generation time}
 
 输入需求：{input model}，{input quantity}台
+排产测算数量：{production quantity}台（不良率5%，按5台的倍数向上取整）
+库存覆盖口径：按排产测算数量判断
 数据时间戳：{API data timestamp or generation time}
 状态：试运行版输出 / 内部候选
 判断等级：AM516算法预测参考，需人工复核
@@ -182,7 +194,8 @@ Use this field order for a successful Feishu report:
 - 主导因素：{main factor}
 
 1. {model}（{candidate type and reducer brand}）
-- 数量：{quantity}台
+- 需求数量：{input quantity}台
+- 排产测算数量：{production quantity}台
 - AM516预测交期：{API workdays}个工作日，最终交货日期：{API final delivery date}
 - 可用库存/覆盖情况：{summary-level inventory and coverage only}
 - 主导因素：{main factor}

configs/material_cache_erp.local.example

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

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 = (
@@ -78,6 +83,8 @@ ALLOWED_VERSIONS = {
     "170H": {"V3"},
 }
 CANDIDATE_BATCH_SIZE = 4
+DEFECT_RATE_PERCENT = 5
+PRODUCTION_BATCH_SIZE = 5
 CANDIDATE_CONTINUATION_TTL_SECONDS = 6 * 60 * 60
 RECEIVED_REACTION_TYPE = "OnIt"
 
@@ -175,6 +182,15 @@ class CandidateUnavailableError(RuntimeError):
     pass
 
 
+def production_quantity_for_demand(quantity: int) -> int:
+    """Cover 5% defects, then round up to a batch of five without float error."""
+    if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity <= 0:
+        raise InputError("需求数量必须为正整数。")
+    numerator = quantity * 100
+    denominator = (100 - DEFECT_RATE_PERCENT) * PRODUCTION_BATCH_SIZE
+    return ((numerator + denominator - 1) // denominator) * PRODUCTION_BATCH_SIZE
+
+
 _candidate_continuations: dict[str, CandidateContinuation] = {}
 _candidate_continuation_lock = threading.Lock()
 
@@ -599,6 +615,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 +625,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 +642,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}"
 
@@ -797,11 +830,12 @@ def overall_judgment_lines(candidates: list[dict[str, Any]]) -> list[str]:
     ]
 
 
-def candidate_report_lines(candidate: dict[str, Any], quantity: int) -> list[str]:
+def candidate_report_lines(candidate: dict[str, Any], quantity: int, production_quantity: int) -> list[str]:
     details = candidate["details"]
     return [
         f"{candidate['index']}. {candidate['model']}（{candidate['type']}）",
-        f"- 数量：{quantity}台",
+        f"- 需求数量：{quantity}台",
+        f"- 排产测算数量：{production_quantity}台",
         f"- {details['prediction']}",
         f"- 可用库存/覆盖情况：{details['inventory']}",
         f"- 主导因素：{details['main_factor']}",
@@ -872,6 +906,9 @@ def render_record(
         "",
         f"**输入型号**：{request.model}",
         f"**输入数量**：{request.quantity}台",
+        f"**排产测算数量（API查询数量）**：{production_quantity_for_demand(request.quantity)}台",
+        "**数量计算规则**：原始需求÷95%良率，再按5台的倍数向上取整。",
+        "**库存覆盖口径**：按排产测算数量判断。",
         f"**生成时间**：{generation_time:%Y-%m-%d %H:%M:%S %z}",
         f"**规则源实际路径**：`{RULE_SOURCE_DISPLAY}`",
         f"**接口依据**：`{API_SOURCE_DISPLAY}`",
@@ -947,6 +984,7 @@ def write_record(
 def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
     if candidate_offset < 0:
         raise InputError("候选批次位置无效，请重新发起型号和数量查询。")
+    production_quantity = production_quantity_for_demand(request.quantity)
 
     all_candidate_specs = generate_candidate_specs(request.model)
     candidate_specs = all_candidate_specs[
@@ -960,12 +998,12 @@ 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_payload = run_controlled_query(RequestInput(request.model, production_quantity))
         except CandidateUnavailableError as exc:
             raise RuntimeError(f"原型号未生成候选结果。\n原因：{exc}") from exc
         baseline = lead_time_details(
             baseline_payload,
-            request.quantity,
+            production_quantity,
         )
         if not baseline["has_lead_time"]:
             raise RuntimeError(
@@ -990,12 +1028,12 @@ 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))
+            payload = run_controlled_query(RequestInput(spec["model"], production_quantity))
         except CandidateUnavailableError as exc:
             return {**spec, "unavailable_reason": str(exc)}, None
         details = lead_time_details(
             payload,
-            request.quantity,
+            production_quantity,
         )
         if not details["has_lead_time"]:
             return {**spec, "unavailable_reason": details["unavailable_reason"]}, None
@@ -1066,6 +1104,8 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
         ),
         "",
         f"输入需求：{request.model}，{request.quantity}台",
+        f"排产测算数量：{production_quantity}台（不良率{DEFECT_RATE_PERCENT}%，按{PRODUCTION_BATCH_SIZE}台的倍数向上取整）",
+        "库存覆盖口径：按排产测算数量判断",
         f"数据时间戳：{data_timestamp}",
         "状态：试运行版输出 / 内部候选",
         "判断等级：AM516算法预测参考，需人工复核",
@@ -1076,7 +1116,7 @@ def build_report(request: RequestInput, candidate_offset: int = 0) -> str:
         "",
     ]
     for candidate_item in candidates:
-        report_lines.extend(candidate_report_lines(candidate_item, request.quantity))
+        report_lines.extend(candidate_report_lines(candidate_item, request.quantity, production_quantity))
     report_lines.append("不可用候选：")
     if unavailable:
         for item in unavailable:

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_feishu_bot_bridge.py

@@ -404,6 +404,66 @@ class TriggerGateTests(unittest.TestCase):
         self.assertNotIn(key, bridge._candidate_continuations)
 
 
+class ProductionQuantityTests(unittest.TestCase):
+    def test_defect_allowance_precedes_batch_rounding_at_exact_boundaries(self) -> None:
+        for demand, expected in ((1, 5), (4, 5), (5, 10), (10, 15), (19, 20), (20, 25), (25, 30), (95, 100), (190, 200)):
+            with self.subTest(demand=demand):
+                self.assertEqual(bridge.production_quantity_for_demand(demand), expected)
+        # Integer arithmetic must also cover quantities beyond float precision.
+        demand = 10**18 + 19
+        planned = bridge.production_quantity_for_demand(demand)
+        self.assertEqual(planned % 5, 0)
+        self.assertGreaterEqual(planned * 95, demand * 100)
+        self.assertLess((planned - 5) * 95, demand * 100)
+
+    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("eRob70H50I-FS-18CN[V5]", value))
+            query.assert_not_called()
+            record.assert_not_called()
+
+    def test_initial_and_follow_up_send_adjusted_quantity_once_to_controlled_script(self) -> None:
+        request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=5)
+        completed = SimpleNamespace(stdout=bridge.json.dumps(successful_payload()), stderr="", returncode=0)
+        for offset, expected_calls in ((0, 5), (bridge.CANDIDATE_BATCH_SIZE, 4)):
+            with (
+                self.subTest(offset=offset),
+                mock.patch.object(bridge.subprocess, "run", return_value=completed) as run,
+                mock.patch.object(bridge, "write_record") as record,
+            ):
+                report = bridge.build_report(request, candidate_offset=offset)
+            self.assertEqual(run.call_count, expected_calls)
+            for call in run.call_args_list:
+                command = call.args[0]
+                self.assertEqual(command[command.index("--quantity") + 1], "10")
+            self.assertIn(f"输入需求：{request.model}，5台", report)
+            self.assertEqual(report.count("- 需求数量：5台"), expected_calls)
+            self.assertEqual(report.count("- 排产测算数量：10台"), expected_calls)
+            self.assertEqual(record.call_args.args[0].quantity, 5)
+            recorded = bridge.render_record(*record.call_args.args)
+            self.assertIn("**输入数量**：5台", recorded)
+            self.assertIn("**排产测算数量（API查询数量）**：10台", recorded)
+
+    def test_stock_coverage_uses_same_quantity_as_api_forecast(self) -> None:
+        payload = successful_payload()
+        payload["data"]["库存信息"]["可用库存(扣除了待产数量)"] = 7
+        with (
+            mock.patch.object(bridge, "run_controlled_query", return_value=payload),
+            mock.patch.object(bridge, "write_record"),
+        ):
+            report = bridge.build_report(bridge.RequestInput("eRob70H50I-BHM-18CTC[V5]", 5))
+        self.assertIn("库存覆盖口径：按排产测算数量判断", report)
+        self.assertIn("部分覆盖，仍需补齐3台", report)
+        self.assertIn("是否完全覆盖数量：否", report)
+        self.assertIn("现货候选：无", report)
+
+
 class CandidateBatchTests(unittest.TestCase):
     def setUp(self) -> None:
         self.request = bridge.RequestInput(model="eRob70H50I-FS-18CN[V5]", quantity=10)
@@ -569,6 +629,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 +713,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}"])

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()