feat(mcp): 飞轮派生数据工具输出 staleness 标注

世界模型自洽:模型可以过期,不能不知道自己过期。
- 新增 sync_state 表(domain 单行 upsert)+ touch/read 依赖注入函数
- 三个写入点:轮询成功(polled>0)、webhook PR 落库、webhook push 落库
- get_pr_events / get_commit_metrics 输出首行标注最近同步时间与距今分钟数
- get_ci_status 为实时 Gitea API 调用,不在范围
- 实锤依据:get_pr_events 延迟致 /ship 误判无 open PR(enterprise-system 2026-07-28 回传)

Feature-Confirmed: true
Rules-Applied: R05, R06, R07
This commit is contained in:
lanrtop 2026-08-23 15:46:03 +09:00
parent 16473e7260
commit 0a8a9e60bc
4 changed files with 72 additions and 2 deletions

View File

@ -549,6 +549,11 @@ pub fn poll_all_repos() -> Result<(usize, usize), String> {
&format!("PR 轮询新采集 {pr_ingested} 条事件"), None); &format!("PR 轮询新采集 {pr_ingested} 条事件"), None);
} }
// staleness 自知:至少成功拉到一个仓库才算一次有效同步
if polled > 0 {
crate::db::touch_sync_state(&conn, "flywheel");
}
Ok((polled, ingested)) Ok((polled, ingested))
} }

View File

@ -564,6 +564,14 @@ fn migrate(conn: &rusqlite::Connection) {
CREATE INDEX IF NOT EXISTS idx_pr_events_project ON pr_events(project_id); CREATE INDEX IF NOT EXISTS idx_pr_events_project ON pr_events(project_id);
").expect("create pr_events table"); ").expect("create pr_events table");
// 飞轮同步状态staleness 自知:派生数据类 MCP 工具据此标注数据新鲜度)
conn.execute_batch("
CREATE TABLE IF NOT EXISTS sync_state (
domain TEXT PRIMARY KEY,
synced_at TEXT NOT NULL DEFAULT (datetime('now'))
);
").expect("create sync_state table");
// 仪表盘全局快启应用(无项目 FK 约束) // 仪表盘全局快启应用(无项目 FK 约束)
conn.execute_batch(" conn.execute_batch("
CREATE TABLE IF NOT EXISTS global_tools ( CREATE TABLE IF NOT EXISTS global_tools (
@ -909,6 +917,26 @@ pub fn conn_with_schema() -> rusqlite::Connection {
conn conn
} }
/// 记录某数据域最近一次成功同步时间upsert。失败时静默同步标记不能反噬业务逻辑
pub fn touch_sync_state(conn: &rusqlite::Connection, domain: &str) {
let _ = conn.execute(
"INSERT INTO sync_state (domain, synced_at) VALUES (?1, datetime('now'))
ON CONFLICT(domain) DO UPDATE SET synced_at = datetime('now')",
rusqlite::params![domain],
);
}
/// 读取某数据域最近同步时间与距今分钟数;无记录返回 None
pub fn read_sync_state(conn: &rusqlite::Connection, domain: &str) -> Option<(String, i64)> {
conn.query_row(
"SELECT synced_at, CAST((julianday('now') - julianday(synced_at)) * 1440 AS INTEGER)
FROM sync_state WHERE domain = ?1",
rusqlite::params![domain],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.ok()
}
/// 写入运行时日志。失败时静默(不能让日志写入反过来崩溃业务逻辑) /// 写入运行时日志。失败时静默(不能让日志写入反过来崩溃业务逻辑)
pub fn log_event(category: &str, level: &str, message: &str, context: Option<&str>) { pub fn log_event(category: &str, level: &str, message: &str, context: Option<&str>) {
let pool = match POOL.get() { let pool = match POOL.get() {
@ -946,6 +974,27 @@ mod tests {
migrate(&conn); migrate(&conn);
} }
#[test]
fn sync_state_touch_then_read() {
let conn = conn_with_schema();
touch_sync_state(&conn, "flywheel");
let (at, mins) = read_sync_state(&conn, "flywheel").expect("应有同步记录");
assert!(!at.is_empty());
assert!(mins >= 0 && mins < 2, "刚 touch 的记录距今应为 0~1 分钟,实际 {mins}");
// upsert重复 touch 不报错、仍只有一行
touch_sync_state(&conn, "flywheel");
let count: i64 = conn
.query_row("SELECT COUNT(*) FROM sync_state WHERE domain='flywheel'", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 1);
}
#[test]
fn sync_state_unknown_domain_returns_none() {
let conn = conn_with_schema();
assert!(read_sync_state(&conn, "nonexistent").is_none());
}
#[test] #[test]
fn core_tables_exist() { fn core_tables_exist() {
let conn = conn_with_schema(); let conn = conn_with_schema();

View File

@ -468,10 +468,16 @@ pub async fn tools_call(params: Option<&Value>) -> Value {
format!("{icon} PR #{num} [{action}] {title}\n {head}{base} by {author} @ {at}") format!("{icon} PR #{num} [{action}] {title}\n {head}{base} by {author} @ {at}")
}) })
.collect(); .collect();
let fresh_line = match db::read_sync_state(&conn, "flywheel") {
Some((at, mins)) => format!(
"⏱ 飞轮数据最近同步:{at} UTC约 {mins} 分钟前。数据可能滞后——PR 是否已存在以 create_pull_request 返回的 already_exists 为准。\n\n"
),
None => "⏱ 飞轮数据同步时间未知(尚无同步记录,可先调 poll_now\n\n".to_string(),
};
if rows.is_empty() { if rows.is_empty() {
Ok(format!("项目 {} 暂无 PR 事件记录", pid)) Ok(format!("{fresh_line}项目 {} 暂无 PR 事件记录", pid))
} else { } else {
Ok(format!("PR 事件(最近 {} 条):\n\n{}", rows.len(), rows.join("\n\n"))) Ok(format!("{fresh_line}PR 事件(最近 {} 条):\n\n{}", rows.len(), rows.join("\n\n")))
} }
} }
"get_commit_metrics" => { "get_commit_metrics" => {
@ -509,6 +515,12 @@ pub async fn tools_call(params: Option<&Value>) -> Value {
) )
.unwrap_or(0); .unwrap_or(0);
let mut out = format!("## 提交飞轮数据(最近 {} 天)\n\n", days); let mut out = format!("## 提交飞轮数据(最近 {} 天)\n\n", days);
match db::read_sync_state(&conn, "flywheel") {
Some((at, mins)) => out.push_str(&format!(
"⏱ 飞轮数据最近同步:{at} UTC约 {mins} 分钟前)\n\n"
)),
None => out.push_str("⏱ 飞轮数据同步时间未知(尚无同步记录,可先调 poll_now\n\n"),
}
out.push_str(&format!("- **总提交数**{}\n", total)); out.push_str(&format!("- **总提交数**{}\n", total));
out.push_str(&format!( out.push_str(&format!(
"- **返工提交**{} 条({:.0}%\n", "- **返工提交**{} 条({:.0}%\n",

View File

@ -151,6 +151,9 @@ async fn handle_push(project_id: &str, body: &Bytes) -> Response {
received += 1; received += 1;
} }
} }
if received > 0 {
db::touch_sync_state(&conn, "flywheel");
}
log_event( log_event(
"gitea_webhook", "gitea_webhook",
"info", "info",
@ -215,6 +218,7 @@ async fn handle_pull_request(project_id: &str, body: &Bytes) -> Response {
match result { match result {
Ok(_) => { Ok(_) => {
db::touch_sync_state(&conn, "flywheel");
log_event( log_event(
"gitea_webhook", "gitea_webhook",
"info", "info",