feat(ai): implement AI worker with NATS JetStream integration and queue configuration

Added app/ai/src/main.rs +81 −0 View app/ai/src/main.rs

@@ -0,0 +1,81 @@

1use anyhow::Context;

2use tracing::{info, warn};

3

4mod worker;

5

6#[tokio::main]

7async fn main() -> anyhow::Result<()> {

8 init_tracing();

9

10 let cfg = config::AppConfig::load().context("loading configuration")?;

11 cfg.validate().map_err(|e| anyhow::anyhow!(e.to_string()))?;

12 anyhow::ensure!(cfg.agent.is_enabled(), "agent configuration is not enabled");

13 anyhow::ensure!(cfg.nats.is_enabled(), "nats url must be set");

14

15 info!("connecting to database / redis / storage / nats");

16 let queue = cfg.agent.queue.clone();

17 let state = export::AppState::with_config(cfg.clone())

18 .await

19 .context("building app state")?;

20 let runtime =

21 agent::runtime::AgentRuntime::new(state.clone()).map_err(|e| anyhow::anyhow!(e))?;

22 let nats = export::AppNats::connect(&cfg.nats)

23 .await

24 .context("connecting to nats")?;

25 info!("dependencies ready");

26

27 let running = worker::start(state, runtime, nats, queue.clone())

28 .await

29 .context("starting ai worker")?;

30 info!(

31 concurrency = queue.concurrency,

32 subject = %queue.subject,

33 "ai worker running"

34 );

35

36 wait_for_shutdown().await;

37 info!("ai worker shutting down");

38

39 running

40 .shutdown()

41 .await

42 .map_err(|e| anyhow::anyhow!("worker shutdown: {e}"))?;

43

44 info!("ai worker shut down");

45 Ok(())

46}

47

48fn init_tracing() {

49 use tracing_subscriber::EnvFilter;

50 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));

51 tracing_subscriber::fmt().with_env_filter(filter).init();

52}

53

54async fn wait_for_shutdown() {

55 let ctrl_c = async {

56 if let Err(err) = tokio::signal::ctrl_c().await {

57 warn!(error = %err, "failed to install ctrl_c handler");

58 }

59 };

60

61 #[cfg(unix)]

62 let terminate = async {

63 use tokio::signal::unix::{SignalKind, signal};

64 match signal(SignalKind::terminate()) {

65 Ok(mut s) => {

66 s.recv().await;

67 }

68 Err(err) => {

69 warn!(error = %err, "failed to install SIGTERM handler");

70 }

71 }

72 };

73

74 #[cfg(not(unix))]

75 let terminate = std::future::pending::<()>();

76

77 tokio::select! {

78 _ = ctrl_c => info!("received SIGINT"),

79 _ = terminate => info!("received SIGTERM"),

80 }

81}

Added app/ai/src/worker.rs +255 −0 View app/ai/src/worker.rs

@@ -0,0 +1,255 @@

1use std::collections::HashSet;

2use std::sync::Arc;

3use std::time::Duration;

4

5use agent::runtime::{AgentRuntime, RunTask, RuntimeConfig, sweep};

6use anyhow::Context;

7use async_nats::jetstream::consumer::{self, AckPolicy, pull};

8use async_nats::jetstream::{AckKind, Message, stream};

9use config::agent::AgentQueueConfig;

10use export::{AppNats, AppState};

11use futures::StreamExt;

12use models::agent::run::{self, AgentRunState};

13use sea_orm::EntityTrait;

14use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore};

15use tokio::task::JoinSet;

16use tokio_util::sync::CancellationToken;

17use tracing::{info, warn};

18use uuid::Uuid;

19

20type PullConsumer = consumer::Consumer<pull::Config>;

21

22pub struct RunningWorker {

23 cancel: CancellationToken,

24 consume: tokio::task::JoinHandle<anyhow::Result<()>>,

25 reaper: tokio::task::JoinHandle<()>,

26}

27

28impl RunningWorker {

29 pub async fn shutdown(self) -> anyhow::Result<()> {

30 self.cancel.cancel();

31 let consume = self.consume.await;

32 let _ = self.reaper.await;

33 match consume {

34 Ok(result) => result.context("consume loop"),

35 Err(err) => Err(anyhow::anyhow!("consume loop panicked: {err}")),

36 }

37 }

38}

39

40pub async fn start(

41 state: AppState,

42 runtime: AgentRuntime,

43 nats: AppNats,

44 queue: AgentQueueConfig,

45) -> anyhow::Result<RunningWorker> {

46 let js = nats.jetstream();

47 let stream = js

48 .get_or_create_stream(stream::Config {

49 name: queue.stream.clone(),

50 subjects: vec![queue.subject.clone()],

51 storage: stream::StorageType::File,

52 max_age: Duration::from_secs(queue.stream_max_age_secs),

53 ..Default::default()

54 })

55 .await

56 .map_err(|e| anyhow::anyhow!("creating stream {}: {e}", queue.stream))?;

57 let consumer = stream

58 .get_or_create_consumer(

59 &queue.durable,

60 pull::Config {

61 durable_name: Some(queue.durable.clone()),

62 ack_policy: AckPolicy::Explicit,

63 ack_wait: Duration::from_secs(queue.ack_wait_secs),

64 max_deliver: queue.max_deliver,

65 ..Default::default()

66 },

67 )

68 .await

69 .map_err(|e| anyhow::anyhow!("creating consumer {}: {e}", queue.durable))?;

70

71 let cancel = CancellationToken::new();

72 let inflight: Arc<Mutex<HashSet<Uuid>>> = Arc::new(Mutex::new(HashSet::new()));

73 let runtime_config = runtime.config().clone();

74 let consume = tokio::spawn(consume_loop(

75 state.clone(),

76 runtime,

77 consumer,

78 queue.clone(),

79 inflight.clone(),

80 cancel.clone(),

81 ));

82 let reaper = tokio::spawn(reaper_loop(

83 state,

84 runtime_config,

85 Duration::from_secs(queue.sweep_interval_secs),

86 cancel.clone(),

87 ));

88 Ok(RunningWorker {

89 cancel,

90 consume,

91 reaper,

92 })

93}

94

95async fn consume_loop(

96 state: AppState,

97 runtime: AgentRuntime,

98 consumer: PullConsumer,

99 queue: AgentQueueConfig,

100 inflight: Arc<Mutex<HashSet<Uuid>>>,

101 cancel: CancellationToken,

102) -> anyhow::Result<()> {

103 let semaphore = Arc::new(Semaphore::new(queue.concurrency));

104 let nak_delay = Duration::from_secs(queue.nak_delay_secs);

105 let mut tasks: JoinSet<()> = JoinSet::new();

106

107 'outer: loop {

108 let mut messages = tokio::select! {

109 result = consumer.messages() => {

110 result.map_err(|e| anyhow::anyhow!("opening consumer stream: {e}"))?

111 }

112 _ = cancel.cancelled() => break 'outer,

113 };

114 loop {

115 let message = tokio::select! {

116 _ = cancel.cancelled() => break 'outer,

117 item = messages.next() => item,

118 };

119 match message {

120 Some(Ok(message)) => {

121 let permit = tokio::select! {

122 permit = semaphore.clone().acquire_owned() => permit,

123 _ = cancel.cancelled() => break 'outer,

124 };

125 let permit = permit.map_err(|e| anyhow::anyhow!(e.to_string()))?;

126 tasks.spawn(handle_message(

127 state.clone(),

128 runtime.clone(),

129 message,

130 permit,

131 inflight.clone(),

132 nak_delay,

133 ));

134 }

135 Some(Err(err)) => {

136 warn!(%err, "consumer stream error, reopening");

137 break;

138 }

139 None => {

140 warn!("consumer stream ended, reopening");

141 break;

142 }

143 }

144 }

145 tokio::select! {

146 _ = cancel.cancelled() => break 'outer,

147 _ = tokio::time::sleep(Duration::from_secs(5)) => {}

148 }

149 }

150

151 let ids: Vec<Uuid> = inflight.lock().await.iter().copied().collect();

152 for run_id in ids {

153 if let Err(err) = runtime.request_cancel(run_id).await {

154 warn!(%run_id, %err, "failed to request cancel during drain");

155 }

156 }

157 let drain = async {

158 while let Some(result) = tasks.join_next().await {

159 if let Err(err) = result {

160 warn!(%err, "run task panicked during drain");

161 }

162 }

163 };

164 if tokio::time::timeout(Duration::from_secs(queue.drain_timeout_secs), drain)

165 .await

166 .is_err()

167 {

168 warn!("drain timeout exceeded, aborting in-flight run tasks");

169 tasks.abort_all();

170 while tasks.join_next().await.is_some() {}

171 }

172 Ok(())

173}

174

175async fn handle_message(

176 state: AppState,

177 runtime: AgentRuntime,

178 message: Message,

179 _permit: OwnedSemaphorePermit,

180 inflight: Arc<Mutex<HashSet<Uuid>>>,

181 nak_delay: Duration,

182) {

183 let task: RunTask = match serde_json::from_slice(&message.message.payload) {

184 Ok(task) => task,

185 Err(err) => {

186 warn!(%err, "dropping malformed agent task message");

187 let _ = message.ack().await;

188 return;

189 }

190 };

191 let run_id = task.run_id;

192 inflight.lock().await.insert(run_id);

193 info!(%run_id, "executing agent run");

194 let nak = match runtime.execute(run_id).await {

195 Ok(outcome) => {

196 info!(

197 %run_id,

198 state = ?outcome.state,

199 stop_reason = ?outcome.stop_reason,

200 "agent run finished"

201 );

202 false

203 }

204 Err(err) => {

205 warn!(%run_id, %err, "agent run execution failed");

206 should_retry(&state, run_id).await

207 }

208 };

209 if nak {

210 let _ = message.ack_with(AckKind::Nak(Some(nak_delay))).await;

211 } else {

212 let _ = message.ack().await;

213 }

214 inflight.lock().await.remove(&run_id);

215}

216

217async fn should_retry(state: &AppState, run_id: Uuid) -> bool {

218 match run::Entity::find_by_id(run_id)

219 .one(state.database.write())

220 .await

221 {

222 Ok(Some(row)) => row.state == AgentRunState::Queued,

223 _ => false,

224 }

225}

226

227async fn reaper_loop(

228 state: AppState,

229 config: RuntimeConfig,

230 interval: Duration,

231 cancel: CancellationToken,

232) {

233 let mut ticker = tokio::time::interval(interval);

234 loop {

235 tokio::select! {

236 _ = cancel.cancelled() => break,

237 _ = ticker.tick() => {

238 match sweep(&state, &config).await {

239 Ok(report) => {

240 if report.killed + report.lost + report.orphaned + report.failed_runs > 0 {

241 info!(

242 killed = report.killed,

243 lost = report.lost,

244 orphaned = report.orphaned,

245 failed_runs = report.failed_runs,

246 "reaper sweep"

247 );

248 }

249 }

250 Err(err) => warn!(%err, "reaper sweep failed"),

251 }

252 }

253 }

254 }

255}

Added app/ai/Cargo.toml +21 −0 View app/ai/Cargo.toml

@@ -0,0 +1,21 @@

1[package]

2name = "ai"

3version.workspace = true

4edition.workspace = true

5publish = false

6

7[dependencies]

8agent = { workspace = true }

9config = { workspace = true }

10export = { workspace = true }

11models = { workspace = true }

12async-nats = { workspace = true }

13sea-orm = { workspace = true }

14tokio = { workspace = true, features = ["signal", "rt-multi-thread", "macros", "time"] }

15tokio-util = { workspace = true }

16futures = { workspace = true }

17serde_json = { workspace = true }

18tracing = { workspace = true }

19tracing-subscriber = { workspace = true }

20anyhow = { workspace = true }

21uuid = { workspace = true }

Added chart/templates/ai.yaml +52 −0 View chart/templates/ai.yaml

@@ -0,0 +1,52 @@

1{{- if .Values.ai.enabled }}

2apiVersion: apps/v1

3kind: Deployment

4metadata:

5 name: {{ printf "%s-ai" (include "gitdata.fullname" .) }}

6 labels:

7 {{- include "gitdata.labels" . | nindent 4 }}

8 app.kubernetes.io/component: ai

9spec:

10 replicas: {{ .Values.ai.replicaCount }}

11 revisionHistoryLimit: 3

12 selector:

13 matchLabels:

14 {{- include "gitdata.selectorLabels" . | nindent 6 }}

15 app.kubernetes.io/component: ai

16 template:

17 metadata:

18 labels:

19 {{- include "gitdata.selectorLabels" . | nindent 8 }}

20 app.kubernetes.io/component: ai

21 {{- with .Values.podLabels }}

22 {{- toYaml . | nindent 8 }}

23 {{- end }}

24 annotations:

25 checksum/env: {{ toYaml .Values.config.env | sha256sum }}

26 {{- with .Values.podAnnotations }}

27 {{- toYaml . | nindent 8 }}

28 {{- end }}

29 spec:

30 serviceAccountName: {{ include "gitdata.serviceAccountName" . }}

31 {{- include "gitdata.imagePullSecrets" . | nindent 6 }}

32 # 停机时取消并排空在飞 run（drain_timeout 120s + 余量）

33 terminationGracePeriodSeconds: 300

34 securityContext:

35 {{- toYaml .Values.podSecurityContext | nindent 8 }}

36 containers:

37 - name: ai

38 image: {{ include "gitdata.image" . }}

39 imagePullPolicy: {{ .Values.image.pullPolicy }}

40 args: ["ai"]

41 {{- include "gitdata.envFrom" . | nindent 10 }}

42 env:

43 {{- include "gitdata.env" . | nindent 12 }}

44 {{- with .Values.ai.extraEnv }}

45 {{- toYaml . | nindent 12 }}

46 {{- end }}

47 # 无监听端口（NATS JetStream worker），不做端口探针；执行全在 cubesandbox，不挂 repos PVC

48 securityContext:

49 {{- toYaml .Values.securityContext | nindent 12 }}

50 resources:

51 {{- toYaml .Values.ai.resources | nindent 12 }}

52{{- end }}

Modified chart/templates/NOTES.txt +3 −0 View chart/templates/NOTES.txt

@@ -23,3 +23,6 @@

2323 生产环境建议 --set secret.existingSecret=<name> 并用外部 Secret 管理。

2424 - repos PVC 需 RWX（NFS/CephFS/EFS）；无 RWX 时将 githost/synchost 各缩到 1 副本并固定同节点。

2525 - 磁盘上的 config/local.toml 含真实密钥，部署前请轮换。

26{{- if .Values.ai.enabled }}

27 - ai worker 需要外部 NATS（APP_NATS__URL）与 APP_AGENT__* provider/sandbox 配置，缺一启动即失败。

28{{- end }}

Modified chart/Chart.yaml +1 −1 View chart/Chart.yaml

@@ -1,6 +1,6 @@

11apiVersion: v2

22name: gitdata

3description: GitData — Rust workspace (apihost / githost / synchost / mails / migrate)

3description: GitData — Rust workspace (apihost / githost / synchost / mails / migrate / ai)

44type: application

55version: 0.1.0

66appVersion: "1.0.0"

Modified chart/values.yaml +10 −0 View chart/values.yaml

@@ -139,6 +139,16 @@

139139 limits: { memory: 256Mi }

140140 extraEnv: []

141141

142# AI agent worker：消费 NATS JetStream 任务并执行 agent run。

143# 依赖外部 NATS（APP_NATS__URL）与 [agent] 配置就绪，故默认关闭。

144ai:

145 enabled: false

146 replicaCount: 1

147 resources:

148 requests: { cpu: 200m, memory: 256Mi }

149 limits: { memory: 1Gi }

150 extraEnv: []

151

142152# ---------------------------------------------------------------------------

143153# 存储：githost 与 synchost 共享 bare 仓库，需要 RWX

144154# ---------------------------------------------------------------------------

Modified config/default.toml +18 −0 View config/default.toml

@@ -16,6 +16,12 @@

1616username = ""

1717password = ""

1818

19[nats]

20url = ""

21token = ""

22user = ""

23password = ""

24

1925[cache]

2026l1_capacity = 10000

2127l1_ttl_secs = 0

@@ -272,3 +278,15 @@

272278artifact_dir = ""

273279context_max_tokens = 128000

274280compact_keep_recent = 16

281

282[agent.queue]

283stream = "AGENT"

284subject = "agent.run.execute"

285durable = "ai-executor"

286concurrency = 4

287ack_wait_secs = 7200

288max_deliver = 8

289nak_delay_secs = 30

290drain_timeout_secs = 120

291sweep_interval_secs = 300

292stream_max_age_secs = 86400

Modified lib/agent/src/runtime/engine.rs +20 −14 View lib/agent/src/runtime/engine.rs

@@ -89,17 +89,13 @@

8989 &self.config

9090 }

9191

92 pub async fn submit(&self, conversation_id: Uuid, prompt: String) -> Result<RunHandle> {

92 pub async fn enqueue(&self, conversation_id: Uuid, prompt: String) -> Result<Uuid> {

9393 let conv = conversation::Entity::find_by_id(conversation_id)

9494 .one(self.state.database.read())

9595 .await?

96 .ok_or_else(|| {

97 AgentError::NotFound(format!("agent conversation {conversation_id}"))

98 })?;

96 .ok_or_else(|| AgentError::NotFound(format!("agent conversation {conversation_id}")))?;

9997 if conv.closed_at.is_some() {

100 return Err(AgentError::Conflict(

101 "conversation is closed".to_string(),

102 ));

98 return Err(AgentError::Conflict("conversation is closed".to_string()));

10399 }

104100 let now = OffsetDateTime::now_utc();

105101 let run_id = Uuid::now_v7();

@@ -119,6 +115,11 @@

119115 }

120116 .insert(self.state.database.write())

121117 .await?;

118 Ok(run_id)

119 }

120

121 pub async fn submit(&self, conversation_id: Uuid, prompt: String) -> Result<RunHandle> {

122 let run_id = self.enqueue(conversation_id, prompt).await?;

122123 let (tx, rx) = mpsc::unbounded_channel();

123124 let rt = self.clone();

124125 let outcome = tokio::spawn(async move { rt.drive(run_id, tx, true).await });

@@ -283,9 +284,7 @@

283284 match item {

284285 Ok(MultiTurnStreamItem::StreamAssistantItem(content)) => match content {

285286 StreamedAssistantContent::Text(t) => {

286 emitter

287 .emit(RunEvent::TextDelta { text: t.text })

288 .await;

287 emitter.emit(RunEvent::TextDelta { text: t.text }).await;

289288 }

290289 StreamedAssistantContent::ReasoningDelta { reasoning, .. } => {

291290 emitter

@@ -456,9 +455,13 @@

456455

457456 let mut out_history = terminal.history;

458457 if compact::estimate_tokens(&out_history) > compact::trigger_threshold(context_max_tokens) {

459 let compaction =

460 compact::compact_history(&summarizer, out_history, context_max_tokens / 2, keep_recent)

461 .await;

458 let compaction = compact::compact_history(

459 &summarizer,

460 out_history,

461 context_max_tokens / 2,

462 keep_recent,

463 )

464 .await;

462465 emit_compacted(&emitter, &compaction).await;

463466 out_history = compaction.history;

464467 }

@@ -612,7 +615,10 @@

612615 let Some(ce) = completion_error else {

613616 return false;

614617 };

615 let body = ce.provider_response_body().unwrap_or_default().to_lowercase();

618 let body = ce

619 .provider_response_body()

620 .unwrap_or_default()

621 .to_lowercase();

616622 body.contains("context_length_exceeded")

617623 || body.contains("maximum context length")

618624 || body.contains("input is too long")

Modified lib/agent/src/runtime/mod.rs +2 −0 View lib/agent/src/runtime/mod.rs

@@ -7,6 +7,7 @@

77pub mod reaper;

88pub mod sandbox;

99pub mod sink;

10pub mod task;

1011

1112pub use compact::{Compaction, Summarizer};

1213pub use config::RuntimeConfig;

@@ -14,3 +15,4 @@

1415pub use events::{CompactMethod, RunEvent, StopReason};

1516pub use reaper::{SweepReport, sweep};

1617pub use sink::{EventEmitter, EventSink, StreamEntry};

18pub use task::RunTask;

Modified lib/agent/src/runtime/persist.rs +8 −12 View lib/agent/src/runtime/persist.rs

@@ -26,7 +26,9 @@

2626 let conversation = conversation::Entity::find_by_id(run.conversation_id)

2727 .one(db.read())

2828 .await?

29 .ok_or_else(|| AgentError::NotFound(format!("agent conversation {}", run.conversation_id)))?;

29 .ok_or_else(|| {

30 AgentError::NotFound(format!("agent conversation {}", run.conversation_id))

31 })?;

3032 let agent = agent::Entity::find_by_id(conversation.agent_id)

3133 .one(db.read())

3234 .await?

@@ -79,11 +81,11 @@

7981

8082pub async fn save_sandbox_id(db: &AppDatabase, run_id: Uuid, sandbox_id: &str) -> Result<()> {

8183 run::Entity::update_many()

84 .col_expr(run::Column::SandboxId, Expr::value(sandbox_id.to_string()))

8285 .col_expr(

83 run::Column::SandboxId,

84 Expr::value(sandbox_id.to_string()),

86 run::Column::UpdatedAt,

87 Expr::value(OffsetDateTime::now_utc()),

8588 )

86 .col_expr(run::Column::UpdatedAt, Expr::value(OffsetDateTime::now_utc()))

8789 .filter(run::Column::Id.eq(run_id))

8890 .exec(db.write())

8991 .await?;

@@ -119,14 +121,8 @@

119121 .map_err(|e| AgentError::Internal(format!("serialize conversation history: {e}")))?;

120122 conversation::Entity::update_many()

121123 .col_expr(conversation::Column::History, Expr::value(history))

122 .col_expr(

123 conversation::Column::LastRunAt,

124 Expr::value(Some(now)),

125 )

126 .col_expr(

127 conversation::Column::UpdatedAt,

128 Expr::value(Some(now)),

129 )

124 .col_expr(conversation::Column::LastRunAt, Expr::value(Some(now)))

125 .col_expr(conversation::Column::UpdatedAt, Expr::value(Some(now)))

130126 .filter(conversation::Column::Id.eq(conversation_id))

131127 .exec(db.write())

132128 .await?;

Modified lib/agent/src/runtime/reaper.rs +4 −13 View lib/agent/src/runtime/reaper.rs

@@ -112,10 +112,8 @@

112112 .all(db.read())

113113 .await?;

114114 for r in running {

115 let stranded_long_ago = r

116 .started_at

117 .is_some_and(|started| started < cutoff)

118 || r.started_at.is_none();

115 let stranded_long_ago =

116 r.started_at.is_some_and(|started| started < cutoff) || r.started_at.is_none();

119117 if !stranded_long_ago {

120118 continue;

121119 }

@@ -126,10 +124,7 @@

126124 .await?;

127125 let stranded = match &sb {

128126 None => true,

129 Some(s) => matches!(

130 s.state,

131 AgentSandboxState::Killed | AgentSandboxState::Lost

132 ),

127 Some(s) => matches!(s.state, AgentSandboxState::Killed | AgentSandboxState::Lost),

133128 };

134129 if !stranded {

135130 continue;

@@ -164,11 +159,7 @@

164159 Ok(report)

165160}

166161

167async fn mark_sandbox(

168 db: &export::AppDatabase,

169 row_id: uuid::Uuid,

170 state: AgentSandboxState,

171) {

162async fn mark_sandbox(db: &export::AppDatabase, row_id: uuid::Uuid, state: AgentSandboxState) {

172163 if let Err(e) =

173164 super::persist::set_sandbox_state(db, row_id, state, Some(OffsetDateTime::now_utc())).await

174165 {

Added lib/agent/src/runtime/task.rs +13 −0 View lib/agent/src/runtime/task.rs

@@ -0,0 +1,13 @@

1use serde::{Deserialize, Serialize};

2use uuid::Uuid;

3

4#[derive(Clone, Debug, Serialize, Deserialize)]

5pub struct RunTask {

6 pub run_id: Uuid,

7}

8

9impl RunTask {

10 pub fn new(run_id: Uuid) -> Self {

11 Self { run_id }

12 }

13}

Modified lib/config/src/agent.rs +60 −0 View lib/config/src/agent.rs

@@ -58,12 +58,45 @@

5858 }

5959}

6060

61#[derive(Clone, Debug, Deserialize)]

62#[serde(default)]

63pub struct AgentQueueConfig {

64 pub stream: String,

65 pub subject: String,

66 pub durable: String,

67 pub concurrency: usize,

68 pub ack_wait_secs: u64,

69 pub max_deliver: i64,

70 pub nak_delay_secs: u64,

71 pub drain_timeout_secs: u64,

72 pub sweep_interval_secs: u64,

73 pub stream_max_age_secs: u64,

74}

75

76impl Default for AgentQueueConfig {

77 fn default() -> Self {

78 Self {

79 stream: "AGENT".to_string(),

80 subject: "agent.run.execute".to_string(),

81 durable: "ai-executor".to_string(),

82 concurrency: 4,

83 ack_wait_secs: 7200,

84 max_deliver: 8,

85 nak_delay_secs: 30,

86 drain_timeout_secs: 120,

87 sweep_interval_secs: 300,

88 stream_max_age_secs: 86_400,

89 }

90 }

91}

92

6193#[derive(Clone, Debug, Default, Deserialize)]

6294#[serde(default)]

6395pub struct AgentConfig {

6496 pub provider: AgentProviderConfig,

6597 pub sandbox: AgentSandboxConfig,

6698 pub runtime: AgentRuntimeConfig,

99 pub queue: AgentQueueConfig,

67100}

68101

69102impl AgentConfig {

@@ -117,6 +150,33 @@

117150 if self.runtime.compact_keep_recent == 0 {

118151 return Err("agent runtime compact_keep_recent must be positive".to_string());

119152 }

153 if self.queue.stream.is_empty() {

154 return Err("agent queue stream must be set".to_string());

155 }

156 if self.queue.subject.is_empty() {

157 return Err("agent queue subject must be set".to_string());

158 }

159 if self.queue.durable.is_empty() {

160 return Err("agent queue durable must be set".to_string());

161 }

162 if self.queue.concurrency == 0 {

163 return Err("agent queue concurrency must be positive".to_string());

164 }

165 if self.queue.ack_wait_secs == 0 {

166 return Err("agent queue ack_wait_secs must be positive".to_string());

167 }

168 if self.queue.max_deliver <= 0 {

169 return Err("agent queue max_deliver must be positive".to_string());

170 }

171 if self.queue.drain_timeout_secs == 0 {

172 return Err("agent queue drain_timeout_secs must be positive".to_string());

173 }

174 if self.queue.sweep_interval_secs == 0 {

175 return Err("agent queue sweep_interval_secs must be positive".to_string());

176 }

177 if self.queue.stream_max_age_secs == 0 {

178 return Err("agent queue stream_max_age_secs must be positive".to_string());

179 }

120180 Ok(())

121181 }

122182}

Modified lib/config/src/lib.rs +3 −0 View lib/config/src/lib.rs

@@ -9,6 +9,7 @@

99pub mod git;

1010pub mod githost;

1111pub mod issues;

12pub mod nats;

1213pub mod pow;

1314pub mod quota;

1415pub mod redis;

@@ -34,6 +35,7 @@

3435use crate::git::GitConfig;

3536use crate::githost::GithostConfig;

3637use crate::issues::IssueConfig;

38use crate::nats::NatsConfig;

3739use crate::pow::PowConfig;

3840use crate::redis::RedisConfig;

3941use crate::s3::S3Config;

@@ -48,6 +50,7 @@

4850pub struct AppConfig {

4951 pub s3: S3Config,

5052 pub redis: RedisConfig,

53 pub nats: NatsConfig,

5154 pub cache: CacheConfig,

5255 pub chat: ChatConfig,

5356 pub database: DatabaseConfig,

Added lib/config/src/nats.rs +16 −0 View lib/config/src/nats.rs

@@ -0,0 +1,16 @@

1use serde::Deserialize;

2

3#[derive(Clone, Debug, Default, Deserialize)]

4#[serde(default)]

5pub struct NatsConfig {

6 pub url: String,

7 pub token: String,

8 pub user: String,

9 pub password: String,

10}

11

12impl NatsConfig {

13 pub fn is_enabled(&self) -> bool {

14 !self.url.is_empty()

15 }

16}

Modified lib/export/src/lib.rs +2 −0 View lib/export/src/lib.rs

@@ -2,6 +2,7 @@

22pub mod error;

33pub mod health;

44pub mod middleware;

5pub mod nats;

56pub mod notify;

67pub mod pagination;

78pub mod redis;

@@ -13,6 +14,7 @@

1314use config::AppConfig;

1415pub use database::{AppDatabase, DatabaseRole};

1516pub use error::ExportError;

17pub use nats::AppNats;

1618pub use redis::{AppRedis, RedisConnection};

1719pub use storage::AppStorage;

1820

Added lib/export/src/nats.rs +41 −0 View lib/export/src/nats.rs

@@ -0,0 +1,41 @@

1use config::nats::NatsConfig;

2

3use crate::error::ExportError;

4

5#[derive(Clone)]

6pub struct AppNats {

7 client: async_nats::Client,

8}

9

10impl AppNats {

11 pub async fn connect(cfg: &NatsConfig) -> Result<Self, ExportError> {

12 if cfg.url.is_empty() {

13 return Err(ExportError::InvalidConfig(

14 "nats url must be set".to_string(),

15 ));

16 }

17 let options = if !cfg.token.is_empty() {

18 async_nats::ConnectOptions::with_token(cfg.token.clone())

19 } else if !cfg.user.is_empty() {

20 async_nats::ConnectOptions::with_user_and_password(

21 cfg.user.clone(),

22 cfg.password.clone(),

23 )

24 } else {

25 async_nats::ConnectOptions::new()

26 };

27 let client = options

28 .connect(cfg.url.clone())

29 .await

30 .map_err(|e| ExportError::Connection(format!("nats: {e}")))?;

31 Ok(Self { client })

32 }

33

34 pub fn client(&self) -> &async_nats::Client {

35 &self.client

36 }

37

38 pub fn jetstream(&self) -> async_nats::jetstream::Context {

39 async_nats::jetstream::new(self.client.clone())

40 }

41}

Modified lib/export/Cargo.toml +1 −0 View lib/export/Cargo.toml

@@ -8,6 +8,7 @@

88storage = { workspace = true }

99sea-orm = { workspace = true }

1010redis = { workspace = true }

11async-nats = { workspace = true }

1112tokio = { workspace = true }

1213email = { workspace = true }

1314axum = { workspace = true }

Modified Cargo.lock +184 −7 View Cargo.lock

@@ -143,6 +143,26 @@

143143]

144144

145145[[package]]

146name = "ai"

147version = "1.0.0"

148dependencies = [

149 "agent",

150 "anyhow",

151 "async-nats",

152 "config",

153 "export",

154 "futures",

155 "models",

156 "sea-orm",

157 "serde_json",

158 "tokio",

159 "tokio-util",

160 "tracing",

161 "tracing-subscriber",

162 "uuid",

163]

164

165[[package]]

146166name = "aliasable"

147167version = "0.1.3"

148168source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -669,6 +689,42 @@

669689]

670690

671691[[package]]

692name = "async-nats"

693version = "0.50.0"

694source = "registry+https://github.com/rust-lang/crates.io-index"

695checksum = "d83a251fa1a4c9d0fe6e816b7acd60549e473e08d14f27a1d992c2675abff05f"

696dependencies = [

697 "base64 0.22.1",

698 "bytes",

699 "futures-util",

700 "memchr",

701 "nkeys",

702 "nuid",

703 "pin-project",

704 "portable-atomic",

705 "rand 0.10.2",

706 "regex",

707 "ring",

708 "rustls-native-certs",

709 "rustls-pki-types",

710 "rustls-webpki",

711 "serde",

712 "serde_json",

713 "serde_nanos",

714 "serde_repr",

715 "thiserror 2.0.20",

716 "time 0.3.55",

717 "tokio",

718 "tokio-rustls",

719 "tokio-stream",

720 "tokio-util",

721 "tokio-websockets",

722 "tracing",

723 "tryhard",

724 "url 2.5.8",

725]

726

727[[package]]

672728name = "async-recursion"

673729version = "1.1.1"

674730source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -2031,6 +2087,21 @@

20312087

20322088[[package]]

20332089name = "curve25519-dalek"

2090version = "4.1.3"

2091source = "registry+https://github.com/rust-lang/crates.io-index"

2092checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be"

2093dependencies = [

2094 "cfg-if",

2095 "cpufeatures 0.2.17",

2096 "curve25519-dalek-derive",

2097 "digest 0.10.7",

2098 "fiat-crypto 0.2.9",

2099 "rustc_version",

2100 "subtle",

2101]

2102

2103[[package]]

2104name = "curve25519-dalek"

20342105version = "5.0.0"

20352106source = "registry+https://github.com/rust-lang/crates.io-index"

20362107checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23"

@@ -2039,7 +2110,7 @@

20392110 "cpufeatures 0.3.0",

20402111 "curve25519-dalek-derive",

20412112 "digest 0.11.3",

2042 "fiat-crypto",

2113 "fiat-crypto 0.3.0",

20432114 "rand_core 0.10.1",

20442115 "rustc_version",

20452116 "subtle",

@@ -2879,6 +2950,7 @@

28792950checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"

28802951dependencies = [

28812952 "const-oid 0.9.6",

2953 "pem-rfc7468 0.7.0",

28822954 "zeroize",

28832955]

28842956

@@ -3118,6 +3190,15 @@

31183190

31193191[[package]]

31203192name = "ed25519"

3193version = "2.2.3"

3194source = "registry+https://github.com/rust-lang/crates.io-index"

3195checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53"

3196dependencies = [

3197 "signature 2.2.0",

3198]

3199

3200[[package]]

3201name = "ed25519"

31213202version = "3.0.0"

31223203source = "registry+https://github.com/rust-lang/crates.io-index"

31233204checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a"

@@ -3128,12 +3209,25 @@

31283209

31293210[[package]]

31303211name = "ed25519-dalek"

3212version = "2.2.0"

3213source = "registry+https://github.com/rust-lang/crates.io-index"

3214checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9"

3215dependencies = [

3216 "curve25519-dalek 4.1.3",

3217 "ed25519 2.2.3",

3218 "sha2 0.10.9",

3219 "signature 2.2.0",

3220 "subtle",

3221]

3222

3223[[package]]

3224name = "ed25519-dalek"

31313225version = "3.0.0"

31323226source = "registry+https://github.com/rust-lang/crates.io-index"

31333227checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de"

31343228dependencies = [

3135 "curve25519-dalek",

3136 "ed25519",

3229 "curve25519-dalek 5.0.0",

3230 "ed25519 3.0.0",

31373231 "rand_core 0.10.1",

31383232 "serde",

31393233 "sha2 0.11.0",

@@ -3358,6 +3452,7 @@

33583452name = "export"

33593453version = "1.0.0"

33603454dependencies = [

3455 "async-nats",

33613456 "axum",

33623457 "bytes",

33633458 "config",

@@ -3506,6 +3601,12 @@

35063601

35073602[[package]]

35083603name = "fiat-crypto"

3604version = "0.2.9"

3605source = "registry+https://github.com/rust-lang/crates.io-index"

3606checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"

3607

3608[[package]]

3609name = "fiat-crypto"

35093610version = "0.3.0"

35103611source = "registry+https://github.com/rust-lang/crates.io-index"

35113612checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24"

@@ -7746,6 +7847,21 @@

77467847]

77477848

77487849[[package]]

7850name = "nkeys"

7851version = "0.4.5"

7852source = "registry+https://github.com/rust-lang/crates.io-index"

7853checksum = "879011babc47a1c7fdf5a935ae3cfe94f34645ca0cac1c7f6424b36fc743d1bf"

7854dependencies = [

7855 "data-encoding",

7856 "ed25519 2.2.3",

7857 "ed25519-dalek 2.2.0",

7858 "getrandom 0.2.17",

7859 "log 0.4.33",

7860 "rand 0.8.7",

7861 "signatory",

7862]

7863

7864[[package]]

77497865name = "no_std_io2"

77507866version = "0.9.4"

77517867source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -7804,6 +7920,15 @@

78047920]

78057921

78067922[[package]]

7923name = "nuid"

7924version = "0.5.0"

7925source = "registry+https://github.com/rust-lang/crates.io-index"

7926checksum = "fc895af95856f929163a0aa20c26a78d26bfdc839f51b9d5aa7a5b79e52b7e83"

7927dependencies = [

7928 "rand 0.8.7",

7929]

7930

7931[[package]]

78077932name = "num-bigint"

78087933version = "0.4.8"

78097934source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -8280,7 +8405,7 @@

82808405dependencies = [

82818406 "ecdsa 0.17.0",

82828407 "elliptic-curve 0.14.1",

8283 "fiat-crypto",

8408 "fiat-crypto 0.3.0",

82848409 "primefield",

82858410 "primeorder 0.14.0",

82868411 "sha2 0.11.0",

@@ -10008,13 +10133,13 @@

1000810133 "cipher 0.5.2",

1000910134 "crypto-bigint 0.7.5",

1001010135 "ctr 0.10.1",

10011 "curve25519-dalek",

10136 "curve25519-dalek 5.0.0",

1001210137 "data-encoding",

1001310138 "delegate",

1001410139 "der 0.8.1",

1001510140 "digest 0.11.3",

1001610141 "ecdsa 0.17.0",

10017 "ed25519-dalek",

10142 "ed25519-dalek 3.0.0",

1001810143 "elliptic-curve 0.14.1",

1001910144 "enum_dispatch",

1002010145 "flate2",

@@ -10696,6 +10821,15 @@

1069610821]

1069710822

1069810823[[package]]

10824name = "serde_nanos"

10825version = "0.1.4"

10826source = "registry+https://github.com/rust-lang/crates.io-index"

10827checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985"

10828dependencies = [

10829 "serde",

10830]

10831

10832[[package]]

1069910833name = "serde_path_to_error"

1070010834version = "0.1.20"

1070110835source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -10947,6 +11081,18 @@

1094711081]

1094811082

1094911083[[package]]

11084name = "signatory"

11085version = "0.27.1"

11086source = "registry+https://github.com/rust-lang/crates.io-index"

11087checksum = "c1e303f8205714074f6068773f0e29527e0453937fe837c9717d066635b65f31"

11088dependencies = [

11089 "pkcs8 0.10.2",

11090 "rand_core 0.6.4",

11091 "signature 2.2.0",

11092 "zeroize",

11093]

11094

11095[[package]]

1095011096name = "signature"

1095111097version = "2.2.0"

1095211098source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -11469,7 +11615,7 @@

1146911615 "argon2",

1147011616 "bcrypt-pbkdf",

1147111617 "ctutils",

11472 "ed25519-dalek",

11618 "ed25519-dalek 3.0.0",

1147311619 "hex",

1147411620 "hmac 0.13.0",

1147511621 "p256 0.14.0",

@@ -12028,6 +12174,27 @@

1202812174]

1202912175

1203012176[[package]]

12177name = "tokio-websockets"

12178version = "0.10.1"

12179source = "registry+https://github.com/rust-lang/crates.io-index"

12180checksum = "f591660438b3038dd04d16c938271c79e7e06260ad2ea2885a4861bfb238605d"

12181dependencies = [

12182 "base64 0.22.1",

12183 "bytes",

12184 "futures-core",

12185 "futures-sink",

12186 "http 1.5.0",

12187 "httparse",

12188 "rand 0.8.7",

12189 "ring",

12190 "rustls-pki-types",

12191 "tokio",

12192 "tokio-rustls",

12193 "tokio-util",

12194 "webpki-roots 0.26.11",

12195]

12196

12197[[package]]

1203112198name = "toml"

1203212199version = "0.8.23"

1203312200source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -12254,6 +12421,16 @@

1225412421checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"

1225512422

1225612423[[package]]

12424name = "tryhard"

12425version = "0.5.2"

12426source = "registry+https://github.com/rust-lang/crates.io-index"

12427checksum = "9fe58ebd5edd976e0fe0f8a14d2a04b7c81ef153ea9a54eebc42e67c2c23b4e5"

12428dependencies = [

12429 "pin-project-lite",

12430 "tokio",

12431]

12432

12433[[package]]

1225712434name = "tungstenite"

1225812435version = "0.29.0"

1225912436source = "registry+https://github.com/rust-lang/crates.io-index"

Modified Cargo.toml +1 −0 View Cargo.toml

@@ -49,6 +49,7 @@

4949dotenvy = "0.15"

5050rust-s3-async = "0.35.0"

5151redis = { version = "1.6.0", features = ["aio", "tokio-comp", "connection-manager", "cluster-async"] }

52async-nats = "0.50"

5253sea-orm = { version = "2.0.2", features = ["sqlx-postgres", "runtime-tokio-rustls"] }

5354oxcache = { version = "0.4.3", features = ["core"] }

5455async-trait = "0.1.92"

Modified Dockerfile +2 −2 View Dockerfile

@@ -27,7 +27,7 @@

2727 --mount=type=cache,target=/build/target \

2828 cargo build --release --workspace --bins && \

2929 mkdir -p /out && \

30 for b in apihost githost synchost mails migrate openapi; do \

30 for b in apihost githost synchost mails migrate openapi ai; do \

3131 cp "target/release/$b" /out/; \

3232 done

3333

@@ -48,7 +48,7 @@

4848WORKDIR /app

4949

5050COPY --chown=gitdata:gitdata config/default.toml config/default.toml

51COPY --from=builder /out/apihost /out/githost /out/synchost /out/mails /out/migrate /out/openapi /usr/local/bin/

51COPY --from=builder /out/apihost /out/githost /out/synchost /out/mails /out/migrate /out/openapi /out/ai /usr/local/bin/

5252

5353USER gitdata

5454