测试处方上传

2026-06-05 16:36:39 +08:00
parent 5703b93a66
commit 16aebb45b6
10 changed files with 523 additions and 64 deletions
--- a/internal/hyfile/upload.go
+++ b/internal/hyfile/upload.go
@@ -21,6 +21,7 @@ import (

 const uploadFormFieldFile = "file"
 const defaultUploadFilename = "prescription.pdf"
+const uploadRawBodyMaxLen = 4096

 var (
 	uploadInsecureSkipVerify bool
@@ -52,6 +53,27 @@ func uploadHTTPClient() *http.Client {
 	return &http.Client{Timeout: 120 * time.Second, Transport: transport}
 }

+// UploadPDFResult 文件上传 HTTP 结果（供联调展示 raw 响应）。
+type UploadPDFResult struct {
+	FileID     string `json:"fileId,omitempty"`
+	HTTPStatus int    `json:"httpStatus"`
+	RawBody    string `json:"rawBody"`
+	Success    bool   `json:"success"`
+	Message    string `json:"message,omitempty"`
+}
+
+// UploadPDFDetail 上传 PDF 并返回 fileId 与原始响应（与 UploadPDF 同一 HTTP 实现）。
+func UploadPDFDetail(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (*UploadPDFResult, error) {
+	res, err := doUploadPDF(pdf, filename, uploadURL, uploadToken, forwardToken)
+	if err != nil {
+		if res != nil {
+			return res, err
+		}
+		return nil, err
+	}
+	return res, nil
+}
+
 // UploadPDF 上传 PDF 至监管文件服务（通常经 forward-go），成功时返回 record.fileId。
 //
 // uploadURL：FILE_UPLOAD_VIA_FORWARD=true 时为 forward 地址；否则为云端 config 返回的政务云 URL。
@@ -59,24 +81,32 @@ func uploadHTTPClient() *http.Client {
 // filename：multipart 表单 file 字段文件名；空时默认 prescription.pdf。
 // forwardToken：非空时设置 X-Forward-Token（与 forward-go FORWARD_SHARED_SECRET 对应）。
 func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (string, error) {
+	res, err := doUploadPDF(pdf, filename, uploadURL, uploadToken, forwardToken)
+	if err != nil {
+		return "", err
+	}
+	return res.FileID, nil
+}
+
+func doUploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string) (*UploadPDFResult, error) {
 	if len(pdf) == 0 {
-		return "", fmt.Errorf("empty pdf")
+		return nil, fmt.Errorf("empty pdf")
 	}
 	if uploadURL == "" {
-		return "", fmt.Errorf("file upload url empty")
+		return nil, fmt.Errorf("file upload url empty")
 	}
 	if uploadToken == "" {
-		return "", fmt.Errorf("upload token empty")
+		return nil, fmt.Errorf("upload token empty")
 	}

 	contentType, body, err := buildMultipartPDFBody(pdf, filename)
 	if err != nil {
-		return "", err
+		return nil, err
 	}

 	req, err := http.NewRequest(http.MethodPost, uploadURL, body)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 	req.Header.Set("Content-Type", contentType)
 	req.Header.Set("X-Authorization", uploadToken)
@@ -87,15 +117,39 @@ func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string
 	acquireUploadSlot()
 	resp, err := uploadHTTPClient().Do(req)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 	defer resp.Body.Close()
 	raw, _ := io.ReadAll(resp.Body)
-
-	if resp.StatusCode != http.StatusOK {
-		return "", formatUploadHTTPError(resp.StatusCode, raw, uploadURL)
+	out := &UploadPDFResult{
+		HTTPStatus: resp.StatusCode,
+		RawBody:    string(truncate(raw, uploadRawBodyMaxLen)),
 	}

+	if resp.StatusCode != http.StatusOK {
+		return out, formatUploadHTTPError(resp.StatusCode, raw, uploadURL)
+	}
+
+	fileID, success, msg, parseErr := parseUploadResponseBody(raw)
+	out.Success = success
+	out.Message = msg
+	out.FileID = fileID
+	if parseErr != nil {
+		return out, parseErr
+	}
+	if !success {
+		if msg == "" {
+			msg = out.RawBody
+		}
+		return out, fmt.Errorf("upload failed: %s", msg)
+	}
+	if fileID == "" {
+		return out, fmt.Errorf("upload response missing record.fileId: %s", truncate(raw, 512))
+	}
+	return out, nil
+}
+
+func parseUploadResponseBody(raw []byte) (fileID string, success bool, message string, err error) {
 	var parsed struct {
 		Success bool `json:"success"`
 		Record  struct {
@@ -104,20 +158,9 @@ func UploadPDF(pdf []byte, filename, uploadURL, uploadToken, forwardToken string
 		Message string `json:"message"`
 	}
 	if err := json.Unmarshal(raw, &parsed); err != nil {
-		return "", fmt.Errorf("parse upload response: %w", err)
+		return "", false, "", fmt.Errorf("parse upload response: %w", err)
 	}
-	if !parsed.Success {
-		msg := parsed.Message
-		if msg == "" {
-			msg = string(truncate(raw, 512))
-		}
-		return "", fmt.Errorf("upload failed: %s", msg)
-	}
-	fileID := parsed.Record.FileID
-	if fileID == "" {
-		return "", fmt.Errorf("upload response missing record.fileId: %s", truncate(raw, 512))
-	}
-	return fileID, nil
+	return parsed.Record.FileID, parsed.Success, parsed.Message, nil
 }

 func buildMultipartPDFBody(pdf []byte, filename string) (contentType string, body io.Reader, err error) {
--- a/internal/hyfile/upload_detail_test.go
+++ b/internal/hyfile/upload_detail_test.go
@@ -0,0 +1,53 @@
+package hyfile
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestUploadPDFDetail_returnsRawBody(t *testing.T) {
+	disableUploadThrottle(t)
+	rawResp := `{"success":true,"record":{"fileId":"fid-99"},"message":"ok"}`
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_, _ = w.Write([]byte(rawResp))
+	}))
+	defer srv.Close()
+
+	res, err := UploadPDFDetail([]byte("%PDF-test"), "a.pdf", srv.URL, "tok", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.FileID != "fid-99" || !res.Success {
+		t.Fatalf("res=%+v", res)
+	}
+	if res.RawBody != rawResp {
+		t.Fatalf("raw=%q", res.RawBody)
+	}
+	if res.HTTPStatus != http.StatusOK {
+		t.Fatalf("status=%d", res.HTTPStatus)
+	}
+}
+
+func TestUploadPDFDetail_httpErrorPreservesBody(t *testing.T) {
+	disableUploadThrottle(t)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		_ = json.NewEncoder(w).Encode(map[string]string{"message": "denied"})
+	}))
+	defer srv.Close()
+
+	res, err := UploadPDFDetail([]byte("x"), "x.pdf", srv.URL, "tok", "")
+	if err == nil {
+		t.Fatal("expected error")
+	}
+	if res == nil || res.HTTPStatus != 403 {
+		t.Fatalf("res=%+v err=%v", res, err)
+	}
+	if res.RawBody == "" {
+		t.Fatal("expected raw body")
+	}
+}
--- a/internal/logweb/server.go
+++ b/internal/logweb/server.go
@@ -67,6 +67,7 @@ func (s *Server) routes() {
 	s.mux.HandleFunc("/runs", s.handlePage("runs.html"))
 	s.mux.HandleFunc("/runs/record", s.handlePage("runs_record.html"))
 	s.mux.HandleFunc("/test", s.handlePage("test.html"))
+	s.mux.HandleFunc("/supervise-step", s.handleSuperviseStepPage)
 	s.mux.HandleFunc("/config", s.handlePage("config.html"))
 	s.mux.HandleFunc("/upload", s.handlePage("upload.html"))
 	s.mux.HandleFunc("/fileauth", s.handlePage("fileauth.html"))
@@ -84,6 +85,7 @@ func (s *Server) routes() {
 	s.mux.HandleFunc("/api/runs/retry", s.handleAPIRunsRetry)
 	s.mux.HandleFunc("/api/test/status", s.handleAPITestStatus)
 	s.mux.HandleFunc("/api/test/sync", s.handleAPITestSync)
+	s.mux.HandleFunc("/api/supervise/step-test", s.handleAPISuperviseStepTest)

 	s.mux.Handle("/static/", http.StripPrefix("/static/", fileServer))
 }
--- a/internal/logweb/supervise_handlers.go
+++ b/internal/logweb/supervise_handlers.go
@@ -0,0 +1,68 @@
+package logweb
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+
+	syncer "xk-hy-transit-go/internal/sync"
+	"xk-hy-transit-go/internal/syncgate"
+)
+
+func (s *Server) handleSuperviseStepPage(w http.ResponseWriter, r *http.Request) {
+	s.handlePage("supervise_step.html")(w, r)
+}
+
+func (s *Server) handleAPISuperviseStepTest(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		writeErr(w, 405, "POST only")
+		return
+	}
+	sr := asSyncRunner(s.deps.Runner)
+	if sr == nil {
+		writeErr(w, 503, "未配置 Runner（请使用 transit serve）")
+		return
+	}
+	var body struct {
+		Step  string `json:"step"`
+		Date  string `json:"date"`
+		Limit *int   `json:"limit"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+		writeErr(w, 400, "invalid json")
+		return
+	}
+	step := strings.TrimSpace(body.Step)
+	if step == "" {
+		writeErr(w, 400, "step required (recipe|verification)")
+		return
+	}
+	limit := 1
+	if body.Limit != nil {
+		limit = *body.Limit
+	}
+	anchor := syncer.ResolveAnchorDate(s.deps.Cfg, strings.TrimSpace(body.Date))
+
+	var result *syncer.StepTestResult
+	var runErr error
+	if !syncgate.TryRun(step, anchor, func() error {
+		result, runErr = sr.TestSuperviseStep(step, anchor, limit)
+		return runErr
+	}) {
+		writeErr(w, 409, "已有同步任务在执行中")
+		return
+	}
+	if runErr != nil {
+		payload := map[string]any{"error": runErr.Error()}
+		if result != nil {
+			payload["result"] = result
+		}
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
+		w.WriteHeader(502)
+		enc := json.NewEncoder(w)
+		enc.SetIndent("", "  ")
+		_ = enc.Encode(payload)
+		return
+	}
+	writeJSON(w, result)
+}
--- a/internal/logweb/web/app.js
+++ b/internal/logweb/web/app.js
@@ -102,6 +102,7 @@ const MAIN_NAV = [
  { href: '/config', id: 'config', label: '配置' },
  { href: '/fileauth', id: 'fileauth', label: '上传凭证' },
  { href: '/upload', id: 'upload', label: '上传' },
+  { href: '/supervise-step', id: 'supervise-step', label: '处方核销' },
  { href: '/runs', id: 'runs', label: '流水' },
  { href: '/logs', id: 'logs', label: '日志' },
  { href: '/test', id: 'test', label: '测试' },
--- a/internal/logweb/web/index.html
+++ b/internal/logweb/web/index.html
@@ -31,6 +31,10 @@
      <h2>文件日志</h2>
      <p>查看 app / pull / push 按日滚动日志，支持搜索与高亮。</p>
    </a>
+    <a class="card" href="/supervise-step">
+      <h2>处方/核销联调</h2>
+      <p>在线处方与处方核销单步测试：组包、文件上传 raw、上报、云端回调。</p>
+    </a>
    <a class="card" href="/test">
      <h2>测试执行</h2>
      <p>触发一次 sync（拉取 → 上传 → 回调），用于联调测试。</p>
--- a/internal/sync/step_result.go
+++ b/internal/sync/step_result.go
@@ -0,0 +1,159 @@
+package syncer
+
+import (
+	"xk-hy-transit-go/internal/hy"
+	"xk-hy-transit-go/internal/hyfile"
+	"xk-hy-transit-go/internal/xkapi"
+)
+
+// StepRunOptions 单 step 执行选项（自动 sync 与联调共用 executeStep）。
+type StepRunOptions struct {
+	ItemLimit    int  // 0=处理 Pull 全部条数；>0 只处理前 N 条
+	CollectTrace bool // true 时填充 StepTestResult
+}
+
+// StepTestResult 联调页返回（recipe / verification）。
+type StepTestResult struct {
+	Step          string             `json:"step"`
+	StepLabel     string             `json:"step_label"`
+	AnchorDate    string             `json:"anchor_date"`
+	BatchID       int                `json:"batch_id"`
+	PullSummary   PullSummary        `json:"pull_summary"`
+	Items         []ItemStepTrace    `json:"items"`
+	CallbackError string             `json:"callback_error,omitempty"`
+	Summary       StepOutcomeSummary `json:"summary"`
+}
+
+// PullSummary 拉取摘要。
+type PullSummary struct {
+	Total            int `json:"total"`
+	ValidationFailed int `json:"validation_failed"`
+	Processed        int `json:"processed"`
+}
+
+// StepOutcomeSummary 本 step 处理结果统计。
+type StepOutcomeSummary struct {
+	Success int `json:"success"`
+	Failed  int `json:"failed"`
+	Skipped int `json:"skipped"`
+}
+
+// ItemStepTrace 单条联调追踪（与 processItem 一一对应）。
+type ItemStepTrace struct {
+	RecordID         int                    `json:"record_id"`
+	BizKey           string                 `json:"biz_key"`
+	ValidationErrors []string               `json:"validation_errors,omitempty"`
+	PayloadSnapshot  string                 `json:"payload_snapshot,omitempty"`
+	PushPlainJSON    string                 `json:"push_plain_json,omitempty"`
+	PushHeaders      map[string]string      `json:"push_headers,omitempty"`
+	RecipeFileUpload *RecipeFileUploadTrace `json:"recipe_file_upload,omitempty"`
+	Push             PushTrace              `json:"push"`
+	Callback         xkapi.CallbackItem     `json:"callback"`
+}
+
+// RecipeFileUploadTrace 处方 PDF 28211 上传追踪。
+type RecipeFileUploadTrace struct {
+	Executed     bool   `json:"executed"`
+	Skipped      bool   `json:"skipped"`
+	SkipReason   string `json:"skip_reason,omitempty"`
+	FileID       string `json:"fileId,omitempty"`
+	HTTPStatus   int    `json:"httpStatus,omitempty"`
+	RawBody      string `json:"rawBody,omitempty"`
+	Success      bool   `json:"success"`
+	Message      string `json:"message,omitempty"`
+	LocalPDFPath string `json:"localPdfPath,omitempty"`
+	Error        string `json:"error,omitempty"`
+}
+
+// PushTrace 监管 JSON 上报结果摘要。
+type PushTrace struct {
+	HTTPCode     int    `json:"http_code,omitempty"`
+	MsgCode      int    `json:"msg_code,omitempty"`
+	Msg          string `json:"msg,omitempty"`
+	TraceID      string `json:"trace_id,omitempty"`
+	ResponseBody string `json:"response_body,omitempty"`
+}
+
+// itemTraceCtx processItem 内部填充，最终写入 ItemStepTrace。
+type itemTraceCtx struct {
+	out *ItemStepTrace
+}
+
+func newItemTraceCtx(item xkapi.PullItem) *itemTraceCtx {
+	return &itemTraceCtx{out: &ItemStepTrace{
+		RecordID:         item.RecordID,
+		BizKey:           item.BizKey,
+		ValidationErrors: append([]string(nil), item.ValidationErrors...),
+	}}
+}
+
+func (c *itemTraceCtx) setPayloadSnapshot(payload map[string]any) {
+	if c == nil || c.out == nil || payload == nil {
+		return
+	}
+	c.out.PayloadSnapshot = mustJSON(payload)
+}
+
+func (c *itemTraceCtx) setPushReq(req *hy.UploadRequest) {
+	if c == nil || c.out == nil || req == nil {
+		return
+	}
+	c.out.PushPlainJSON = req.PlainJSON
+	c.out.PushHeaders = copyHeadersForTrace(req.Headers)
+}
+
+func copyHeadersForTrace(h map[string]string) map[string]string {
+	if h == nil {
+		return nil
+	}
+	out := make(map[string]string, len(h))
+	for k, v := range h {
+		if k == "secret" {
+			continue
+		}
+		out[k] = v
+	}
+	return out
+}
+
+func (c *itemTraceCtx) setRecipeUpload(up *hyfile.UploadPDFResult, localPath string, err error) {
+	if c == nil || c.out == nil {
+		return
+	}
+	tr := &RecipeFileUploadTrace{Executed: true, LocalPDFPath: localPath}
+	if up != nil {
+		tr.FileID = up.FileID
+		tr.HTTPStatus = up.HTTPStatus
+		tr.RawBody = up.RawBody
+		tr.Success = up.Success
+		tr.Message = up.Message
+	}
+	if err != nil {
+		tr.Error = err.Error()
+	}
+	c.out.RecipeFileUpload = tr
+}
+
+func (c *itemTraceCtx) setRecipeSkipped(reason string) {
+	if c == nil || c.out == nil {
+		return
+	}
+	c.out.RecipeFileUpload = &RecipeFileUploadTrace{
+		Skipped:    true,
+		SkipReason: reason,
+	}
+}
+
+func (c *itemTraceCtx) finish(cb xkapi.CallbackItem) {
+	if c == nil || c.out == nil {
+		return
+	}
+	c.out.Callback = cb
+	c.out.Push = PushTrace{
+		HTTPCode:     cb.HTTPCode,
+		MsgCode:      cb.MsgCode,
+		Msg:          cb.Msg,
+		TraceID:      cb.TraceID,
+		ResponseBody: cb.ResponseBody,
+	}
+}
--- a/internal/sync/step_test_run_test.go
+++ b/internal/sync/step_test_run_test.go
@@ -0,0 +1,14 @@
+package syncer
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestTestSuperviseStep_rejectsOtherSteps(t *testing.T) {
+	r := &Runner{}
+	_, err := r.TestSuperviseStep("referral", "2026-05-18", 1)
+	if err == nil || !strings.Contains(err.Error(), "recipe or verification") {
+		t.Fatalf("err=%v", err)
+	}
+}
--- a/internal/sync/supervise_step.go
+++ b/internal/sync/supervise_step.go
@@ -0,0 +1,23 @@
+package syncer
+
+import (
+	"fmt"
+	"strings"
+)
+
+// TestSuperviseStep 联调：仅 recipe / verification，与自动 sync 共用 executeStep。
+func (r *Runner) TestSuperviseStep(step, anchorDate string, limit int) (*StepTestResult, error) {
+	step = strings.TrimSpace(step)
+	switch step {
+	case "recipe", "verification":
+	default:
+		return nil, fmt.Errorf("step must be recipe or verification, got %q", step)
+	}
+	if limit < 0 {
+		limit = 0
+	}
+	return r.executeStep(step, anchorDate, StepRunOptions{
+		ItemLimit:    limit,
+		CollectTrace: true,
+	})
+}
--- a/internal/sync/sync.go
+++ b/internal/sync/sync.go
@@ -117,30 +117,55 @@ func (r *Runner) Run(step, anchorDate string) error {

 // runStep 执行单个监管类型的完整同步（见包注释 ①～⑤）。
 func (r *Runner) runStep(step, anchorDate string) error {
+	_, err := r.executeStep(step, anchorDate, StepRunOptions{ItemLimit: 0, CollectTrace: false})
+	return err
+}
+
+// executeStep 单 step 完整链路；自动与联调共用，开头刷新云端配置与 uploadToken。
+func (r *Runner) executeStep(step, anchorDate string, opts StepRunOptions) (*StepTestResult, error) {
 	method, ok := hy.StepMethods[step]
 	if !ok {
-		return fmt.Errorf("unknown step: %s", step)
+		return nil, fmt.Errorf("unknown step: %s", step)
+	}
+	if err := r.ReloadCloudConfig(); err != nil {
+		return nil, fmt.Errorf("reload config: %w", err)
+	}
+
+	var result *StepTestResult
+	if opts.CollectTrace {
+		result = &StepTestResult{
+			Step:       step,
+			StepLabel:  hy.LabelStep(step),
+			AnchorDate: anchorDate,
+		}
 	}

 	applog.Appf("sync start step=%s date=%s", step, anchorDate)
-	_, _ = r.store.UpsertJob(anchorDate, step, "running", 0, 0, 0, "")
+	if r.store != nil {
+		_, _ = r.store.UpsertJob(anchorDate, step, "running", 0, 0, 0, "")
+	}

-	// ① 云端建批次
 	batchID, err := r.xk.BatchCreate(anchorDate, step)
 	if err != nil {
-		_ = r.store.FinishJob(anchorDate, step, "failed")
+		if r.store != nil {
+			_ = r.store.FinishJob(anchorDate, step, "failed")
+		}
 		applog.Pullf("batch create failed step=%s date=%s err=%v", step, anchorDate, err)
-		return fmt.Errorf("batch create: %w", err)
+		return result, fmt.Errorf("batch create: %w", err)
+	}
+	if result != nil {
+		result.BatchID = batchID
 	}
 	applog.Pullf("batch created step=%s date=%s batch_id=%d", step, anchorDate, batchID)

-	// ② 云端组包拉取（含 validation_errors、meta.needs_recipe_upload 等）
 	pull, err := r.xk.Pull(step, anchorDate, batchID)
 	if err != nil {
 		_ = r.xk.BatchFinish(batchID, "failed")
-		_ = r.store.FinishJob(anchorDate, step, "failed")
+		if r.store != nil {
+			_ = r.store.FinishJob(anchorDate, step, "failed")
+		}
 		applog.Pullf("pull failed step=%s batch_id=%d err=%v", step, batchID, err)
-		return err
+		return result, err
 	}

 	validationCount := 0
@@ -150,14 +175,37 @@ func (r *Runner) runStep(step, anchorDate string) error {
 		}
 	}
 	applog.Pullf("pull done step=%s batch_id=%d items=%d validation_failed=%d", step, batchID, len(pull.Items), validationCount)
+	if result != nil {
+		result.PullSummary = PullSummary{
+			Total:            len(pull.Items),
+			ValidationFailed: validationCount,
+		}
+	}
+
+	items := pull.Items
+	if opts.ItemLimit > 0 && len(items) > opts.ItemLimit {
+		items = items[:opts.ItemLimit]
+	}
+	if result != nil {
+		result.PullSummary.Processed = len(items)
+	}

 	var callbacks []xkapi.CallbackItem
 	success, failed, skipped := 0, 0, 0
-
-	// ③ 逐条：recipe 可能先 PDF 上传，再加密经 forward 报政务云
-	for _, item := range pull.Items {
-		cb := r.processItem(step, anchorDate, method, batchID, item)
+	for _, item := range items {
+		var tr *itemTraceCtx
+		if opts.CollectTrace {
+			tr = newItemTraceCtx(item)
+			if item.BizKey == "" && item.Payload != nil {
+				tr.out.BizKey = hy.BizKey(step, anchorDate, item.Payload)
+			}
+		}
+		cb := r.processItem(step, anchorDate, method, batchID, item, tr)
 		callbacks = append(callbacks, cb)
+		if tr != nil {
+			tr.finish(cb)
+			result.Items = append(result.Items, *tr.out)
+		}
 		switch cb.PushStatus {
 		case "success":
 			success++
@@ -168,22 +216,35 @@ func (r *Runner) runStep(step, anchorDate string) error {
 		}
 	}

-	// ④ 整批回调云端（写 push_status / callback_status）
 	if err := r.xk.BatchCallback(batchID, callbacks); err != nil {
-		applog.Appf("batch callback warning batch_id=%d err=%v", batchID, err)
+		applog.Appf("batch callback failed step=%s batch_id=%d err=%v", step, batchID, err)
+		if r.store != nil {
+			_, _ = r.store.UpsertJob(anchorDate, step, "failed", success+failed+skipped, success, failed, err.Error())
+			_ = r.store.FinishJob(anchorDate, step, "failed")
+		}
+		if result != nil {
+			result.CallbackError = err.Error()
+			result.Summary = StepOutcomeSummary{Success: success, Failed: failed, Skipped: skipped}
+		}
+		_ = r.xk.BatchFinish(batchID, "failed")
+		return result, fmt.Errorf("batch callback: %w", err)
 	}
-	// ⑤ 标记批次结束
-	_ = r.xk.BatchFinish(batchID, "done")

+	_ = r.xk.BatchFinish(batchID, "done")
 	total := success + failed + skipped
-	_, _ = r.store.UpsertJob(anchorDate, step, "done", total, success, failed, "")
-	_ = r.store.FinishJob(anchorDate, step, "done")
+	if r.store != nil {
+		_, _ = r.store.UpsertJob(anchorDate, step, "done", total, success, failed, "")
+		_ = r.store.FinishJob(anchorDate, step, "done")
+	}
 	applog.Pullf("step done step=%s batch_id=%d total=%d success=%d failed=%d skipped=%d", step, batchID, total, success, failed, skipped)
-	return nil
+	if result != nil {
+		result.Summary = StepOutcomeSummary{Success: success, Failed: failed, Skipped: skipped}
+	}
+	return result, nil
 }

 // processItem 处理单条：校验失败则 skipped；否则 BuildUpload + PostForward，组装 CallbackItem。
-func (r *Runner) processItem(step, anchorDate, method string, batchID int, item xkapi.PullItem) xkapi.CallbackItem {
+func (r *Runner) processItem(step, anchorDate, method string, batchID int, item xkapi.PullItem, tr *itemTraceCtx) xkapi.CallbackItem {
 	payload := item.Payload
 	bizKey := item.BizKey
 	if bizKey == "" && payload != nil {
@@ -203,7 +264,9 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item
 			bussID = fmt.Sprintf("%v", payload["bussID"])
 		}
 		payloadJSON, _ := json.Marshal(payload)
-		_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, bussID, bizKey, "", string(payloadJSON), item.ValidationErrors)
+		if r.store != nil {
+			_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, bussID, bizKey, "", string(payloadJSON), item.ValidationErrors)
+		}
 		cb.PushStatus = "skipped"
 		cb.CallbackStatus = "failed"
 		cb.ErrorMessage = "validation failed"
@@ -214,39 +277,57 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item
 		return cb
 	}

-	// recipe 且需上传处方 PDF：先经 forward 拿到 recipeFileId，再加密上报监管
-	if step == "recipe" && metaBool(item.Meta, "needs_recipe_upload") {
-		if err := r.ensureRecipeFileID(item, payload); err != nil {
-			_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", mustJSON(payload), nil)
-			cb.PushStatus = "failed"
-			cb.CallbackStatus = "failed"
-			cb.ErrorMessage = "recipe file upload: " + err.Error()
-			applog.Pushf("recipe upload fail biz_key=%s err=%v", bizKey, err)
-			return cb
+	if tr != nil {
+		tr.setPayloadSnapshot(payload)
+	}
+
+	if step == "recipe" {
+		if metaBool(item.Meta, "needs_recipe_upload") {
+			if err := r.ensureRecipeFileID(item, payload, tr); err != nil {
+				if r.store != nil {
+					_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", mustJSON(payload), nil)
+				}
+				cb.PushStatus = "failed"
+				cb.CallbackStatus = "failed"
+				cb.ErrorMessage = "recipe file upload: " + err.Error()
+				applog.Pushf("recipe upload fail biz_key=%s err=%v", bizKey, err)
+				return cb
+			}
+		} else if tr != nil {
+			tr.setRecipeSkipped("needs_recipe_upload=false")
 		}
 	}

 	payloadJSON, _ := json.Marshal(payload)
-	_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", string(payloadJSON), nil)
+	if r.store != nil {
+		_ = r.store.SaveRecord(int64(batchID), anchorDate, method, method, fmt.Sprintf("%v", payload["bussID"]), bizKey, "", string(payloadJSON), nil)
+	}

 	r.configMu.RLock()
 	appKey, appSecret, aesKey := r.hyAppKey, r.hyAppSecret, r.hyAesKey
 	r.configMu.RUnlock()
 	req, err := hy.BuildUpload(method, payload, r.organ, appKey, appSecret, aesKey)
 	if err != nil {
-		_ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error())
+		if r.store != nil {
+			_ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error())
+		}
 		cb.PushStatus = "failed"
 		cb.CallbackStatus = "failed"
 		cb.ErrorMessage = err.Error()
 		applog.Pushf("build upload fail biz_key=%s err=%v", bizKey, err)
 		return cb
 	}
+	if tr != nil {
+		tr.setPushReq(req)
+	}

 	start := time.Now()
 	result, err := hy.PostForward(r.cfg.ForwardBaseURL, req, r.cfg.ForwardSharedSecret)
 	duration := int(time.Since(start).Milliseconds())
 	if err != nil {
-		_ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error())
+		if r.store != nil {
+			_ = r.store.UpdateRecordStatus(bizKey, "failed", err.Error())
+		}
 		cb.PushStatus = "failed"
 		cb.CallbackStatus = "failed"
 		cb.ErrorMessage = err.Error()
@@ -254,9 +335,11 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item
 		return cb
 	}

-	recordLocalID, _ := r.store.GetRecordID(bizKey)
-	headersJSON, _ := json.Marshal(req.Headers)
-	_ = r.store.SaveLog(recordLocalID, result.HTTPCode, result.MsgCode, result.Msg, result.TraceID, len(req.Body), req.PlainJSON, string(headersJSON), result.Body, duration)
+	if r.store != nil {
+		recordLocalID, _ := r.store.GetRecordID(bizKey)
+		headersJSON, _ := json.Marshal(req.Headers)
+		_ = r.store.SaveLog(recordLocalID, result.HTTPCode, result.MsgCode, result.Msg, result.TraceID, len(req.Body), req.PlainJSON, string(headersJSON), result.Body, duration)
+	}

 	cb.HTTPCode = result.HTTPCode
 	cb.MsgCode = result.MsgCode
@@ -265,13 +348,17 @@ func (r *Runner) processItem(step, anchorDate, method string, batchID int, item
 	cb.ResponseBody = result.Body

 	if hy.IsSuperviseOK(result) {
-		_ = r.store.UpdateRecordStatus(bizKey, "success", "")
+		if r.store != nil {
+			_ = r.store.UpdateRecordStatus(bizKey, "success", "")
+		}
 		cb.PushStatus = "success"
 		cb.CallbackStatus = "success"
 		applog.Pushf("ok biz_key=%s method=%s code=%d msgCode=%d trace=%s %dms", bizKey, method, result.GatewayCode, result.MsgCode, result.TraceID, duration)
 	} else {
 		errText := hy.SuperviseErrorSummary(result)
-		_ = r.store.UpdateRecordStatus(bizKey, "failed", errText)
+		if r.store != nil {
+			_ = r.store.UpdateRecordStatus(bizKey, "failed", errText)
+		}
 		cb.PushStatus = "failed"
 		cb.CallbackStatus = "failed"
 		cb.ErrorMessage = errText
@@ -289,7 +376,7 @@ func ResolveAnchorDate(cfg config.Config, dateArg string) string {
 }

 // ensureRecipeFileID 处方 PDF 子流程：HTML→PDF→本地落盘→经 forward 上传→写回 recipeFileId。
-func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any) error {
+func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any, tr *itemTraceCtx) error {
 	// 步骤 1：解析处方 ID，必要时从云端拉取打印 HTML
 	prescriptionID := metaInt(item.Meta, "prescription_id")
 	if prescriptionID <= 0 {
@@ -334,11 +421,15 @@ func (r *Runner) ensureRecipeFileID(item xkapi.PullItem, payload map[string]any)
 	r.configMu.RLock()
 	url := r.fileUploadURL
 	r.configMu.RUnlock()
-	fileID, err := hyfile.UploadPDF(pdf, filename, url, token, r.cfg.ForwardSharedSecret)
+	upRes, err := hyfile.UploadPDFDetail(pdf, filename, url, token, r.cfg.ForwardSharedSecret)
+	if tr != nil {
+		tr.setRecipeUpload(upRes, localPath, err)
+	}
 	if err != nil {
 		applog.Pushf("upload pdf fail prescription_no=%s local=%s err=%v", pdfName.PrescriptionNo, localPath, err)
 		return err
 	}
+	fileID := upRes.FileID
 	applog.Pushf("upload pdf ok prescription_no=%s fileId=%s local=%s url=%s", pdfName.PrescriptionNo, fileID, localPath, r.fileUploadURL)

 	// 步骤 5：写入组包 payload，并回写云端处方表
@@ -528,9 +619,10 @@ func (r *Runner) RetryRecord(localRecordID int64) error {
 		ValidationErrors: validationErrors,
 	}

-	cb := r.processItem(step, rec.AnchorDate, method, batchID, item)
+	cb := r.processItem(step, rec.AnchorDate, method, batchID, item, nil)
 	if err := r.xk.BatchCallback(batchID, []xkapi.CallbackItem{cb}); err != nil {
-		applog.Appf("retry batch callback warning record_id=%d err=%v", localRecordID, err)
+		applog.Appf("retry batch callback failed record_id=%d err=%v", localRecordID, err)
+		return fmt.Errorf("batch callback: %w", err)
 	}
 	_ = r.store.IncRetry(rec.BizKey)
 	if cb.PushStatus != "success" {