新增: 知识图谱Phase2事件流(父⑥) — project_events统一事件流+埋点+timeline

父⑥ Phase 2(对标设计 §2.4):
- V30 迁移:project_events 追加型审计表(id/project_id/event_type/entity_type/entity_id/from_state/to_state/context_json/source/conversation_id/created_at)+ 双索引(project,time / entity)
- ProjectEventRecord + ProjectEventRepo(insert 追加型/get_by_project/get_by_entity/list_recent)
- 事件埋点(best-effort,hook/after,失败 warn 不阻断主操作):
  - task.rs emit_event 辅助 + create_task/advance_task/move_task_queue 埋点
  - idea.rs promote_idea 埋点(idea_promoted,project_id=新项目)
  - project.rs create/delete 埋点
  - 注:idea_created 暂不埋点(idea 无 project_id,待 Inbox 项目落地,NOT NULL+FK 约束)
- get_project_timeline IPC(events 模块)+ lib.rs 注册 + AI 工具(基线 +N)
- source 语义:IPC 标 human,AI 工具路径标 ai

注:workflow 脚本中文变量名 const 致验证 agent 未跑,主控独立 cargo check EXIT 0 + grep 落地齐全核验。
This commit is contained in:
2026-06-27 00:23:17 +08:00
parent 744b68da5b
commit 5a893680b7
12 changed files with 853 additions and 20 deletions

View File

@@ -7,6 +7,7 @@
//! - [`mod@project_repo`]:ProjectRepo/BranchRepo/ReleaseRepo/WorkflowRepo/NodeExecutionRepo //! - [`mod@project_repo`]:ProjectRepo/BranchRepo/ReleaseRepo/WorkflowRepo/NodeExecutionRepo
//! - [`mod@task_repo`]:TaskRepo(含 advance_status_atomic 状态机收口) //! - [`mod@task_repo`]:TaskRepo(含 advance_status_atomic 状态机收口)
//! - [`mod@task_link_repo`]:TaskLinkRepo(任务横向关联 task_links,V29,知识图谱 Phase 1) //! - [`mod@task_link_repo`]:TaskLinkRepo(任务横向关联 task_links,V29,知识图谱 Phase 1)
//! - [`mod@project_event_repo`]:ProjectEventRepo(统一事件流 project_events,V30,知识图谱 Phase 2)
//! - [`mod@conversation_repo`]:AiProviderRepo/AiConversationRepo/AiToolExecutionRepo //! - [`mod@conversation_repo`]:AiProviderRepo/AiConversationRepo/AiToolExecutionRepo
//! - [`mod@idea_repo`]:IdeaRepo/KnowledgeRepo/KnowledgeEventsRepo + 向量工具 //! - [`mod@idea_repo`]:IdeaRepo/KnowledgeRepo/KnowledgeEventsRepo + 向量工具
//! - [`mod@idea_eval_repo`]:IdeaEvalRepo(灵感评估历史追加型审计表 idea_evaluations,V22) //! - [`mod@idea_eval_repo`]:IdeaEvalRepo(灵感评估历史追加型审计表 idea_evaluations,V22)
@@ -19,6 +20,7 @@ mod conversation_repo;
mod idea_eval_repo; mod idea_eval_repo;
mod idea_repo; mod idea_repo;
mod message_repo; mod message_repo;
mod project_event_repo;
mod project_repo; mod project_repo;
mod settings; mod settings;
mod task_link_repo; mod task_link_repo;
@@ -28,6 +30,7 @@ pub use conversation_repo::*;
pub use idea_eval_repo::*; pub use idea_eval_repo::*;
pub use idea_repo::*; pub use idea_repo::*;
pub use message_repo::*; pub use message_repo::*;
pub use project_event_repo::*;
pub use project_repo::*; pub use project_repo::*;
pub use settings::*; pub use settings::*;
pub use task_link_repo::*; pub use task_link_repo::*;
@@ -297,6 +300,7 @@ mod baseline_tests {
let _ = ProjectRepo::new(&db); let _ = ProjectRepo::new(&db);
let _ = TaskRepo::new(&db); let _ = TaskRepo::new(&db);
let _ = TaskLinkRepo::new(&db); let _ = TaskLinkRepo::new(&db);
let _ = ProjectEventRepo::new(&db);
let _ = BranchRepo::new(&db); let _ = BranchRepo::new(&db);
let _ = ReleaseRepo::new(&db); let _ = ReleaseRepo::new(&db);
let _ = WorkflowRepo::new(&db); let _ = WorkflowRepo::new(&db);

View File

@@ -0,0 +1,391 @@
//! 项目事件流 Repo:ProjectEventRepo(project_events 表,知识图谱 Phase 2 V30)
//!
//! 统一事件流——一张表承载跨实体(idea/task/workflow/knowledge/module/service/project)的
//! 全部事件,AI 精准检索的基础(回答"上周做了什么 / 这个任务为何 blocked / 决策何时做出")。
//! 对标设计 docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.4。
//!
//! 关键设计(对标 TaskLinkRepo 全专用路径):
//! - **追加型审计表**:只 INSERT,无 UPDATE/DELETE(历史不改可追溯,对标 knowledge_events V10 /
//! idea_evaluations V22)。故不走 `impl_repo!` 宏(宏生成的 update/delete/update_field 对审计
//! 语义是错的),手写专用 insert + 三个查询方法。
//! - 不登记通用列白名单(`allowed_columns_for("project_events")` 返回 None):同 task_links,
//! 无通用 update_field 路径故无需白名单。`baseline_tests::all_known_tables_have_column_whitelist`
//! 的已知表数组不含本表(与 task_links 同款豁免)。
//! - `event_type` / `entity_type` 白名单**应用层校验**(类比 task_links.link_type),不进 DB 约束
//! (保留扩展性,新增事件/实体类型无需迁移)。本 Repo insert **不做白名单校验**——埋点层
//! (commands/IPC hook)是事件唯一生产者,字符串字面量自带约束;开放校验留给未来若开放外部写入。
//! - 埋点策略(设计 §2.4 hook/after)与 knowledge_events 双写均在调用方(commands 层)实现,
//! 非 Repo 职责。事件写入失败 best-effort 不阻断主操作(设计 §10.1)。
use std::sync::Arc;
use rusqlite::{params, Row};
use tokio::sync::Mutex;
use df_types::error::Result;
use crate::db::Database;
use crate::models::ProjectEventRecord;
use super::{now_millis_str, storage_err};
// ============================================================
// from_row 辅助函数
// ============================================================
fn project_event_from_row(row: &Row<'_>) -> std::result::Result<ProjectEventRecord, rusqlite::Error> {
Ok(ProjectEventRecord {
id: row.get("id")?,
project_id: row.get("project_id")?,
event_type: row.get("event_type")?,
entity_type: row.get("entity_type")?,
entity_id: row.get("entity_id")?,
from_state: row.get("from_state")?,
to_state: row.get("to_state")?,
context_json: row.get("context_json")?,
source: row.get("source")?,
conversation_id: row.get("conversation_id")?,
created_at: row.get("created_at")?,
})
}
/// project_events SELECT 列清单(11 字段,防 COLS 漂移,对标 KNOWLEDGE_COLS)。
const PROJECT_EVENT_COLS: &str = "id,project_id,event_type,entity_type,entity_id,from_state,\
to_state,context_json,source,conversation_id,created_at";
// ============================================================
// ProjectEventRepo
// ============================================================
/// 项目统一事件流表 Repo(project_events,V30,追加型审计表)。
///
/// 不走 `impl_repo!` 宏:① 追加型审计表无 UPDATE/DELETE(宏生成的 update/update_field/delete
/// 对审计语义是错的);② 表无 `updated_at`(宏的 update_field 硬编码 `SET ... updated_at = ?`
/// 会触发 "no such column: updated_at")。故手写专用方法,对标 TaskLinkRepo 全专用路径。
pub struct ProjectEventRepo {
conn: Arc<Mutex<rusqlite::Connection>>,
}
impl ProjectEventRepo {
pub fn new(db: &Database) -> Self {
Self { conn: db.conn() }
}
/// 追加一条事件(只 INSERT,无 UPDATE,审计语义不可篡改)。
///
/// 返回插入的事件 id。`created_at` 由本方法内部取当前毫秒时间戳填充(调用方无需传),
/// 与 IdeaRepo::insert(now_millis_str)同款收口——事件生产时机即落库时机,时间由存储层定。
///
/// 不做 event_type / entity_type 白名单校验(埋点层是唯一生产者,字符串字面量自带约束)。
pub async fn insert(&self, record: ProjectEventRecord) -> Result<String> {
let conn = self.conn.clone();
// created_at 由存储层覆盖(调用方传入的不可信),保证「事件生产时机即落库时机」。
let now = now_millis_str();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
guard
.execute(
"INSERT INTO project_events \
(id, project_id, event_type, entity_type, entity_id, from_state, to_state, \
context_json, source, conversation_id, created_at) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
params![
record.id,
record.project_id,
record.event_type,
record.entity_type,
record.entity_id,
record.from_state,
record.to_state,
record.context_json,
record.source,
record.conversation_id,
now,
],
)
.map_err(storage_err)?;
Ok(record.id)
})
.await
.map_err(storage_err)?
}
/// 按项目查询事件流(time DESC,id DESC 兜底同毫秒并列,limit 钳制上限 200,
/// 命中 idx_project_events_project)。
///
/// AI / Dashboard 项目时间线视图用:取某项目最近 N 条事件(设计 §2.4 / §10.3.4)。
/// 同毫秒插入的事件(高频埋点常见)以 id DESC 兜底——生产 id 是 ULID(字典序单调递增),
/// 字典序 DESC 等价插入逆序,保证「最后发生的事件在最前」语义稳定(纯 created_at DESC
/// 同毫秒并列时顺序不确定,审计表必须可复现)。
pub async fn get_by_project(
&self,
project_id: &str,
limit: u32,
) -> Result<Vec<ProjectEventRecord>> {
let conn = self.conn.clone();
let project_id = project_id.to_owned();
// 钳制 limit 防滥用(最大 200,对标 KnowledgeEventsRepo::list_recent)
let safe_limit = limit.min(200) as i64;
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let sql = format!(
"SELECT {PROJECT_EVENT_COLS} FROM project_events \
WHERE project_id = ?1 ORDER BY created_at DESC, id DESC LIMIT ?2"
);
let mut stmt = guard.prepare(&sql).map_err(storage_err)?;
let rows = stmt
.query_map(params![project_id, safe_limit], project_event_from_row)
.map_err(storage_err)?;
let mut results = Vec::new();
for r in rows {
results.push(r.map_err(storage_err)?);
}
Ok(results)
})
.await
.map_err(storage_err)?
}
/// 按实体反查事件流(命中 idx_project_events_entity)。
///
/// AI「这个任务/灵感发生过什么」反查:取某实体的全部事件(时间倒序)。跨项目——
/// 同一实体(如被移动过项目的任务)的全部历史事件都返回。同毫秒并列以 id DESC 兜底
/// (语义稳定,见 get_by_project 说明)。
pub async fn get_by_entity(
&self,
entity_type: &str,
entity_id: &str,
) -> Result<Vec<ProjectEventRecord>> {
let conn = self.conn.clone();
let entity_type = entity_type.to_owned();
let entity_id = entity_id.to_owned();
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let sql = format!(
"SELECT {PROJECT_EVENT_COLS} FROM project_events \
WHERE entity_type = ?1 AND entity_id = ?2 ORDER BY created_at DESC, id DESC"
);
let mut stmt = guard.prepare(&sql).map_err(storage_err)?;
let rows = stmt
.query_map(params![entity_type, entity_id], project_event_from_row)
.map_err(storage_err)?;
let mut results = Vec::new();
for r in rows {
results.push(r.map_err(storage_err)?);
}
Ok(results)
})
.await
.map_err(storage_err)?
}
/// 跨项目列最近 N 条事件(全表 created_at DESC,id DESC 兜底,top-N,命中
/// idx_project_events_project 的 created_at 维度)。
///
/// 对标 KnowledgeEventsRepo::list_recent:供「全局事件流 / 跨项目时间线」视图使用
/// (设计 §10.3.4 模式 B 事件流)。limit 上限钳制 200,防恶意/失误传超大值。
/// 同毫秒并列以 id DESC 兜底(语义稳定,见 get_by_project 说明)。
pub async fn list_recent(&self, limit: u32) -> Result<Vec<ProjectEventRecord>> {
let conn = self.conn.clone();
let safe_limit = limit.min(200) as i64;
tokio::task::spawn_blocking(move || {
let guard = conn.blocking_lock();
let sql = format!(
"SELECT {PROJECT_EVENT_COLS} FROM project_events \
ORDER BY created_at DESC, id DESC LIMIT ?1"
);
let mut stmt = guard.prepare(&sql).map_err(storage_err)?;
let rows = stmt
.query_map(params![safe_limit], project_event_from_row)
.map_err(storage_err)?;
let mut results = Vec::new();
for r in rows {
results.push(r.map_err(storage_err)?);
}
Ok(results)
})
.await
.map_err(storage_err)?
}
}
// ============================================================
// 单元测试 — ProjectEventRepo insert + get_by_project + get_by_entity(内存 DB,
// 对标 task_link_repo 测试:先建占位 project 满足 FK,再插事件)
// ============================================================
#[cfg(test)]
mod tests {
use super::*;
use crate::crud::ProjectRepo;
use crate::models::ProjectRecord;
/// 构造内存 DB + 占位 project(project_events.project_id FK 要求 projects 存在)。
async fn setup() -> (crate::db::Database, ProjectEventRepo) {
let db = crate::db::Database::open_in_memory()
.await
.expect("open_in_memory");
let repo = ProjectEventRepo::new(&db);
let project_repo = ProjectRepo::new(&db);
project_repo
.insert(ProjectRecord {
id: "proj-1".to_string(),
name: "proj-1".to_string(),
description: String::new(),
status: "active".to_string(),
idea_id: None,
path: None,
stack: None,
created_at: "1700000000000".to_string(),
updated_at: "1700000000000".to_string(),
})
.await
.unwrap();
(db, repo)
}
/// 构造一条 ProjectEventRecord fixture(可指定 created_at 便于排序断言)。
fn erec(id: &str, project_id: &str, created_at: &str) -> ProjectEventRecord {
ProjectEventRecord {
id: id.to_string(),
project_id: project_id.to_string(),
event_type: "task_advanced".to_string(),
entity_type: Some("task".to_string()),
entity_id: Some("task-A".to_string()),
from_state: Some("todo".to_string()),
to_state: Some("in_progress".to_string()),
context_json: None,
source: Some("ai".to_string()),
conversation_id: Some("conv-1".to_string()),
created_at: created_at.to_string(),
}
}
#[tokio::test]
async fn insert_and_read_back_full_fields() {
let (_db, repo) = setup().await;
let id = repo.insert(erec("e1", "proj-1", "1700000000001")).await.unwrap();
assert_eq!(id, "e1");
let got = repo
.get_by_project("proj-1", 10)
.await
.unwrap()
.into_iter()
.next()
.expect("应有 1 条事件");
// 全字段回读校验
assert_eq!(got.id, "e1");
assert_eq!(got.project_id, "proj-1");
assert_eq!(got.event_type, "task_advanced");
assert_eq!(got.entity_type.as_deref(), Some("task"));
assert_eq!(got.entity_id.as_deref(), Some("task-A"));
assert_eq!(got.from_state.as_deref(), Some("todo"));
assert_eq!(got.to_state.as_deref(), Some("in_progress"));
assert!(got.context_json.is_none());
assert_eq!(got.source.as_deref(), Some("ai"));
assert_eq!(got.conversation_id.as_deref(), Some("conv-1"));
// created_at 由存储层覆盖为当前毫秒(非 fixture 传入的旧值)
assert_ne!(got.created_at, "1700000000001", "created_at 应由存储层覆盖");
assert!(!got.created_at.is_empty());
}
#[tokio::test]
async fn get_by_project_orders_desc_and_filters() {
let (_db, repo) = setup().await;
// 三条事件不同时间(create_at 由存储层覆盖,但插入顺序决定落库 created_at 单调递增,
// 故 DESC 应按插入逆序返回——最后插的最前)
repo.insert(erec("e1", "proj-1", "1")).await.unwrap();
repo.insert(erec("e2", "proj-1", "1")).await.unwrap();
repo.insert(erec("e3", "proj-1", "1")).await.unwrap();
let got = repo.get_by_project("proj-1", 10).await.unwrap();
assert_eq!(got.len(), 3, "应返回全部 3 条事件");
// DESC:最后插入的 e3 在前
assert_eq!(got[0].id, "e3");
assert_eq!(got[1].id, "e2");
assert_eq!(got[2].id, "e1");
// 过滤:不存在的项目返回空
let empty = repo.get_by_project("no-such-proj", 10).await.unwrap();
assert!(empty.is_empty());
}
#[tokio::test]
async fn get_by_project_limit_clamped() {
let (_db, repo) = setup().await;
for i in 0..5 {
repo.insert(erec(&format!("e{i}"), "proj-1", "1"))
.await
.unwrap();
}
// limit=2 只返回 2 条(最新两条)
let got = repo.get_by_project("proj-1", 2).await.unwrap();
assert_eq!(got.len(), 2);
// limit=99999 钳制为 200,但只有 5 条,返回 5 条
let got_all = repo.get_by_project("proj-1", 99999).await.unwrap();
assert_eq!(got_all.len(), 5, "钳制后仍返回全部存在的事件");
}
#[tokio::test]
async fn get_by_entity_cross_project_and_filters() {
let (_db, repo) = setup().await;
// 同一实体 task-A 在 proj-1 下产生 2 条事件
repo.insert(erec("e1", "proj-1", "1")).await.unwrap();
repo.insert(erec("e2", "proj-1", "1")).await.unwrap();
// 另一实体 task-B 1 条(改 entity_id)
let mut other = erec("e3", "proj-1", "1");
other.entity_id = Some("task-B".to_string());
repo.insert(other).await.unwrap();
// task-A 反查:返回 2 条(DESC)
let got_a = repo.get_by_entity("task", "task-A").await.unwrap();
assert_eq!(got_a.len(), 2);
assert_eq!(got_a[0].id, "e2");
assert_eq!(got_a[1].id, "e1");
// task-B 反查:返回 1 条
let got_b = repo.get_by_entity("task", "task-B").await.unwrap();
assert_eq!(got_b.len(), 1);
assert_eq!(got_b[0].id, "e3");
// 不存在的实体返回空
let empty = repo.get_by_entity("task", "no-such").await.unwrap();
assert!(empty.is_empty());
}
#[tokio::test]
async fn list_recent_cross_project() {
let (_db, repo) = setup().await;
for i in 0..3 {
repo.insert(erec(&format!("e{i}"), "proj-1", "1"))
.await
.unwrap();
}
// list_recent 跨项目取最近 N 条(此处仅 proj-1)
let got = repo.list_recent(10).await.unwrap();
assert_eq!(got.len(), 3);
// DESC:最后插入的最前
assert_eq!(got[0].id, "e2");
}
/// entity_type/entity_id 为 None(纯决策日志无明确实体)也能正常插入与查询。
#[tokio::test]
async fn insert_with_null_entity_fields() {
let (_db, repo) = setup().await;
let mut rec = erec("e1", "proj-1", "1");
rec.event_type = "decision_made".to_string();
rec.entity_type = None;
rec.entity_id = None;
rec.from_state = None;
rec.to_state = None;
repo.insert(rec).await.unwrap();
let got = repo.get_by_project("proj-1", 10).await.unwrap();
assert_eq!(got.len(), 1);
assert!(got[0].entity_type.is_none());
assert!(got[0].entity_id.is_none());
assert!(got[0].from_state.is_none());
assert!(got[0].to_state.is_none());
}
}

View File

@@ -37,7 +37,10 @@ pub fn run(conn: &Connection) -> Result<()> {
// 项目知识图谱与任务队列系统-2026-06-26.md §2.1/§2.2):tasks 加 queue(管理池)/ // 项目知识图谱与任务队列系统-2026-06-26.md §2.1/§2.2):tasks 加 queue(管理池)/
// parent_id(父任务纵向关联)/content_json(结构化需求规格)三列 + task_links 表 // parent_id(父任务纵向关联)/content_json(结构化需求规格)三列 + task_links 表
// (横向关联 depends_on/blocks/relates_to),为 AI 编排地基打底。 // (横向关联 depends_on/blocks/relates_to),为 AI 编排地基打底。
let steps: [(i32, fn(&Connection) -> Result<()>); 29] = [ // V30 = 知识图谱 Phase 2 统一事件流(对标设计 §2.4):project_events 追加型审计表,
// 跨实体(idea/task/workflow/knowledge/module/service/project)事件流,AI 精准检索
// 的基础(回答"上周做了什么/这个任务为何 blocked/决策何时做出")。
let steps: [(i32, fn(&Connection) -> Result<()>); 30] = [
(1, migrate_v1), (1, migrate_v1),
(2, migrate_v2), (2, migrate_v2),
(3, migrate_v3), (3, migrate_v3),
@@ -67,6 +70,7 @@ pub fn run(conn: &Connection) -> Result<()> {
(27, migrate_v27), (27, migrate_v27),
(28, migrate_v28), (28, migrate_v28),
(29, migrate_v29), (29, migrate_v29),
(30, migrate_v30),
]; ];
for (version, migrate_fn) in steps { for (version, migrate_fn) in steps {
@@ -773,6 +777,74 @@ fn migrate_v29(conn: &Connection) -> Result<()> {
Ok(()) Ok(())
} }
/// V30:知识图谱 Phase 2 统一事件流 — project_events 追加型审计表
///
/// 对标 docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.4,
/// 为「AI 精准检索项目事件流」打数据层地基。一张表承载跨实体(idea/task/workflow/
/// knowledge/module/service/project)的全部事件,回答"上周做了什么 / 这个任务为何 blocked /
/// 这个决策何时做出"——此前需跨 6 张表 JOIN。
///
/// **追加型审计表**(对标 knowledge_events V10 / idea_evaluations V22):只 INSERT 不 UPDATE/
/// DELETE,历史不改可追溯。故表无 updated_at,无通用 update/delete 路径,无列白名单登记
/// (ProjectEventRepo 走专用方法,与 TaskLinkRepo 同款全专用路径)。
///
/// **与 knowledge_events 的关系**(对标设计 D9 / §2.4):knowledge_events 保持不变(知识库专属,
/// 向后兼容),project_events 是全局事件流;knowledge 相关事件**同时写入两者**——本表只管写入,
/// 双写策略在埋点层(commands/IPC hook)实现,非本表职责。
///
/// 列语义(对标设计 §2.4):
/// - `event_type`:事件类型白名单(idea_created/idea_promoted/...task_created/task_advanced/
/// .../decision_made 等,见设计 §2.4 注释枚举)。白名单**应用层校验**(类比 task_links
/// link_type),不进 DB 约束(保留扩展性,新增事件类型无需迁移)。
/// - `entity_type` / `entity_id`:事件指向的实体(可空——部分事件无明确实体,如纯决策日志)。
/// entity_type 白名单同样应用层校验(idea/project/task/workflow/knowledge/module/service)。
/// - `from_state` / `to_state`:状态变化前后(仅状态变化类事件有值,如 task_advanced;
/// created/referenced 类为 NULL)。
/// - `context_json`:事件附加上下文(JSON 字符串,因 event_type 而异)。
/// - `source`:ai / human / system(AI Working 溯源——区分是 AI 自主操作还是人操作)。
/// - `conversation_id`:触发事件的对应对话(AI Working 溯源链:事件→对话→决策,可空)。
///
/// **埋点策略**(设计 §2.4 hook/after):在现有 IPC 命令(create_task/advance_task/create_project
/// /idea_promote 等)执行后追加事件写入,不侵入业务逻辑。事件写入失败 best-effort 不阻断主操作
/// (设计 §10.1 风险已识别),由埋点层实现,非本表职责。
///
/// **CREATE TABLE IF NOT EXISTS 幂等**:新库建表、老库(V30 之前的库)已有则跳过,均安全。
/// 索引覆盖最高频查询:
/// - `idx_project_events_project(project_id, created_at)`:按项目查事件流(时间倒序,Dashboard 时间线)
/// - `idx_project_events_entity(entity_type, entity_id)`:按实体反查(AI「这个任务发生过什么」)
fn migrate_v30(conn: &Connection) -> Result<()> {
conn.execute(
"CREATE TABLE IF NOT EXISTS project_events (
id TEXT PRIMARY KEY,
project_id TEXT NOT NULL REFERENCES projects(id),
event_type TEXT NOT NULL,
entity_type TEXT,
entity_id TEXT,
from_state TEXT,
to_state TEXT,
context_json TEXT,
source TEXT,
conversation_id TEXT,
created_at TEXT NOT NULL
)",
[],
)?;
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_project_events_project \
ON project_events(project_id, created_at)",
[],
)?;
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_project_events_entity \
ON project_events(entity_type, entity_id)",
[],
)?;
tracing::info!("v30: 建 project_events 表 + 索引(统一事件流,知识图谱 Phase 2)");
conn.execute("INSERT INTO schema_version (version) VALUES (?)", [30])?;
tracing::info!("迁移 v30 完成");
Ok(())
}
/// V21 建表 SQL — 消息拆分存储 ai_messages 表 /// V21 建表 SQL — 消息拆分存储 ai_messages 表
/// ///
/// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。 /// 与 V9_SQL 中的 ai_messages 镜像(V9 给新库,此 const 给老库 V21 迁移用 IF NOT EXISTS)。

View File

@@ -146,6 +146,41 @@ pub struct TaskLinkRecord {
pub created_at: String, pub created_at: String,
} }
/// 项目事件流记录(project_events 表,知识图谱 Phase 2 V30,追加型审计表)。
///
/// 一张表承载跨实体(idea/task/workflow/knowledge/module/service/project)的全部事件,
/// 是 AI 精准检索项目事件流的基础(回答"上周做了什么 / 这个任务为何 blocked / 决策何时做出")。
/// 对标设计 docs/02-架构设计/专项设计/项目知识图谱与任务队列系统-2026-06-26.md §2.4。
///
/// **追加型审计**(对标 KnowledgeEventRecord / IdeaEvaluationRecord):只 INSERT 不 UPDATE/
/// DELETE,历史不改可追溯。无 updated_at 字段,无通用 update/delete 路径。
///
/// - `project_id`:所属项目(FK projects.id)。
/// - `event_type`:事件类型白名单(idea_created/task_advanced/decision_made 等,应用层校验)。
/// - `entity_type` / `entity_id`:事件指向的实体(可空——纯决策日志等无明确实体)。entity_type
/// 白名单应用层校验(idea/project/task/workflow/knowledge/module/service)。
/// - `from_state` / `to_state`:状态变化前后(仅状态变化类事件有值;created/referenced 类为 None)。
/// - `context_json`:事件附加上下文(JSON 字符串,因 event_type 而异)。
/// - `source`:ai / human / system(AI Working 溯源——区分 AI 自主操作还是人操作)。
/// - `conversation_id`:触发事件的对应对话(AI Working 溯源链:事件→对话→决策,可空)。
///
/// **与 knowledge_events 关系**(设计 D9):knowledge_events 保持不变(知识库专属),knowledge
/// 相关事件同时写入两者,双写在埋点层(commands/IPC hook)实现。
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProjectEventRecord {
pub id: String,
pub project_id: String,
pub event_type: String,
pub entity_type: Option<String>,
pub entity_id: Option<String>,
pub from_state: Option<String>,
pub to_state: Option<String>,
pub context_json: Option<String>,
pub source: Option<String>,
pub conversation_id: Option<String>,
pub created_at: String,
}
/// 分支记录 /// 分支记录
#[derive(Debug, Clone, Serialize, Deserialize)] #[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BranchRecord { pub struct BranchRecord {

View File

@@ -41,7 +41,7 @@ graph TD
P3["父③ AI对话体验<br/>📋待办"]:::todo P3["父③ AI对话体验<br/>📋待办"]:::todo
P4["父④ F-09 per-conv<br/>📋待办"]:::todo P4["父④ F-09 per-conv<br/>📋待办"]:::todo
P5["父⑤ 灵感模块<br/>✅完成(⑤.1/①.4/⑤.2)"]:::done P5["父⑤ 灵感模块<br/>✅完成(⑤.1/①.4/⑤.2)"]:::done
P6["父⑥ Phase2-5<br/>📋待办"]:::todo P6["父⑥ Phase2-5<br/>🔨Phase2✅·3-5待办"]:::doing
P7["父⑦ 技术债<br/>📋待办"]:::todo P7["父⑦ 技术债<br/>📋待办"]:::todo
P1 -.先行.-> P2 P1 -.先行.-> P2
@@ -59,7 +59,7 @@ graph TD
| **父③** AI对话体验 | 📋 待办 | ③.1 B-260619-04 ToolCard / ③.2 REFACTOR-260619-04 审批状态机 | — | | **父③** AI对话体验 | 📋 待办 | ③.1 B-260619-04 ToolCard / ③.2 REFACTOR-260619-04 审批状态机 | — |
| **父④** F-09 per-conv | 📋 待办 | ④.1 streaming/currentText per-conv | — | | **父④** F-09 per-conv | 📋 待办 | ④.1 streaming/currentText per-conv | — |
| **父⑤** 灵感模块 | ✅ 完成 | ⑤.1 软删除✅ / ①.4 雷达图✅ / ⑤.2 #05✅/#06拆const✅/#09/#10表单(逗号tags)✅ / #07 DEC-02保留purge(不改) / 附:priority_from_i32跨层映射修复(对齐前端0=critical) | #07→②.1 | | **父⑤** 灵感模块 | ✅ 完成 | ⑤.1 软删除✅ / ①.4 雷达图✅ / ⑤.2 #05✅/#06拆const✅/#09/#10表单(逗号tags)✅ / #07 DEC-02保留purge(不改) / 附:priority_from_i32跨层映射修复(对齐前端0=critical) | #07→②.1 |
| **父⑥** Phase2-5 | 📋 待办 | ⑥.1事件流 / ⑥.2基础设施 / ⑥.3注入 / ⑥.4前端 | ⑥→父②, ⑥.3→父②+父④+G1 | | **父⑥** Phase2-5 | 🔨 Phase2✅ | ⑥.1事件流✅(V30 project_events+埋点9处+timeline IPC+AI工具) / ⑥.2基础设施待办 / ⑥.3注入(依赖父②+父④) / ⑥.4前端 | ⑥→父②, ⑥.3→父②+父④+G1 |
| **父⑦** 技术债 | 📋 待办 | SMELL-P1-6 / conditions / CR缓存 / UX分页 / 审批超时 / miniapp / 双监听器 | — (穿插) | | **父⑦** 技术债 | 📋 待办 | SMELL-P1-6 / conditions / CR缓存 / UX分页 / 审批超时 / miniapp / 双监听器 | — (穿插) |
> **推进路径**:父①先行(攒批提交)→ 父②主线 workflow → 父③/④并行 → 父⑤/⑥/⑦穿插。每父任务一个 workflow 批,子任务相关文件批量读改减少交互。**状态更新约定**:子项完成→父状态 🔨;全子完成→父 ✅;每批提交后同步此表。 > **推进路径**:父①先行(攒批提交)→ 父②主线 workflow → 父③/④并行 → 父⑤/⑥/⑦穿插。每父任务一个 workflow 批,子任务相关文件批量读改减少交互。**状态更新约定**:子项完成→父状态 🔨;全子完成→父 ✅;每批提交后同步此表。

View File

@@ -490,7 +490,7 @@ fn register_http_tools(registry: &mut AiToolRegistry) {
); );
} }
/// 数据层 AI 工具注册(24 个持 db 的 CRUD/状态机/工作流/知识图谱工具)——从 build_ai_tool_registry 抽出。 /// 数据层 AI 工具注册(25 个持 db 的 CRUD/状态机/工作流/知识图谱工具)——从 build_ai_tool_registry 抽出。
/// ///
/// 工具闭包捕获 `db: &Arc<Database>` Arc 重建 Repo(列表/创建/更新/删除/状态推进/工作流/任务关联)。 /// 工具闭包捕获 `db: &Arc<Database>` Arc 重建 Repo(列表/创建/更新/删除/状态推进/工作流/任务关联)。
/// 例外:run_workflow handler 防御返回 Err(CR-52),不持 db 不 clone,真正执行经 /// 例外:run_workflow handler 防御返回 Err(CR-52),不持 db 不 clone,真正执行经
@@ -498,6 +498,8 @@ fn register_http_tools(registry: &mut AiToolRegistry) {
/// ///
/// 知识图谱 Phase 1(2026-06-26):新增 register_task_graph_tools 子函数(6 工具:task_link CRUD /// 知识图谱 Phase 1(2026-06-26):新增 register_task_graph_tools 子函数(6 工具:task_link CRUD
/// + 父子树 + 跨池移动 + content 更新),data 层 18→24。 /// + 父子树 + 跨池移动 + content 更新),data 层 18→24。
/// 知识图谱 Phase 2(2026-06-26):register_task_graph_tools 增 get_project_timeline(项目事件流查询),
/// data 层 24→25。
/// ///
/// SMELL-P0-2:抽自原 build_ai_tool_registry 1091 行单函数(数据+文件混合)。 /// SMELL-P0-2:抽自原 build_ai_tool_registry 1091 行单函数(数据+文件混合)。
/// 18 个 register 调用【原样移入】,零行为变更,仅机械搬运。 /// 18 个 register 调用【原样移入】,零行为变更,仅机械搬运。
@@ -510,8 +512,10 @@ fn register_http_tools(registry: &mut AiToolRegistry) {
/// - register_trash_tools(1):list_trash /// - register_trash_tools(1):list_trash
/// 知识图谱 Phase 1(2026-06-26)新增第 6 子函数: /// 知识图谱 Phase 1(2026-06-26)新增第 6 子函数:
/// - register_task_graph_tools(6):task_link CRUD + 父子树 + 跨池移动 + content 更新 /// - register_task_graph_tools(6):task_link CRUD + 父子树 + 跨池移动 + content 更新
/// AiToolRegistry 底层 HashMap(注册顺序无关),拆分后工具集合与原一致 + 新增 6, /// 知识图谱 Phase 2(2026-06-26):register_task_graph_tools 增 get_project_timeline(项目事件流,只读),
/// 行为零变更(create_task 扩展参数 + 6 新工具,基线测试 test_build_ai_tool_registry_baseline_tool_count 守护) /// 该子函数 6→7 工具,data 层 24→25
/// AiToolRegistry 底层 HashMap(注册顺序无关),拆分后工具集合与原一致 + 新增,
/// 行为零变更(create_task 扩展参数 + 7 新工具,基线测试 test_build_ai_tool_registry_baseline_tool_count 守护)。
fn register_data_tools(registry: &mut AiToolRegistry, db: &Arc<Database>) { fn register_data_tools(registry: &mut AiToolRegistry, db: &Arc<Database>) {
register_project_tools(registry, db); register_project_tools(registry, db);
register_task_tools(registry, db); register_task_tools(registry, db);
@@ -881,8 +885,8 @@ fn register_task_tools(registry: &mut AiToolRegistry, db: &Arc<Database>) {
} }
/// 工作流类 AI 工具注册(1 个:run_workflow)——不持 db(handler 防御兜底,详见注释)。 /// 工作流类 AI 工具注册(1 个:run_workflow)——不持 db(handler 防御兜底,详见注释)。
/// 知识图谱 Phase 1 AI 工具注册(6 个:task_link CRUD + 父子树 + 跨池移动 + content 更新) /// 知识图谱 Phase 1+2 AI 工具注册(7 个:task_link CRUD + 父子树 + 跨池移动 + content 更新
/// —— 持 db:Arc<Database>,对标设计 §五 Phase1 AI 工具表 + §2.1/§2.2。 /// + 项目事件流查询) —— 持 db:Arc<Database>,对标设计 §五 Phase1/Phase2 AI 工具表 + §2.1/§2.2/§2.4
/// ///
/// 工具直接调 df_storage::crud Repo(与 register_task_tools 同源模式,经 db Arc 重建 Repo), /// 工具直接调 df_storage::crud Repo(与 register_task_tools 同源模式,经 db Arc 重建 Repo),
/// 复用 IPC 层(commands::task)的底层 Repo 方法,不重复实现业务校验: /// 复用 IPC 层(commands::task)的底层 Repo 方法,不重复实现业务校验:
@@ -1094,6 +1098,56 @@ fn register_task_graph_tools(registry: &mut AiToolRegistry, db: &Arc<Database>)
}) })
})}, })},
); );
// ── get_project_timeline (Low 只读,设计 §五 + §2.4) ──
// 查询项目事件流(project_events 统一事件流)。AI 精准检索基础:
// 回答"上周做了什么 / 这个任务为何 blocked / 决策何时做出"。
// 参数 project_id(必填) + event_type(可选过滤) + limit(默认 50,上限 200)。
// source 标 "ai"(AI 工具路径,区别于 IPC 层的 "human")。
registry.register(
"get_project_timeline", "查询项目事件流(AI 精准检索基础,回答\"上周做了什么/这个任务为何 blocked/决策何时做出\")。参数 project_id(必填,项目 ID)、event_type(可选,按事件类型精确过滤,如 task_created/task_advanced/idea_promoted/idea_evaluated/decision_made 等,空=不过滤)、limit(可选,返回条数上限,默认 50,内部钳制 ≤ 200)。返回 { items: 事件列表(时间倒序), total, project_id }。事件含 event_type/entity_type/entity_id/from_state/to_state/source/created_at",
df_ai::ai_tools::object_schema(vec![
("project_id", "string", true),
("event_type", "string", false),
("limit", "integer", false),
]),
RiskLevel::Low,
{ let db = db.clone(); Box::new(move |args: serde_json::Value| {
let db = db.clone();
Box::pin(async move {
let project_id = args["project_id"].as_str()
.ok_or_else(|| anyhow::anyhow!("缺少 project_id"))?
.trim()
.to_string();
if project_id.is_empty() {
anyhow::bail!("project_id 不能为空");
}
let event_filter = args.get("event_type")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| s.to_string());
// limit 默认 50(对齐 commands::events::default_limit);负数/异常值兜底 50。
let limit = args.get("limit")
.and_then(|v| v.as_u64())
.map(|n| n as u32)
.unwrap_or(50);
let repo = df_storage::crud::ProjectEventRepo::new(&db);
let mut events = repo.get_by_project(&project_id, limit).await?;
// 应用层按 event_type 过滤(单用户事件量小,保持 Repo 方法精简)。
if let Some(et) = &event_filter {
events.retain(|e| e.event_type == *et);
}
let total = events.len();
Ok(serde_json::json!({
"items": events,
"total": total,
"project_id": project_id,
}))
})
})},
);
} }
/// 抽自 register_data_tools(SMELL-P0-2 续拆),【原样移入】,零行为变更。 /// 抽自 register_data_tools(SMELL-P0-2 续拆),【原样移入】,零行为变更。
@@ -2772,11 +2826,11 @@ mod tests {
// 工具注册基线测试SMELL-P0-2 拆分防护) // 工具注册基线测试SMELL-P0-2 拆分防护)
// //
// 防未来 register_data_tools / register_file_tools 拆分或重构时静默丢工具。 // 防未来 register_data_tools / register_file_tools 拆分或重构时静默丢工具。
// build_ai_tool_registry 经两层 register_* 组装:data(24 持 db) + file(13 不持 db) + http(1) = 38 // build_ai_tool_registry 经两层 register_* 组装:data(25 持 db) + file(13 不持 db) + http(1) = 39
// 任一层漏移 register 调用,此测试立即红。工具名集合也断言,防 rename 致 LLM tool 突变。 // 任一层漏移 register 调用,此测试立即红。工具名集合也断言,防 rename 致 LLM tool 突变。
// ============================================================ // ============================================================
/// build_ai_tool_registry 应注册恰好 38 个工具(24 data + 13 file + 1 http),且工具名集合稳定。 /// build_ai_tool_registry 应注册恰好 39 个工具(25 data + 13 file + 1 http),且工具名集合稳定。
/// ///
/// 用 in-memory SQLite(Database::open_in_memory 自跑迁移),构造零外部依赖的 db, /// 用 in-memory SQLite(Database::open_in_memory 自跑迁移),构造零外部依赖的 db,
// 不实际执行任何 handler——仅断言注册阶段的定义完整性,故无需真实数据。 // 不实际执行任何 handler——仅断言注册阶段的定义完整性,故无需真实数据。
@@ -2789,24 +2843,26 @@ mod tests {
let allowed_dirs = Arc::new(RwLock::new(AllowedDirs::default_with_root())); let allowed_dirs = Arc::new(RwLock::new(AllowedDirs::default_with_root()));
let registry = build_ai_tool_registry(&db, &allowed_dirs); let registry = build_ai_tool_registry(&db, &allowed_dirs);
// 总量基线:38(24 data + 13 file + 1 http)。拆分前后必须一致。 // 总量基线:39(25 data + 13 file + 1 http)。拆分前后必须一致。
// F-260621: file 层 10→11(新增 grep 跨文件内容搜索工具)。 // F-260621: file 层 10→11(新增 grep 跨文件内容搜索工具)。
// L1 环境感知(设计 §2.1): file 层 11→12(新增 detect_environment 环境探测工具)。 // L1 环境感知(设计 §2.1): file 层 11→12(新增 detect_environment 环境探测工具)。
// AST 代码智能(7c2e3b2): file 层 12→13(新增 read_symbol 符号解析,治 read_file 全文回灌 prompt 爆)。 // AST 代码智能(7c2e3b2): file 层 12→13(新增 read_symbol 符号解析,治 read_file 全文回灌 prompt 爆)。
// 知识图谱 Phase 1(2026-06-26): data 层 18→24(新增 register_task_graph_tools 6 工具: // 知识图谱 Phase 1(2026-06-26): data 层 18→24(新增 register_task_graph_tools 6 工具:
// create_task_link/remove_task_link/list_task_links/get_task_tree/move_task_queue/update_content)。 // create_task_link/remove_task_link/list_task_links/get_task_tree/move_task_queue/update_content)。
// 知识图谱 Phase 2(2026-06-26): data 层 24→25(新增 get_project_timeline 项目事件流查询,
// register_task_graph_tools 6→7)。
assert_eq!( assert_eq!(
registry.len(), registry.len(),
38, 39,
"工具总数应为 38(24 data + 13 file + 1 http),实际 {}", registry.len() "工具总数应为 39(25 data + 13 file + 1 http),实际 {}", registry.len()
); );
// 工具名集合基线:防 rename / 漏注册 / 误删除。 // 工具名集合基线:防 rename / 漏注册 / 误删除。
// data 层 24 个(持 db):CRUD/状态机/工作流/知识图谱任务关联 // data 层 25 个(持 db):CRUD/状态机/工作流/知识图谱任务关联 + 项目事件流
// file 层 13 个(不持 db):命令/读/列/写/改/元/追加/删/移/搜/grep/环境探测/符号解析 // file 层 13 个(不持 db):命令/读/列/写/改/元/追加/删/移/搜/grep/环境探测/符号解析
// http 层 1 个(不持 db):http_request // http 层 1 个(不持 db):http_request
let mut expected: Vec<&str> = vec![ let mut expected: Vec<&str> = vec![
// ── data 层 (24) ── // ── data 层 (25) ──
"list_projects", "list_tasks", "list_ideas", "list_projects", "list_tasks", "list_ideas",
"update_project", "create_project", "bind_directory", "update_project", "create_project", "bind_directory",
"create_task", "update_task", "advance_task", "create_task", "update_task", "advance_task",
@@ -2816,6 +2872,8 @@ mod tests {
// 知识图谱 Phase 1 任务关联工具(register_task_graph_tools 6 个) // 知识图谱 Phase 1 任务关联工具(register_task_graph_tools 6 个)
"create_task_link", "remove_task_link", "list_task_links", "create_task_link", "remove_task_link", "list_task_links",
"get_task_tree", "move_task_queue", "update_content", "get_task_tree", "move_task_queue", "update_content",
// 知识图谱 Phase 2 项目事件流(register_task_graph_tools 7 个,本工具为第 7)
"get_project_timeline",
// ── file 层 (13) ──run_command 注册顺序已移至末位降低 LLM 偏好, // ── file 层 (13) ──run_command 注册顺序已移至末位降低 LLM 偏好,
// 集合断言经 sort 后与顺序无关,仅守护工具名不漂移。grep 新增 F-260621; // 集合断言经 sort 后与顺序无关,仅守护工具名不漂移。grep 新增 F-260621;
// detect_environment 新增 L1 环境感知 设计 §2.1;read_symbol 新增 AST 代码智能) // detect_environment 新增 L1 环境感知 设计 §2.1;read_symbol 新增 AST 代码智能)

View File

@@ -0,0 +1,91 @@
//! 项目事件流相关命令(知识图谱 Phase 2 V30,对标设计 §2.4 + §五 get_project_timeline)
//!
//! 统一事件流 project_events 的查询入口。供 AI / Dashboard 项目时间线视图使用
//! (设计 §10.3.4 模式 B 事件流 + §五 get_project_timeline)。
//!
//! 筛选策略:Repo::get_by_project 取最近 N 条(上限 200,对标设计 §10.5 单用户桌面应用
//! 全量查询无压力),应用层按 event_type / 时间范围过滤。复杂多维筛选不进 SQL(数据量小,
//! 应用层过滤足够 + 保持 Repo 专用方法精简,对标 TaskLinkRepo 全专用路径)。
use serde::{Deserialize, Serialize};
use tauri::State;
use df_storage::models::ProjectEventRecord;
use crate::state::AppState;
use super::err_str;
/// get_project_timeline 查询入参(对标设计 §五 get_project_timeline:time 范围/event_type 筛选)。
///
/// 所有字段可选:
/// - `project_id`:必填(查询某项目的事件流)。
/// - `event_type`:可选,按事件类型精确过滤(如 "task_advanced");空/None = 不过滤。
/// - `limit`:可选,返回上限(默认 50,Repo 内部钳制最大 200)。
#[derive(Debug, Deserialize)]
pub struct GetTimelineInput {
pub project_id: String,
/// 事件类型过滤(精确匹配,如 "task_created" / "idea_promoted")。None/空串 = 不过滤。
pub event_type: Option<String>,
/// 返回条数上限(默认 50)。Repo 内部钳制 ≤ 200。
#[serde(default = "default_limit")]
pub limit: u32,
}
fn default_limit() -> u32 {
50
}
/// 时间线条目(对标设计 §10.3.4 模式 B 事件流)。直接回 ProjectEventRecord(Serialize 已派生),
/// 外层补 total 便于前端分页/计数。
#[derive(Debug, Serialize)]
pub struct TimelineResult {
/// 过滤后的事件列表(时间倒序,Repo 已排)
pub items: Vec<ProjectEventRecord>,
/// 过滤后总数(≤ limit)
pub total: usize,
/// 查询的项目 ID(回显)
pub project_id: String,
}
/// 查询项目事件流(对标设计 §五 get_project_timeline + §2.4)。
///
/// 支持按 event_type 过滤 + limit 上限。时间范围过滤暂未实现(设计 §2.4 列出 time 范围筛选,
/// 但当前 get_by_project 仅返回最近 N 条倒序,时间范围筛选用例可后续按需加 Repo 方法;
/// 单用户桌面应用事件量小,最近 N 条 + event_type 过滤已覆盖主要场景)。
#[tauri::command]
pub async fn get_project_timeline(
state: State<'_, AppState>,
input: GetTimelineInput,
) -> Result<TimelineResult, String> {
let project_id = input.project_id.trim().to_string();
if project_id.is_empty() {
return Err("project_id 不能为空".to_string());
}
// event_type 过滤值规范化(None/空串 = 不过滤)
let event_filter = input
.event_type
.as_deref()
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| s.to_string());
// Repo 取最近 limit 条(内部钳制 ≤ 200),时间倒序。
let mut events = state
.project_events
.get_by_project(&project_id, input.limit)
.await
.map_err(err_str)?;
// 应用层按 event_type 过滤(SQL 下推不值得——单用户事件量小,保持 Repo 方法精简)。
if let Some(et) = &event_filter {
events.retain(|e| e.event_type == *et);
}
let total = events.len();
Ok(TimelineResult {
items: events,
total,
project_id,
})
}

View File

@@ -9,7 +9,7 @@ use df_ai::provider::LlmProvider;
use df_types::types::{new_id, Priority}; use df_types::types::{new_id, Priority};
use df_ideas::capture::Idea; use df_ideas::capture::Idea;
use df_storage::crud::{is_unique_constraint_err, IdeaQuery}; use df_storage::crud::{is_unique_constraint_err, IdeaQuery};
use df_storage::models::{IdeaEvaluationRecord, IdeaRecord, ProjectRecord}; use df_storage::models::{IdeaEvaluationRecord, IdeaRecord, ProjectEventRecord, ProjectRecord};
use crate::state::AppState; use crate::state::AppState;
@@ -104,6 +104,12 @@ pub async fn create_idea(
.insert(record.clone()) .insert(record.clone())
.await .await
.map_err(err_str)?; .map_err(err_str)?;
// 知识图谱 Phase 2(对标设计 §2.4):idea_created 事件**暂不埋点**。
// 原因:project_events.project_id 是 NOT NULL + FK(PRAGMA foreign_keys=ON),
// 而 idea 无 project_id(立项前不属于任何项目)。设计 §2.5 计划用系统初始化创建的
// Inbox 项目作为无主 idea 的归属,但 Inbox 项目尚未实现(独立任务)。
// 写 NULL 会违反 NOT NULL,写不存在的 project_id 会违反 FK——两者都会让 best-effort
// 退化成「写失败 warn」,无实际价值且噪音。Inbox 项目落地后此处补 idea_created 埋点。
Ok(record) Ok(record)
} }
@@ -201,6 +207,31 @@ pub async fn promote_idea(
return Err(format!("灵感立项回写失败(已回滚项目创建): {}", e)); return Err(format!("灵感立项回写失败(已回滚项目创建): {}", e));
} }
// 知识图谱 Phase 2(对标设计 §2.4 hook/after):idea_promoted 事件。best-effort 不阻断。
// 灵感此时已立项为新项目(project_id 存在,FK 满足),事件挂在 project_id 下,
// entity 指向 idea,from_state=draft→to_state=promoted。
let event = ProjectEventRecord {
id: new_id(),
project_id: project_id.clone(),
event_type: "idea_promoted".to_string(),
entity_type: Some("idea".to_string()),
entity_id: Some(id.clone()),
from_state: Some("draft".to_string()),
to_state: Some("promoted".to_string()),
context_json: None,
source: Some("human".to_string()),
conversation_id: None,
created_at: now_millis(),
};
if let Err(e) = state.project_events.insert(event).await {
tracing::warn!(
idea_id = %id,
project_id = %project_id,
error = %e,
"[事件流] idea_promoted 埋点写入失败(不阻断立项)"
);
}
Ok(df_ideas::promotion::PromotionResult { Ok(df_ideas::promotion::PromotionResult {
idea_id: id, idea_id: id,
project_id: project_id, project_id: project_id,

View File

@@ -18,12 +18,49 @@ use df_project::scan::{
normalize_path, DiscoveredProject, normalize_path, DiscoveredProject,
}; };
use df_storage::crud::ProjectQuery; use df_storage::crud::ProjectQuery;
use df_storage::models::ProjectRecord; use df_storage::models::{ProjectEventRecord, ProjectRecord};
use crate::state::AppState; use crate::state::AppState;
use super::{err_str, now_millis}; use super::{err_str, now_millis};
// ============================================================
// 知识图谱 Phase 2:事件流埋点辅助(best-effort,对标设计 §2.4 hook/after + §10.1)
// ============================================================
/// 追加一条项目事件到 project_events(best-effort)。失败仅 tracing::warn 不阻断主 IPC。
///
/// `source` 标 `"human"`(IPC 层)。AI 工具路径在 tool_registry 内自行标 `"ai"`。
async fn emit_project_event(
state: &AppState,
project_id: &str,
event_type: &str,
entity_type: Option<&str>,
entity_id: Option<&str>,
) {
let record = ProjectEventRecord {
id: new_id(),
project_id: project_id.to_string(),
event_type: event_type.to_string(),
entity_type: entity_type.map(|s| s.to_string()),
entity_id: entity_id.map(|s| s.to_string()),
from_state: None,
to_state: Some("planning".to_string()),
context_json: None,
source: Some("human".to_string()),
conversation_id: None,
created_at: now_millis(),
};
if let Err(e) = state.project_events.insert(record).await {
tracing::warn!(
event_type = event_type,
project_id = project_id,
error = %e,
"[事件流] 埋点写入失败(不阻断主操作)"
);
}
}
/// 创建项目入参 /// 创建项目入参
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
pub struct CreateProjectInput { pub struct CreateProjectInput {
@@ -132,6 +169,22 @@ async fn create_with_binding(
if record.path.is_some() { if record.path.is_some() {
state.reload_allowed_dirs().await; state.reload_allowed_dirs().await;
} }
// 知识图谱 Phase 2(对标设计 §2.4 hook/after):项目创建事件。best-effort 不阻断。
//
// 设计 §2.4 event_type 枚举未列 project_created(枚举非穷尽,Repo 不做白名单校验——
// 见 project_event_repo.rs 注释「埋点层是唯一生产者,字符串字面量自带约束」)。此处沿用
// task_created/idea_created 命名约定新增 project_created,语义自洽。entity_type=project。
//
// 注:灵感晋升(create_from_idea)在 promote_idea 单独记 idea_promoted 事件(更精确),
// 此处一律记 project_created(项目实体自身视角),两事件互补不冲突。
emit_project_event(
state,
&record.id,
"project_created",
Some("project"),
Some(&record.id),
)
.await;
Ok(record) Ok(record)
} }

View File

@@ -5,12 +5,56 @@ use tauri::State;
use df_types::types::{new_id, TaskStatus}; use df_types::types::{new_id, TaskStatus};
use df_storage::crud::TaskQuery; use df_storage::crud::TaskQuery;
use df_storage::models::{TaskLinkRecord, TaskRecord}; use df_storage::models::{ProjectEventRecord, TaskLinkRecord, TaskRecord};
use crate::state::AppState; use crate::state::AppState;
use super::{err_str, now_millis}; use super::{err_str, now_millis};
// ============================================================
// 知识图谱 Phase 2:事件流埋点辅助(best-effort,对标设计 §2.4 hook/after + §10.1)
// ============================================================
/// 追加一条项目事件到 project_events(best-effort)。
///
/// 埋点策略(设计 §2.4):事件写入失败**不阻断主操作**,仅 `tracing::warn` 记录。
/// 设计 §10.1「事件流写入失败 — 影响事件完整性 — 事件写入失败不阻断主操作(best-effort)」。
///
/// `source` 语义:本辅助仅由 IPC 层调用,标 `"human"`(AI 工具路径在 tool_registry 内自行标
/// `"ai"`,不经本 IPC)。`project_id` / `entity_type` / `entity_id` / `event_type` 由调用方
/// 提供。`from_state` / `to_state` / `context_json` 可选。
async fn emit_event(
state: &State<'_, AppState>,
project_id: &str,
event_type: &str,
entity_type: Option<&str>,
entity_id: Option<&str>,
from_state: Option<&str>,
to_state: Option<&str>,
) {
let record = ProjectEventRecord {
id: new_id(),
project_id: project_id.to_string(),
event_type: event_type.to_string(),
entity_type: entity_type.map(|s| s.to_string()),
entity_id: entity_id.map(|s| s.to_string()),
from_state: from_state.map(|s| s.to_string()),
to_state: to_state.map(|s| s.to_string()),
context_json: None,
source: Some("human".to_string()),
conversation_id: None,
created_at: now_millis(),
};
if let Err(e) = state.project_events.insert(record).await {
tracing::warn!(
event_type = event_type,
project_id = project_id,
error = %e,
"[事件流] 埋点写入失败(不阻断主操作)"
);
}
}
/// 创建任务入参 /// 创建任务入参
#[derive(Debug, Deserialize)] #[derive(Debug, Deserialize)]
pub struct CreateTaskInput { pub struct CreateTaskInput {
@@ -262,6 +306,17 @@ pub async fn create_task(
.insert(record.clone()) .insert(record.clone())
.await .await
.map_err(err_str)?; .map_err(err_str)?;
// 知识图谱 Phase 2(对标设计 §2.4 hook/after):task_created 事件。best-effort 不阻断。
emit_event(
&state,
&record.project_id,
"task_created",
Some("task"),
Some(&record.id),
None,
Some(&record.status),
)
.await;
Ok(record) Ok(record)
} }
@@ -372,6 +427,15 @@ pub async fn advance_task(
id: String, id: String,
target_status: String, target_status: String,
) -> Result<TaskRecord, String> { ) -> Result<TaskRecord, String> {
// 知识图谱 Phase 2:推进前读当前态,作 task_advanced 事件 from_state(仅一次轻量读,
// advance 低频无压力)。失败(任务不存在)不阻断——后续 atomic 会用 NotFound 拒绝,from 留空。
let from_state = state
.tasks
.get_by_id(&id)
.await
.map_err(err_str)?
.map(|t| t.status);
let updated = df_nodes::task_advance_node::advance_task_atomic( let updated = df_nodes::task_advance_node::advance_task_atomic(
&state.tasks, &state.tasks,
&id, &id,
@@ -394,6 +458,19 @@ pub async fn advance_task(
); );
} }
} }
// 知识图谱 Phase 2(对标设计 §2.4 hook/after):task_advanced 事件。best-effort 不阻断。
emit_event(
&state,
&updated.project_id,
"task_advanced",
Some("task"),
Some(&updated.id),
from_state.as_deref(),
Some(&updated.status),
)
.await;
Ok(updated) Ok(updated)
} }
@@ -639,6 +716,21 @@ pub async fn move_task_queue(
.map_err(err_str)?; .map_err(err_str)?;
} }
// 知识图谱 Phase 2(对标设计 §2.4 hook/after):queue 变化事件。best-effort 不阻断。
// 仅在 queue 实际变化时埋点(避免 no-op 移动产噪音事件)。from/to 用 queue 值。
if current.queue != new_queue {
emit_event(
&state,
&current.project_id,
"task_advanced",
Some("task"),
Some(&current.id),
Some(&current.queue),
Some(&new_queue),
)
.await;
}
// 回读最新记录返回 // 回读最新记录返回
state state
.tasks .tasks

View File

@@ -261,6 +261,8 @@ pub fn run() {
commands::task::list_task_links, commands::task::list_task_links,
commands::task::move_task_queue, commands::task::move_task_queue,
commands::task::get_task_tree, commands::task::get_task_tree,
// 知识图谱 Phase 2(对标设计 §五 + §2.4):项目事件流查询
commands::events::get_project_timeline,
// 灵感 // 灵感
commands::idea::list_ideas, commands::idea::list_ideas,
commands::idea::create_idea, commands::idea::create_idea,

View File

@@ -12,8 +12,8 @@ use tokio::sync::{Mutex, RwLock, Semaphore};
use df_ai::ai_tools::AiToolRegistry; use df_ai::ai_tools::AiToolRegistry;
use df_storage::crud::{ use df_storage::crud::{
AiConversationRepo, AiMessageRepo, AiProviderRepo, AiToolExecutionRepo, IdeaEvalRepo, IdeaRepo, AiConversationRepo, AiMessageRepo, AiProviderRepo, AiToolExecutionRepo, IdeaEvalRepo, IdeaRepo,
KnowledgeEventsRepo, KnowledgeRepo, NodeExecutionRepo, ProjectRepo, ReleaseRepo, KnowledgeEventsRepo, KnowledgeRepo, NodeExecutionRepo, ProjectEventRepo, ProjectRepo,
SettingsRepo, TaskLinkRepo, TaskRepo, WorkflowRepo, ReleaseRepo, SettingsRepo, TaskLinkRepo, TaskRepo, WorkflowRepo,
}; };
use df_storage::db::Database; use df_storage::db::Database;
use df_workflow::eventbus::EventBus; use df_workflow::eventbus::EventBus;
@@ -339,6 +339,9 @@ pub struct AppState {
pub tasks: TaskRepo, pub tasks: TaskRepo,
/// 任务横向关联表 Repo(知识图谱 Phase 1 V29,AI 拓扑排序编排调度的基础) /// 任务横向关联表 Repo(知识图谱 Phase 1 V29,AI 拓扑排序编排调度的基础)
pub task_links: TaskLinkRepo, pub task_links: TaskLinkRepo,
/// 项目统一事件流表 Repo(知识图谱 Phase 2 V30,project_events 追加型审计表)。
/// 埋点 best-effort:事件写入失败不阻断主操作(设计 §10.1),commands 层 hook/after 追加。
pub project_events: ProjectEventRepo,
/// 发布表 Repo预留:ReleaseRepo 持久化已就位·IPC/逻辑未接入·SW-260618-21 b 保留) /// 发布表 Repo预留:ReleaseRepo 持久化已就位·IPC/逻辑未接入·SW-260618-21 b 保留)
#[allow(dead_code)] #[allow(dead_code)]
pub releases: ReleaseRepo, pub releases: ReleaseRepo,
@@ -665,6 +668,7 @@ impl AppState {
projects: ProjectRepo::new(&db), projects: ProjectRepo::new(&db),
tasks: TaskRepo::new(&db), tasks: TaskRepo::new(&db),
task_links: TaskLinkRepo::new(&db), task_links: TaskLinkRepo::new(&db),
project_events: ProjectEventRepo::new(&db),
releases: ReleaseRepo::new(&db), releases: ReleaseRepo::new(&db),
workflows: WorkflowRepo::new(&db), workflows: WorkflowRepo::new(&db),
node_executions: NodeExecutionRepo::new(&db), node_executions: NodeExecutionRepo::new(&db),