ZhenYi/gitdataai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: ZhenYi:108dd714d3fae438ef8b76cc7d246d56067783a9
Branches Tags
ZhenYi:main
ZhenYi:refactor
..
compare: ZhenYi:18917b6de197be0890077202ea23b673e4dc7289
Branches Tags
ZhenYi:main
ZhenYi:refactor

No commits in common. "108dd714d3fae438ef8b76cc7d246d56067783a9" and "18917b6de197be0890077202ea23b673e4dc7289" have entirely different histories.
108dd714d3 ... 18917b6de1

14 changed files with 163 additions and 330 deletions
Show Stats Expand all files Collapse all files

15
libs/agent/tool/registry.rs
View File

@ -11,13 +11,6 @@ use super::call::ToolError;
use super::context::ToolContext; use super::context::ToolContext;
use super::definition::ToolDefinition; use super::definition::ToolDefinition;
/// Error type for tool registry operations.
#[derive(Debug, Clone, thiserror::Error)]
pub enum ToolRegistryError {
#[error("tool already registered: {0}")]
AlreadyRegistered(String),
}
/// Inner function pointer type for tool handlers. /// Inner function pointer type for tool handlers.
type InnerHandlerFn = dyn Fn( type InnerHandlerFn = dyn Fn(
ToolContext, ToolContext,
@ -82,8 +75,7 @@ impl ToolRegistry {
pub fn register(&mut self, def: ToolDefinition, handler: ToolHandler) -> &mut Self { pub fn register(&mut self, def: ToolDefinition, handler: ToolHandler) -> &mut Self {
let name = def.name.clone(); let name = def.name.clone();
if self.handlers.contains_key(&name) { if self.handlers.contains_key(&name) {
tracing::warn!("tool already registered (skipping duplicate): {}", name); panic!("tool already registered: {}", name);
return self;
} }
self.handlers.insert(name.clone(), handler); self.handlers.insert(name.clone(), handler);
self.definitions.insert(name, def); self.definitions.insert(name, def);
@ -115,12 +107,11 @@ impl ToolRegistry {
} }
/// Merges another registry's tools into this one. /// Merges another registry's tools into this one.
/// Skips tools with duplicate names and logs a warning. /// Panics if a tool with the same name already exists.
pub fn merge(&mut self, other: ToolRegistry) { pub fn merge(&mut self, other: ToolRegistry) {
for (name, handler) in other.handlers { for (name, handler) in other.handlers {
if self.handlers.contains_key(&name) { if self.handlers.contains_key(&name) {
tracing::warn!("merge skipped duplicate tool: {}", name); panic!("tool already registered: {}", name);
continue;
} }
self.handlers.insert(name, handler); self.handlers.insert(name, handler);
} }

16
libs/config/lib.rs
View File

@ -13,9 +13,7 @@ impl AppConfig {
pub fn load() -> AppConfig { pub fn load() -> AppConfig {
let mut env = HashMap::new(); let mut env = HashMap::new();
for env_file in AppConfig::ENV_FILES { for env_file in AppConfig::ENV_FILES {
if let Err(e) = dotenvy::from_path(env_file) { dotenvy::from_path(env_file).ok();
eprintln!("dotenv skipped: {} ({})", env_file, e);
}
if let Ok(env_file_content) = std::fs::read_to_string(env_file) { if let Ok(env_file_content) = std::fs::read_to_string(env_file) {
for line in env_file_content.lines() { for line in env_file_content.lines() {
if let Some((key, value)) = line.split_once('=') { if let Some((key, value)) = line.split_once('=') {
@ -27,13 +25,13 @@ impl AppConfig {
// Environment variables (e.g. K8s injected APP_DOMAIN_URL) take precedence over .env files // Environment variables (e.g. K8s injected APP_DOMAIN_URL) take precedence over .env files
env = std::env::vars().chain(env).collect(); env = std::env::vars().chain(env).collect();
let this = AppConfig { env }; let this = AppConfig { env };
// Handle the race condition: if another thread already set the global, return it. if let Err(config) = GLOBAL_CONFIG.set(this) {
// This is safe because config is immutable after load. eprintln!("Failed to set global config: {:?}", config);
if GLOBAL_CONFIG.get().is_some() { }
GLOBAL_CONFIG.get().unwrap().clone() if let Some(config) = GLOBAL_CONFIG.get() {
config.clone()
} else { } else {
let _ = GLOBAL_CONFIG.set(this); panic!("Failed to get global config");
GLOBAL_CONFIG.get().expect("global config should be set after load").clone()
} }
} }
} }

95
libs/git/hook/sync/mod.rs
View File

@ -8,11 +8,11 @@ pub mod tag;
use db::cache::AppCache; use db::cache::AppCache;
use db::database::AppDatabase; use db::database::AppDatabase;
use models::projects::project_skill::ActiveModel as SkillActiveModel;
use models::projects::project_skill::{Column as SkillCol, Entity as SkillEntity}; use models::projects::project_skill::{Column as SkillCol, Entity as SkillEntity};
use models::projects::project_skill::ActiveModel as SkillActiveModel;
use models::repos::repo::Model as RepoModel; use models::repos::repo::Model as RepoModel;
use models::ActiveModelTrait;
use models::RepoId; use models::RepoId;
use models::ActiveModelTrait;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, Set}; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, Set};
use std::collections::HashMap; use std::collections::HashMap;
use std::path::Path; use std::path::Path;
@ -43,14 +43,8 @@ fn scan_skills_from_dir(
let path = entry.path(); let path = entry.path();
if path.is_dir() { if path.is_dir() {
stack.push(path); stack.push(path);
} else if path } else if path.file_name().and_then(|n| n.to_str()) == Some("SKILL.md") {
.file_name() if let Some(dir_name) = path.parent()
.and_then(|n| n.to_str())
.map(|x| x.to_lowercase())
== Some("skill.md".to_string())
{
if let Some(dir_name) = path
.parent()
.and_then(|p| p.file_name()) .and_then(|p| p.file_name())
.and_then(|n| n.to_str()) .and_then(|n| n.to_str())
.filter(|s| !s.starts_with('.')) .filter(|s| !s.starts_with('.'))
@ -141,7 +135,11 @@ pub struct HookMetaDataSync {
} }
impl HookMetaDataSync { impl HookMetaDataSync {
pub fn new(db: AppDatabase, cache: AppCache, repo: RepoModel) -> Result<Self, crate::GitError> { pub fn new(
db: AppDatabase,
cache: AppCache,
repo: RepoModel,
) -> Result<Self, crate::GitError> {
let domain = GitDomain::from_model(repo.clone())?; let domain = GitDomain::from_model(repo.clone())?;
Ok(Self { Ok(Self {
db, db,
@ -187,16 +185,18 @@ impl HookMetaDataSync {
/// Full sync pipeline (no locking — caller is responsible). /// Full sync pipeline (no locking — caller is responsible).
async fn sync_work(&self) -> Result<(), crate::GitError> { async fn sync_work(&self) -> Result<(), crate::GitError> {
let mut txn = let mut txn = self
self.db.begin().await.map_err(|e| { .db
crate::GitError::IoError(format!("failed to begin transaction: {}", e)) .begin()
})?; .await
.map_err(|e| crate::GitError::IoError(format!("failed to begin transaction: {}", e)))?;
self.sync_refs(&mut txn).await?; self.sync_refs(&mut txn).await?;
self.sync_commits(&mut txn).await?; self.sync_commits(&mut txn).await?;
self.sync_tags(&mut txn).await?; self.sync_tags(&mut txn).await?;
self.sync_lfs_objects(&mut txn).await?; self.sync_lfs_objects(&mut txn).await?;
self.run_fsck_and_rollback_if_corrupt(&mut txn).await?; self.run_fsck_and_rollback_if_corrupt(&mut txn)
.await?;
txn.commit().await.map_err(|e| { txn.commit().await.map_err(|e| {
crate::GitError::IoError(format!("failed to commit transaction: {}", e)) crate::GitError::IoError(format!("failed to commit transaction: {}", e))
@ -210,12 +210,14 @@ impl HookMetaDataSync {
/// Fsck only work (no locking — caller is responsible). /// Fsck only work (no locking — caller is responsible).
async fn fsck_work(&self) -> Result<(), crate::GitError> { async fn fsck_work(&self) -> Result<(), crate::GitError> {
let mut txn = let mut txn = self
self.db.begin().await.map_err(|e| { .db
crate::GitError::IoError(format!("failed to begin transaction: {}", e)) .begin()
})?; .await
.map_err(|e| crate::GitError::IoError(format!("failed to begin transaction: {}", e)))?;
self.run_fsck_and_rollback_if_corrupt(&mut txn).await?; self.run_fsck_and_rollback_if_corrupt(&mut txn)
.await?;
txn.commit().await.map_err(|e| { txn.commit().await.map_err(|e| {
crate::GitError::IoError(format!("failed to commit transaction: {}", e)) crate::GitError::IoError(format!("failed to commit transaction: {}", e))
@ -330,40 +332,22 @@ impl HookMetaDataSync {
} }
}; };
// Deduplicate by {repo_id}+{blob_hash}, keep latest by commit_sha let existing_by_slug: HashMap<_, _> = existing
let mut deduped: std::collections::HashMap<String, DiscoveredSkill> = std::collections::HashMap::new();
for skill in discovered {
let key = format!("{}:{}", self.repo.id, skill.blob_hash.as_ref().unwrap_or(&skill.slug));
match deduped.get(&key) {
Some(existing) => {
if skill.commit_sha.as_ref().unwrap_or(&String::new()) > existing.commit_sha.as_ref().unwrap_or(&String::new()) {
deduped.insert(key, skill);
}
}
None => {
deduped.insert(key, skill);
}
}
}
let existing_by_hash: HashMap<_, _> = existing
.into_iter() .into_iter()
.map(|s| { .map(|s| (s.slug.clone(), s))
let key = format!("{}:{}", s.repo_id.unwrap_or_default(), s.blob_hash.clone().unwrap_or_default());
(key, s)
})
.collect(); .collect();
let mut seen_keys = std::collections::HashSet::new(); let mut seen_slugs = std::collections::HashSet::new();
for (key, skill) in deduped { for skill in discovered {
seen_keys.insert(key.clone()); seen_slugs.insert(skill.slug.clone());
let json_meta = serde_json::to_value(&skill.metadata).unwrap_or_default(); let json_meta = serde_json::to_value(&skill.metadata).unwrap_or_default();
if let Some(existing_skill) = existing_by_hash.get(&key) { if let Some(existing_skill) = existing_by_slug.get(&skill.slug) {
if existing_skill.content != skill.content if existing_skill.content != skill.content
|| existing_skill.metadata != json_meta || existing_skill.metadata != json_meta
|| existing_skill.commit_sha != skill.commit_sha || existing_skill.commit_sha.as_ref() != skill.commit_sha.as_ref()
|| existing_skill.blob_hash.as_ref() != skill.blob_hash.as_ref()
{ {
let mut active: SkillActiveModel = existing_skill.clone().into(); let mut active: SkillActiveModel = existing_skill.clone().into();
active.content = Set(skill.content); active.content = Set(skill.content);
@ -399,25 +383,16 @@ impl HookMetaDataSync {
} }
} }
for (key, old_skill) in existing_by_hash { for (slug, old_skill) in existing_by_slug {
if !seen_keys.contains(&key) { if !seen_slugs.contains(&slug) {
if SkillEntity::delete_by_id(old_skill.id) if SkillEntity::delete_by_id(old_skill.id).exec(&self.db).await.is_ok() {
.exec(&self.db)
.await
.is_ok()
{
removed += 1; removed += 1;
} }
} }
} }
if created > 0 || updated > 0 || removed > 0 { if created > 0 || updated > 0 || removed > 0 {
tracing::info!( tracing::info!("skills synced created={} updated={} removed={}", created, updated, removed);
"skills synced created={} updated={} removed={}",
created,
updated,
removed
);
} }
} }
} }

70
libs/git/ssh/handle.rs
View File

@ -118,9 +118,7 @@ impl SSHandle {
if let Some(mut stdin) = self.stdin.remove(&channel_id) { if let Some(mut stdin) = self.stdin.remove(&channel_id) {
tokio::spawn(async move { tokio::spawn(async move {
let _ = tokio::time::timeout(Duration::from_secs(5), async { let _ = tokio::time::timeout(Duration::from_secs(5), async {
if let Err(e) = stdin.flush().await { stdin.flush().await.ok();
tracing::warn!(error = %e, "ssh_cleanup_flush_failed channel={:?}", channel_id);
}
let _ = stdin.shutdown().await; let _ = stdin.shutdown().await;
}) })
.await; .await;
@ -298,9 +296,7 @@ impl russh::server::Handler for SSHandle {
tracing::info!("Closing stdin channel={:?} client={:?}", channel, self.client_addr); tracing::info!("Closing stdin channel={:?} client={:?}", channel, self.client_addr);
// Use timeout so we never block the SSH event loop waiting for git. // Use timeout so we never block the SSH event loop waiting for git.
let _ = tokio::time::timeout(Duration::from_secs(5), async { let _ = tokio::time::timeout(Duration::from_secs(5), async {
if let Err(e) = stdin.flush().await { stdin.flush().await.ok();
tracing::warn!(error = %e, "ssh_eof_flush_failed channel={:?}", channel);
}
let _ = stdin.shutdown().await; let _ = stdin.shutdown().await;
}) })
.await; .await;
@ -322,9 +318,7 @@ impl russh::server::Handler for SSHandle {
.map(|addr| format!("{}", addr)) .map(|addr| format!("{}", addr))
.unwrap_or_else(|| "unknown".to_string()); .unwrap_or_else(|| "unknown".to_string());
tracing::info!("channel_open_session channel={:?} client={}", channel, client_info); tracing::info!("channel_open_session channel={:?} client={}", channel, client_info);
if let Err(e) = session.flush() { let _ = session.flush().ok();
tracing::warn!(error = %e, "ssh_session_flush_failed");
}
Ok(true) Ok(true)
} }
@ -340,9 +334,7 @@ impl russh::server::Handler for SSHandle {
session: &mut Session, session: &mut Session,
) -> Result<(), Self::Error> { ) -> Result<(), Self::Error> {
tracing::warn!("pty_request not supported channel={:?} term={} cols={} rows={}", channel, term, col_width, row_height); tracing::warn!("pty_request not supported channel={:?} term={} cols={} rows={}", channel, term, col_width, row_height);
if let Err(e) = session.flush() { let _ = session.flush().ok();
tracing::warn!(error = %e, "ssh_session_flush_failed");
}
Ok(()) Ok(())
} }
@ -355,9 +347,7 @@ impl russh::server::Handler for SSHandle {
tracing::info!("subsystem_request channel={:?} subsystem={}", channel, name); tracing::info!("subsystem_request channel={:?} subsystem={}", channel, name);
// git-clients may send "subsystem" for git protocol over ssh. // git-clients may send "subsystem" for git protocol over ssh.
// We don't use subsystem; exec_request handles it directly. // We don't use subsystem; exec_request handles it directly.
if let Err(e) = session.flush() { let _ = session.flush().ok();
tracing::warn!(error = %e, "ssh_session_flush_failed");
}
Ok(()) Ok(())
} }
async fn data( async fn data(
@ -477,21 +467,23 @@ impl russh::server::Handler for SSHandle {
); );
tracing::info!("shell_request user={}", user.username); tracing::info!("shell_request user={}", user.username);
let _ = session session
.data(channel_id, CryptoVec::from_slice(welcome_msg.as_bytes())); .data(channel_id, CryptoVec::from_slice(welcome_msg.as_bytes()))
let _ = session.exit_status_request(channel_id, 0); .ok();
let _ = session.eof(channel_id); session.exit_status_request(channel_id, 0).ok();
let _ = session.close(channel_id); session.eof(channel_id).ok();
let _ = session.flush(); session.close(channel_id).ok();
let _ = session.flush().ok();
} else { } else {
tracing::warn!("shell_request_unauthenticated channel={:?}", channel_id); tracing::warn!("shell_request_unauthenticated channel={:?}", channel_id);
let msg = "Authentication required\r\n"; let msg = "Authentication required\r\n";
let _ = session session
.data(channel_id, CryptoVec::from_slice(msg.as_bytes())); .data(channel_id, CryptoVec::from_slice(msg.as_bytes()))
let _ = session.exit_status_request(channel_id, 1); .ok();
let _ = session.eof(channel_id); session.exit_status_request(channel_id, 1).ok();
let _ = session.close(channel_id); session.eof(channel_id).ok();
let _ = session.flush(); session.close(channel_id).ok();
let _ = session.flush().ok();
} }
Ok(()) Ok(())
} }
@ -512,12 +504,13 @@ impl russh::server::Handler for SSHandle {
Ok(cmd) => cmd.trim(), Ok(cmd) => cmd.trim(),
Err(e) => { Err(e) => {
tracing::error!("invalid_command_encoding error={}", e); tracing::error!("invalid_command_encoding error={}", e);
let _ = session session
.disconnect( .disconnect(
Disconnect::ServiceNotAvailable, Disconnect::ServiceNotAvailable,
"Invalid command encoding", "Invalid command encoding",
"", "",
); )
.ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }
}; };
@ -526,8 +519,9 @@ impl russh::server::Handler for SSHandle {
None => { None => {
tracing::error!("invalid_git_command command={}", git_shell_cmd); tracing::error!("invalid_git_command command={}", git_shell_cmd);
let msg = format!("Invalid git command: {}", git_shell_cmd); let msg = format!("Invalid git command: {}", git_shell_cmd);
let _ = session session
.disconnect(Disconnect::ServiceNotAvailable, &msg, ""); .disconnect(Disconnect::ServiceNotAvailable, &msg, "")
.ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }
}; };
@ -537,8 +531,9 @@ impl russh::server::Handler for SSHandle {
None => { None => {
let msg = format!("Invalid repository path: {}", path); let msg = format!("Invalid repository path: {}", path);
tracing::error!("invalid_repo_path path={}", path); tracing::error!("invalid_repo_path path={}", path);
let _ = session session
.disconnect(Disconnect::ServiceNotAvailable, &msg, ""); .disconnect(Disconnect::ServiceNotAvailable, &msg, "")
.ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }
}; };
@ -549,8 +544,9 @@ impl russh::server::Handler for SSHandle {
Err(e) => { Err(e) => {
// Log the detailed error internally; client receives generic message. // Log the detailed error internally; client receives generic message.
tracing::error!("repo_fetch_error error={}", e); tracing::error!("repo_fetch_error error={}", e);
let _ = session session
.disconnect(Disconnect::ServiceNotAvailable, "Repository not found", ""); .disconnect(Disconnect::ServiceNotAvailable, "Repository not found", "")
.ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }
}; };
@ -561,7 +557,7 @@ impl russh::server::Handler for SSHandle {
None => { None => {
let msg = "Authentication error: no authenticated user"; let msg = "Authentication error: no authenticated user";
tracing::error!("exec_no_authenticated_user channel={:?}", channel_id); tracing::error!("exec_no_authenticated_user channel={:?}", channel_id);
let _ = session.disconnect(Disconnect::ByApplication, msg, ""); session.disconnect(Disconnect::ByApplication, msg, "").ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }
}; };
@ -580,7 +576,7 @@ impl russh::server::Handler for SSHandle {
repo.repo_name repo.repo_name
); );
tracing::error!("access_denied user={} repo={} is_write={}", operator.username, repo.repo_name, is_write); tracing::error!("access_denied user={} repo={} is_write={}", operator.username, repo.repo_name, is_write);
let _ = session.disconnect(Disconnect::ByApplication, &msg, ""); session.disconnect(Disconnect::ByApplication, &msg, "").ok();
return Err(russh::Error::Disconnect); return Err(russh::Error::Disconnect);
} }

56
libs/room/src/service/history.rs
View File

@ -2,7 +2,6 @@ use db::database::AppDatabase;
use models::repos::repo; use models::repos::repo;
use models::rooms::room_ai; use models::rooms::room_ai;
use models::rooms::room_message::{Column as RmCol, Entity as RoomMessage}; use models::rooms::room_message::{Column as RmCol, Entity as RoomMessage};
use models::users::user;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect}; use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use uuid::Uuid; use uuid::Uuid;
@ -60,18 +59,6 @@ pub async fn get_room_ai_config(
Ok(ai_config) Ok(ai_config)
} }
pub async fn get_room_ai_configs(
db: &AppDatabase,
room_id: Uuid,
) -> Result<Vec<room_ai::Model>, RoomError> {
let ai_configs = room_ai::Entity::find()
.filter(room_ai::Column::Room.eq(room_id))
.all(db)
.await?;
Ok(ai_configs)
}
pub async fn extract_mention_context( pub async fn extract_mention_context(
db: &AppDatabase, db: &AppDatabase,
project_id: Uuid, project_id: Uuid,
@ -79,43 +66,24 @@ pub async fn extract_mention_context(
) -> Vec<agent::chat::Mention> { ) -> Vec<agent::chat::Mention> {
let mut mentions: Vec<agent::chat::Mention> = Vec::new(); let mut mentions: Vec<agent::chat::Mention> = Vec::new();
let mut seen_repos: Vec<String> = Vec::new(); let mut seen_repos: Vec<String> = Vec::new();
let mut seen_users: Vec<String> = Vec::new();
for cap in mention_bracket_re().captures_iter(content) { for cap in mention_bracket_re().captures_iter(content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) { if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
let type_str = type_m.as_str(); if type_m.as_str() == "repo" {
let id = id_m.as_str().trim(); let repo_name = id_m.as_str().trim().to_string();
if repo_name.is_empty() || seen_repos.contains(&repo_name) {
match type_str { continue;
"repo" => {
let repo_name = id.to_string();
if repo_name.is_empty() || seen_repos.contains(&repo_name) {
continue;
}
seen_repos.push(repo_name.clone());
if let Ok(Some(repo_model)) = repo::Entity::find()
.filter(repo::Column::Project.eq(project_id))
.filter(repo::Column::RepoName.eq(&repo_name))
.one(db)
.await
{
mentions.push(agent::chat::Mention::Repo(repo_model));
}
} }
"user" => { seen_repos.push(repo_name.clone());
if seen_users.contains(&id.to_string()) {
continue;
}
seen_users.push(id.to_string());
if let Ok(uuid) = Uuid::parse_str(id) { if let Ok(Some(repo_model)) = repo::Entity::find()
if let Ok(Some(user_model)) = user::Entity::find_by_id(uuid).one(db).await { .filter(repo::Column::Project.eq(project_id))
mentions.push(agent::chat::Mention::User(user_model)); .filter(repo::Column::RepoName.eq(&repo_name))
} .one(db)
} .await
{
mentions.push(agent::chat::Mention::Repo(repo_model));
} }
_ => {}
} }
} }
} }

83
libs/room/src/service/mod.rs
View File

@ -258,20 +258,22 @@ impl RoomService {
} }
pub async fn should_ai_respond(&self, room_id: Uuid, content: &str) -> Result<bool, RoomError> { pub async fn should_ai_respond(&self, room_id: Uuid, content: &str) -> Result<bool, RoomError> {
let ai_configs = history::get_room_ai_configs(&self.db, room_id).await?; let ai_config = history::get_room_ai_config(&self.db, room_id).await?;
if ai_configs.is_empty() {
return Ok(false); let config = match ai_config {
Some(c) => c,
None => return Ok(false),
};
if !config.use_exact {
return Ok(true);
} }
// Collect all model IDs in this room let model_id_str = config.model.to_string();
let model_ids: std::collections::HashSet<String> = ai_configs
.iter()
.map(|c| c.model.to_string())
.collect();
for cap in mention_bracket_re().captures_iter(content) { for cap in mention_bracket_re().captures_iter(content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) { if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
if type_m.as_str() == "ai" && model_ids.contains(id_m.as_str().trim()) { if type_m.as_str() == "ai" && id_m.as_str().trim() == model_id_str {
return Ok(true); return Ok(true);
} }
} }
@ -279,7 +281,7 @@ impl RoomService {
for cap in mention_tag_re().captures_iter(content) { for cap in mention_tag_re().captures_iter(content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) { if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
if type_m.as_str() == "ai" && model_ids.contains(id_m.as_str().trim()) { if type_m.as_str() == "ai" && id_m.as_str().trim() == model_id_str {
return Ok(true); return Ok(true);
} }
} }
@ -321,48 +323,8 @@ impl RoomService {
return Ok(()); return Ok(());
}; };
// Extract mentioned AI model ID from content let Some(ai_config) = self.get_room_ai_config(room_id).await? else {
let mentioned_model_id = { return Ok(());
let mut found = None;
for cap in mention_bracket_re().captures_iter(&content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
if type_m.as_str() == "ai" {
if let Ok(uuid) = Uuid::parse_str(id_m.as_str().trim()) {
found = Some(uuid);
break;
}
}
}
}
if found.is_none() {
for cap in mention_tag_re().captures_iter(&content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
if type_m.as_str() == "ai" {
if let Ok(uuid) = Uuid::parse_str(id_m.as_str().trim()) {
found = Some(uuid);
break;
}
}
}
}
}
found
};
let model_id = match mentioned_model_id {
Some(id) => id,
None => return Ok(()), // No @ai mention, don't respond
};
// Verify the mentioned AI exists in this room
let ai_config = match room_ai::Entity::find()
.filter(room_ai::Column::Room.eq(room_id))
.filter(room_ai::Column::Model.eq(model_id))
.one(&self.db)
.await?
{
Some(c) => c,
None => return Ok(()), // Mentioned AI not in this room
}; };
let Some(lock_guard) = let Some(lock_guard) =
@ -377,6 +339,23 @@ impl RoomService {
.one(&self.db) .one(&self.db)
.await? .await?
.ok_or_else(|| RoomError::NotFound("Project not found".to_string()))?; .ok_or_else(|| RoomError::NotFound("Project not found".to_string()))?;
let mentioned_model_id = {
let mut found = None;
for cap in mention_bracket_re().captures_iter(&content) {
if let (Some(type_m), Some(id_m)) = (cap.get(1), cap.get(2)) {
if type_m.as_str() == "ai" {
if let Ok(uuid) = Uuid::parse_str(id_m.as_str().trim()) {
found = Some(uuid);
break;
}
}
}
}
found
};
let model_id = mentioned_model_id.unwrap_or(ai_config.model);
let model = models::agents::model::Entity::find_by_id(model_id) let model = models::agents::model::Entity::find_by_id(model_id)
.one(&self.db) .one(&self.db)
.await? .await?

11
libs/service/skill/scan.rs
View File

@ -24,16 +24,17 @@ impl AppService {
project_uid: Uuid, project_uid: Uuid,
_caller_uid: Uuid, _caller_uid: Uuid,
) -> Result<scanner::ScanSyncResult, AppError> { ) -> Result<scanner::ScanSyncResult, AppError> {
let mut total_created = 0i64; // Collect all repo IDs for this project
let mut total_updated = 0i64;
let mut total_removed = 0i64;
let mut total_discovered = 0i64;
let repos: Vec<_> = RepoEntity::find() let repos: Vec<_> = RepoEntity::find()
.filter(RCol::Project.eq(project_uid)) .filter(RCol::Project.eq(project_uid))
.all(&self.db) .all(&self.db)
.await?; .await?;
let mut total_created = 0i64;
let mut total_updated = 0i64;
let mut total_removed = 0i64;
let mut total_discovered = 0i64;
for repo in repos { for repo in repos {
let result = scanner::scan_and_sync_skills(&self.db, project_uid, &repo).await?; let result = scanner::scan_and_sync_skills(&self.db, project_uid, &repo).await?;
total_created += result.created; total_created += result.created;

101
libs/service/skill/scanner.rs
View File

@ -4,7 +4,6 @@
use crate::error::AppError; use crate::error::AppError;
use chrono::Utc; use chrono::Utc;
use git2::Repository;
use models::ActiveModelTrait; use models::ActiveModelTrait;
use models::projects::project_skill::ActiveModel as SkillActiveModel; use models::projects::project_skill::ActiveModel as SkillActiveModel;
use models::projects::project_skill::Column as C; use models::projects::project_skill::Column as C;
@ -109,12 +108,7 @@ pub fn scan_repo_for_skills(
let path = entry.path(); let path = entry.path();
if path.is_dir() { if path.is_dir() {
stack.push(path); stack.push(path);
} else if path } else if path.file_name().and_then(|n| n.to_str()) == Some("SKILL.md") {
.file_name()
.and_then(|n| n.to_str())
.map(|s| s.to_lowercase())
== Some("skill.md".to_string())
{
if let Some(dir_name) = path.parent() if let Some(dir_name) = path.parent()
.and_then(|p| p.file_name()) .and_then(|p| p.file_name())
.and_then(|n| n.to_str()) .and_then(|n| n.to_str())
@ -142,40 +136,10 @@ pub async fn scan_and_sync_skills(
project_uuid: Uuid, project_uuid: Uuid,
repo: &RepoModel, repo: &RepoModel,
) -> Result<ScanSyncResult, AppError> { ) -> Result<ScanSyncResult, AppError> {
// Open with git2 to get the actual workdir // Resolve the repo path
let git_repo = match Repository::open(&repo.storage_path) { let storage_path = Path::new(&repo.storage_path);
Ok(r) => r, let discovered = scan_repo_for_skills(storage_path, repo.id)?;
Err(e) => {
tracing::warn!("failed to open git repo {}: {:?}", repo.storage_path, e);
return Ok(ScanSyncResult {
discovered: 0,
created: 0,
updated: 0,
removed: 0,
});
}
};
let workdir = git_repo.workdir().map(|p| p.to_path_buf()).unwrap_or_else(|| Path::new(&repo.storage_path).to_path_buf());
let commit_sha = git_repo.head().ok().and_then(|h| h.target()).map(|oid| oid.to_string());
let mut discovered = scan_repo_for_skills(&workdir, repo.id)?;
// Fill in commit_sha for discovered skills
for skill in &mut discovered {
skill.commit_sha = commit_sha.clone();
}
sync_discovered_skills(db, project_uuid, repo.id, discovered).await
}
/// Sync discovered skills with deduplication by {repo_id}+{blob_hash}.
async fn sync_discovered_skills(
db: &db::database::AppDatabase,
project_uuid: Uuid,
repo_id: Uuid,
discovered: Vec<DiscoveredSkill>,
) -> Result<ScanSyncResult, AppError> {
if discovered.is_empty() { if discovered.is_empty() {
return Ok(ScanSyncResult { return Ok(ScanSyncResult {
discovered: 0, discovered: 0,
@ -189,56 +153,37 @@ async fn sync_discovered_skills(
let mut created = 0i64; let mut created = 0i64;
let mut updated = 0i64; let mut updated = 0i64;
// Deduplicate by {repo_id}+{blob_hash}, keep latest by commit_sha // Collect all repo-sourced skills in this repo for this project
let mut deduped: std::collections::HashMap<String, DiscoveredSkill> = std::collections::HashMap::new();
for skill in discovered {
let key = format!("{}:{}", repo_id, skill.blob_hash.as_ref().unwrap_or(&skill.slug));
match deduped.get(&key) {
Some(existing) => {
// Keep the one with the later commit_sha
if skill.commit_sha.as_ref().unwrap_or(&String::new()) > existing.commit_sha.as_ref().unwrap_or(&String::new()) {
deduped.insert(key, skill);
}
}
None => {
deduped.insert(key, skill);
}
}
}
// Query existing skills for this repo
let existing: Vec<_> = SkillEntity::find() let existing: Vec<_> = SkillEntity::find()
.filter(C::ProjectUuid.eq(project_uuid)) .filter(C::ProjectUuid.eq(project_uuid))
.filter(C::Source.eq("repo")) .filter(C::Source.eq("repo"))
.filter(C::RepoId.eq(repo_id)) .filter(C::RepoId.eq(repo.id))
.all(db) .all(db)
.await?; .await?;
let existing_by_hash: std::collections::HashMap<_, _> = existing let existing_by_slug: std::collections::HashMap<_, _> = existing
.into_iter() .into_iter()
.map(|s| { .map(|s| (s.slug.clone(), s))
let key = format!("{}:{}", s.repo_id.unwrap_or_default(), s.blob_hash.clone().unwrap_or_default());
(key, s)
})
.collect(); .collect();
let mut seen_keys = std::collections::HashSet::new(); let mut seen_slugs = std::collections::HashSet::new();
let discovered_count = discovered.len() as i64;
for skill in discovered {
seen_slugs.insert(skill.slug.clone());
let discovered_count = deduped.len() as i64;
for (key, skill) in deduped {
seen_keys.insert(key.clone());
let json_meta = serde_json::to_value(&skill.metadata).unwrap_or_default(); let json_meta = serde_json::to_value(&skill.metadata).unwrap_or_default();
if let Some(existing_skill) = existing_by_hash.get(&key) { if let Some(existing_skill) = existing_by_slug.get(&skill.slug) {
if existing_skill.content != skill.content if existing_skill.content != skill.content
|| existing_skill.metadata != json_meta || existing_skill.metadata != json_meta
|| existing_skill.commit_sha != skill.commit_sha || existing_skill.blob_hash != skill.blob_hash
{ {
let mut active: SkillActiveModel = existing_skill.clone().into(); let mut active: SkillActiveModel = existing_skill.clone().into();
active.content = Set(skill.content); active.content = Set(skill.content);
active.metadata = Set(json_meta); active.metadata = Set(json_meta);
active.commit_sha = Set(skill.commit_sha); active.commit_sha = Set(skill.commit_sha.clone());
active.blob_hash = Set(skill.blob_hash); active.blob_hash = Set(skill.blob_hash.clone());
active.updated_at = Set(now); active.updated_at = Set(now);
active.update(db).await?; active.update(db).await?;
updated += 1; updated += 1;
@ -247,13 +192,13 @@ async fn sync_discovered_skills(
let active = SkillActiveModel { let active = SkillActiveModel {
id: Set(0), id: Set(0),
project_uuid: Set(project_uuid), project_uuid: Set(project_uuid),
slug: Set(skill.slug), slug: Set(skill.slug.clone()),
name: Set(skill.name), name: Set(skill.name),
description: Set(skill.description), description: Set(skill.description),
source: Set("repo".to_string()), source: Set("repo".to_string()),
repo_id: Set(Some(repo_id)), repo_id: Set(Some(repo.id)),
commit_sha: Set(skill.commit_sha), commit_sha: Set(skill.commit_sha.clone()),
blob_hash: Set(skill.blob_hash), blob_hash: Set(skill.blob_hash.clone()),
content: Set(skill.content), content: Set(skill.content),
metadata: Set(json_meta), metadata: Set(json_meta),
enabled: Set(true), enabled: Set(true),
@ -268,8 +213,8 @@ async fn sync_discovered_skills(
// Remove skills that no longer exist in the repo // Remove skills that no longer exist in the repo
let mut removed = 0i64; let mut removed = 0i64;
for (key, old_skill) in existing_by_hash { for (slug, old_skill) in existing_by_slug {
if !seen_keys.contains(&key) { if !seen_slugs.contains(&slug) {
SkillEntity::delete_by_id(old_skill.id).exec(db).await?; SkillEntity::delete_by_id(old_skill.id).exec(db).await?;
removed += 1; removed += 1;
} }

16
src/contexts/room-context.tsx
View File

@ -313,14 +313,10 @@ export function RoomProvider({
// Subscribe to room events. connect() is already called at the provider // Subscribe to room events. connect() is already called at the provider
// level — subscribe/unsubscribe only manage per-room event routing. // level — subscribe/unsubscribe only manage per-room event routing.
client.subscribeRoom(activeRoomId).catch((err) => { client.subscribeRoom(activeRoomId).catch(() => {});
console.warn('[RoomContext] subscribeRoom failed:', err);
});
return () => { return () => {
client.unsubscribeRoom(activeRoomId).catch((err) => { client.unsubscribeRoom(activeRoomId).catch(() => {});
console.warn('[RoomContext] unsubscribeRoom failed:', err);
});
}; };
}, [activeRoomId, wsClient]); }, [activeRoomId, wsClient]);
@ -361,9 +357,7 @@ export function RoomProvider({
); );
} }
}; };
doLoad().catch((err) => { doLoad().catch(() => {});
console.warn('[RoomContext] loadReactions failed:', err);
});
}; };
const loadMore = useCallback( const loadMore = useCallback(
@ -859,9 +853,7 @@ export function RoomProvider({
const client = wsClientRef.current; const client = wsClientRef.current;
if (client && client.getStatus() !== 'open') { if (client && client.getStatus() !== 'open') {
console.debug('[RoomContext] Tab visible, reconnecting WS...'); console.debug('[RoomContext] Tab visible, reconnecting WS...');
client.connect().catch(() => { client.connect().catch(() => {});
// connect() has its own retry logic; ignore here to avoid duplicate warnings
});
} }
} }
}; };

7
src/contexts/theme-context.tsx
View File

@ -9,10 +9,8 @@ interface ThemeContextType {
setTheme: (theme: ThemePreference) => void; setTheme: (theme: ThemePreference) => void;
} }
const getSystemTheme = (): ResolvedTheme => { const getSystemTheme = (): ResolvedTheme =>
if (typeof window === 'undefined') return 'light'; window.matchMedia('(prefers-color-scheme: dark)').matches ? 'dark' : 'light';
return window.matchMedia('(prefers-color-scheme: dark)').matches ? 'dark' : 'light';
};
const ThemeContext = createContext<ThemeContextType | undefined>(undefined); const ThemeContext = createContext<ThemeContextType | undefined>(undefined);
@ -42,7 +40,6 @@ export function ThemeProvider({ children }: { children: React.ReactNode }) {
); );
useEffect(() => { useEffect(() => {
if (typeof window === 'undefined') return;
const root = window.document.documentElement; const root = window.document.documentElement;
root.classList.remove('light', 'dark'); root.classList.remove('light', 'dark');
root.classList.add(resolvedTheme); root.classList.add(resolvedTheme);

1
src/hooks/use-mobile.ts
View File

@ -6,7 +6,6 @@ export function useIsMobile() {
const [isMobile, setIsMobile] = React.useState<boolean | undefined>(undefined) const [isMobile, setIsMobile] = React.useState<boolean | undefined>(undefined)
React.useEffect(() => { React.useEffect(() => {
if (typeof window === 'undefined') return;
const mql = window.matchMedia(`(max-width: ${MOBILE_BREAKPOINT - 1}px)`) const mql = window.matchMedia(`(max-width: ${MOBILE_BREAKPOINT - 1}px)`)
const onChange = () => { const onChange = () => {
setIsMobile(window.innerWidth < MOBILE_BREAKPOINT) setIsMobile(window.innerWidth < MOBILE_BREAKPOINT)

4
src/lib/link-unfurl.ts
View File

@ -94,9 +94,7 @@ export function detectLinkType(url: string): UnfurlResult | null {
// External URL // External URL
try { try {
const parsed = new URL(url); const parsed = new URL(url);
const isExternal = typeof window === 'undefined' const isExternal = !parsed.hostname.includes(window.location.hostname);
? true
: !parsed.hostname.includes(window.location.hostname);
return { return {
type: 'external', type: 'external',
url, url,

12
src/lib/room-ws-client.ts
View File

@ -235,9 +235,7 @@ export class RoomWsClient {
this.reconnectAttempt = 0; this.reconnectAttempt = 0;
this.setStatus('open'); this.setStatus('open');
this.startHeartbeat(); this.startHeartbeat();
this.resubscribeAll().catch((err) => { this.resubscribeAll().catch(() => {});
console.warn('[RoomWs] resubscribe failed:', err);
});
resolve(); resolve();
}; };
@ -1004,9 +1002,7 @@ export class RoomWsClient {
sendTyping(roomId: string, action: 'start' | 'stop'): void { sendTyping(roomId: string, action: 'start' | 'stop'): void {
if (this.ws && this.status === 'open') { if (this.ws && this.status === 'open') {
const wsAction = action === 'start' ? 'typing.start' as WsAction : 'typing.stop' as WsAction; const wsAction = action === 'start' ? 'typing.start' as WsAction : 'typing.stop' as WsAction;
this.requestWs<void>(wsAction, { room_id: roomId, typing: action }).catch((err) => { this.requestWs<void>(wsAction, { room_id: roomId, typing: action }).catch(() => {});
console.debug('[RoomWs] typing event failed:', err);
});
} }
} }
@ -1226,9 +1222,7 @@ export class RoomWsClient {
this.wsToken = null; this.wsToken = null;
console.debug('[RoomWs] Clearing token after 3 reconnect failures, will fetch fresh'); console.debug('[RoomWs] Clearing token after 3 reconnect failures, will fetch fresh');
} }
this.connect().catch(() => { this.connect().catch(() => {});
// connect() has its own retry logic; ignore here to avoid duplicate warnings
});
}, delay); }, delay);
} }
} }

6
src/lib/universal-ws.ts
View File

@ -64,8 +64,8 @@ export class UniversalWsClient {
// Re-subscribe to rooms after reconnect // Re-subscribe to rooms after reconnect
for (const roomId of this.subscribedRooms) { for (const roomId of this.subscribedRooms) {
this.request('room.subscribe', { room_id: roomId }).catch((err) => { this.request('room.subscribe', { room_id: roomId }).catch(() => {
console.warn('[UniversalWs] re-subscribe failed:', err); // ignore re-subscribe errors
}); });
} }
@ -202,7 +202,7 @@ export class UniversalWsClient {
this.reconnectTimer = setTimeout(() => { this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null; this.reconnectTimer = null;
this.connect().catch(() => { this.connect().catch(() => {
// connect() has its own retry logic; ignore here to avoid duplicate warnings // connect() will retry on its own
}); });
}, delay); }, delay);
} }