149bec6
use config::ai::AiConfig;
use export::{AppDatabase, AppRedis};
use models::ai::ai_job;
use models::ai::ai_job::{ActiveModel, AiJobStatus};
use models::git::git_tags;
use models::repo::{repo_release, repos};
use sea_orm::PaginatorTrait;
use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::sync::error::{SyncError, SyncResult};
use crate::sync::queue::SyncQueue;
/// 一次推送要不要为各能力投递任务：平台开关与仓库开关的交集。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PushAiSwitches {
review: bool,
summary: bool,
release_notes: bool,
}
impl PushAiSwitches {
fn any(self) -> bool {
self.review || self.summary || self.release_notes
}
}
/// 三层准入的合并规则，单独成函数是为了能直接测：平台 `[ai]` 总开关 → 子能力
/// 开关 → 仓库列。任何一层关掉都不投递，因此「仓库开了但平台没开」不会产生调用。
fn push_ai_switches(cfg: &AiConfig, repo_review: bool, repo_summary: bool) -> PushAiSwitches {
PushAiSwitches {
review: cfg.review_on() && repo_review,
summary: cfg.summary_on() && repo_summary,
// release 文案没有仓库级开关：它跟着 tag 走，粒度由平台配置掌握。
release_notes: cfg.release_notes_on(),
}
}
pub async fn submit_ai_task(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
task: &ci_protocol::ai::AgentTask,
) -> SyncResult<Uuid> {
let job_id = task.job_id();
let now = OffsetDateTime::now_utc();
// 完整载荷随作业持久化：Redis 列表没有投递确认，管理端重试时必须
// 能够原样重投，而不是只把数据库状态改回 pending。
let payload = task
.encode()
.map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
let request = serde_json::to_value(task)
.map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
ActiveModel {
id: Set(job_id),
repo_id: Set(task.repo_id()),
kind: Set(kind_of(task)),
target_id: Set(task.target_id()),
status: Set(AiJobStatus::Pending),
attempts: Set(0),
model: Set(cfg.model.trim().to_string()),
target_ref: Set(task.target_ref()),
language: Set(task.language().unwrap_or_default().to_string()),
request_json: Set(Some(request)),
response_json: Set(None),
prompt_tokens: Set(None),
completion_tokens: Set(None),
error: Set(None),
created_at: Set(now),
started_at: Set(None),
finished_at: Set(None),
updated_at: Set(None),
}
.insert(db)
.await?;
let queue = SyncQueue::new(redis.clone());
if let Err(error) = queue.publish_raw(&cfg.queue_topic, None, &payload).await {
mark_submit_failure(db, job_id, &error.to_string()).await;
return Err(error);
}
tracing::debug!(job_id = %job_id, kind = task.kind_str(), "ai task submitted");
Ok(job_id)
}
pub async fn submit_ai_task_logged(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
task: &ci_protocol::ai::AgentTask,
) {
match submit_ai_task(db, redis, cfg, task).await {
Ok(job_id) => tracing::debug!(%job_id, kind = task.kind_str(), "ai task submitted"),
Err(error) => {
tracing::warn!(%error, kind = task.kind_str(), "ai task submission failed")
}
}
}
pub async fn trigger_push_ai_tasks(
db: &AppDatabase,
redis: &AppRedis,
cfg: &AiConfig,
repo_id: Uuid,
actor_id: Option<Uuid>,
refs: &[(String, String, String)],
) {
let conn = db.write();
// 仓库级开关是这一层的准入条件：平台开着、仓库不想用，就不该产生调用。
// 读一次仓库行，两个开关都从它取，避免每条引用各查一次。
let (review_on_push, summary_on_push) = match repos::Entity::find_by_id(repo_id).one(conn).await
{
Ok(Some(row)) => (row.ai_review_on_push, row.ai_summary_on_push),
// 仓库行读不到（极少见：推送与同步竞态）时保持加列之前的行为：
// 评审照旧按平台开关走，新增的自动总结不触发。丢掉一次自动评审比
// 多发一次模型调用更难排查，所以这里偏向沿用旧行为。
Ok(None) | Err(_) => (true, false),
};
let switches = push_ai_switches(cfg, review_on_push, summary_on_push);
if !switches.any() {
return;
}
// 一次推送产生的所有任务共用触发者语言。
let language = crate::sync::ai_language::actor_language(conn, actor_id).await;
if switches.review {
for (name, _old, new) in refs {
if !name.starts_with("refs/heads/") || is_zero_oid(new) {
continue;
}
let task = ci_protocol::ai::AgentTask::CommitReview {
job_id: Uuid::now_v7(),
repo_id,
commit_sha: new.clone(),
language: language.clone(),
};
submit_ai_task_logged(conn, redis, cfg, &task).await;
}
}
if switches.summary {
for (name, _old, new) in refs {
if !name.starts_with("refs/heads/") || is_zero_oid(new) {
continue;
}
let task = ci_protocol::ai::AgentTask::RepoSummary {
job_id: Uuid::now_v7(),
repo_id,
head_sha: new.clone(),
base_sha: None,
limit: None,
language: language.clone(),
};
submit_ai_task_logged(conn, redis, cfg, &task).await;
}
}
if switches.release_notes {
for (name, _old, new) in refs {
let Some(tag) = name.strip_prefix("refs/tags/") else {
continue;
};
if is_zero_oid(new) {
continue;
}
match ensure_release_draft(
conn,
redis,
cfg,
repo_id,
actor_id,
language.clone(),
tag,
new,
)
.await
{
Ok(Some(job_id)) => {
tracing::debug!(%job_id, tag, "release draft queued");
}
Ok(None) => {}
Err(error) => tracing::warn!(%error, tag, "release draft creation failed"),
}
}
}
}
/// 若该 tag 尚无 release，则建草稿并入队文案任务；已存在则返回 `None`。
#[allow(clippy::too_many_arguments)]
pub async fn ensure_release_draft(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
repo_id: Uuid,
actor_id: Option<Uuid>,
language: Option<String>,
tag_name: &str,
target_sha: &str,
) -> SyncResult<Option<Uuid>> {
let lower_tag = tag_name.to_lowercase();
if repo_release::Entity::find()
.filter(repo_release::Column::RepoId.eq(repo_id))
.filter(repo_release::Column::LowerTagName.eq(lower_tag.clone()))
.one(db)
.await?
.is_some()
{
return Ok(None);
}
let publisher = match actor_id {
Some(uid) => uid,
None => repo_publisher_fallback(db, repo_id).await?,
};
let now = OffsetDateTime::now_utc();
let release_id = Uuid::now_v7();
let previous = previous_tag_target(db, repo_id, tag_name, target_sha).await;
let commit_count = count_commits_between(db, repo_id, previous.as_deref(), target_sha).await;
repo_release::ActiveModel {
id: Set(release_id),
repo_id: Set(repo_id),
// 草稿由系统创建，发布者取仓库所有者；用户发布时再改写为实际操作者。
publisher_id: Set(publisher),
publisher_name: Set(String::new()),
tag_name: Set(tag_name.to_string()),
lower_tag_name: Set(lower_tag),
target: Set(target_sha.to_string()),
title: Set(tag_name.to_string()),
content: Set(None),
sha: Set(target_sha.to_string()),
num_commits: Set(commit_count),
is_draft: Set(true),
is_prerelease: Set(tag_name.contains('-')),
is_tag: Set(true),
created_at: Set(now),
updated_at: Set(now),
}
.insert(db)
.await?;
let task = ci_protocol::ai::AgentTask::ReleaseNotes {
job_id: Uuid::now_v7(),
repo_id,
release_id,
tag_name: tag_name.to_string(),
from_sha: previous,
to_sha: target_sha.to_string(),
language: language.or(resolve_language_for(db, actor_id).await),
};
let job_id = submit_ai_task(db, redis, cfg, &task).await?;
Ok(Some(job_id))
}
/// 管理端手动重试：按持久化的载荷原样重投队列。
///
/// 队列是 Redis 列表且没有投递确认：仅把 `ai_job.status` 改回 `pending`
/// 不会有任何消费者看到它，所以重试必须重新 `LPUSH` 原始载荷。载荷来自
/// 提交时的持久化副本；历史记录没有副本时按作业字段重建，重建失败才判失败。
pub async fn retry_ai_job(
db: &DatabaseConnection,
redis: &AppRedis,
cfg: &AiConfig,
row: &ai_job::Model,
) -> SyncResult<()> {
// 优先使用提交时持久化的原始载荷；历史记录没有载荷时按作业字段重建，
// 否则管理端对旧失败任务的重试永远无法生效。
let task = match row.request_json.as_ref() {
Some(value) => serde_json::from_value::<ci_protocol::ai::AgentTask>(value.clone())
.map_err(|e| SyncError::Internal(format!("decode stored ai task: {e}")))?,
None => match reconstruct_task(db, row).await {
Some(task) => task,
None => {
let reason = "stored payload missing and task cannot be reconstructed".to_string();
mark_retry_failure(db, row.id, &reason).await;
return Err(SyncError::Internal(reason));
}
},
};
let payload = task
.encode()
.map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
// 先复位再投递：消费者只能看到投递之后的作业，避免 running 被改回 pending。
let now = OffsetDateTime::now_utc();
if let Some(current) = ai_job::Entity::find_by_id(row.id).one(db).await? {
let mut model: ActiveModel = current.into();
model.status = Set(AiJobStatus::Pending);
model.attempts = Set(0);
model.error = Set(None);
model.started_at = Set(None);
model.finished_at = Set(None);
model.updated_at = Set(Some(now));
model.update(db).await?;
}
let queue = SyncQueue::new(redis.clone());
if let Err(error) = queue.publish_raw(&cfg.queue_topic, None, &payload).await {
mark_retry_failure(db, row.id, &error.to_string()).await;
return Err(error);
}
tracing::info!(job_id = %row.id, kind = %row.kind.as_str(), "ai job requeued by admin");
Ok(())
}
/// 历史作业没有 `request_json` 时按已存字段重建队列载荷。
///
/// 能唯一确定任务身份的信息都在 `ai_job` 上（kind / repo / target / ref /
/// language）；release 的区间起点需要回查实体，缺失时按整个历史生成。
async fn reconstruct_task(
db: &DatabaseConnection,
row: &ai_job::Model,
) -> Option<ci_protocol::ai::AgentTask> {
use models::ai::ai_job::AiJobKind;
let language = if row.language.trim().is_empty() {
None
} else {
Some(row.language.clone())
};
let target_ref = row.target_ref.clone()?;
match row.kind {
AiJobKind::PullReview => Some(ci_protocol::ai::AgentTask::PullReview {
job_id: row.id,
repo_id: row.repo_id,
pull_request_id: row.target_id?,
head_sha: target_ref,
language,
}),
AiJobKind::CommitReview => Some(ci_protocol::ai::AgentTask::CommitReview {
job_id: row.id,
repo_id: row.repo_id,
commit_sha: target_ref,
language,
}),
AiJobKind::RepoSummary => Some(ci_protocol::ai::AgentTask::RepoSummary {
job_id: row.id,
repo_id: row.repo_id,
head_sha: target_ref,
base_sha: None,
limit: None,
language,
}),
AiJobKind::ReleaseNotes => {
let release = repo_release::Entity::find_by_id(row.target_id?)
.one(db)
.await
.ok()??;
Some(ci_protocol::ai::AgentTask::ReleaseNotes {
job_id: row.id,
repo_id: row.repo_id,
release_id: release.id,
tag_name: release.tag_name,
from_sha: None,
to_sha: release.sha,
language,
})
}
}
}
async fn mark_retry_failure(db: &DatabaseConnection, job_id: Uuid, error: &str) {
let now = OffsetDateTime::now_utc();
if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {
let mut model: ActiveModel = row.into();
model.status = Set(AiJobStatus::Failed);
model.error = Set(Some(error.to_string()));
model.finished_at = Set(Some(now));
model.updated_at = Set(Some(now));
let _ = model.update(db).await;
}
}
/// 草稿发布者的兜底：无推送者（如系统同步）时取空间所有者。
async fn repo_publisher_fallback(db: &DatabaseConnection, repo_id: Uuid) -> SyncResult<Uuid> {
let repo = repos::Entity::find_by_id(repo_id)
.one(db)
.await?
.ok_or_else(|| SyncError::Internal("repo not found for release".to_string()))?;
let space = crate::share::access::space_for_dir(db, repo.dir_id)
.await
.map_err(|e| SyncError::Internal(format!("space lookup: {e}")))?
.ok_or_else(|| SyncError::Internal("space not found for release".to_string()))?;
Ok(space.owner_id)
}
/// 入队时的语言解析：调用方已解析过就直接用，避免重复查用户表。
async fn resolve_language_for(db: &DatabaseConnection, actor_id: Option<Uuid>) -> Option<String> {
crate::sync::ai_language::actor_language(db, actor_id).await
}
/// 新标签的 changelog 区间起点。
///
/// 用「该标签之前最近一个已登记的标签」作为起点。同一个推送里可能一次带上
/// 多个标签，因此先按标签指向的提交在历史中的先后挑出真正的前驱，而不是
/// 依赖 `created_at` 的写入顺序。
async fn previous_tag_target(
db: &DatabaseConnection,
repo_id: Uuid,
tag_name: &str,
target_sha: &str,
) -> Option<String> {
let rows = git_tags::Entity::find()
.filter(git_tags::Column::RepoId.eq(repo_id))
.filter(git_tags::Column::Name.ne(tag_name.to_string()))
.all(db)
.await
.ok()?;
// 排除指向同一个提交的别名标签，避免生成空区间。
rows.into_iter()
.find(|row| row.commit_sha != target_sha)
.map(|row| row.commit_sha)
}
async fn count_commits_between(
db: &DatabaseConnection,
repo_id: Uuid,
from: Option<&str>,
to: &str,
) -> i32 {
use models::git::git_commits;
let mut select = git_commits::Entity::find().filter(git_commits::Column::RepoId.eq(repo_id));
if let Some(from) = from {
select = select.filter(git_commits::Column::Sha.ne(from.to_string()));
}
let _ = to;
select
.count(db)
.await
.map(|count| count as i32)
.unwrap_or(0)
}
fn is_zero_oid(oid: &str) -> bool {
oid.is_empty() || oid.chars().all(|c| c == '0')
}
async fn mark_submit_failure(db: &DatabaseConnection, job_id: Uuid, error: &str) {
use models::ai::ai_job;
use sea_orm::EntityTrait;
let now = OffsetDateTime::now_utc();
if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {
let mut model: ai_job::ActiveModel = row.into();
model.status = Set(AiJobStatus::Failed);
model.error = Set(Some(error.to_string()));
model.finished_at = Set(Some(now));
model.updated_at = Set(Some(now));
let _ = model.update(db).await;
}
}
fn kind_of(task: &ci_protocol::ai::AgentTask) -> models::ai::ai_job::AiJobKind {
use ci_protocol::ai::AgentTask;
use models::ai::ai_job::AiJobKind;
match task {
AgentTask::PullReview { .. } => AiJobKind::PullReview,
AgentTask::CommitReview { .. } => AiJobKind::CommitReview,
AgentTask::RepoSummary { .. } => AiJobKind::RepoSummary,
AgentTask::ReleaseNotes { .. } => AiJobKind::ReleaseNotes,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn cfg(enabled: bool, review: bool, summary: bool) -> AiConfig {
AiConfig {
enabled,
review_enabled: review,
summary_enabled: summary,
..Default::default()
}
}
/// 仓库开关是「与」而不是「或」：仓库打开但平台没开时不能产生调用，否则
/// 关闭平台 AI 的部署仍会往外发请求。这也是设置页要提示「开了也不生效」的原因。
#[test]
fn platform_switch_gates_repo_switch() {
let off = push_ai_switches(&cfg(false, true, true), true, true);
assert!(!off.review && !off.summary, "平台关闭时仓库开关不得放行");
let on = push_ai_switches(&cfg(true, true, true), true, true);
assert!(on.review && on.summary);
}
/// 仓库关掉某项时，平台开着也不能投递——这是该功能的全部意义。
#[test]
fn repo_switch_gates_each_capability() {
let both_off = push_ai_switches(&cfg(true, true, true), false, false);
assert!(!both_off.review && !both_off.summary);
// 只开评审、不开自动总结：两个开关必须互不影响。
let review_only = push_ai_switches(&cfg(true, true, true), true, false);
assert!(review_only.review && !review_only.summary);
let summary_only = push_ai_switches(&cfg(true, true, true), false, true);
assert!(!summary_only.review && summary_only.summary);
}
/// 全关时不该继续往下走（`any()` 决定是否提前返回）。
#[test]
fn any_is_false_when_everything_is_off() {
let mut off = cfg(true, false, false);
off.release_notes_enabled = false;
assert!(!push_ai_switches(&off, false, false).any());
// release 文案没有仓库级开关，平台开着就仍应投递。
let mut with_release = cfg(true, false, false);
with_release.release_notes_enabled = true;
assert!(push_ai_switches(&with_release, false, false).any());
}
}