fix: stabilize API routing and simplify delivery reports

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

@@ -14,9 +14,11 @@ 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;
+- 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;
@@ -25,6 +27,7 @@ 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 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;
@@ -84,7 +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;
-- the continuation retains original demand 10 and again uses forecast quantity 15; it never compounds the adjustment to 20;
+- 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 `不可用候选`;
@@ -244,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.
@@ -268,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/production_quantity.md

@@ -1,36 +1,43 @@
-# 交期预测数量口径
+# 交期预测数量传递规则
 
-用户输入为原始需求数量。交期回复机器人在查询原型号、候选型号及后续批次前，统一按以下规则计算 API 输入数量：
+## 计算责任
+
+AR51 于 2026-10-15 检查上游 API 的交期预测逻辑后确认：接口已核算可用库存；并明确取消机器人本地的扣库存、95% 合格率折算和向上取 5 台倍数规则。此确认是本次修正依据，不代表已经实时验证服务端或运行电脑。
 
 ```text
-排产测算数量 = 5 × ceil(原始需求数量 ÷ 0.95 ÷ 5)
+D = 用户原始需求数量（正整数）
+API quantity = D
 ```
 
-固定不良率为 5%，对应良率 95%。先除以良率，再向上取到 5 的倍数。代码使用整数运算，避免恰好落在倍数边界时因浮点误差多加一个批次。
+每个型号只通过受控脚本查询一次，原型号和各候选型号都传原始需求 D。机器人保留型号推荐、结果整理和记录能力；库存扣减及交期测算由上游 API 处理。
 
-| 原始需求 | 除以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 |
+## 示例与输出
 
-回复示例中的数量部分：
+| 原始需求 | API 返回可用库存 | API 查询数量 | 库存展示 |
+|---:|---:|---:|---|
+| 12 | 10 | 12 | 部分覆盖，仍需补齐 2 台 |
+| 8 | 5 | 8 | 部分覆盖，仍需补齐 3 台 |
+| 24 | 0 | 24 | 需生产/采购 |
+| 24 | 24 | 24 | 完全覆盖 |
+| 24 | 30 | 24 | 完全覆盖 |
 
-```text
-输入需求：eRob70H120I-BHM-18CT[V5]，5台
-排产测算数量：10台（不良率5%，按5台的倍数向上取整）
-库存覆盖口径：按排产测算数量判断
+库存覆盖按原始需求与该型号 API 返回的库存比较，仅用于展示。展示差额不回传 API，不作为机器人计算的排产数量。每个候选展示自己的库存，不能沿用原型号或其他候选的库存。
 
-1. eRob70H120I-BHM-18CT[V5]（原型号，DZ）
-- 需求数量：5台
-- 排产测算数量：10台
-```
+库存缺失或无效时显示待复核，不视作 0；如果 API 已返回有效工作日交期和最终交货日期，机器人保留该交期并标明库存覆盖无法判断。缺失有效交期字段或请求失败时继续执行原有异常处理。
 
-实际交期继续读取 API 返回的工作日交期和最终交货日期。库存覆盖也以本次 API 使用的排产测算数量判断，例如原始需求 5、API 数量 10、库存 7 时，显示部分覆盖，仍需补齐 3 台。
+飞书报告展示需求数量、API 返回交期和库存覆盖情况。按 AR51 于 2026-10-16 的显示要求，首次及续查回复均不显示“数量传递规则”“库存覆盖口径”和各型号下的“API查询数量”，也不输出本地净需求或排产测算数量。内部记录仍保存数量规则、原始需求及每个型号实际 API 查询数量，历史记录不改写。
 
-对话上下文和本地记录保留原始需求；记录元数据另列排产测算数量和计算规则。继续推荐仍从原始需求重新计算一次，不把 10 台再次放大为 15 台。低层受控脚本的 `--quantity` 参数就是最终发送给 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/skills/joint-module-leadtime-report/SKILL.md

@@ -40,14 +40,14 @@ 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
+## Original Demand 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.
+- 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
 
@@ -88,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 <production_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.
@@ -179,8 +179,6 @@ 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算法预测参考，需人工复核
@@ -195,7 +193,6 @@ Use this field order for a successful Feishu report:
 
 1. {model}（{candidate type and reducer brand}）
 - 需求数量：{input quantity}台
-- 排产测算数量：{production quantity}台
 - AM516预测交期：{API workdays}个工作日，最终交货日期：{API final delivery date}
 - 可用库存/覆盖情况：{summary-level inventory and coverage only}
 - 主导因素：{main factor}
@@ -219,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.

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

@@ -83,8 +83,6 @@ 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"
 
@@ -182,13 +180,10 @@ 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."""
+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("需求数量必须为正整数。")
-    numerator = quantity * 100
-    denominator = (100 - DEFECT_RATE_PERCENT) * PRODUCTION_BATCH_SIZE
-    return ((numerator + denominator - 1) // denominator) * PRODUCTION_BATCH_SIZE
 
 
 _candidate_continuations: dict[str, CandidateContinuation] = {}
@@ -716,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
@@ -830,12 +832,11 @@ def overall_judgment_lines(candidates: list[dict[str, Any]]) -> list[str]:
     ]
 
 
-def candidate_report_lines(candidate: dict[str, Any], quantity: int, production_quantity: int) -> list[str]:
+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"- 排产测算数量：{production_quantity}台",
         f"- {details['prediction']}",
         f"- 可用库存/覆盖情况：{details['inventory']}",
         f"- 主导因素：{details['main_factor']}",
@@ -906,9 +907,13 @@ def render_record(
         "",
         f"**输入型号**：{request.model}",
         f"**输入数量**：{request.quantity}台",
-        f"**排产测算数量（API查询数量）**：{production_quantity_for_demand(request.quantity)}台",
-        "**数量计算规则**：原始需求÷95%良率，再按5台的倍数向上取整。",
-        "**库存覆盖口径**：按排产测算数量判断。",
+        "**数量传递规则**：原始需求原样传入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}`",
@@ -981,10 +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("候选批次位置无效，请重新发起型号和数量查询。")
-    production_quantity = production_quantity_for_demand(request.quantity)
+    validate_demand_quantity(request.quantity)
 
     all_candidate_specs = generate_candidate_specs(request.model)
     candidate_specs = all_candidate_specs[
@@ -998,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(RequestInput(request.model, production_quantity))
+            baseline = query_model_for_demand(request)
         except CandidateUnavailableError as exc:
             raise RuntimeError(f"原型号未生成候选结果。\n原因：{exc}") from exc
-        baseline = lead_time_details(
-            baseline_payload,
-            production_quantity,
-        )
         if not baseline["has_lead_time"]:
             raise RuntimeError(
-                "原型号接口未返回有效工作日交期和最终交货日期，未生成候选结果。\n"
+                "原型号未生成可用交期结果，未生成候选结果。\n"
                 f"原因：{baseline['unavailable_reason']}\n"
-                "处理建议：请接口维护人员核对上述字段及服务端计算结果后重试。"
+                "处理建议：请接口维护人员核对上述原因、库存口径及服务端计算结果后重试。"
             )
         candidates.append(
             {
@@ -1028,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"], production_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,
-            production_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:
@@ -1104,8 +1113,6 @@ 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算法预测参考，需人工复核",
@@ -1116,7 +1123,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, production_quantity))
+        report_lines.extend(candidate_report_lines(candidate_item, request.quantity))
     report_lines.append("不可用候选：")
     if unavailable:
         for item in unavailable:
@@ -1147,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 {"总体判断：", "本批次判断：", "不可用候选："}

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,18 +404,7 @@ 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)
-
+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 (
@@ -424,44 +413,117 @@ class ProductionQuantityTests(unittest.TestCase):
                 mock.patch.object(bridge, "write_record") as record,
             ):
                 with self.assertRaises(bridge.InputError):
-                    bridge.build_report(bridge.RequestInput("eRob70H50I-FS-18CN[V5]", value))
+                    bridge.build_report(bridge.RequestInput("eRob70F50I-FHM-18ET[V4]", 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)):
+    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.subprocess, "run", return_value=completed) as run,
-                mock.patch.object(bridge, "write_record") as record,
+                mock.patch.object(bridge, 'run_controlled_query', return_value=successful_payload()) as query,
+                mock.patch.object(bridge, 'write_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)
+                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_stock_coverage_uses_same_quantity_as_api_forecast(self) -> None:
-        payload = successful_payload()
-        payload["data"]["库存信息"]["可用库存(扣除了待产数量)"] = 7
+    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", return_value=payload),
-            mock.patch.object(bridge, "write_record"),
+            mock.patch.object(bridge, 'run_controlled_query', side_effect=query) as run,
+            mock.patch.object(bridge, 'write_record') as 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)
+            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):
@@ -871,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()