194c44c
use std::sync::Arc;
use ci_protocol::ai::AgentTask as Task;
use config::AppConfig;
use export::AppState;
use git::service::GitService;
use models::ai::ai_review_result::{ActiveModel as ResultModel, AiReviewVerdict};
use models::ai::repo_ai_summary::ActiveModel as SummaryModel;
use models::pull_request::{pull_request_comment, pull_request_review, pull_requests};
use models::repo::repo_release;
use sea_orm::{
ActiveModelTrait, ActiveValue::Set, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::client::{AgentClient, AgentError, AgentResult, extract_json};
use crate::prompt::{
PromptContext, ReleaseNotesOutput, ReviewOutput, SummaryOutput, detect_languages,
release_system, release_user_prompt, review_system, review_user_prompt, summary_system,
summary_user_prompt,
};
/// 总结最多纳入的提交数（与 `[ai].summary_commit_limit` 取较小值）。
const MAX_SUMMARY_COMMITS: u64 = 200;
/// release 文案最多纳入的提交数。
const MAX_RELEASE_COMMITS: u32 = 200;
/// AI 运行核心：持有单一模型客户端与共享状态，执行队列中的任务。
#[derive(Clone)]
pub struct AgentService {
pub state: AppState,
pub git: GitService,
client: Arc<AgentClient>,
}
impl AgentService {
pub fn new(state: AppState) -> AgentResult<Self> {
let client = AgentClient::new(&state.config.ai)?;
let git = GitService::new(state.clone());
Ok(Self {
state,
git,
client: Arc::new(client),
})
}
pub fn config(&self) -> &AppConfig {
&self.state.config
}
pub fn db_read(&self) -> &DatabaseConnection {
self.state.database.read()
}
pub fn db_write(&self) -> &DatabaseConnection {
self.state.database.write()
}
pub fn client(&self) -> &AgentClient {
&self.client
}
/// 执行一条队列任务。返回 `Err` 时由调用方决定重投或落 DLQ。
pub async fn run_task(&self, task: &Task) -> AgentResult<()> {
let job_id = task.job_id();
self.mark_running(job_id, task.kind_str())
.await
.map_err(|e| AgentError::InvalidTask(e.to_string()))?;
let outcome = match task {
Task::PullReview { .. } => self.run_pull_review(task).await,
Task::CommitReview { .. } => self.run_commit_review(task).await,
Task::RepoSummary { .. } => self.run_repo_summary(task).await,
Task::ReleaseNotes { .. } => self.run_release_notes(task).await,
};
if let Err(error) = &outcome {
tracing::warn!(%error, %job_id, "ai task failed");
}
outcome
}
/// 评审提示词上下文：输出语言 + 本次补丁涉及的语言。
fn review_context(&self, language: &str, diff: &str) -> PromptContext {
PromptContext::new(language, detect_languages(diff))
}
/// 总结与 release 提示词上下文：输出语言 + 仓库已登记的语言统计。
///
/// 提交列表本身不含路径，因此复用 sync 阶段写入的 `git_language` 统计；
/// 统计缺失时只保留语言指令，不影响生成。
async fn repo_context(&self, language: &str, repo_id: Uuid) -> PromptContext {
let languages = self.repo_languages(repo_id).await;
PromptContext::new(language, languages)
}
/// 读取仓库语言统计（按占比降序，最多 5 种）。
async fn repo_languages(&self, repo_id: Uuid) -> Vec<String> {
use models::git::git_languages;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
let rows = git_languages::Entity::find()
.filter(git_languages::Column::RepoId.eq(repo_id))
.order_by_desc(git_languages::Column::PercentageBp)
.limit(5)
.all(self.db_read())
.await;
match rows {
Ok(rows) => rows.into_iter().map(|row| row.language).collect(),
Err(error) => {
tracing::debug!(%error, "repo language lookup failed; omitting language hint");
Vec::new()
}
}
}
/// 仓库全名（space/repo），用于提示词与通知文案。
pub(crate) async fn repo_label(&self, repo_id: Uuid) -> String {
use models::repo::repos;
let repo = repos::Entity::find_by_id(repo_id)
.one(self.db_read())
.await
.ok()
.flatten();
match repo {
Some(row) => row.name,
None => repo_id.to_string(),
}
}
pub(crate) async fn space_name_for(&self, repo_id: Uuid) -> Option<String> {
use models::repo::repos;
let repo = repos::Entity::find_by_id(repo_id)
.one(self.db_read())
.await
.ok()??;
git::share::access::space_for_dir(self.db_read(), repo.dir_id)
.await
.ok()?
.map(|space| space.lower_name)
}
// ---- PR 评审 -------------------------------------------------------
async fn run_pull_review(&self, task: &Task) -> AgentResult<()> {
let Task::PullReview {
job_id,
repo_id,
pull_request_id,
head_sha,
language,
..
} = task
else {
return Err(AgentError::InvalidTask("not a pull review".to_string()));
};
let pull = pull_requests::Entity::find_by_id(*pull_request_id)
.filter(pull_requests::Column::RepoId.eq(*repo_id))
.one(self.db_read())
.await
.map_err(db_err)?
.ok_or_else(|| AgentError::InvalidTask("pull request not found".to_string()))?;
let space = self
.space_name_for(*repo_id)
.await
.ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;
let repo_name = self.repo_label(*repo_id).await;
let (diff, label) = self
.build_pull_diff(&space, &repo_name, head_sha, &pull.base_branch)
.await
.map_err(|e| AgentError::Execution(format!("build diff: {e}")))?;
let lang = self
.resolve_language(Some(pull.author_id), language.as_deref(), *repo_id)
.await;
let ctx = self.review_context(&lang, &diff.text);
let prompt = review_user_prompt(
&repo_name,
&label,
&diff.text,
diff.truncated,
self.config().ai.max_findings,
);
let completion = self
.client
.complete(&review_system(&ctx), &prompt, true)
.await?;
let parsed: ReviewOutput = parse_output(&completion.content)?;
let (findings, dropped) = crate::diff::validate_findings(
&diff.text,
parsed.findings,
self.config().ai.max_findings,
);
if dropped > 0 {
tracing::info!(dropped, "dropped ai findings outside changed lines");
}
let verdict = match parsed.verdict.as_deref() {
Some("changes_requested") => AiReviewVerdict::ChangesRequested,
_ => AiReviewVerdict::Commented,
};
self.persist_pull_review(
job_id,
repo_id,
&pull,
head_sha,
&lang,
verdict,
&parsed.summary,
&findings,
diff.truncated,
&completion.raw,
completion.prompt_tokens,
completion.completion_tokens,
)
.await
}
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments)]
async fn persist_pull_review(
&self,
job_id: &Uuid,
repo_id: &Uuid,
pull: &pull_requests::Model,
head_sha: &str,
language: &str,
verdict: AiReviewVerdict,
summary: &str,
findings: &[crate::prompt::ReviewFinding],
truncated: bool,
raw: &serde_json::Value,
prompt_tokens: Option<i32>,
completion_tokens: Option<i32>,
) -> AgentResult<()> {
let bot = crate::bot::resolve_bot_user(&self.state, &self.config().ai)
.await
.map_err(db_err)?;
let now = OffsetDateTime::now_utc();
let state = match verdict {
AiReviewVerdict::ChangesRequested => pull_request_review::ReviewState::ChangesRequested,
AiReviewVerdict::Commented => pull_request_review::ReviewState::Commented,
};
let review_id = Uuid::now_v7();
pull_request_review::ActiveModel {
id: Set(review_id),
pull_request_id: Set(pull.id),
user_id: Set(bot),
commit_sha: Set(head_sha.to_string()),
official: Set(false),
content: Set(Some(summary.to_string())),
state: Set(state),
created_at: Set(now),
updated_at: Set(None),
}
.insert(self.db_write())
.await
.map_err(db_err)?;
for finding in findings {
pull_request_comment::ActiveModel {
id: Set(Uuid::now_v7()),
pull_request_id: Set(pull.id),
user_id: Set(bot),
review_id: Set(Some(review_id)),
parent_comment_id: Set(None),
commit_sha: Set(Some(head_sha.to_string())),
original_commit_sha: Set(Some(head_sha.to_string())),
path: Set(Some(finding.path.clone())),
side: Set(finding.side.clone()),
line: Set(Some(finding.line)),
original_line: Set(Some(finding.line)),
body: Set(finding_body(finding)),
resolved_at: Set(None),
resolved_by: Set(None),
created_at: Set(now),
updated_at: Set(None),
}
.insert(self.db_write())
.await
.map_err(db_err)?;
}
let findings_value = serde_json::to_value(findings).unwrap_or(serde_json::Value::Null);
ResultModel {
id: Set(Uuid::now_v7()),
repo_id: Set(*repo_id),
pull_request_id: Set(Some(pull.id)),
commit_sha: Set(head_sha.to_string()),
verdict: Set(verdict),
summary: Set(summary.to_string()),
findings_json: Set(findings_value),
truncated: Set(truncated),
language: Set(language.to_string()),
job_id: Set(*job_id),
created_at: Set(now),
}
.insert(self.db_write())
.await
.map_err(db_err)?;
self.mark_success(
*job_id,
None,
Some(raw.clone()),
prompt_tokens,
completion_tokens,
)
.await;
self.notify_pull_ai(pull, bot, language, summary, findings.len())
.await;
Ok(())
}
// ---- Commit 评审 ---------------------------------------------------
async fn run_commit_review(&self, task: &Task) -> AgentResult<()> {
let Task::CommitReview {
job_id,
repo_id,
commit_sha,
language,
..
} = task
else {
return Err(AgentError::InvalidTask("not a commit review".to_string()));
};
let space = self
.space_name_for(*repo_id)
.await
.ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;
let repo_name = self.repo_label(*repo_id).await;
let (diff, label) = self
.build_commit_diff(&space, &repo_name, commit_sha)
.await
.map_err(|e| AgentError::Execution(format!("build diff: {e}")))?;
let lang = self
.resolve_language(None, language.as_deref(), *repo_id)
.await;
let ctx = self.review_context(&lang, &diff.text);
let prompt = review_user_prompt(
&repo_name,
&label,
&diff.text,
diff.truncated,
self.config().ai.max_findings,
);
let completion = self
.client
.complete(&review_system(&ctx), &prompt, true)
.await?;
let parsed: ReviewOutput = parse_output(&completion.content)?;
let (findings, _dropped) = crate::diff::validate_findings(
&diff.text,
parsed.findings,
self.config().ai.max_findings,
);
let verdict = match parsed.verdict.as_deref() {
Some("changes_requested") => AiReviewVerdict::ChangesRequested,
_ => AiReviewVerdict::Commented,
};
let now = OffsetDateTime::now_utc();
let findings_value = serde_json::to_value(&findings).unwrap_or(serde_json::Value::Null);
ResultModel {
id: Set(Uuid::now_v7()),
repo_id: Set(*repo_id),
pull_request_id: Set(None),
commit_sha: Set(commit_sha.clone()),
verdict: Set(verdict),
summary: Set(parsed.summary.clone()),
findings_json: Set(findings_value),
truncated: Set(diff.truncated),
language: Set(lang.clone()),
job_id: Set(*job_id),
created_at: Set(now),
}
.insert(self.db_write())
.await
.map_err(db_err)?;
self.mark_success(
*job_id,
None,
Some(completion.raw.clone()),
completion.prompt_tokens,
completion.completion_tokens,
)
.await;
self.publish_commit_status(
*repo_id,
commit_sha,
verdict,
&parsed.summary,
findings.len(),
)
.await;
Ok(())
}
/// 把提交级评审结论写成 `context=ai-review` 的提交状态。
///
/// 默认不进入 `required_status_contexts`；是否作为合并门禁由管理员配置。
async fn publish_commit_status(
&self,
repo_id: Uuid,
commit_sha: &str,
verdict: AiReviewVerdict,
summary: &str,
findings: usize,
) {
use git::service::commits::{CommitStatusDtoState, SetCommitStatusCmd};
let (space, repo) = match self.repo_location(repo_id).await {
Some(pair) => pair,
None => {
tracing::warn!(%repo_id, "skip commit status: repo location unknown");
return;
}
};
let bot = match crate::bot::resolve_bot_user(&self.state, &self.config().ai).await {
Ok(uid) => uid,
Err(error) => {
tracing::warn!(%error, "skip commit status: bot user unavailable");
return;
}
};
let (state, description) = match verdict {
AiReviewVerdict::ChangesRequested => (
CommitStatusDtoState::Failure,
format!("AI 评审要求修改（{findings} 条建议）"),
),
AiReviewVerdict::Commented => (CommitStatusDtoState::Success, summary.to_string()),
};
let cmd = SetCommitStatusCmd {
sha: commit_sha.to_string(),
state,
context: git::service::ai_review::AI_REVIEW_CONTEXT.to_string(),
description: Some(crate::client::truncate(&description, 200)),
target_url: None,
};
if let Err(error) = self.git.set_commit_status(space, repo, bot, cmd).await {
tracing::warn!(%error, %commit_sha, "failed to publish ai-review commit status");
}
}
/// 由仓库 id 解析 (space, repo) 名称对。
async fn repo_location(&self, repo_id: Uuid) -> Option<(String, String)> {
use models::repo::repos;
let repo = repos::Entity::find_by_id(repo_id)
.one(self.db_read())
.await
.ok()??;
let space = self.space_name_for(repo_id).await?;
Some((space, repo.name))
}
// ---- 仓库总结 ------------------------------------------------------
async fn run_repo_summary(&self, task: &Task) -> AgentResult<()> {
let Task::RepoSummary {
job_id,
repo_id,
head_sha,
base_sha,
limit,
language,
..
} = task
else {
return Err(AgentError::InvalidTask("not a repo summary".to_string()));
};
let space = self
.space_name_for(*repo_id)
.await
.ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;
let repo_name = self.repo_label(*repo_id).await;
// 总结的口径是「最近 N 个提交」，不是两点区间：
// 有显式 base 时按 base..head 取，否则直接从 head 起步。
let limit = limit
.unwrap_or(self.config().ai.summary_commit_limit)
.clamp(1, MAX_SUMMARY_COMMITS);
let trimmed = match base_sha {
Some(base) => self
.git
.commits_between(
space.clone(),
repo_name.clone(),
base,
head_sha,
limit as u32,
)
.await
.map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,
None => self
.git
.recent_commits(
space.clone(),
repo_name.clone(),
Some(head_sha),
limit as u32,
0,
)
.await
.map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,
};
let commit_text = Self::format_commits(&trimmed);
// 区间级的改动规模：让模型知道这批更新是「大量工作」还是「零星几次
// 提交」，据此决定详略。取不到就省略，不影响生成。
let scale = match base_sha {
Some(base) => self
.git
.diff_short_stat(
space.clone(),
repo_name.clone(),
base.to_string(),
head_sha.to_string(),
)
.await
.ok()
.map(|stat| {
format!(
"{} 个文件，+{} / -{} 行",
stat.files_changed, stat.additions, stat.deletions
)
}),
None => None,
};
let prompt = summary_user_prompt(
&repo_name,
head_sha,
trimmed.len(),
scale.as_deref(),
&commit_text,
);
let lang = self
.resolve_language(None, language.as_deref(), *repo_id)
.await;
let ctx = self.repo_context(&lang, *repo_id).await;
let completion = self
.client
.complete(&summary_system(&ctx), &prompt, true)
.await?;
let parsed: SummaryOutput = parse_output(&completion.content)?;
let now = OffsetDateTime::now_utc();
let highlights =
serde_json::to_value(&parsed.highlights).unwrap_or(serde_json::Value::Null);
let risks = serde_json::to_value(&parsed.risks).unwrap_or(serde_json::Value::Null);
let existing = models::ai::repo_ai_summary::Entity::find()
.filter(models::ai::repo_ai_summary::Column::RepoId.eq(*repo_id))
.filter(models::ai::repo_ai_summary::Column::HeadSha.eq(head_sha.clone()))
.one(self.db_read())
.await
.map_err(db_err)?;
match existing {
Some(row) => {
let mut model: SummaryModel = row.into();
model.content = Set(parsed.content.clone());
model.highlights_json = Set(highlights);
model.risks_json = Set(risks);
model.commit_count = Set(trimmed.len() as i32);
model.model = Set(self.client.model().to_string());
model.language = Set(lang.clone());
model.job_id = Set(Some(*job_id));
model.updated_at = Set(now);
model.update(self.db_write()).await.map_err(db_err)?;
}
None => {
SummaryModel {
id: Set(Uuid::now_v7()),
repo_id: Set(*repo_id),
head_sha: Set(head_sha.clone()),
base_sha: Set(base_sha.clone()),
commit_count: Set(trimmed.len() as i32),
content: Set(parsed.content.clone()),
highlights_json: Set(highlights),
risks_json: Set(risks),
model: Set(self.client.model().to_string()),
language: Set(lang.clone()),
job_id: Set(Some(*job_id)),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await
.map_err(db_err)?;
}
}
self.mark_success(
*job_id,
None,
Some(completion.raw.clone()),
completion.prompt_tokens,
completion.completion_tokens,
)
.await;
Ok(())
}
// ---- Release 文案 --------------------------------------------------
async fn run_release_notes(&self, task: &Task) -> AgentResult<()> {
let Task::ReleaseNotes {
job_id,
repo_id,
release_id,
tag_name,
from_sha,
to_sha,
language,
..
} = task
else {
return Err(AgentError::InvalidTask("not a release notes".to_string()));
};
let release = repo_release::Entity::find_by_id(*release_id)
.filter(repo_release::Column::RepoId.eq(*repo_id))
.one(self.db_read())
.await
.map_err(db_err)?
.ok_or_else(|| AgentError::InvalidTask("release not found".to_string()))?;
let space = self
.space_name_for(*repo_id)
.await
.ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;
let repo_name = self.repo_label(*repo_id).await;
// 有上一个标签时按区间取；首个版本没有起点，则回退到「最近若干提交」，
// 而不是留空——否则首个 release 的文案会没有任何素材。
let commits = match from_sha {
Some(from) => self
.git
.commits_between(
space.clone(),
repo_name.clone(),
from,
to_sha,
MAX_RELEASE_COMMITS,
)
.await
.map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,
None => self
.git
.recent_commits(
space,
repo_name.clone(),
Some(to_sha),
MAX_RELEASE_COMMITS,
0,
)
.await
.map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,
};
let commit_text = Self::format_commits(&commits);
let prompt = release_user_prompt(&repo_name, tag_name, from_sha.as_deref(), &commit_text);
let lang = self
.resolve_language(None, language.as_deref(), *repo_id)
.await;
let ctx = self.repo_context(&lang, *repo_id).await;
let completion = self
.client
.complete(&release_system(&ctx), &prompt, true)
.await?;
let parsed: ReleaseNotesOutput = parse_output(&completion.content)?;
let now = OffsetDateTime::now_utc();
// 已发布的 release 不再被 AI 覆盖，只记录作业结果。
if release.is_draft {
let mut model: repo_release::ActiveModel = release.into();
let title = if parsed.title.trim().is_empty() {
tag_name.clone()
} else {
parsed.title.clone()
};
model.title = Set(title);
model.content = Set(Some(parsed.content.clone()));
model.num_commits = Set(commits.len() as i32);
model.updated_at = Set(now);
model.update(self.db_write()).await.map_err(db_err)?;
}
self.mark_success(
*job_id,
None,
Some(completion.raw.clone()),
completion.prompt_tokens,
completion.completion_tokens,
)
.await;
Ok(())
}
/// 记录 release 草稿的作业请求体（便于运维排查）。
pub async fn record_release_request(&self, job_id: Uuid, request: serde_json::Value) {
use models::ai::ai_job;
use sea_orm::EntityTrait;
if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id)
.one(self.db_write())
.await
{
let mut model: ai_job::ActiveModel = row.into();
model.request_json = Set(Some(request));
model.updated_at = Set(Some(OffsetDateTime::now_utc()));
let _ = model.update(self.db_write()).await;
}
}
}
fn finding_body(finding: &crate::prompt::ReviewFinding) -> String {
let severity = finding.severity.as_deref().unwrap_or("minor");
format!("**[{severity}]** {}", finding.body.trim())
}
fn parse_output<T: serde::de::DeserializeOwned>(content: &str) -> AgentResult<T> {
let value = extract_json(content)?;
serde_json::from_value(value).map_err(|e| AgentError::Decode(format!("shape mismatch: {e}")))
}
/// 数据库抖动是可重试的：连接中断、死锁等都会在下一轮恢复。
/// 记录本身缺失另有专门的 `InvalidTask` 分支（`ok_or_else`），不走这里。
fn db_err(error: sea_orm::DbErr) -> AgentError {
AgentError::Execution(format!("database error: {error}"))
}