feat: poller心跳+诊断探针 bid_probe.dspy；日志双通道（appPublic.debug 进 pipeline.log）

pipeline_bidding/bid_flow.py

@@ -26,7 +26,7 @@ from .bid_common import (
     HT_TENDER_FILE, HT_HUMAN_DOCS, HT_DELIVERY_CONFIRM,
 )
 
-logger = logging.getLogger("pipeline.bidding.flow")
+logger = logging.getLogger("pipeline.bidding")
 
 R_ANALYST = "agent.tender_analyst"
 R_PREP = "agent.bid_prep"
@@ -39,6 +39,9 @@ MAX_ANALYST_ATTEMPT = 2      # 解析阶段最多重派次数（防死循环）
 MAX_PREP_ATTEMPT = 2
 POLL_INTERVAL = 15
 
+# poller 心跳（供 /pipeline-bidding/api/bid_probe.dspy 诊断；服务内可见）
+POLLER_STATE = {"last_run": "", "last_error": "", "rounds": 0}
+
 
 async def _count(sor, sql, p):
     recs = await sor.sqlExe(sql, p)
@@ -338,17 +341,34 @@ def start_poller():
                     async def _once():
                         async with db.sqlorContext(dbname) as sor:
                             res = await _reconcile_all_with(sor)
-                        for pid, acts in (res or {}).items():
-                            for a in acts:
-                                if a.startswith(("created", "escalated", "completed", "error")):
-                                    logger.info("bid_flow %s: %s", pid, a)
+                        return res
+
+                    import datetime
                     # watchdog：每轮最多 60 秒，防连接池/MDL 锁卡死导致 poller 永久停摆
-                    await asyncio.wait_for(_once(), timeout=60)
+                    res = await asyncio.wait_for(_once(), timeout=60)
+                    POLLER_STATE["rounds"] += 1
+                    POLLER_STATE["last_run"] = datetime.datetime.now().strftime("%H:%M:%S")
+                    POLLER_STATE["last_error"] = ""
+                    for pid, acts in (res or {}).items():
+                        for a in acts:
+                            if a.startswith(("created", "escalated", "completed", "error")):
+                                try:
+                                    from appPublic.log import debug as _dbg
+                                    _dbg("bid_flow %s: %s" % (pid, a))
+                                except Exception:
+                                    pass
+                                logger.info("bid_flow %s: %s", pid, a)
                 except Exception as e:
+                    POLLER_STATE["last_error"] = "%s: %s" % (type(e).__name__, str(e)[:160])
                     logger.warning("bid_poller error/timeout: %s", str(e)[:200])
                 await asyncio.sleep(POLL_INTERVAL)
 
         asyncio.create_task(_loop())
+        try:
+            from appPublic.log import debug as _dbg
+            _dbg("bid_flow poller started (interval=%ss)" % POLL_INTERVAL)
+        except Exception:
+            pass
         logger.info("bid_flow poller started (interval=%ss)", POLL_INTERVAL)
 
     add_startup(_bid_poller)

scripts/load_path.py

@@ -34,9 +34,10 @@ for t in TABLES:
         "/%s/%s/update_%s.dspy" % (MOD, t, t),
         "/%s/%s/delete_%s.dspy" % (MOD, t, t),
     ]
-# 产线业务 API（当前无自写 dspy；保留结构，后续新增在此登记）
+# 产线业务 API
 PATHS_LOGINED += [
     "/%s/api/" % MOD,
+    "/%s/api/bid_probe.dspy" % MOD,
 ]
 
 

wwwroot/api/bid_probe.dspy

@@ -0,0 +1,36 @@
+# bid_probe.dspy — 投标产线服务内探针（诊断用）
+# GET/POST: action=status | reconcile
+#   status    → 模块装载状态 + poller 心跳 + 活跃投标项目数
+#   reconcile → 在服务进程内跑一轮全量对账，返回动作明细
+
+import json
+
+action = (params_kw or {}).get('action', 'status')
+dbname = get_module_dbname('pipeline-bidding')
+
+try:
+    import pipeline_bidding
+    from pipeline_bidding import bid_flow
+    loaded = True
+    ver = getattr(pipeline_bidding, '__version__', '?')
+except Exception as e:
+    loaded = False
+    ver = str(e)[:100]
+
+if action == 'reconcile':
+    try:
+        res = await bid_flow.reconcile_all()
+        return json.dumps({"success": True, "projects": res}, ensure_ascii=False, default=str)
+    except Exception as e:
+        return json.dumps({"success": False, "error": str(e)[:300]}, ensure_ascii=False)
+
+out = {"loaded": loaded, "version": ver,
+       "poller_state": getattr(bid_flow, 'POLLER_STATE', None) if loaded else None}
+if loaded:
+    async with DBPools().sqlorContext(dbname) as sor:
+        recs = await sor.sqlExe(
+            "SELECT COUNT(*) AS c FROM sd_projects WHERE pipeline_id='bidding_general' "
+            "AND status NOT IN ('paused','archived','completed','cancelled')", {})
+        await sor.sqlExe("COMMIT", {})
+        out["active_projects"] = getattr(recs[0], 'c', 0) if recs else 0
+return json.dumps(out, ensure_ascii=False, default=str)