新增: 断点续跑与日志租约底座

This commit is contained in:
lxy
2026-10-05 23:41:18 +08:00
parent 8233ba095e
commit cb260aba4b
22 changed files with 1132 additions and 136 deletions
@@ -110,6 +110,8 @@ fn ai_conversation_from_row(row: &Row<'_>) -> std::result::Result<AiConversation
origin: row.get("origin").unwrap_or(None),
// boundary_json 列老库经 v45 迁移补建。unwrap_or(None) 兜底(NULL=未声明任务边界)。
boundary_json: row.get("boundary_json").unwrap_or(None),
// loop_state 列老库经 v49 迁移补建。unwrap_or(None) 兜底(NULL=非任务会话)。
loop_state: row.get("loop_state").unwrap_or(None),
})
}
@@ -185,25 +187,25 @@ impl_repo!(
from_row => |row| ai_conversation_from_row(row),
insert => |conn, rec| {
conn.execute(
"INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at, origin, boundary_json)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)",
"INSERT INTO ai_conversations (id, title, messages, provider_id, model, models, archived, pinned, prompt_tokens, completion_tokens, pinned_goals, pending_approvals, created_at, updated_at, origin, boundary_json, loop_state)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)",
params![
rec.id, rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
if rec.pinned { 1i32 } else { 0i32 },
rec.prompt_tokens, rec.completion_tokens,
rec.pinned_goals, rec.pending_approvals, rec.created_at, rec.updated_at, rec.origin,
rec.boundary_json
rec.boundary_json, rec.loop_state
],
)
},
update => |conn, rec| {
conn.execute(
"UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, pinned_goals = ?10, pending_approvals = ?11, updated_at = ?12, boundary_json = ?13 WHERE id = ?14",
"UPDATE ai_conversations SET title = ?1, messages = ?2, provider_id = ?3, model = ?4, models = ?5, archived = ?6, pinned = ?7, prompt_tokens = ?8, completion_tokens = ?9, pinned_goals = ?10, pending_approvals = ?11, updated_at = ?12, boundary_json = ?13, loop_state = ?14 WHERE id = ?15",
params![
rec.title, rec.messages, rec.provider_id, rec.model, rec.models, rec.archived,
if rec.pinned { 1i32 } else { 0i32 },
rec.prompt_tokens, rec.completion_tokens,
rec.pinned_goals, rec.pending_approvals, rec.updated_at, rec.boundary_json, rec.id
rec.pinned_goals, rec.pending_approvals, rec.updated_at, rec.boundary_json, rec.loop_state, rec.id
],
)
}
@@ -809,6 +811,72 @@ impl AiConversationRepo {
.map_err(storage_err)?
}
/// loop 生命周期标记写入(V49 loop_state,断点续跑底座)。
///
/// state: Some("running")=任务 loop spawn 时置起;Some("parked")=重启扫描改写;
/// None=正常收敛/用户停止时清理。纯元数据标记不进 update_full(防 save_conversation
/// 高频全量写覆盖 loop 生命周期语义),对齐 set_boundary_json/set_archived 模式。
/// 不动 updated_at(纯标记)。对话未落库(懒创建)时影响 0 行,调用方忽略。
pub async fn set_loop_state(&self, id: &str, state: Option<&str>) -> Result<bool> {
let conn = self.conn.clone();
let id = id.to_owned();
let value = state.map(|s| s.to_owned());
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let affected = guard
.execute(
"UPDATE ai_conversations SET loop_state = ?1 WHERE id = ?2",
params![value, id],
)
.map_err(storage_err)?;
Ok(affected > 0)
})
.await
.map_err(storage_err)?
}
/// 重启扫描:把所有 loop_state='running' 残留改写为 'parked'。
///
/// 进程启动时调用一次:上次运行中 loop_state=running 的会话,进程重启后 loop 必死
/// (内存态不复存在),残留 running 是假活。改写 parked 后侧栏读 list_recoverable
/// 展示「任务中断可恢复」。返回改写条数(0=无中断任务,静默)。
pub async fn mark_running_as_parked(&self) -> Result<u64> {
let conn = self.conn.clone();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let affected = guard
.execute(
"UPDATE ai_conversations SET loop_state = 'parked' WHERE loop_state = 'running'",
[],
)
.map_err(storage_err)?;
Ok(affected as u64)
})
.await
.map_err(storage_err)?
}
/// 列出可恢复会话(loop_state='parked',重启中断待一键恢复)。
pub async fn list_parked(&self) -> Result<Vec<AiConversationRecord>> {
let conn = self.conn.clone();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let mut stmt = guard
.prepare(
"SELECT * FROM ai_conversations WHERE loop_state = 'parked' ORDER BY updated_at DESC",
)
.map_err(storage_err)?;
let rows = stmt
.query_map([], ai_conversation_from_row)
.map_err(storage_err)?
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(storage_err)?;
Ok(rows)
})
.await
.map_err(storage_err)?
}
/// G1.3: 删除对话 + 其全部 ai_messages 子行(单事务原子)。
///
/// 背景:原宏生成 `delete` 只删 ai_conversations 主行,而 ai_messages 表无外键级联
+28 -5
View File
@@ -57,7 +57,9 @@ pub fn run(conn: &Connection) -> Result<()> {
// V46/V47 = 任务/灵感归档(archived INTEGER NOT NULL DEFAULT 0,归档功能 P1,
// 对标 ai_conversations.archived V4;归档与软删正交:回收站=删除缓冲,归档=低频可见保检索)。
// V48 = 项目归档(projects.archived 列,归档功能 P2,同款三方对齐)。
let steps: [(i32, fn(&Connection) -> Result<()>); 48] = [
// V49 = loop 生命周期标记(ai_conversations.loop_state TEXT,断点续跑底座:
// running/parked,重启扫描 running→parked 改写,侧栏「可恢复」标记)。
let steps: [(i32, fn(&Connection) -> Result<()>); 49] = [
(1, migrate_v1),
(2, migrate_v2),
(3, migrate_v3),
@@ -109,6 +111,8 @@ pub fn run(conn: &Connection) -> Result<()> {
(47, migrate_v47),
// V48 = 项目归档(projects.archived 列,归档功能 P2,对标 V46/V47 同款)。
(48, migrate_v48),
// V49 = loop 生命周期标记(ai_conversations.loop_state,断点续跑底座)。
(49, migrate_v49),
];
for (version, migrate_fn) in steps {
@@ -1427,6 +1431,25 @@ fn migrate_v48(conn: &Connection) -> Result<()> {
Ok(())
}
/// V49: 幂等补 ai_conversations.loop_state 列(断点续跑底座)
///
/// loop_state TEXT NULL:None=非任务会话/已收敛;"running"=任务 loop 在跑(spawn 时置起);
/// "parked"=重启扫描时 running 残留改写(侧栏标「任务中断可恢复」+ 一键恢复)。
/// 对齐无人值守长任务设计 §4.5 拍板 A(不自动续跑,标记+一键恢复)。
/// - **幂等**:column_exists 守卫;**可逆**:`ALTER TABLE ai_conversations DROP COLUMN loop_state`。
fn migrate_v49(conn: &Connection) -> Result<()> {
if !column_exists(conn, "ai_conversations", "loop_state") {
conn.execute(
"ALTER TABLE ai_conversations ADD COLUMN loop_state TEXT",
[],
)?;
tracing::info!("v49: 补建 ai_conversations.loop_state 列(断点续跑标记)");
}
conn.execute("INSERT OR IGNORE INTO schema_version (version) VALUES (?)", [49])?;
tracing::info!("迁移 v49 完成: ai_conversations 加 loop_state 列");
Ok(())
}
/// V21 建表 SQL — 消息拆分存储 ai_messages 表
///
/// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。
@@ -2455,8 +2478,8 @@ mod tests {
)
.expect("查 schema_version 应成功");
assert_eq!(
max_version, 48,
"全量迁移后 schema_version 应为 48(实际 {}),说明某条 migrate_vN 链路断在中间",
max_version, 49,
"全量迁移后 schema_version 应为 49(实际 {}),说明某条 migrate_vN 链路断在中间",
max_version
);
@@ -2533,7 +2556,7 @@ mod tests {
|r| r.get(0),
)
.unwrap();
assert_eq!(max_v, 48, "首轮应推进到 48");
assert_eq!(max_v, 49, "首轮应推进到 49");
// 清空版本表强制全链第二遍(每步 execute 第二次)
conn.execute("DELETE FROM schema_version", []).unwrap();
@@ -2545,7 +2568,7 @@ mod tests {
|r| r.get(0),
)
.unwrap();
assert_eq!(max_v2, 48, "重跑后应重新推进到 48");
assert_eq!(max_v2, 49, "重跑后应重新推进到 49");
}
// ============================================================
+6
View File
@@ -390,6 +390,12 @@ pub struct AiConversationRecord {
/// 无人值守任务边界快照(V45):TaskBoundary JSON;None=未声明(现状逐次审批)。
/// 生命周期=会话存续(发起时一次声明,随会话同亡),save 随行写/switch 读还原。
pub boundary_json: Option<String>,
/// loop 生命周期标记(V49,断点续跑底座):None=非任务会话/已收敛清理;
/// Some("running")=任务 loop 在跑(spawn 时置起);Some("parked")=进程重启时检测到
/// running 残留 → 启动扫描改写,侧栏标「任务中断可恢复」+ 一键恢复。
/// 纯元数据标记不进 update_full 通用 UPDATE,走专用 set_loop_state。
#[serde(default)]
pub loop_state: Option<String>,
}
/// AI 工具执行审计记录