Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e2b46faa05 | |||
| ba35e251be | |||
| 341b7e3990 | |||
| 30a95ef08a |
@@ -1,165 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""提交信息规范校验(CI 侧权威闸门)。
|
||||
|
||||
规则与本机 ~/.githooks/commit-msg 完全一致,防止 --no-verify 绕过。
|
||||
放行:Merge / Revert " / fixup! / squash! / amend! 开头的自动生成提交。
|
||||
|
||||
存量历史口径:合并一条早已推送到别的远端分支的lineage(例如把生产分支合进 main)时,
|
||||
被带进来的提交并非本次新增,它们的标题不合规属于既有事实。判定为「已在其它远端分支存在」
|
||||
的提交只告警、不判红;本次真正新造的提交一律照旧严格校验。
|
||||
豁免必须是可证明的:先抓全部远端分支头,用 `git branch -r --contains <sha>` 取证;
|
||||
抓不到 / 证不出 ⇒ 不豁免(宁可红,也不假绿)。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
TYPES = "feat|fix|docs|style|refactor|perf|test|build|ci|chore|revert"
|
||||
RE = re.compile(r"^(?:" + TYPES + r")(?:\([a-zA-Z0-9._/-]+\))?!?: .{3,}$")
|
||||
ALLOW = re.compile(r'^(?:Merge |Revert "|fixup! |squash! |amend! )')
|
||||
|
||||
# 本次推送的分支名,用来把「当前分支自己」排除在豁免证据之外
|
||||
PUSH_BRANCH = (os.environ.get("GITHUB_REF_NAME") or "").strip()
|
||||
|
||||
HELP = """
|
||||
\u274c 提交信息不合规,已拦截。
|
||||
|
||||
规范 : <类型>(<范围>): <简短描述>
|
||||
类型 : feat 新功能 | fix 修缺陷 | docs 文档 | style 格式(不改逻辑)
|
||||
refactor 重构 | perf 性能 | test 测试 | build 构建/依赖
|
||||
ci CI 配置 | chore 杂务 | revert 回滚
|
||||
示例 : feat(asset): 新增按歌曲生成应援序列
|
||||
fix(parser): 修复表格解析越界导致的 500
|
||||
docs: 补充 OSS 短签名接口说明
|
||||
|
||||
要点 : 类型小写、冒号后一个空格、描述 \u22653 字符
|
||||
标题 \u226472 字符、结尾不加句号
|
||||
破坏性变更类型后加 !,正文写 BREAKING CHANGE:
|
||||
详见 : docs/提交规范.md
|
||||
"""
|
||||
|
||||
|
||||
def sh(args):
|
||||
return subprocess.run(args, capture_output=True, text=True)
|
||||
|
||||
|
||||
def collect():
|
||||
"""返回 [(sha, message)]:优先 Gitea 事件负载,取不到则退回 git log。"""
|
||||
items = []
|
||||
ev = os.environ.get("GITHUB_EVENT_PATH") or ""
|
||||
if ev and os.path.exists(ev):
|
||||
try:
|
||||
with open(ev, encoding="utf-8") as fh:
|
||||
d = json.load(fh)
|
||||
for c in (d.get("commits") or []):
|
||||
m = (c.get("message") or "").strip()
|
||||
if m:
|
||||
items.append(((c.get("id") or "").strip(), m))
|
||||
except Exception as e:
|
||||
print(" 事件负载解析失败(%s),回退到 git log" % e)
|
||||
if not items:
|
||||
out = sh(["git", "log", "-n", "5", "--format=%H%x1f%B%x1e"]).stdout
|
||||
for rec in out.split("\x1e"):
|
||||
rec = rec.strip()
|
||||
if not rec:
|
||||
continue
|
||||
sha, _, msg = rec.partition("\x1f")
|
||||
if msg.strip():
|
||||
items.append((sha.strip(), msg.strip()))
|
||||
return items
|
||||
|
||||
|
||||
_fetched = None
|
||||
|
||||
|
||||
def ensure_remote_heads():
|
||||
"""抓全部远端分支头(浅历史足够判断 lineage 末端)。"""
|
||||
global _fetched
|
||||
if _fetched is None:
|
||||
r = sh(["git", "fetch", "-q", "--depth", "200", "origin", "+refs/heads/*:refs/remotes/origin/*"])
|
||||
_fetched = (r.returncode == 0, (r.stderr or "").strip())
|
||||
if not _fetched[0]:
|
||||
print(" \u26a0 无法抓取远端分支,存量豁免不可用(将严格校验): %s" % _fetched[1][:200])
|
||||
return _fetched[0]
|
||||
|
||||
|
||||
def refs_containing(sha):
|
||||
"""除本次推送分支外,还有哪些远端分支包含该提交。"""
|
||||
if not ensure_remote_heads():
|
||||
return []
|
||||
r = sh(["git", "branch", "-r", "--contains", sha])
|
||||
if r.returncode != 0:
|
||||
return []
|
||||
refs = []
|
||||
for line in r.stdout.splitlines():
|
||||
ref = line.strip().lstrip("*").strip()
|
||||
if not ref or "->" in ref:
|
||||
continue
|
||||
if PUSH_BRANCH and ref == "origin/" + PUSH_BRANCH:
|
||||
continue
|
||||
refs.append(ref)
|
||||
return refs
|
||||
|
||||
|
||||
def subject_of(m):
|
||||
for line in m.splitlines():
|
||||
s = line.strip()
|
||||
if not s or s.startswith("#"):
|
||||
continue
|
||||
return s
|
||||
return ""
|
||||
|
||||
|
||||
def judge(subj):
|
||||
if not subj:
|
||||
return "空提交信息"
|
||||
if ALLOW.match(subj):
|
||||
return ""
|
||||
if len(subj) > 72:
|
||||
return "标题 %d 字符 > 72" % len(subj)
|
||||
if not RE.match(subj):
|
||||
return "不符合 <类型>(<范围>): <描述>"
|
||||
if subj.endswith(".") or subj.endswith("\u3002"):
|
||||
return "标题结尾不要加句号"
|
||||
return ""
|
||||
|
||||
|
||||
def main():
|
||||
items = collect()
|
||||
bad = []
|
||||
excused = []
|
||||
print("待校验提交 %d 条:" % len(items))
|
||||
for sha, msg in items:
|
||||
subj = subject_of(msg)
|
||||
why = judge(subj)
|
||||
if not why:
|
||||
print(" \u2705 %s" % subj)
|
||||
continue
|
||||
refs = refs_containing(sha) if sha else []
|
||||
if refs:
|
||||
excused.append((sha, subj, why, refs))
|
||||
print(" \u21aa [%s] %s ← 已存在于 %s,按存量历史放行" % (why, subj, ",".join(refs[:3])))
|
||||
continue
|
||||
bad.append((subj, why))
|
||||
print(" \u274c [%s] %s" % (why, subj))
|
||||
if excused:
|
||||
print("")
|
||||
print("存量豁免 %d 条(提交早于本次合并,已在其它远端分支存在):" % len(excused))
|
||||
for sha, subj, why, refs in excused:
|
||||
print(" %s %s [%s]" % (sha[:10], subj, ",".join(refs[:3])))
|
||||
try:
|
||||
with open("/tmp/ci-gate-status", "w") as fh:
|
||||
fh.write("failure" if bad else "success")
|
||||
except Exception:
|
||||
pass
|
||||
if bad:
|
||||
print(HELP)
|
||||
return 1
|
||||
print("\u2705 提交信息全部合规")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -1,82 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""把 CI 结果上报给主机侧 ci-bridge(bridge 负责落报告页 + 推微信)。"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import urllib.request
|
||||
|
||||
|
||||
def rd(path, default=""):
|
||||
try:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return fh.read().strip()
|
||||
except Exception:
|
||||
return default
|
||||
|
||||
|
||||
def git(*args):
|
||||
try:
|
||||
return subprocess.run(["git"] + list(args), capture_output=True,
|
||||
text=True).stdout.strip()
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
def main():
|
||||
repo = os.environ.get("GITHUB_REPOSITORY", "unknown/unknown")
|
||||
sha = rd("/tmp/ci-sha") or git("rev-parse", "HEAD")
|
||||
branch = os.environ.get("GITHUB_REF_NAME") or \
|
||||
os.environ.get("GITHUB_REF", "").replace("refs/heads/", "")
|
||||
gate = rd("/tmp/ci-gate-status", "skipped")
|
||||
build = rd("/tmp/ci-build-status", "failure")
|
||||
status = "success" if (gate == "success" and build == "success") else "failure"
|
||||
|
||||
log = rd("/tmp/ci-build-tail")
|
||||
if gate != "success":
|
||||
log = "提交信息规范校验未通过\n" + log
|
||||
|
||||
try:
|
||||
count = int(rd("/tmp/ci-commits-count", "1") or 1)
|
||||
except Exception:
|
||||
count = 1
|
||||
|
||||
rep = {
|
||||
"repo": repo,
|
||||
"sha": sha,
|
||||
"branch": branch,
|
||||
"actor": os.environ.get("GITHUB_ACTOR", ""),
|
||||
"status": status,
|
||||
"commit_subject": git("log", "-1", "--format=%s"),
|
||||
"commit_count": count,
|
||||
"duration_s": rd("/tmp/ci-duration", "?"),
|
||||
"preview_url": rd("/tmp/ci-preview-url"),
|
||||
"log": log[-3500:],
|
||||
"steps": [
|
||||
{"name": "提交信息规范校验", "status": gate},
|
||||
{"name": "编译与自检", "status": build},
|
||||
],
|
||||
}
|
||||
if rep["preview_url"]:
|
||||
print("\U0001F310 网页预览: %s" % rep["preview_url"])
|
||||
print("结果: %s | 构建: %s | 提交闸门: %s | 耗时: %ss"
|
||||
% (status, build, gate, rep["duration_s"]))
|
||||
|
||||
base = os.environ.get("CI_BRIDGE_RESOLVED", "").rstrip("/")
|
||||
if not base:
|
||||
print("跳过上报:未找到可达的 ci-bridge")
|
||||
return 0
|
||||
data = json.dumps(rep, ensure_ascii=False).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
base + "/api/ci/report", data=data, method="POST",
|
||||
headers={"Content-Type": "application/json",
|
||||
"X-CI-Token": os.environ.get("BRIDGE_TOKEN", "")})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
print("\u2705 已上报 ci-bridge:", resp.read().decode()[:300])
|
||||
except Exception as e:
|
||||
print("\u274c 上报 ci-bridge 失败:", e)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,156 +0,0 @@
|
||||
name: build
|
||||
|
||||
# 提交 -> 本机 Runner 编译 -> 微信通知 + 网页预览
|
||||
# 实测结论(勿随意改动):
|
||||
# 1. 不用任何第三方 action(作业容器不联网拉 action,全用镜像自带工具)
|
||||
# 2. 必须自己 clone —— Runner 不会自动检出代码,工作区初始为空
|
||||
# 3. act 以 `bash -e -o pipefail` 执行每步 -> 禁止 `cmd | head` 这种会 SIGPIPE 提前退出的管道
|
||||
# 4. 工具链来自主机只读挂载 /opt/citools(go/uv),缓存写 /opt/citools-cache
|
||||
# 5. bridge 从容器内走 docker 网关 172.17.0.1:8899(实测可达)
|
||||
# 6. 校验/上报逻辑放在 .gitea/scripts/ 下的脚本里,YAML 只负责调用(避免 YAML 里嵌深缩进代码)
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- "**"
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
BRIDGE_URL: ${{ vars.CI_BRIDGE_URL }}
|
||||
BRIDGE_TOKEN: ${{ secrets.CI_BRIDGE_TOKEN }}
|
||||
GIT_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
concurrency:
|
||||
group: build-${{ github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 40
|
||||
steps:
|
||||
- name: 检出代码
|
||||
run: |
|
||||
set -uo pipefail
|
||||
SRV="${GITHUB_SERVER_URL:-http://192.168.50.223:3000}"
|
||||
HP="${SRV#*://}"
|
||||
BR="${GITHUB_REF_NAME:-${GITHUB_REF#refs/heads/}}"
|
||||
cd "${GITHUB_WORKSPACE:-$(pwd)}"
|
||||
rm -rf ./* .git 2>/dev/null || true
|
||||
git init -q .
|
||||
git remote add origin "http://x-access-token:${GIT_TOKEN}@${HP}/${GITHUB_REPOSITORY}.git"
|
||||
git fetch -q --depth 50 origin "$BR"
|
||||
git checkout -q -B "$BR" FETCH_HEAD
|
||||
git rev-parse HEAD > /tmp/ci-sha
|
||||
date +%s > /tmp/ci-start
|
||||
: > /tmp/ci-preview-url
|
||||
echo skipped > /tmp/ci-build-status
|
||||
git rev-list --count FETCH_HEAD > /tmp/ci-commits-count 2>/dev/null || echo 1 > /tmp/ci-commits-count
|
||||
echo "已检出 ${GITHUB_REPOSITORY} 分支 ${BR}"
|
||||
git log -1 --format=' %h %an %s'
|
||||
|
||||
- name: 提交信息规范校验
|
||||
run: python3 .gitea/scripts/ci_gate.py
|
||||
|
||||
- name: 编译与自检
|
||||
run: |
|
||||
set +e; set +o pipefail
|
||||
LOG=/tmp/ci-build.log; : > "$LOG"
|
||||
run() { echo "-- $* --"; "$@" 2>&1 | tee -a "$LOG"; return "${PIPESTATUS[0]}"; }
|
||||
runin(){ local d="$1"; shift; echo "-- ($d) $* --"; ( cd "$d" && "$@" ) 2>&1 | tee -a "$LOG"; return "${PIPESTATUS[0]}"; }
|
||||
|
||||
export PATH=/opt/citools/go/bin:/opt/citools/bin:$PATH
|
||||
export GOCACHE=/opt/citools-cache/gocache GOMODCACHE=/opt/citools-cache/gomod UV_CACHE_DIR=/opt/citools-cache/uv
|
||||
export GOPROXY=https://goproxy.cn,direct GOTOOLCHAIN=local GOFLAGS=-mod=mod GOPATH=/opt/citools-cache/gopath
|
||||
mkdir -p "$GOCACHE" "$GOMODCACHE" "$UV_CACHE_DIR" "$GOPATH" 2>/dev/null
|
||||
|
||||
echo "工具链: $(go version 2>&1) | $(uv --version 2>&1) | $(node -v 2>&1) | $(python3 -V 2>&1)"
|
||||
fail=0; ran=""
|
||||
|
||||
if [ -f go.mod ]; then
|
||||
ran="$ran go"
|
||||
run go build ./... || fail=1
|
||||
run go vet ./... || fail=1
|
||||
run go test ./... || fail=1
|
||||
fi
|
||||
|
||||
if [ -f pyproject.toml ] || [ -f requirements.txt ] || [ -f setup.py ]; then
|
||||
ran="$ran python"
|
||||
if grep -qE '^check:' Makefile 2>/dev/null; then
|
||||
run make UV=uv check || fail=1
|
||||
else
|
||||
run uv sync || fail=1
|
||||
run uv run pytest -q || fail=1
|
||||
fi
|
||||
fi
|
||||
|
||||
WEBPKG=""
|
||||
for p in package.json web/package.json frontend/package.json ui/package.json; do
|
||||
if [ -f "$p" ]; then WEBPKG="$p"; break; fi
|
||||
done
|
||||
if [ -n "$WEBPKG" ]; then
|
||||
ran="$ran node"
|
||||
D="$(dirname "$WEBPKG")"
|
||||
if [ -f "$D/package-lock.json" ]; then
|
||||
runin "$D" npm ci || fail=1
|
||||
else
|
||||
runin "$D" npm install || fail=1
|
||||
fi
|
||||
if grep -q '"build"[[:space:]]*:' "$WEBPKG" 2>/dev/null; then
|
||||
runin "$D" npm run build || fail=1
|
||||
fi
|
||||
fi
|
||||
|
||||
if [ -z "$ran" ] && [ -f Makefile ]; then
|
||||
ran="$ran make"
|
||||
run make || fail=1
|
||||
fi
|
||||
if [ -z "$ran" ]; then
|
||||
echo "未识别到构建入口(go.mod / pyproject.toml / package.json / Makefile 都没有),跳过编译"
|
||||
fi
|
||||
|
||||
if [ "$fail" -eq 0 ]; then echo success > /tmp/ci-build-status; else echo failure > /tmp/ci-build-status; fi
|
||||
tail -c 4000 "$LOG" > /tmp/ci-build-tail
|
||||
echo "构建结果: $(cat /tmp/ci-build-status)(识别:$ran)"
|
||||
[ "$fail" -eq 0 ] || exit 1
|
||||
|
||||
- name: 发布网页预览
|
||||
if: always()
|
||||
run: |
|
||||
set +e; set +o pipefail
|
||||
: > /tmp/ci-preview-url
|
||||
if [ "$(cat /tmp/ci-build-status 2>/dev/null)" != "success" ]; then
|
||||
echo "构建未成功,跳过预览发布"; exit 0
|
||||
fi
|
||||
WEB=""
|
||||
for d in web/dist dist build out public web/build frontend/dist; do
|
||||
if [ -f "$d/index.html" ]; then WEB="$d"; break; fi
|
||||
done
|
||||
if [ -z "$WEB" ]; then echo "未发现静态网页产物目录,跳过预览发布"; exit 0; fi
|
||||
tar -czf /tmp/ci-web.tgz -C "$WEB" . || { echo "打包失败"; exit 0; }
|
||||
B="${BRIDGE_URL:-}"
|
||||
for cand in "$B" "http://172.17.0.1:8899" "http://192.168.50.134:8899" "http://100.92.49.78:8899"; do
|
||||
if [ -n "$cand" ]; then
|
||||
curl -s -m 5 -o /dev/null "$cand/healthz" && { B="$cand"; break; }
|
||||
fi
|
||||
done
|
||||
SHA="$(cat /tmp/ci-sha)"
|
||||
echo "发布预览到 $B(产物目录 $WEB)"
|
||||
curl -sS -m 90 -X POST "$B/api/ci/publish?repo=${GITHUB_REPOSITORY}&sha=${SHA}" -H "X-CI-Token: ${BRIDGE_TOKEN}" -H 'Content-Type: application/gzip' --data-binary @/tmp/ci-web.tgz -o /tmp/ci-pub.json
|
||||
cat /tmp/ci-pub.json; echo
|
||||
URL="$(python3 -c "import json;print(json.load(open('/tmp/ci-pub.json')).get('url',''))" 2>/dev/null)"
|
||||
if [ -n "$URL" ]; then echo "$URL" > /tmp/ci-preview-url; echo "网页预览: $URL"; fi
|
||||
|
||||
- name: 通知微信
|
||||
if: always()
|
||||
run: |
|
||||
set +e; set +o pipefail
|
||||
B="${BRIDGE_URL:-}"
|
||||
for cand in "$B" "http://172.17.0.1:8899" "http://192.168.50.134:8899" "http://100.92.49.78:8899"; do
|
||||
if [ -n "$cand" ]; then
|
||||
curl -s -m 5 -o /dev/null "$cand/healthz" && { B="$cand"; break; }
|
||||
fi
|
||||
done
|
||||
echo "使用 ci-bridge: $B"
|
||||
S=$(date +%s); T=$(cat /tmp/ci-start 2>/dev/null); [ -n "$T" ] || T=$S
|
||||
echo $((S - T)) > /tmp/ci-duration
|
||||
CI_BRIDGE_RESOLVED="$B" python3 .gitea/scripts/ci_report.py
|
||||
@@ -0,0 +1,51 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
|
||||
)
|
||||
|
||||
// cardPartialUpdatesMode returns the configured card delivery mode for a
|
||||
// finished answer (see dingtalk.CardPartialUpdatesMode):
|
||||
//
|
||||
// unset / on / partial → placeholder card + throttled partial frames (historical)
|
||||
// off / final → placeholder card, and only the finished answer
|
||||
// direct → no placeholder card; one card with the finished answer
|
||||
//
|
||||
// The default stays "partial" so an environment without the variable behaves
|
||||
// exactly as before; DINGTALK_CARD_PARTIAL_UPDATES selects the other modes. The
|
||||
// resolved mode is logged so the startup log always states what is running.
|
||||
func cardPartialUpdatesMode() dingtalk.CardPartialUpdatesMode {
|
||||
mode := dingtalk.ParseCardPartialUpdatesMode(os.Getenv("DINGTALK_CARD_PARTIAL_UPDATES"))
|
||||
log.Printf("DingTalk card partial updates mode=%s", mode)
|
||||
return mode
|
||||
}
|
||||
|
||||
// cardFinalWriteMode returns the configured terminal-write mode for a finished
|
||||
// answer (see dingtalk.CardFinalWriteMode):
|
||||
//
|
||||
// unset / streaming → the finished answer is written with the card streaming
|
||||
// API and isFinalize=true (historical behaviour)
|
||||
// update / plain → the same answer is written once more through the plain
|
||||
// card update API (PUT /v1.0/card/instances) AFTER the
|
||||
// streaming finalize (observed to blank the delivered card)
|
||||
// refresh → the answer goes through the plain card update API FIRST
|
||||
// and the stream is then closed with the very same text,
|
||||
// so the client has no delta left to animate
|
||||
// close-refresh → the stream is closed on the text the card already shows
|
||||
// (zero delta) and the answer is refreshed afterwards
|
||||
//
|
||||
// DingTalk renders streamed card content with a typewriter animation the client
|
||||
// drives itself, so an answer that is already complete on the server can still be
|
||||
// replayed character by character - and a client that is not in the foreground
|
||||
// pauses that animation, which reads as "streaming stopped, it continues when I
|
||||
// switch back". The default stays "streaming" so an environment without the
|
||||
// variable behaves exactly as before; DINGTALK_CARD_FINAL_WRITE selects the other
|
||||
// mode. The resolved mode is logged so the startup log always states what ran.
|
||||
func cardFinalWriteMode() dingtalk.CardFinalWriteMode {
|
||||
mode := dingtalk.ParseCardFinalWriteMode(os.Getenv("DINGTALK_CARD_FINAL_WRITE"))
|
||||
log.Printf("DingTalk card final write mode=%s", mode)
|
||||
return mode
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
|
||||
)
|
||||
|
||||
func TestCardPartialUpdatesMode(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
raw string
|
||||
want dingtalk.CardPartialUpdatesMode
|
||||
}{
|
||||
{"unset keeps streaming", "", dingtalk.CardPartialUpdatesPartial},
|
||||
{"on keeps streaming", "on", dingtalk.CardPartialUpdatesPartial},
|
||||
{"partial keeps streaming", "PARTIAL", dingtalk.CardPartialUpdatesPartial},
|
||||
{"unknown keeps streaming", "hammer", dingtalk.CardPartialUpdatesPartial},
|
||||
{"final", " final ", dingtalk.CardPartialUpdatesFinal},
|
||||
{"off maps to final", "off", dingtalk.CardPartialUpdatesFinal},
|
||||
{"direct", "direct", dingtalk.CardPartialUpdatesDirect},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Setenv("DINGTALK_CARD_PARTIAL_UPDATES", tt.raw)
|
||||
if got := cardPartialUpdatesMode(); got != tt.want {
|
||||
t.Fatalf("cardPartialUpdatesMode(%q)=%q want %q", tt.raw, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteMode(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
raw string
|
||||
want dingtalk.CardFinalWriteMode
|
||||
}{
|
||||
{"unset keeps the streaming finalize", "", dingtalk.CardFinalWriteStreaming},
|
||||
{"streaming stays streaming", "streaming", dingtalk.CardFinalWriteStreaming},
|
||||
{"unknown keeps the streaming finalize", "hammer", dingtalk.CardFinalWriteStreaming},
|
||||
{"update", " update ", dingtalk.CardFinalWriteUpdate},
|
||||
{"plain alias", "plain", dingtalk.CardFinalWriteUpdate},
|
||||
{"on alias", "ON", dingtalk.CardFinalWriteUpdate},
|
||||
{"1 alias", "1", dingtalk.CardFinalWriteUpdate},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Setenv("DINGTALK_CARD_FINAL_WRITE", tt.raw)
|
||||
if got := cardFinalWriteMode(); got != tt.want {
|
||||
t.Fatalf("cardFinalWriteMode(%q)=%q want %q", tt.raw, got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -12,13 +12,15 @@ import (
|
||||
// unset / on → grouped list on every frame (historical behaviour)
|
||||
// off → real tables on every frame
|
||||
// stream → grouped list while streaming, real table on the final frame
|
||||
// smooth → like "on", and every table is grouped while streaming so a
|
||||
// table forming mid-answer no longer freezes the card
|
||||
//
|
||||
// The default stays "on" so an environment without the variable renders exactly
|
||||
// as before; DINGTALK_CARD_WIDE_TABLE_GROUP selects the other modes.
|
||||
func cardWideTableGroupingMode() string {
|
||||
mode := strings.ToLower(strings.TrimSpace(os.Getenv("DINGTALK_CARD_WIDE_TABLE_GROUP")))
|
||||
switch mode {
|
||||
case "off", "0", "false", "no", "stream", "streaming", "partial":
|
||||
case "off", "0", "false", "no", "stream", "streaming", "partial", "smooth":
|
||||
log.Printf("DingTalk wide-table grouping mode=%s", mode)
|
||||
return mode
|
||||
default:
|
||||
|
||||
+3
-4
@@ -71,10 +71,7 @@ func main() {
|
||||
if strings.TrimSpace(settingsRuntime.clientID) == "" || strings.TrimSpace(settingsRuntime.secret) == "" {
|
||||
log.Fatal("DingTalk credentials are required; configure DINGTALK_CLIENT_ID and DINGTALK_CLIENT_SECRET in the protected runtime environment")
|
||||
}
|
||||
// 0 = no client-side total timeout. WeKnora's smart-reasoning agent can run
|
||||
// 20+ tool rounds and stream far longer than 120s; the total-time cap cut the
|
||||
// SSE mid-answer and the failure branch overwrote the streamed card.
|
||||
client, err := weknora.NewClient(os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"), 0)
|
||||
client, err := weknora.NewClient(os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"), 120*time.Second)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
@@ -301,6 +298,7 @@ func main() {
|
||||
cardTemplateID,
|
||||
nil,
|
||||
dingtalk.WithCardDebugUserID(cardDebugUserID),
|
||||
dingtalk.WithCardFinalWrite(cardFinalWriteMode()),
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatalf("DingTalk card sender init failed: %v", err)
|
||||
@@ -328,6 +326,7 @@ func main() {
|
||||
|
||||
OutboundRewriteTimeout: outboundImages.rewriteTimeout,
|
||||
WideTableGrouping: cardWideTableGroupingMode(),
|
||||
PartialUpdates: cardPartialUpdatesMode(),
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
|
||||
@@ -232,6 +232,10 @@ func TestCardWideTableGroupingMode(t *testing.T) {
|
||||
if got := cardWideTableGroupingMode(); got != "" {
|
||||
t.Fatalf("on is the default and must not be forwarded as a mode, got %q", got)
|
||||
}
|
||||
t.Setenv("DINGTALK_CARD_WIDE_TABLE_GROUP", "SMOOTH")
|
||||
if got := cardWideTableGroupingMode(); got != "smooth" {
|
||||
t.Fatalf("smooth must be normalised, got %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMainDoesNotExposeWebAdministrationOrSetupOnlyMode(t *testing.T) {
|
||||
|
||||
@@ -20,6 +20,7 @@ import (
|
||||
const (
|
||||
cardCreateURL = "https://api.dingtalk.com/v1.0/card/instances/createAndDeliver"
|
||||
cardStreamingURL = "https://api.dingtalk.com/v1.0/card/streaming"
|
||||
cardUpdateURL = "https://api.dingtalk.com/v1.0/card/instances"
|
||||
cardDebugPrivateKey = "_CARD_DEBUG_TOOL_ENTRY"
|
||||
|
||||
// cardWideScreenJSONKey is DingTalk's reserved system field inside
|
||||
@@ -57,13 +58,22 @@ type CardSender struct {
|
||||
tokenURL string
|
||||
createURL string
|
||||
streamURL string
|
||||
updateURL string
|
||||
apiClient *http.Client
|
||||
allowTestURL bool
|
||||
debugUserID string
|
||||
finalWrite CardFinalWriteMode
|
||||
|
||||
mu sync.Mutex
|
||||
token string
|
||||
tokenTill time.Time
|
||||
|
||||
// streamed keeps the text every live card currently shows, so a terminal
|
||||
// write can close the stream with a frame the client has nothing to animate
|
||||
// (see CardFinalWriteMode). Bounded by the number of cards in flight in one
|
||||
// process; entries are dropped once the card is finalized.
|
||||
contentMu sync.Mutex
|
||||
streamed map[string]string
|
||||
}
|
||||
|
||||
// CardSenderOption configures optional CardSender behavior.
|
||||
@@ -77,20 +87,31 @@ func WithCardDebugUserID(userID string) CardSenderOption {
|
||||
}
|
||||
}
|
||||
|
||||
// WithCardFinalWrite selects how the terminal write of a finished answer lands
|
||||
// in the card; see CardFinalWriteMode. The zero value keeps the historical
|
||||
// streaming finalize.
|
||||
func WithCardFinalWrite(mode CardFinalWriteMode) CardSenderOption {
|
||||
return func(sender *CardSender) {
|
||||
sender.finalWrite = mode
|
||||
}
|
||||
}
|
||||
|
||||
// NewCardSender creates a CardSender. All three config values are required
|
||||
// after trimming whitespace. A nil http.Client installs a hardened default
|
||||
// (no redirects, Proxy=nil, 15s timeout).
|
||||
func NewCardSender(clientID, clientSecret, cardTemplateID string, client *http.Client, options ...CardSenderOption) (*CardSender, error) {
|
||||
return newCardSender(clientID, clientSecret, cardTemplateID, client, oauthTokenURL, cardCreateURL, cardStreamingURL, false, options...)
|
||||
return newCardSender(clientID, clientSecret, cardTemplateID, client, oauthTokenURL, cardCreateURL, cardStreamingURL, cardUpdateURL, false, options...)
|
||||
}
|
||||
|
||||
// newCardSenderForTest is used by tests to inject httptest URLs.
|
||||
func newCardSenderForTest(clientID, clientSecret, cardTemplateID string, client *http.Client, tokenURL, createURL string, options ...CardSenderOption) (*CardSender, error) {
|
||||
streamURL := strings.TrimSuffix(createURL, "/create") + "/stream"
|
||||
return newCardSender(clientID, clientSecret, cardTemplateID, client, tokenURL, createURL, streamURL, true, options...)
|
||||
base := strings.TrimSuffix(createURL, "/create")
|
||||
streamURL := base + "/stream"
|
||||
updateURL := base + "/update"
|
||||
return newCardSender(clientID, clientSecret, cardTemplateID, client, tokenURL, createURL, streamURL, updateURL, true, options...)
|
||||
}
|
||||
|
||||
func newCardSender(clientID, clientSecret, cardTemplateID string, client *http.Client, tokenURL, createURL, streamURL string, allowTest bool, options ...CardSenderOption) (*CardSender, error) {
|
||||
func newCardSender(clientID, clientSecret, cardTemplateID string, client *http.Client, tokenURL, createURL, streamURL, updateURL string, allowTest bool, options ...CardSenderOption) (*CardSender, error) {
|
||||
clientID = strings.TrimSpace(clientID)
|
||||
clientSecret = strings.TrimSpace(clientSecret)
|
||||
cardTemplateID = strings.TrimSpace(cardTemplateID)
|
||||
@@ -130,6 +151,7 @@ func newCardSender(clientID, clientSecret, cardTemplateID string, client *http.C
|
||||
tokenURL: tokenURL,
|
||||
createURL: createURL,
|
||||
streamURL: streamURL,
|
||||
updateURL: updateURL,
|
||||
apiClient: &apiClient,
|
||||
allowTestURL: allowTest,
|
||||
}
|
||||
@@ -232,7 +254,115 @@ func (s *CardSender) UpdateCard(ctx context.Context, outTrackID, content string,
|
||||
return err
|
||||
}
|
||||
|
||||
return s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError)
|
||||
if !isFinalize || isError {
|
||||
if err := s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError); err != nil {
|
||||
return err
|
||||
}
|
||||
if !isFinalize {
|
||||
// Remember what the card now shows: a terminal write can close the
|
||||
// stream with a frame that carries no new text.
|
||||
s.rememberStreamedContent(outTrackID, content)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Terminal write. The order of the two available writes decides whether the
|
||||
// client replays the finished answer with its own typewriter animation, so
|
||||
// each mode sequences them explicitly - see CardFinalWriteMode.
|
||||
defer s.forgetStreamedContent(outTrackID)
|
||||
switch {
|
||||
case s.finalWrite.refreshBeforeFinalize():
|
||||
// Show the finished answer first (plain write, no streaming semantics)
|
||||
// and close the stream with the very same text afterwards: the client
|
||||
// has no delta left to animate, so the answer is simply there.
|
||||
s.refreshCardPlain(ctx, outTrackID, content, token)
|
||||
return s.finalizeCard(ctx, outTrackID, content, token, true, false)
|
||||
case s.finalWrite.closeThenRefresh():
|
||||
shown := s.lastStreamedContent(outTrackID)
|
||||
if shown == "" {
|
||||
// Nothing was streamed into this card: there is no zero-delta frame
|
||||
// to close on, so refresh first and finalize after.
|
||||
s.refreshCardPlain(ctx, outTrackID, content, token)
|
||||
return s.finalizeCard(ctx, outTrackID, content, token, true, false)
|
||||
}
|
||||
// Close the stream on the text the card already shows (zero delta,
|
||||
// nothing to animate), then refresh with the finished answer.
|
||||
if err := s.finalizeCard(ctx, outTrackID, shown, token, true, false); err != nil {
|
||||
return err
|
||||
}
|
||||
s.refreshCardPlain(ctx, outTrackID, content, token)
|
||||
return nil
|
||||
default:
|
||||
if err := s.finalizeCard(ctx, outTrackID, content, token, isFinalize, isError); err != nil {
|
||||
return err
|
||||
}
|
||||
// Optionally let the terminal content land through the plain card update
|
||||
// API as well: content written through the streaming endpoint is replayed
|
||||
// by the client with a typewriter animation, while a plain write has no
|
||||
// streaming semantics. The streaming finalize above already closed the
|
||||
// card, so a failing follow-up write is logged and swallowed - it must
|
||||
// never turn a delivered answer into an error.
|
||||
if s.finalWrite.alsoPlainUpdates() {
|
||||
if err := s.replaceCardContent(ctx, outTrackID, content, token); err != nil {
|
||||
log.Printf("DingTalk card final write: plain update failed error=%v", err)
|
||||
} else {
|
||||
log.Printf("DingTalk card final write: plain update ok bytes=%d", len(content))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// refreshCardPlain writes a finished answer through the plain card update API
|
||||
// and logs the outcome. The write is best effort: the answer is already on its
|
||||
// way through the streaming channel, so a failure here is never turned into an
|
||||
// error for the caller.
|
||||
func (s *CardSender) refreshCardPlain(ctx context.Context, outTrackID, content, token string) {
|
||||
if err := s.replaceCardContent(ctx, outTrackID, content, token); err != nil {
|
||||
log.Printf("DingTalk card final write: plain refresh failed error=%v", err)
|
||||
return
|
||||
}
|
||||
log.Printf("DingTalk card final write: plain refresh ok bytes=%d", len(content))
|
||||
}
|
||||
|
||||
// rememberStreamedContent records the text a card currently shows.
|
||||
func (s *CardSender) rememberStreamedContent(outTrackID, content string) {
|
||||
s.contentMu.Lock()
|
||||
defer s.contentMu.Unlock()
|
||||
if s.streamed == nil {
|
||||
s.streamed = make(map[string]string)
|
||||
}
|
||||
s.streamed[outTrackID] = content
|
||||
}
|
||||
|
||||
// lastStreamedContent returns the text a card currently shows, or "" when the
|
||||
// card was created but never received a partial frame.
|
||||
func (s *CardSender) lastStreamedContent(outTrackID string) string {
|
||||
s.contentMu.Lock()
|
||||
defer s.contentMu.Unlock()
|
||||
return s.streamed[outTrackID]
|
||||
}
|
||||
|
||||
// forgetStreamedContent drops the remembered text of a finalized card.
|
||||
func (s *CardSender) forgetStreamedContent(outTrackID string) {
|
||||
s.contentMu.Lock()
|
||||
defer s.contentMu.Unlock()
|
||||
delete(s.streamed, outTrackID)
|
||||
}
|
||||
|
||||
// cardParamMap builds the template variables for a given answer. The plain card
|
||||
// update must receive exactly the same variables as the create call: the update
|
||||
// replaces the whole param map, so anything omitted here would be cleared on the
|
||||
// delivered card.
|
||||
func cardParamMap(content string) map[string]any {
|
||||
return map[string]any{
|
||||
"content": content,
|
||||
"feedbackSelected": "",
|
||||
"feedbackReasonSelected": "",
|
||||
"showFeedbackReasons": "false",
|
||||
"feedbackMessage": "",
|
||||
cardWideScreenJSONKey: cardWideScreenJSON,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *CardSender) buildCardBody(req CardRequest, outTrackID string) map[string]any {
|
||||
@@ -241,14 +371,7 @@ func (s *CardSender) buildCardBody(req CardRequest, outTrackID string) map[strin
|
||||
"outTrackId": outTrackID,
|
||||
"callbackType": "STREAM",
|
||||
"cardData": map[string]any{
|
||||
"cardParamMap": map[string]any{
|
||||
"content": req.Content,
|
||||
"feedbackSelected": "",
|
||||
"feedbackReasonSelected": "",
|
||||
"showFeedbackReasons": "false",
|
||||
"feedbackMessage": "",
|
||||
cardWideScreenJSONKey: cardWideScreenJSON,
|
||||
},
|
||||
"cardParamMap": cardParamMap(req.Content),
|
||||
},
|
||||
"userIdType": 1,
|
||||
}
|
||||
@@ -370,6 +493,29 @@ func (s *CardSender) finalizeCard(ctx context.Context, outTrackID, content, toke
|
||||
return validateDingTalkBusinessResponse(payload)
|
||||
}
|
||||
|
||||
// replaceCardContent writes an already finished answer through the plain card
|
||||
// update API (PUT /v1.0/card/instances). The payload carries the same template
|
||||
// variables as the create call and replaces the whole param map, so the card
|
||||
// keeps its wide-screen configuration and feedback placeholders. Because this
|
||||
// write does not go through the streaming endpoint, the client applies the
|
||||
// terminal content without running a typewriter animation over it.
|
||||
func (s *CardSender) replaceCardContent(ctx context.Context, outTrackID, content, token string) error {
|
||||
body := map[string]any{
|
||||
"outTrackId": outTrackID,
|
||||
"cardData": map[string]any{
|
||||
"cardParamMap": cardParamMap(content),
|
||||
},
|
||||
"cardUpdateOptions": map[string]any{
|
||||
"updateCardDataByKey": false,
|
||||
},
|
||||
}
|
||||
payload, err := s.doJSON(ctx, http.MethodPut, s.updateURL, body, token)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return validateDingTalkBusinessResponse(payload)
|
||||
}
|
||||
|
||||
func (s *CardSender) postJSON(ctx context.Context, endpoint string, value any, token string) ([]byte, error) {
|
||||
return s.doJSON(ctx, http.MethodPost, endpoint, value, token)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,131 @@
|
||||
package dingtalk
|
||||
|
||||
import "strings"
|
||||
|
||||
// CardPartialUpdatesMode selects how much of an answer is written into the card
|
||||
// while the answer is still being produced.
|
||||
//
|
||||
// partial (default) → the card is created with the "processing" placeholder and
|
||||
// partial answer/progress frames are pushed as they arrive,
|
||||
// throttled by cardUpdateInterval (historical behaviour)
|
||||
// final → same placeholder card, but no partial frame at all: the
|
||||
// finished answer lands in the single final update
|
||||
// direct → no placeholder card; one card is created carrying the
|
||||
// finished answer
|
||||
//
|
||||
// "final" and "direct" remove the redraw loop (and with it the frozen card and
|
||||
// the end-of-answer jump) at the cost of not showing interim progress.
|
||||
type CardPartialUpdatesMode string
|
||||
|
||||
const (
|
||||
CardPartialUpdatesPartial CardPartialUpdatesMode = "partial"
|
||||
CardPartialUpdatesFinal CardPartialUpdatesMode = "final"
|
||||
CardPartialUpdatesDirect CardPartialUpdatesMode = "direct"
|
||||
)
|
||||
|
||||
// ParseCardPartialUpdatesMode maps DINGTALK_CARD_PARTIAL_UPDATES onto a mode.
|
||||
// Unset or unrecognised values keep the historical partial streaming so an
|
||||
// environment without the variable renders exactly as before.
|
||||
func ParseCardPartialUpdatesMode(raw string) CardPartialUpdatesMode {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case "final", "off", "0", "false", "no", "answer", "answer_only":
|
||||
return CardPartialUpdatesFinal
|
||||
case "direct", "once", "single":
|
||||
return CardPartialUpdatesDirect
|
||||
default:
|
||||
return CardPartialUpdatesPartial
|
||||
}
|
||||
}
|
||||
|
||||
// streamsPartialFrames reports whether partial answer/progress frames may be
|
||||
// pushed into the card. Only the explicit "final" and "direct" modes suppress
|
||||
// them, so an unset or unknown mode keeps the historical streaming.
|
||||
func (m CardPartialUpdatesMode) streamsPartialFrames() bool {
|
||||
return m != CardPartialUpdatesFinal && m != CardPartialUpdatesDirect
|
||||
}
|
||||
|
||||
// placeholderCardRequested reports whether the card is created up front with the
|
||||
// "processing" placeholder. Only "direct" waits for the finished answer before
|
||||
// creating the card.
|
||||
func (m CardPartialUpdatesMode) placeholderCardRequested() bool {
|
||||
return m != CardPartialUpdatesDirect
|
||||
}
|
||||
|
||||
// CardFinalWriteMode selects how the terminal write of a finished answer lands
|
||||
// in the card.
|
||||
//
|
||||
// streaming (default) → the finished answer is written with the card streaming
|
||||
// API and isFinalize=true (historical behaviour). DingTalk
|
||||
// renders content that arrives through the streaming
|
||||
// endpoint with a typewriter animation driven by the
|
||||
// client itself: an answer that is already complete on the
|
||||
// server can still be replayed character by character, and
|
||||
// a client that is not in the foreground pauses that
|
||||
// animation - which reads as "the streaming stopped and it
|
||||
// resumes when I switch back", long after the server has
|
||||
// pushed the final frame.
|
||||
// update → after the streaming finalize, the same finished answer is
|
||||
// written once more through the plain card update API
|
||||
// (PUT /v1.0/card/instances). That write carries no
|
||||
// streaming semantics, so the terminal state is applied to
|
||||
// the card without leaving a typewriter queue behind.
|
||||
// NOTE: this order is the one that blanked a live card on
|
||||
// 2026-09-20 (the finalize frame starts the client's
|
||||
// typewriter and the plain write lands in the middle of it);
|
||||
// kept only to reproduce that regression.
|
||||
// refresh → the finished answer is written through the plain card
|
||||
// update API FIRST, so the card shows it without any
|
||||
// typewriter queue, and only then is the stream closed by a
|
||||
// finalize frame that carries exactly the same text. The
|
||||
// client has no delta left to animate, so the answer is
|
||||
// simply there when it arrives.
|
||||
// close-refresh → the stream is closed with a frame carrying the text the
|
||||
// card already shows (zero delta, nothing to animate) and
|
||||
// the finished answer is written through the plain update
|
||||
// API afterwards. Used when the plain write must land on a
|
||||
// stream that is already closed and quiet.
|
||||
//
|
||||
// The default stays "streaming" so an environment without the variable behaves
|
||||
// exactly as before; DINGTALK_CARD_FINAL_WRITE selects the other mode.
|
||||
type CardFinalWriteMode string
|
||||
|
||||
const (
|
||||
CardFinalWriteStreaming CardFinalWriteMode = "streaming"
|
||||
CardFinalWriteUpdate CardFinalWriteMode = "update"
|
||||
CardFinalWriteRefresh CardFinalWriteMode = "refresh"
|
||||
CardFinalWriteCloseRefresh CardFinalWriteMode = "close-refresh"
|
||||
)
|
||||
|
||||
// ParseCardFinalWriteMode maps DINGTALK_CARD_FINAL_WRITE onto a mode. Unset or
|
||||
// unrecognised values keep the historical streaming finalize.
|
||||
func ParseCardFinalWriteMode(raw string) CardFinalWriteMode {
|
||||
switch strings.ToLower(strings.TrimSpace(raw)) {
|
||||
case "update", "plain", "plain_update", "on", "true", "1", "yes":
|
||||
return CardFinalWriteUpdate
|
||||
case "refresh", "refresh_first", "plain_first", "refresh_then_finalize", "refresh-first", "plain-first":
|
||||
return CardFinalWriteRefresh
|
||||
case "close-refresh", "close_refresh", "finalize_then_refresh", "refresh_after_finalize", "close-then-refresh":
|
||||
return CardFinalWriteCloseRefresh
|
||||
default:
|
||||
return CardFinalWriteStreaming
|
||||
}
|
||||
}
|
||||
|
||||
// alsoPlainUpdates reports whether the finished answer must additionally be
|
||||
// written through the plain (non-streaming) card update endpoint after the
|
||||
// streaming finalize.
|
||||
func (m CardFinalWriteMode) alsoPlainUpdates() bool {
|
||||
return m == CardFinalWriteUpdate
|
||||
}
|
||||
|
||||
// refreshBeforeFinalize reports whether the finished answer goes through the
|
||||
// plain card update endpoint before the stream is closed.
|
||||
func (m CardFinalWriteMode) refreshBeforeFinalize() bool {
|
||||
return m == CardFinalWriteRefresh
|
||||
}
|
||||
|
||||
// closeThenRefresh reports whether the stream is closed on the text already on
|
||||
// the card and the finished answer is refreshed afterwards.
|
||||
func (m CardFinalWriteMode) closeThenRefresh() bool {
|
||||
return m == CardFinalWriteCloseRefresh
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
package dingtalk
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestParseCardPartialUpdatesMode(t *testing.T) {
|
||||
tests := []struct {
|
||||
raw string
|
||||
want CardPartialUpdatesMode
|
||||
}{
|
||||
{"", CardPartialUpdatesPartial},
|
||||
{"on", CardPartialUpdatesPartial},
|
||||
{"PARTIAL", CardPartialUpdatesPartial},
|
||||
{"sometimes", CardPartialUpdatesPartial},
|
||||
{"final", CardPartialUpdatesFinal},
|
||||
{" FINAL ", CardPartialUpdatesFinal},
|
||||
{"off", CardPartialUpdatesFinal},
|
||||
{"0", CardPartialUpdatesFinal},
|
||||
{"false", CardPartialUpdatesFinal},
|
||||
{"no", CardPartialUpdatesFinal},
|
||||
{"answer", CardPartialUpdatesFinal},
|
||||
{"answer_only", CardPartialUpdatesFinal},
|
||||
{"direct", CardPartialUpdatesDirect},
|
||||
{"Single", CardPartialUpdatesDirect},
|
||||
{"once", CardPartialUpdatesDirect},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if got := ParseCardPartialUpdatesMode(tt.raw); got != tt.want {
|
||||
t.Fatalf("ParseCardPartialUpdatesMode(%q)=%q want %q", tt.raw, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardPartialUpdatesModeFramesAndPlaceholder(t *testing.T) {
|
||||
tests := []struct {
|
||||
mode CardPartialUpdatesMode
|
||||
streamsPartials bool
|
||||
placeholderIssued bool
|
||||
}{
|
||||
{CardPartialUpdatesPartial, true, true},
|
||||
{CardPartialUpdatesFinal, false, true},
|
||||
{CardPartialUpdatesDirect, false, false},
|
||||
// An unset mode must keep the historical streaming behaviour.
|
||||
{"", true, true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if got := tt.mode.streamsPartialFrames(); got != tt.streamsPartials {
|
||||
t.Fatalf("mode=%q streamsPartialFrames=%v want %v", tt.mode, got, tt.streamsPartials)
|
||||
}
|
||||
if got := tt.mode.placeholderCardRequested(); got != tt.placeholderIssued {
|
||||
t.Fatalf("mode=%q placeholderCardRequested=%v want %v", tt.mode, got, tt.placeholderIssued)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseCardFinalWriteMode(t *testing.T) {
|
||||
tests := []struct {
|
||||
raw string
|
||||
want CardFinalWriteMode
|
||||
}{
|
||||
{"", CardFinalWriteStreaming},
|
||||
{"streaming", CardFinalWriteStreaming},
|
||||
{"stream", CardFinalWriteStreaming},
|
||||
{"sometimes", CardFinalWriteStreaming},
|
||||
{"0", CardFinalWriteStreaming},
|
||||
{"off", CardFinalWriteStreaming},
|
||||
{"update", CardFinalWriteUpdate},
|
||||
{" UPDATE ", CardFinalWriteUpdate},
|
||||
{"plain", CardFinalWriteUpdate},
|
||||
{"plain_update", CardFinalWriteUpdate},
|
||||
{"on", CardFinalWriteUpdate},
|
||||
{"true", CardFinalWriteUpdate},
|
||||
{"yes", CardFinalWriteUpdate},
|
||||
{"1", CardFinalWriteUpdate},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
if got := ParseCardFinalWriteMode(tt.raw); got != tt.want {
|
||||
t.Fatalf("ParseCardFinalWriteMode(%q)=%q want %q", tt.raw, got, tt.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteModeAlsoPlainUpdates(t *testing.T) {
|
||||
// Only the explicit update mode adds the plain write; the zero value and the
|
||||
// historical streaming mode must leave the single-write behaviour untouched.
|
||||
for mode, want := range map[CardFinalWriteMode]bool{
|
||||
CardFinalWriteStreaming: false,
|
||||
CardFinalWriteUpdate: true,
|
||||
"": false,
|
||||
} {
|
||||
if got := mode.alsoPlainUpdates(); got != want {
|
||||
t.Fatalf("mode=%q alsoPlainUpdates=%v want %v", mode, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,461 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helper: test server with token + create + streaming + plain update endpoints
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type finalWriteTestServer struct {
|
||||
server *httptest.Server
|
||||
|
||||
createCalls int
|
||||
streamCalls int
|
||||
updateCalls int
|
||||
|
||||
lastCreateBody map[string]any
|
||||
lastStreamBody map[string]any
|
||||
|
||||
lastUpdateBody map[string]any
|
||||
lastUpdateMethod string
|
||||
lastUpdateHeader string
|
||||
|
||||
// callSeq records the endpoint order ("create"/"stream"/"update") so the
|
||||
// write ORDER of a terminal answer can be asserted, not just the counts.
|
||||
callSeq []string
|
||||
streamBodies []map[string]any
|
||||
|
||||
streamStatus int
|
||||
streamResponse any
|
||||
updateStatus int
|
||||
updateResponse any
|
||||
createResponse any
|
||||
}
|
||||
|
||||
func newFinalWriteTestServer() *finalWriteTestServer {
|
||||
ts := &finalWriteTestServer{
|
||||
streamStatus: http.StatusOK,
|
||||
updateStatus: http.StatusOK,
|
||||
streamResponse: map[string]any{"result": "ok"},
|
||||
updateResponse: map[string]any{"result": true, "success": true},
|
||||
createResponse: map[string]any{"result": "ok"},
|
||||
}
|
||||
ts.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.URL.Path {
|
||||
case "/token":
|
||||
json.NewEncoder(w).Encode(map[string]any{"accessToken": "test-access-token", "expireIn": 3600})
|
||||
case "/create":
|
||||
ts.createCalls++
|
||||
ts.callSeq = append(ts.callSeq, "create")
|
||||
ts.lastCreateBody = make(map[string]any)
|
||||
json.NewDecoder(r.Body).Decode(&ts.lastCreateBody)
|
||||
json.NewEncoder(w).Encode(ts.createResponse)
|
||||
case "/stream":
|
||||
ts.streamCalls++
|
||||
ts.callSeq = append(ts.callSeq, "stream")
|
||||
ts.lastStreamBody = make(map[string]any)
|
||||
json.NewDecoder(r.Body).Decode(&ts.lastStreamBody)
|
||||
ts.streamBodies = append(ts.streamBodies, ts.lastStreamBody)
|
||||
if ts.streamStatus != http.StatusOK {
|
||||
w.WriteHeader(ts.streamStatus)
|
||||
return
|
||||
}
|
||||
json.NewEncoder(w).Encode(ts.streamResponse)
|
||||
case "/update":
|
||||
ts.updateCalls++
|
||||
ts.callSeq = append(ts.callSeq, "update")
|
||||
ts.lastUpdateMethod = r.Method
|
||||
ts.lastUpdateHeader = r.Header.Get("x-acs-dingtalk-access-token")
|
||||
ts.lastUpdateBody = make(map[string]any)
|
||||
json.NewDecoder(r.Body).Decode(&ts.lastUpdateBody)
|
||||
if ts.updateStatus != http.StatusOK {
|
||||
w.WriteHeader(ts.updateStatus)
|
||||
return
|
||||
}
|
||||
json.NewEncoder(w).Encode(ts.updateResponse)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
}))
|
||||
return ts
|
||||
}
|
||||
|
||||
func (ts *finalWriteTestServer) Close() { ts.server.Close() }
|
||||
|
||||
func (ts *finalWriteTestServer) sender(t *testing.T, mode CardFinalWriteMode) *CardSender {
|
||||
t.Helper()
|
||||
sender, err := newCardSenderForTest("id", "secret", "tpl", ts.server.Client(),
|
||||
ts.server.URL+"/token", ts.server.URL+"/create", WithCardFinalWrite(mode))
|
||||
if err != nil {
|
||||
t.Fatalf("newCardSenderForTest: %v", err)
|
||||
}
|
||||
return sender
|
||||
}
|
||||
|
||||
func (ts *finalWriteTestServer) paramMap(t *testing.T) map[string]any {
|
||||
t.Helper()
|
||||
cardData, ok := ts.lastUpdateBody["cardData"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("update body has no cardData: %#v", ts.lastUpdateBody)
|
||||
}
|
||||
paramMap, ok := cardData["cardParamMap"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("update body has no cardParamMap: %#v", cardData)
|
||||
}
|
||||
return paramMap
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Default (streaming) mode keeps the historical single-write behaviour
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCardFinalWriteStreamingModeDoesNotTouchPlainUpdate(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
|
||||
sender := ts.sender(t, CardFinalWriteStreaming)
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "答案", true, false); err != nil {
|
||||
t.Fatalf("UpdateCard: %v", err)
|
||||
}
|
||||
if ts.streamCalls != 1 {
|
||||
t.Fatalf("streamCalls=%d, want 1", ts.streamCalls)
|
||||
}
|
||||
if ts.updateCalls != 0 {
|
||||
t.Fatalf("updateCalls=%d, want 0 in streaming mode", ts.updateCalls)
|
||||
}
|
||||
// The zero value (option not passed at all) must behave the same.
|
||||
ts2 := newFinalWriteTestServer()
|
||||
defer ts2.Close()
|
||||
plain, err := newCardSenderForTest("id", "secret", "tpl", ts2.server.Client(),
|
||||
ts2.server.URL+"/token", ts2.server.URL+"/create")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := plain.UpdateCard(context.Background(), "track-2", "答案", true, false); err != nil {
|
||||
t.Fatalf("UpdateCard: %v", err)
|
||||
}
|
||||
if ts2.updateCalls != 0 {
|
||||
t.Fatalf("updateCalls=%d, want 0 with the zero-value mode", ts2.updateCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// update mode: one plain write after the streaming finalize, same variables
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCardFinalWriteUpdateModeWritesPlainUpdateAfterFinalize(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
|
||||
sender := ts.sender(t, CardFinalWriteUpdate)
|
||||
if _, err := sender.StartCard(context.Background(), CardRequest{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "user-1",
|
||||
Content: "占位",
|
||||
}); err != nil {
|
||||
t.Fatalf("StartCard: %v", err)
|
||||
}
|
||||
|
||||
const answer = "| 型号 | 速度 |\n| --- | --- |\n| CRV3 | 180°/s |"
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", answer, true, false); err != nil {
|
||||
t.Fatalf("UpdateCard: %v", err)
|
||||
}
|
||||
|
||||
if ts.streamCalls != 1 || ts.updateCalls != 1 {
|
||||
t.Fatalf("streamCalls=%d updateCalls=%d, want 1/1", ts.streamCalls, ts.updateCalls)
|
||||
}
|
||||
if ts.lastUpdateMethod != http.MethodPut {
|
||||
t.Fatalf("plain update method=%q, want PUT", ts.lastUpdateMethod)
|
||||
}
|
||||
if ts.lastUpdateHeader != "test-access-token" {
|
||||
t.Fatalf("plain update token header=%q", ts.lastUpdateHeader)
|
||||
}
|
||||
if ts.lastUpdateBody["outTrackId"] != ts.lastStreamBody["outTrackId"] {
|
||||
t.Fatalf("outTrackId mismatch: update=%v stream=%v", ts.lastUpdateBody["outTrackId"], ts.lastStreamBody["outTrackId"])
|
||||
}
|
||||
options, ok := ts.lastUpdateBody["cardUpdateOptions"].(map[string]any)
|
||||
if !ok || options["updateCardDataByKey"] != false {
|
||||
t.Fatalf("cardUpdateOptions=%#v, want updateCardDataByKey=false", ts.lastUpdateBody["cardUpdateOptions"])
|
||||
}
|
||||
|
||||
paramMap := ts.paramMap(t)
|
||||
if paramMap["content"] != answer {
|
||||
t.Fatalf("plain update content=%v, want the finished answer", paramMap["content"])
|
||||
}
|
||||
if paramMap[cardWideScreenJSONKey] != cardWideScreenJSON {
|
||||
t.Fatalf("plain update lost the wide-screen config: %#v", paramMap[cardWideScreenJSONKey])
|
||||
}
|
||||
|
||||
// The plain write replaces the whole param map, so it must carry exactly the
|
||||
// same variables as the create call.
|
||||
createParamMap, ok := ts.lastCreateBody["cardData"].(map[string]any)["cardParamMap"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("create body has no cardParamMap: %#v", ts.lastCreateBody)
|
||||
}
|
||||
if len(createParamMap) != len(paramMap) {
|
||||
t.Fatalf("param map size differs: create=%d update=%d", len(createParamMap), len(paramMap))
|
||||
}
|
||||
for key := range createParamMap {
|
||||
if _, present := paramMap[key]; !present {
|
||||
t.Fatalf("plain update is missing create variable %q", key)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteUpdateModeSkipsNonFinalAndErrorFrames(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
sender := ts.sender(t, CardFinalWriteUpdate)
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil {
|
||||
t.Fatalf("partial UpdateCard: %v", err)
|
||||
}
|
||||
if ts.updateCalls != 0 {
|
||||
t.Fatalf("updateCalls=%d after a partial frame, want 0", ts.updateCalls)
|
||||
}
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "出错了", true, true); err != nil {
|
||||
t.Fatalf("error UpdateCard: %v", err)
|
||||
}
|
||||
if ts.updateCalls != 0 {
|
||||
t.Fatalf("updateCalls=%d after an error card, want 0", ts.updateCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// A failing follow-up write must never fail an already delivered answer
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCardFinalWriteUpdateModeSwallowsPlainUpdateFailure(t *testing.T) {
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
status int
|
||||
response any
|
||||
}{
|
||||
{"http error", http.StatusInternalServerError, nil},
|
||||
{"business failure", http.StatusOK, map[string]any{"success": false, "errcode": 40035}},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
ts.updateStatus = tt.status
|
||||
if tt.response != nil {
|
||||
ts.updateResponse = tt.response
|
||||
}
|
||||
sender := ts.sender(t, CardFinalWriteUpdate)
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "答案", true, false); err != nil {
|
||||
t.Fatalf("UpdateCard must succeed once the streaming finalize landed, got %v", err)
|
||||
}
|
||||
if ts.updateCalls != 1 {
|
||||
t.Fatalf("updateCalls=%d, want 1", ts.updateCalls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteUpdateModeSkipsPlainUpdateWhenFinalizeFails(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
ts.streamStatus = http.StatusInternalServerError
|
||||
sender := ts.sender(t, CardFinalWriteUpdate)
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "答案", true, false); err == nil {
|
||||
t.Fatal("expected the streaming finalize failure to surface")
|
||||
}
|
||||
if ts.updateCalls != 0 {
|
||||
t.Fatalf("updateCalls=%d after a failed finalize, want 0", ts.updateCalls)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// SendFinalCard (direct delivery) honours the mode too
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func TestCardFinalWriteUpdateModeAppliesToSendFinalCard(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
sender := ts.sender(t, CardFinalWriteUpdate)
|
||||
|
||||
if _, err := sender.SendFinalCard(context.Background(), CardRequest{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "user-1",
|
||||
Content: "答案",
|
||||
}); err != nil {
|
||||
t.Fatalf("SendFinalCard: %v", err)
|
||||
}
|
||||
if ts.createCalls != 1 || ts.streamCalls != 1 || ts.updateCalls != 1 {
|
||||
t.Fatalf("create/stream/update=%d/%d/%d, want 1/1/1", ts.createCalls, ts.streamCalls, ts.updateCalls)
|
||||
}
|
||||
if ts.paramMap(t)["content"] != "答案" {
|
||||
t.Fatalf("plain update content=%v", ts.paramMap(t)["content"])
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Order modes: refresh (plain first) and close-refresh (zero-delta finalize
|
||||
// first). Both exist because the client replays streamed text with its own
|
||||
// typewriter animation, so the ORDER of the two terminal writes decides what
|
||||
// the user sees when the answer is ready.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
func wantCallSeq(t *testing.T, got, want []string) {
|
||||
t.Helper()
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("call sequence=%v, want %v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("call sequence=%v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteRefreshModeRefreshesBeforeFinalize(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
sender := ts.sender(t, CardFinalWriteRefresh)
|
||||
|
||||
if _, err := sender.StartCard(context.Background(), CardRequest{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "user-1",
|
||||
Content: "占位",
|
||||
}); err != nil {
|
||||
t.Fatalf("StartCard: %v", err)
|
||||
}
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil {
|
||||
t.Fatalf("partial UpdateCard: %v", err)
|
||||
}
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil {
|
||||
t.Fatalf("terminal UpdateCard: %v", err)
|
||||
}
|
||||
|
||||
// The finished answer goes out through the plain update BEFORE the frame
|
||||
// that closes the stream.
|
||||
wantCallSeq(t, ts.callSeq, []string{"create", "stream", "update", "stream"})
|
||||
if got := ts.paramMap(t)["content"]; got != "完整答案" {
|
||||
t.Fatalf("plain refresh content=%v, want the finished answer", got)
|
||||
}
|
||||
if len(ts.streamBodies) != 2 {
|
||||
t.Fatalf("stream frames=%d, want 2", len(ts.streamBodies))
|
||||
}
|
||||
closing := ts.streamBodies[1]
|
||||
if closing["content"] != "完整答案" || closing["isFinalize"] != true {
|
||||
t.Fatalf("closing frame=%v, want the finished answer with isFinalize=true", closing)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteRefreshModeKeepsDeliveryWhenPlainRefreshFails(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
ts.updateStatus = http.StatusInternalServerError
|
||||
sender := ts.sender(t, CardFinalWriteRefresh)
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil {
|
||||
t.Fatalf("a failed plain refresh must not fail the delivery, got %v", err)
|
||||
}
|
||||
if ts.updateCalls != 1 || ts.streamCalls != 1 {
|
||||
t.Fatalf("update/stream=%d/%d, want 1/1 (the closing frame still goes out)", ts.updateCalls, ts.streamCalls)
|
||||
}
|
||||
if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "完整答案" {
|
||||
t.Fatalf("closing frame=%v", ts.lastStreamBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteCloseRefreshModeClosesOnShownTextThenRefreshes(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
sender := ts.sender(t, CardFinalWriteCloseRefresh)
|
||||
|
||||
if _, err := sender.StartCard(context.Background(), CardRequest{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "user-1",
|
||||
Content: "占位",
|
||||
}); err != nil {
|
||||
t.Fatalf("StartCard: %v", err)
|
||||
}
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil {
|
||||
t.Fatalf("partial UpdateCard: %v", err)
|
||||
}
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil {
|
||||
t.Fatalf("terminal UpdateCard: %v", err)
|
||||
}
|
||||
|
||||
// The closing frame carries the text the card already shows (zero delta, so
|
||||
// the client has nothing to animate) and the finished answer follows through
|
||||
// the plain update.
|
||||
wantCallSeq(t, ts.callSeq, []string{"create", "stream", "stream", "update"})
|
||||
if len(ts.streamBodies) != 2 {
|
||||
t.Fatalf("stream frames=%d, want 2", len(ts.streamBodies))
|
||||
}
|
||||
closing := ts.streamBodies[1]
|
||||
if closing["content"] != "部分" || closing["isFinalize"] != true {
|
||||
t.Fatalf("closing frame=%v, want the previously shown text with isFinalize=true", closing)
|
||||
}
|
||||
if got := ts.paramMap(t)["content"]; got != "完整答案" {
|
||||
t.Fatalf("plain refresh content=%v, want the finished answer", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteCloseRefreshModeRefreshesFirstWhenNothingStreamed(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
sender := ts.sender(t, CardFinalWriteCloseRefresh)
|
||||
|
||||
// A card that never received a partial frame has no zero-delta frame to
|
||||
// close on, so the refresh has to go first.
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil {
|
||||
t.Fatalf("terminal UpdateCard: %v", err)
|
||||
}
|
||||
wantCallSeq(t, ts.callSeq, []string{"update", "stream"})
|
||||
if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "完整答案" {
|
||||
t.Fatalf("closing frame=%v", ts.lastStreamBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteCloseRefreshModeKeepsDeliveryWhenPlainRefreshFails(t *testing.T) {
|
||||
ts := newFinalWriteTestServer()
|
||||
defer ts.Close()
|
||||
ts.updateStatus = http.StatusInternalServerError
|
||||
sender := ts.sender(t, CardFinalWriteCloseRefresh)
|
||||
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "部分", false, false); err != nil {
|
||||
t.Fatalf("partial UpdateCard: %v", err)
|
||||
}
|
||||
if err := sender.UpdateCard(context.Background(), "track-1", "完整答案", true, false); err != nil {
|
||||
t.Fatalf("a failed plain refresh must not fail the delivery, got %v", err)
|
||||
}
|
||||
wantCallSeq(t, ts.callSeq, []string{"stream", "stream", "update"})
|
||||
if ts.lastStreamBody["isFinalize"] != true || ts.lastStreamBody["content"] != "部分" {
|
||||
t.Fatalf("closing frame=%v", ts.lastStreamBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFinalWriteParserAcceptsOrderValues(t *testing.T) {
|
||||
for raw, want := range map[string]CardFinalWriteMode{
|
||||
"": CardFinalWriteStreaming,
|
||||
"streaming": CardFinalWriteStreaming,
|
||||
"off": CardFinalWriteStreaming,
|
||||
"nonsense": CardFinalWriteStreaming,
|
||||
"update": CardFinalWriteUpdate,
|
||||
"refresh": CardFinalWriteRefresh,
|
||||
"plain_first": CardFinalWriteRefresh,
|
||||
"refresh-first": CardFinalWriteRefresh,
|
||||
"close_refresh": CardFinalWriteCloseRefresh,
|
||||
"close-refresh": CardFinalWriteCloseRefresh,
|
||||
"CLOSE-REFRESH ": CardFinalWriteCloseRefresh,
|
||||
} {
|
||||
if got := ParseCardFinalWriteMode(raw); got != want {
|
||||
t.Fatalf("ParseCardFinalWriteMode(%q)=%q, want %q", raw, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
|
||||
)
|
||||
|
||||
// pacedEventQA replays the answer events of a real turn and sleeps between them
|
||||
// so the asynchronous 500ms frame pump gets a chance to fire. It is the
|
||||
// sensitivity control for the frame assertions below: with mode "partial" the
|
||||
// very same fixture must produce partial frames.
|
||||
type pacedEventQA struct {
|
||||
fakeStreamingQA
|
||||
events []weknora.StreamEvent
|
||||
hold time.Duration
|
||||
eventsCalled bool
|
||||
}
|
||||
|
||||
func (f *pacedEventQA) AskEvents(_ context.Context, req core.QARequest, observer weknora.StreamObserver) (core.QAResult, error) {
|
||||
f.eventsCalled = true
|
||||
f.lastReq = req
|
||||
for _, event := range f.events {
|
||||
if observer != nil {
|
||||
observer(event)
|
||||
}
|
||||
time.Sleep(f.hold)
|
||||
}
|
||||
return f.result, f.err
|
||||
}
|
||||
|
||||
func TestStreamingModesControlPartialCardFrames(t *testing.T) {
|
||||
rawAnswer := "最终答案内容"
|
||||
tests := []struct {
|
||||
name string
|
||||
mode CardPartialUpdatesMode
|
||||
wantPartials bool
|
||||
}{
|
||||
{"partial streams frames", CardPartialUpdatesPartial, true},
|
||||
{"final suppresses frames", CardPartialUpdatesFinal, false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
card := &fakeStreamingCard{startID: "track-mode"}
|
||||
qa := &pacedEventQA{
|
||||
fakeStreamingQA: fakeStreamingQA{result: core.QAResult{Answer: rawAnswer}},
|
||||
events: []weknora.StreamEvent{
|
||||
{ResponseType: "tool_call", ToolName: "knowledge_search"},
|
||||
{ResponseType: "references", ReferenceCount: 2},
|
||||
{ResponseType: "answer", FullAnswer: "最终"},
|
||||
{ResponseType: "answer", FullAnswer: rawAnswer, Done: true},
|
||||
},
|
||||
hold: 5 * time.Millisecond,
|
||||
}
|
||||
reply := &fakeReply{}
|
||||
h, err := NewHandlerWithOptions(
|
||||
TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"},
|
||||
qa, reply, HandlerOptions{CardSender: card, PartialUpdates: tt.mode})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
h.cardUpdateInterval = time.Millisecond
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
|
||||
if !card.startCalled || card.startReq.Content != streamingPlaceholder {
|
||||
t.Fatalf("placeholder card must be created: called=%v content=%q", card.startCalled, card.startReq.Content)
|
||||
}
|
||||
var partials, finals int
|
||||
for _, update := range card.updates {
|
||||
if update.isFinalize {
|
||||
finals++
|
||||
continue
|
||||
}
|
||||
partials++
|
||||
}
|
||||
if got := partials > 0; got != tt.wantPartials {
|
||||
t.Fatalf("partial frames=%d want frames=%v updates=%+v", partials, tt.wantPartials, card.updates)
|
||||
}
|
||||
if finals != 1 {
|
||||
t.Fatalf("final updates=%d want 1 updates=%+v", finals, card.updates)
|
||||
}
|
||||
last := card.updates[len(card.updates)-1]
|
||||
if !last.isFinalize || last.isError || last.content != rawAnswer {
|
||||
t.Fatalf("final update=%+v want the complete answer", last)
|
||||
}
|
||||
if reply.called {
|
||||
t.Fatal("SendMarkdown must not be called")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDirectModeDeliversOneCardWithoutPlaceholder(t *testing.T) {
|
||||
rawAnswer := "直接输出答案"
|
||||
card := &fakeStreamingCard{finalID: "track-direct"}
|
||||
qa := &fakeEventQA{
|
||||
fakeStreamingQA: fakeStreamingQA{result: core.QAResult{Answer: rawAnswer, TurnID: "turn-1", AnswerMessageID: "msg-1"}},
|
||||
events: []weknora.StreamEvent{
|
||||
{ResponseType: "tool_call", ToolName: "query_understand"},
|
||||
{ResponseType: "answer", FullAnswer: "直接"},
|
||||
{ResponseType: "answer", FullAnswer: rawAnswer, Done: true},
|
||||
},
|
||||
}
|
||||
reply := &fakeReply{}
|
||||
h, err := NewHandlerWithOptions(
|
||||
TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"},
|
||||
qa, reply, HandlerOptions{CardSender: card, PartialUpdates: CardPartialUpdatesDirect})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
h.cardUpdateInterval = time.Millisecond
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
|
||||
if card.startCalled {
|
||||
t.Fatalf("no placeholder card may be created in direct mode: %q", card.startReq.Content)
|
||||
}
|
||||
if len(card.updates) != 0 {
|
||||
t.Fatalf("no card update may be sent before the answer is complete: %+v", card.updates)
|
||||
}
|
||||
if !card.finalCalled || card.finalReq.Content != rawAnswer {
|
||||
t.Fatalf("final card=%v content=%q want %q", card.finalCalled, card.finalReq.Content, rawAnswer)
|
||||
}
|
||||
if reply.called {
|
||||
t.Fatal("SendMarkdown must not be called")
|
||||
}
|
||||
if len(qa.deliveries) == 0 || qa.deliveries[len(qa.deliveries)-1].DeliveryMethod != core.DeliveryMethodCard {
|
||||
t.Fatalf("delivery audit=%+v", qa.deliveries)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDirectModeFallsBackToMarkdownWhenTheAnswerFails(t *testing.T) {
|
||||
card := &fakeStreamingCard{finalID: "track-direct-error"}
|
||||
qa := &fakeEventQA{fakeStreamingQA: fakeStreamingQA{err: errors.New("weknora down")}}
|
||||
reply := &fakeReply{}
|
||||
h, err := NewHandlerWithOptions(
|
||||
TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"},
|
||||
qa, reply, HandlerOptions{CardSender: card, PartialUpdates: CardPartialUpdatesDirect})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if card.finalCalled || card.startCalled {
|
||||
t.Fatal("a failed answer must not create a card")
|
||||
}
|
||||
if !reply.called || reply.lastText != qaErrorReply {
|
||||
t.Fatalf("markdown fallback: called=%v text=%q", reply.called, reply.lastText)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDirectModeAppliesToTheStreamingAnswerPath(t *testing.T) {
|
||||
rawAnswer := "流式接口直接输出"
|
||||
card := &fakeStreamingCard{finalID: "track-direct-stream"}
|
||||
qa := &fakeStreamingQA{chunks: []string{"流式", "接口直接输出"}, result: core.QAResult{Answer: rawAnswer}}
|
||||
reply := &fakeReply{}
|
||||
h, err := NewHandlerWithOptions(
|
||||
TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"},
|
||||
qa, reply, HandlerOptions{CardSender: card, PartialUpdates: CardPartialUpdatesDirect})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
if err := h.Handle(context.Background(), directTextCallback()); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if !qa.streamCalled {
|
||||
t.Fatal("the streaming asker must be used")
|
||||
}
|
||||
if qa.observerCaptured == nil {
|
||||
t.Fatal("the streaming asker must still receive an observer")
|
||||
}
|
||||
if card.startCalled || len(card.updates) != 0 {
|
||||
t.Fatalf("no placeholder/partial card may be created: start=%v updates=%+v", card.startCalled, card.updates)
|
||||
}
|
||||
if !card.finalCalled || card.finalReq.Content != rawAnswer {
|
||||
t.Fatalf("final card=%v content=%q", card.finalCalled, card.finalReq.Content)
|
||||
}
|
||||
}
|
||||
@@ -169,11 +169,17 @@ type HandlerOptions struct {
|
||||
// delivery timeout, otherwise the remaining images are dropped. Zero keeps
|
||||
// finalDeliveryTimeout.
|
||||
OutboundRewriteTimeout time.Duration
|
||||
// WideTableGrouping selects how over-wide answer tables are adapted, per
|
||||
// frame: "" or "on" groups on every frame (historical), "off" never groups,
|
||||
// "stream" groups only while the answer is still streaming and keeps the
|
||||
// real table on the final frame. See tableGroupingForMode.
|
||||
// WideTableGrouping selects how answer tables are adapted, per frame: ""
|
||||
// or "on" groups over-wide tables on every frame (historical), "off" never
|
||||
// groups, "stream" groups only while the answer is still streaming and
|
||||
// keeps the real table on the final frame, "smooth" also groups every
|
||||
// table on streaming frames so a forming table never freezes the card.
|
||||
// See tableGroupingForMode.
|
||||
WideTableGrouping string
|
||||
// PartialUpdates selects whether the answer is streamed into the card.
|
||||
// Empty (or an unknown value) keeps the historical partial streaming; see
|
||||
// CardPartialUpdatesMode.
|
||||
PartialUpdates CardPartialUpdatesMode
|
||||
}
|
||||
|
||||
type Handler struct {
|
||||
@@ -201,6 +207,12 @@ type Handler struct {
|
||||
// keep the real table, which the PC wide-screen card lays out itself.
|
||||
groupTablesPartial bool
|
||||
groupTablesFinal bool
|
||||
// groupAllTablesPartial groups every table on streaming frames (mode
|
||||
// "smooth"): the card keeps growing while a table is still forming.
|
||||
groupAllTablesPartial bool
|
||||
// partialUpdates decides whether the answer is streamed into the card at
|
||||
// all; see CardPartialUpdatesMode.
|
||||
partialUpdates CardPartialUpdatesMode
|
||||
}
|
||||
|
||||
// NewHandler accepts at most one image downloader. A nil/omitted downloader
|
||||
@@ -232,7 +244,7 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
|
||||
if reply == nil {
|
||||
return nil, errors.New("dingtalk handler: markdown sender is required")
|
||||
}
|
||||
groupPartial, groupFinal := tableGroupingForMode(opts.WideTableGrouping)
|
||||
groupPartial, groupFinal, groupAllPartial := tableGroupingForMode(opts.WideTableGrouping)
|
||||
|
||||
return &Handler{
|
||||
cfg: cfg,
|
||||
@@ -254,6 +266,8 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
|
||||
outboundRewriteTimeout: opts.OutboundRewriteTimeout,
|
||||
groupTablesPartial: groupPartial,
|
||||
groupTablesFinal: groupFinal,
|
||||
groupAllTablesPartial: groupAllPartial,
|
||||
partialUpdates: opts.PartialUpdates,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -464,22 +478,37 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
|
||||
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
|
||||
return nil
|
||||
}
|
||||
content := h.rewriteOutboundContent(ctx, result.Answer)
|
||||
return h.deliverAnswerCardDirect(ctx, webhookURL, req, result)
|
||||
}
|
||||
|
||||
// deliverAnswerCardDirect delivers a finished answer as a single card built from
|
||||
// the final content (no partial frames), falling back to the markdown reply when
|
||||
// the card cannot be delivered. The delivery gets a fresh deadline so a long
|
||||
// answer whose QA context already expired still reaches the user.
|
||||
func (h *Handler) deliverAnswerCardDirect(ctx context.Context, webhookURL string, req core.QARequest, result core.QAResult) error {
|
||||
timeout := h.finalDeliveryTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = defaultHTTPTimeout
|
||||
}
|
||||
deliverCtx, cancelDeliver := context.WithTimeout(context.WithoutCancel(ctx), timeout)
|
||||
defer cancelDeliver()
|
||||
|
||||
content := h.rewriteOutboundContent(deliverCtx, result.Answer)
|
||||
|
||||
if h.card != nil {
|
||||
cardReq := CardRequest{
|
||||
ConversationType: req.ConversationType,
|
||||
ExternalUserID: req.ExternalUserID,
|
||||
ExternalChatID: req.ExternalChatID,
|
||||
Content: h.rewriteCardImages(ctx, content, true),
|
||||
Content: h.rewriteCardImages(deliverCtx, content, true),
|
||||
}
|
||||
if outTrackID, cardErr := h.card.SendFinalCard(ctx, cardReq); cardErr == nil {
|
||||
if outTrackID, cardErr := h.card.SendFinalCard(deliverCtx, cardReq); cardErr == nil {
|
||||
h.recordDelivery(ctx, result, core.DeliverySent, core.DeliveryMethodCard, outTrackID, "")
|
||||
return nil
|
||||
}
|
||||
log.Printf("DingTalk card delivery failed, falling back to markdown")
|
||||
}
|
||||
if err := h.sendReply(ctx, webhookURL, content); err != nil {
|
||||
if err := h.sendReply(deliverCtx, webhookURL, content); err != nil {
|
||||
h.recordDelivery(ctx, result, core.DeliveryFailed, "", "", core.DeliveryFailureChannelDelivery)
|
||||
return err
|
||||
}
|
||||
@@ -488,6 +517,20 @@ func (h *Handler) handleNonStreaming(ctx context.Context, webhookURL string, req
|
||||
}
|
||||
|
||||
func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req core.QARequest, streamingCard StreamingCardSender, streamingQA StreamingQAAsker) error {
|
||||
if !h.partialUpdates.placeholderCardRequested() {
|
||||
// Mode "direct": no placeholder card and no partial frames. The answer
|
||||
// is asked for first and delivered as one finished card.
|
||||
result, err := streamingQA.AskStream(ctx, req, func(string) {})
|
||||
if err != nil || strings.TrimSpace(result.Answer) == "" {
|
||||
logQAFailure("stream_direct", errOrEmptyAnswer(err, result.Answer))
|
||||
return h.sendReply(ctx, webhookURL, qaErrorReply)
|
||||
}
|
||||
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
|
||||
return nil
|
||||
}
|
||||
return h.deliverAnswerCardDirect(ctx, webhookURL, req, result)
|
||||
}
|
||||
|
||||
cardReq := CardRequest{
|
||||
ConversationType: req.ConversationType,
|
||||
ExternalUserID: req.ExternalUserID,
|
||||
@@ -523,6 +566,11 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
|
||||
var lastPartialContent string
|
||||
feed := reasoningmode.NewPartialFeed(req.ReasoningExecution)
|
||||
observer := func(fullContent string) {
|
||||
if !h.partialUpdates.streamsPartialFrames() {
|
||||
// Mode "final": the finished answer is the only card content, so the
|
||||
// throttle and the frame itself are skipped.
|
||||
return
|
||||
}
|
||||
visible := feed.OnAnswer(fullContent)
|
||||
if strings.TrimSpace(visible) == "" {
|
||||
return
|
||||
@@ -555,6 +603,20 @@ func (h *Handler) handleStreaming(ctx context.Context, webhookURL string, req co
|
||||
}
|
||||
|
||||
func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, req core.QARequest, streamingCard StreamingCardSender, eventQA EventQAAsker) error {
|
||||
if !h.partialUpdates.placeholderCardRequested() {
|
||||
// Mode "direct": no placeholder card and no partial frames. The answer
|
||||
// is asked for first and delivered as one finished card.
|
||||
result, err := eventQA.AskEvents(ctx, req, func(weknora.StreamEvent) {})
|
||||
if err != nil || strings.TrimSpace(result.Answer) == "" {
|
||||
logQAFailure("event_direct", errOrEmptyAnswer(err, result.Answer))
|
||||
return h.sendReply(ctx, webhookURL, qaErrorReply)
|
||||
}
|
||||
if result.Replayed && result.DeliveryStatus == core.DeliverySent {
|
||||
return nil
|
||||
}
|
||||
return h.deliverAnswerCardDirect(ctx, webhookURL, req, result)
|
||||
}
|
||||
|
||||
cardReq := CardRequest{
|
||||
ConversationType: req.ConversationType,
|
||||
ExternalUserID: req.ExternalUserID,
|
||||
@@ -581,11 +643,20 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
|
||||
return nil
|
||||
}
|
||||
|
||||
pump := newCardUpdatePump(ctx, h.cardUpdateInterval, func(updateCtx context.Context, content string) error {
|
||||
return streamingCard.UpdateCard(updateCtx, outTrackID, content, false, false)
|
||||
})
|
||||
var pump *cardUpdatePump
|
||||
if h.partialUpdates.streamsPartialFrames() {
|
||||
pump = newCardUpdatePump(ctx, h.cardUpdateInterval, func(updateCtx context.Context, content string) error {
|
||||
return streamingCard.UpdateCard(updateCtx, outTrackID, content, false, false)
|
||||
})
|
||||
}
|
||||
feed := reasoningmode.NewPartialFeed(req.ReasoningExecution)
|
||||
result, err := eventQA.AskEvents(ctx, req, func(event weknora.StreamEvent) {
|
||||
if pump == nil {
|
||||
// Mode "final": the finished answer is the only card content, so the
|
||||
// partial frames and the progress lines are dropped instead of being
|
||||
// drawn on every model chunk.
|
||||
return
|
||||
}
|
||||
if event.ResponseType == "answer" {
|
||||
visible := feed.OnAnswer(event.FullAnswer)
|
||||
if strings.TrimSpace(visible) == "" {
|
||||
@@ -612,7 +683,9 @@ func (h *Handler) handleEventStreaming(ctx context.Context, webhookURL string, r
|
||||
})
|
||||
// Stop and drain the asynchronous worker before the synchronous final update
|
||||
// so no stale progress/partial answer can overwrite the terminal card state.
|
||||
pump.Stop()
|
||||
if pump != nil {
|
||||
pump.Stop()
|
||||
}
|
||||
|
||||
if err != nil || strings.TrimSpace(result.Answer) == "" {
|
||||
logQAFailure("event_streaming", errOrEmptyAnswer(err, result.Answer))
|
||||
@@ -699,7 +772,7 @@ func (h *Handler) sanitizeOutboundPartial(ctx context.Context, content string) s
|
||||
if ok {
|
||||
content = sanitizer.SanitizePartial(content)
|
||||
}
|
||||
options := tableNormalizeOptions{groupWideTables: h.groupTablesPartial}
|
||||
options := tableNormalizeOptions{groupWideTables: h.groupTablesPartial, groupAllTables: h.groupAllTablesPartial}
|
||||
content = normalizeMarkdownTables(sanitizeDisplayAnswer(content), options)
|
||||
return h.rewriteCardImages(ctx, stableMarkdownPrefix(content), false)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// smoothAnswer mirrors a live answer shape: prose, a compact table, more prose,
|
||||
// a second table, then a long tail. With the historical modes the streaming
|
||||
// card froze at the first table and the rest arrived in one final frame.
|
||||
func smoothAnswer() string {
|
||||
return "Nova 是新一代工业协作机器人平台。\n\n" +
|
||||
"| Model | Payload | Reach |\n| :--- | :--- | :--- |\n| Nova 3 | 3 kg | 626 mm |\n| Nova 5 | 5 kg | 900 mm |\n\n" +
|
||||
"选型时先看负载与臂展,再看控制柜与接口。\n\n" +
|
||||
"| Interface | Count |\n| :--- | :--- |\n| DI | 16 |\n| DO | 16 |\n\n" +
|
||||
strings.Repeat("最后一段补充说明。", 30)
|
||||
}
|
||||
|
||||
func streamedFrames(t *testing.T, mode, answer string) (frames, lastPartial int) {
|
||||
t.Helper()
|
||||
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "t", Channel: "dingtalk"}, &fakeQA{}, &fakeReply{}, HandlerOptions{WideTableGrouping: mode})
|
||||
if err != nil {
|
||||
t.Fatalf("handler: %v", err)
|
||||
}
|
||||
prev := ""
|
||||
for n := 1; n <= len(answer); n += 7 { // ~model chunk granularity
|
||||
p := h.sanitizeOutboundPartial(context.Background(), answer[:n])
|
||||
if p != prev {
|
||||
frames++
|
||||
prev = p
|
||||
}
|
||||
}
|
||||
return frames, len(prev)
|
||||
}
|
||||
|
||||
// TestSmoothModeKeepsStreamingThroughFormingTable guards the regression where a
|
||||
// half-formed table truncated every streaming frame (stableMarkdownPrefix), so
|
||||
// the card appeared to stop and dumped the remaining bytes at finalize.
|
||||
func TestSmoothModeKeepsStreamingThroughFormingTable(t *testing.T) {
|
||||
answer := smoothAnswer()
|
||||
smoothFrames, smoothLast := streamedFrames(t, "smooth", answer)
|
||||
streamFrames, streamLast := streamedFrames(t, "stream", answer)
|
||||
if smoothFrames <= streamFrames {
|
||||
t.Fatalf("smooth mode must deliver more streaming frames: smooth=%d stream=%d", smoothFrames, streamFrames)
|
||||
}
|
||||
if smoothLast < len(answer)*9/10 {
|
||||
t.Fatalf("smooth streaming must deliver the growing answer: last=%d answer=%d", smoothLast, len(answer))
|
||||
}
|
||||
if streamLast >= len(answer)*9/10 {
|
||||
t.Fatalf("stream mode is expected to hold the table tail back, got last=%d", streamLast)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSmoothModeKeepsRealCompactTableOnFinalFrame pins that the fix only
|
||||
// changes streaming frames: a compact table still lands as a real table.
|
||||
func TestSmoothModeKeepsRealCompactTableOnFinalFrame(t *testing.T) {
|
||||
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "t", Channel: "dingtalk"}, &fakeQA{}, &fakeReply{}, HandlerOptions{WideTableGrouping: "smooth"})
|
||||
if err != nil {
|
||||
t.Fatalf("handler: %v", err)
|
||||
}
|
||||
final := h.rewriteOutboundContent(context.Background(), smoothAnswer())
|
||||
if !strings.Contains(final, "| Model | Payload | Reach |") {
|
||||
t.Fatalf("final frame must keep the real compact table, got %q", final)
|
||||
}
|
||||
}
|
||||
@@ -17,6 +17,11 @@ type tableNormalizeOptions struct {
|
||||
// lists (bold row heading + "- header:value" lines). Disable it when the
|
||||
// client lays out wide tables itself, so the answer keeps its real table.
|
||||
groupWideTables bool
|
||||
// groupAllTables rewrites every table into a grouped list, compact ones
|
||||
// included. Streaming frames use it so a table that is still forming never
|
||||
// freezes the card; the final frame keeps the real table whenever it lays
|
||||
// out safely.
|
||||
groupAllTables bool
|
||||
}
|
||||
|
||||
func defaultTableNormalizeOptions() tableNormalizeOptions {
|
||||
@@ -30,14 +35,19 @@ func defaultTableNormalizeOptions() tableNormalizeOptions {
|
||||
// ""/"on" group on every frame
|
||||
// "off" never group: real tables everywhere
|
||||
// "stream" group while streaming, real table on the final frame
|
||||
func tableGroupingForMode(mode string) (partial, final bool) {
|
||||
// "smooth" group on every frame and group every table (compact ones too)
|
||||
// while streaming, so a table forming mid-answer keeps the card
|
||||
// growing instead of freezing it until the final frame
|
||||
func tableGroupingForMode(mode string) (partial, final, allPartial bool) {
|
||||
switch strings.ToLower(strings.TrimSpace(mode)) {
|
||||
case "off", "0", "false", "no":
|
||||
return false, false
|
||||
return false, false, false
|
||||
case "stream", "streaming", "partial":
|
||||
return true, false
|
||||
return true, false, false
|
||||
case "smooth":
|
||||
return true, true, true
|
||||
default:
|
||||
return true, true
|
||||
return true, true, false
|
||||
}
|
||||
}
|
||||
|
||||
@@ -98,7 +108,7 @@ func normalizeTableBlock(rows []markdownStreamLine, options tableNormalizeOption
|
||||
}
|
||||
}
|
||||
header := cells[0]
|
||||
if maxWidth <= maxTableCellVisualWidth || !options.groupWideTables {
|
||||
if !options.groupAllTables && (maxWidth <= maxTableCellVisualWidth || !options.groupWideTables) {
|
||||
// Compact tables keep their exact original bytes unless a cell carries
|
||||
// <br>, which the card renderer mishandles inside cells. The same path
|
||||
// serves over-wide tables when grouping is disabled: the table stays a
|
||||
|
||||
@@ -83,9 +83,10 @@ func TestNormalizeKeepsWideTableWhenGroupingDisabled(t *testing.T) {
|
||||
|
||||
func TestTableGroupingForMode(t *testing.T) {
|
||||
cases := []struct {
|
||||
mode string
|
||||
wantPartial bool
|
||||
wantFinal bool
|
||||
mode string
|
||||
wantPartial bool
|
||||
wantFinal bool
|
||||
wantAllPartial bool
|
||||
}{
|
||||
{mode: "", wantPartial: true, wantFinal: true},
|
||||
{mode: "on", wantPartial: true, wantFinal: true},
|
||||
@@ -94,11 +95,13 @@ func TestTableGroupingForMode(t *testing.T) {
|
||||
{mode: "NO", wantPartial: false, wantFinal: false},
|
||||
{mode: "stream", wantPartial: true, wantFinal: false},
|
||||
{mode: " Stream ", wantPartial: true, wantFinal: false},
|
||||
{mode: "smooth", wantPartial: true, wantFinal: true, wantAllPartial: true},
|
||||
{mode: " Smooth ", wantPartial: true, wantFinal: true, wantAllPartial: true},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
partial, final := tableGroupingForMode(tc.mode)
|
||||
if partial != tc.wantPartial || final != tc.wantFinal {
|
||||
t.Fatalf("tableGroupingForMode(%q) = (%v,%v), want (%v,%v)", tc.mode, partial, final, tc.wantPartial, tc.wantFinal)
|
||||
partial, final, allPartial := tableGroupingForMode(tc.mode)
|
||||
if partial != tc.wantPartial || final != tc.wantFinal || allPartial != tc.wantAllPartial {
|
||||
t.Fatalf("tableGroupingForMode(%q) = (%v,%v,%v), want (%v,%v,%v)", tc.mode, partial, final, allPartial, tc.wantPartial, tc.wantFinal, tc.wantAllPartial)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -86,8 +86,9 @@ func NewClient(baseURL, apiKey string, timeout time.Duration) (*Client, error) {
|
||||
if strings.TrimSpace(apiKey) == "" {
|
||||
return nil, errors.New("WEKNORA_API_KEY is required")
|
||||
}
|
||||
// timeout <= 0 keeps Go's default: no client-side total timeout on the HTTP
|
||||
// client, so long-lived agent SSE streams are never cut by a duration cap.
|
||||
if timeout <= 0 {
|
||||
timeout = 120 * time.Second
|
||||
}
|
||||
return &Client{baseURL: parsed.String(), apiKey: apiKey, http: &http.Client{Timeout: timeout}}, nil
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user