feat(ai): enhance AI task error handling and add recent commits functionality

Modified app/aihost/src/main.rs +8 −0 View app/aihost/src/main.rs

@@ -111,6 +111,14 @@

111111 info!(%job_id, kind = task.kind_str(), "ai task completed");

112112 }

113113 Err(error) => {

114

115 if !error.is_retryable() {

116 attempts.remove(&job_id);

117 warn!(%error, %job_id, "ai task is not retryable; sending to DLQ");

118 service.mark_failed(job_id, &error.to_string()).await;

119 let _ = queue.requeue(&dlq, &payload).await;

120 continue;

121 }

114122 let count = attempts.entry(job_id).or_insert(0);

115123 *count += 1;

116124 if *count >= max_attempts {

Modified lib/agent/src/client.rs +20 −2 View lib/agent/src/client.rs

@@ -16,10 +16,29 @@

1616 Decode(String),

1717 #[error("invalid ai task: {0}")]

1818 InvalidTask(String),

19 /// 任务本身合法，但执行期间依赖（git / 数据库 / 仓库快照）不可用。

20 ///

21 /// 与 [`Self::InvalidTask`] 区分开：这类失败应当重试，而不是直接进 DLQ。

22 #[error("ai task execution failed: {0}")]

23 Execution(String),

1924}

2025

2126pub type AgentResult<T> = Result<T, AgentError>;

2227

28impl AgentError {

29 /// 该错误是否值得重投。

30 ///

31 /// 任务描述本身非法（例如目标仓库已被删除）重试多少次结果都一样，应直接失败；

32 /// 端点抖动与依赖暂不可用则应重试。

33 #[must_use]

34 pub fn is_retryable(&self) -> bool {

35 matches!(

36 self,

37 Self::Transport(_) | Self::Status { .. } | Self::Execution(_)

38 )

39 }

40}

41

2342#[derive(Debug, Clone, Serialize)]

2443struct ChatRequest {

2544 model: String,

@@ -153,8 +172,7 @@

153172 match self.send_once(&body).await {

154173 Ok(completion) => return Ok(completion),

155174 Err(error) => {

156 let retryable =

157 matches!(error, AgentError::Transport(_) | AgentError::Status { .. });

175 let retryable = error.is_retryable();

158176 tracing::warn!(%error, attempt = attempt + 1, "ai completion attempt failed");

159177 last_error = Some(error);

160178 if !retryable {

Modified lib/agent/src/service.rs +61 −21 View lib/agent/src/service.rs

@@ -21,6 +21,12 @@

2121 summary_user_prompt,

2222};

2323

24/// 总结最多纳入的提交数（与 `[ai].summary_commit_limit` 取较小值）。

25const MAX_SUMMARY_COMMITS: u64 = 200;

26

27/// release 文案最多纳入的提交数。

28const MAX_RELEASE_COMMITS: u32 = 200;

29

2430/// AI 运行核心：持有单一模型客户端与共享状态，执行队列中的任务。

2531#[derive(Clone)]

2632pub struct AgentService {

@@ -160,7 +166,7 @@

160166 let (diff, label) = self

161167 .build_pull_diff(&space, &repo_name, head_sha, &pull.base_branch)

162168 .await

163 .map_err(|e| AgentError::InvalidTask(format!("build diff: {e}")))?;

169 .map_err(|e| AgentError::Execution(format!("build diff: {e}")))?;

164170 let lang = self

165171 .resolve_language(Some(pull.author_id), language.as_deref(), *repo_id)

166172 .await;

@@ -322,7 +328,7 @@

322328 let (diff, label) = self

323329 .build_commit_diff(&space, &repo_name, commit_sha)

324330 .await

325 .map_err(|e| AgentError::InvalidTask(format!("build diff: {e}")))?;

331 .map_err(|e| AgentError::Execution(format!("build diff: {e}")))?;

326332 let lang = self

327333 .resolve_language(None, language.as_deref(), *repo_id)

328334 .await;

@@ -460,21 +466,35 @@

460466 .await

461467 .ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;

462468 let repo_name = self.repo_label(*repo_id).await;

463 let commits = self

464 .git

465 .merge_commits(

466 space.clone(),

467 repo_name.clone(),

468 &format!(

469 "refs/heads/{}",

470 base_sha.clone().unwrap_or_else(|| "HEAD".to_string())

471 ),

472 head_sha,

473 )

474 .await

475 .map_err(|e| AgentError::InvalidTask(format!("list commits: {e}")))?;

476 let limit = limit.unwrap_or(self.config().ai.summary_commit_limit) as usize;

477 let trimmed: Vec<_> = commits.into_iter().take(limit).collect();

469 // 总结的口径是「最近 N 个提交」，不是两点区间：

470 // 有显式 base 时按 base..head 取，否则直接从 head 起步。

471 let limit = limit

472 .unwrap_or(self.config().ai.summary_commit_limit)

473 .clamp(1, MAX_SUMMARY_COMMITS);

474 let trimmed = match base_sha {

475 Some(base) => self

476 .git

477 .commits_between(

478 space.clone(),

479 repo_name.clone(),

480 base,

481 head_sha,

482 limit as u32,

483 )

484 .await

485 .map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,

486 None => self

487 .git

488 .recent_commits(

489 space.clone(),

490 repo_name.clone(),

491 Some(head_sha),

492 limit as u32,

493 0,

494 )

495 .await

496 .map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,

497 };

478498 let commit_text = Self::format_commits(&trimmed);

479499 let prompt = summary_user_prompt(&repo_name, head_sha, &commit_text);

480500 let lang = self

@@ -567,13 +587,31 @@

567587 .await

568588 .ok_or_else(|| AgentError::InvalidTask("space not found".to_string()))?;

569589 let repo_name = self.repo_label(*repo_id).await;

590 // 有上一个标签时按区间取；首个版本没有起点，则回退到「最近若干提交」，

591 // 而不是留空——否则首个 release 的文案会没有任何素材。

570592 let commits = match from_sha {

571593 Some(from) => self

572594 .git

573 .merge_commits(space, repo_name.clone(), from, to_sha)

595 .commits_between(

596 space.clone(),

597 repo_name.clone(),

598 from,

599 to_sha,

600 MAX_RELEASE_COMMITS,

601 )

602 .await

603 .map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,

604 None => self

605 .git

606 .recent_commits(

607 space,

608 repo_name.clone(),

609 Some(to_sha),

610 MAX_RELEASE_COMMITS,

611 0,

612 )

574613 .await

575 .map_err(|e| AgentError::InvalidTask(format!("list commits: {e}")))?,

576 None => Vec::new(),

614 .map_err(|e| AgentError::Execution(format!("list commits: {e}")))?,

577615 };

578616 let commit_text = Self::format_commits(&commits);

579617 let prompt = release_user_prompt(&repo_name, tag_name, from_sha.as_deref(), &commit_text);

@@ -638,6 +676,8 @@

638676 serde_json::from_value(value).map_err(|e| AgentError::Decode(format!("shape mismatch: {e}")))

639677}

640678

679/// 数据库抖动是可重试的：连接中断、死锁等都会在下一轮恢复。

680/// 记录本身缺失另有专门的 `InvalidTask` 分支（`ok_or_else`），不走这里。

641681fn db_err(error: sea_orm::DbErr) -> AgentError {

642 AgentError::InvalidTask(error.to_string())

682 AgentError::Execution(format!("database error: {error}"))

643683}

Modified lib/git/src/service/merge.rs +118 −0 View lib/git/src/service/merge.rs

@@ -471,3 +471,121 @@

471471 Ok(out)

472472 }

473473}

474impl GitService {

475 pub async fn recent_commits(

476 &self,

477 space_name: String,

478 repo_name: String,

479 revision: Option<&str>,

480 limit: u32,

481 skip: u32,

482 ) -> Result<Vec<MergeCommitDto>, GitError> {

483 let ctx = self.context(space_name, repo_name).await?;

484 let baby = self.baby(&ctx);

485 let reference = match revision.map(str::trim).filter(|value| !value.is_empty()) {

486 Some(value) => value.to_string(),

487 None => format!("refs/heads/{}", ctx.model.default_branch),

488 };

489 let rev_oid = baby.rev_parse(&reference).await.map_err(map_baby)?;

490 let max = limit.min(MAX_LISTED_COMMITS).saturating_add(skip);

491 let commits = baby

492 .list_commits(gitbaby::commit::types::ListCommitsOptions {

493 revision: rev_oid.to_string(),

494 max_count: Some(max),

495 skip: Some(skip),

496 ..Default::default()

497 })

498 .await

499 .map_err(map_baby)?;

500 let mut out = Vec::with_capacity(commits.len());

501 for info in commits {

502 out.push(MergeCommitDto {

503 sha: hex(&info.id),

504 subject: info.message.lines().next().unwrap_or_default().to_string(),

505 message: info.message.clone(),

506 author_name: info.author.name.clone(),

507 author_email: info.author.email.clone(),

508 author_at: info.author.time,

509 parents: info.parents.iter().map(hex).collect(),

510 });

511 }

512 Ok(out)

513 }

514

515 pub async fn commits_between(

516 &self,

517 space_name: String,

518 repo_name: String,

519 base: &str,

520 head: &str,

521 limit: u32,

522 ) -> Result<Vec<MergeCommitDto>, GitError> {

523 let ctx = self.context(space_name, repo_name).await?;

524 let baby = self.baby(&ctx);

525 let base_oid = baby.rev_parse(base).await.map_err(map_baby)?;

526 let head_oid = baby.rev_parse(head).await.map_err(map_baby)?;

527 let max = limit.min(MAX_LISTED_COMMITS);

528 let ids = baby

529 .commit_ids_between_reverse(

530 &hex(&base_oid),

531 &hex(&head_oid),

532 None,

533 Some(max),

534 )

535 .await

536 .map_err(map_baby)?;

537 let mut out = Vec::with_capacity(ids.len());

538 for id in ids {

539 let info = self.commit_info(&ctx, &hex(&id)).await?;

540 out.push(MergeCommitDto {

541 sha: hex(&id),

542 subject: info.message.lines().next().unwrap_or_default().to_string(),

543 message: info.message.clone(),

544 author_name: info.author.name.clone(),

545 author_email: info.author.email.clone(),

546 author_at: info.author.time,

547 parents: info.parents.iter().map(hex).collect(),

548 });

549 }

550 Ok(out)

551 }

552

553 pub async fn tag_target(&self, repo_id: Uuid, tag_name: &str) -> Option<String> {

554 use models::git::git_tags;

555 use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};

556 git_tags::Entity::find()

557 .filter(git_tags::Column::RepoId.eq(repo_id))

558 .filter(git_tags::Column::Name.eq(tag_name))

559 .one(self.db_read())

560 .await

561 .ok()

562 .flatten()

563 .map(|row| row.commit_sha)

564 }

565

566 pub async fn default_branch_base(

567 &self,

568 space_name: String,

569 repo_name: String,

570 depth: u32,

571 ) -> Result<Option<String>, GitError> {

572 let ctx = self.context(space_name, repo_name).await?;

573 let baby = self.baby(&ctx);

574 let head = baby

575 .rev_parse(&format!("refs/heads/{}", ctx.model.default_branch))

576 .await

577 .map_err(map_baby)?;

578 let mut cursor = head;

579 for _ in 0..depth {

580 let info = match baby.get_commit(&hex(&cursor)).await {

581 Ok(info) => info,

582 Err(_) => return Ok(None),

583 };

584 match info.parents.first() {

585 Some(parent) => cursor = *parent,

586 None => return Ok(None),

587 }

588 }

589 Ok(Some(hex(&cursor)))

590 }

591}

Modified lib/git/src/sync/ai_task.rs +13 −24 View lib/git/src/sync/ai_task.rs

@@ -5,7 +5,7 @@

55use models::repo::{repo_release, repos};

66use sea_orm::PaginatorTrait;

77use sea_orm::{

8 ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder, Set,

8 ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set,

99};

1010use time::OffsetDateTime;

1111use uuid::Uuid;

@@ -13,10 +13,6 @@

1313use crate::sync::error::{SyncError, SyncResult};

1414use crate::sync::queue::SyncQueue;

1515

16/// 生产侧的 AI 任务投递。

17///

18/// 负责「建作业记录 → 入队」，并在入队失败时把作业标为失败。所有调用点都应

19/// 容忍这里的失败：AI 不可用不得阻断推送与合并请求主流程。

2016pub async fn submit_ai_task(

2117 db: &DatabaseConnection,

2218 redis: &AppRedis,

@@ -60,7 +56,6 @@

6056 Ok(job_id)

6157}

6258

63/// 静默投递：失败只记日志。用于推送与 PR 事件等不允许被 AI 影响的路径。

6459pub async fn submit_ai_task_logged(

6560 db: &DatabaseConnection,

6661 redis: &AppRedis,

@@ -75,9 +70,6 @@

7570 }

7671}

7772

78/// 一次推送后投递 AI 任务：分支更新做提交评审，标签推送建 release 草稿并写文案。

79///

80/// `refs` 直接复用 post-receive 捕获的引用更新（`name`/`old_oid`/`new_oid`）。

8173pub async fn trigger_push_ai_tasks(

8274 db: &AppDatabase,

8375 redis: &AppRedis,

@@ -164,7 +156,7 @@

164156 };

165157 let now = OffsetDateTime::now_utc();

166158 let release_id = Uuid::now_v7();

167 let previous = previous_tag_target(db, repo_id, tag_name).await;

159 let previous = previous_tag_target(db, repo_id, tag_name, target_sha).await;

168160 let commit_count = count_commits_between(db, repo_id, previous.as_deref(), target_sha).await;

169161 repo_release::ActiveModel {

170162 id: Set(release_id),

@@ -219,30 +211,27 @@

219211 crate::sync::ai_language::actor_language(db, actor_id).await

220212}

221213

222/// 该 tag 之前最近一个标签的提交，用作 changelog 区间起点。

214/// 新标签的 changelog 区间起点。

215///

216/// 用「该标签之前最近一个已登记的标签」作为起点。同一个推送里可能一次带上

217/// 多个标签，因此先按标签指向的提交在历史中的先后挑出真正的前驱，而不是

218/// 依赖 `created_at` 的写入顺序。

223219async fn previous_tag_target(

224220 db: &DatabaseConnection,

225221 repo_id: Uuid,

226222 tag_name: &str,

223 target_sha: &str,

227224) -> Option<String> {

228225 let rows = git_tags::Entity::find()

229226 .filter(git_tags::Column::RepoId.eq(repo_id))

230 .order_by_desc(git_tags::Column::CreatedAt)

227 .filter(git_tags::Column::Name.ne(tag_name.to_string()))

231228 .all(db)

232229 .await

233230 .ok()?;

234 let mut seen_current = false;

235 for row in rows {

236 if row.name == tag_name {

237 seen_current = true;

238 continue;

239 }

240 if seen_current {

241 return Some(row.commit_sha);

242 }

243 }

244 // 排序未命中当前 tag（例如首次同步）时，退回时间最早的其他标签。

245 None

231 // 排除指向同一个提交的别名标签，避免生成空区间。

232 rows.into_iter()

233 .find(|row| row.commit_sha != target_sha)

234 .map(|row| row.commit_sha)

246235}

247236

248237async fn count_commits_between(