1. 1//! Durable channel delivery and the transport adapter boundary.
2. 2
3. 3use async_trait::async_trait;
4. 4use serde_json::Value;
5. 5use std::collections::{BTreeMap, BTreeSet, HashMap};
6. 6use std::sync::{Arc, Mutex, OnceLock};
7. 7use std::time::Duration;
8. 8use vak_core::Core;
9. 9use vak_delivery::client::WorkerClient;
10. 10use vak_delivery::outbox::{Outbox, OutboxRecord};
11. 11use vak_delivery::templates::{ChannelPreference, load_layers};
12. 12use vak_delivery::{
13. 13 AnswerDraft, DeliveryAction, DeliveryContent, DeliveryJob, DeliveryKind, DeliveryPacket,
14. 14 DeliveryProfile, Markup,
15. 15};
16. 16
17. 17const WORKER_TIMEOUT: Duration = Duration::from_secs(5);
18. 18const MAX_CHANNEL_CHARS: usize = 100_000;
19. 19
20. 20/// Build a plugin-merged `PresentationPlanner` (skills + recipes) from all
21. 21/// enabled plugins across the Core's capability roots. Starts with built-in
22. 22/// skills and recipes and layers in any `PresentationSkillManifest` and
23. 23/// `PresentationRecipe` files that plugins declare in their
24. 24/// `components.presentation` list.
25. 25pub(crate) fn merged_presentation_planner(core: &Core) -> vak_delivery::PresentationPlanner {
26. 26 let mut skills = vak_delivery::built_in_skill_registry();
27. 27 let mut recipes = vak_delivery::built_in_recipes();
28. 28 let mut revoked_skills = BTreeSet::new();
29. 29 for root in core.capability_roots() {
30. 30 let Ok(plugins) = vak_plugin::PluginStore::new(&root.path).enabled() else {
31. 31 continue;
32. 32 };
33. 33 for plugin in plugins {
34. 34 for relative in &plugin.capabilities.presentation {
35. 35 let path = plugin.package_path.join(relative);
36. 36 let Ok(bytes) = std::fs::read(&path) else {
37. 37 continue;
38. 38 };
39. 39 if let Ok(json) = std::str::from_utf8(&bytes) {
40. 40 if let Ok(manifest) =
41. 41 serde_json::from_str::<vak_delivery::PresentationSkillManifest>(json)
42. 42 {
43. 43 if !core.capability_revoked(
44. 44 vak_session::types::CapabilityKind::Skill,
45. 45 &manifest.id,
46. 46 ) {
47. 47 let _ = skills.register(manifest);
48. 48 } else {
49. 49 revoked_skills.insert(manifest.id);
50. 50 }
51. 51 } else if let Ok(recipe) =
52. 52 serde_json::from_str::<vak_delivery::PresentationRecipe>(json)
53. 53 {
54. 54 let _ = recipes.register(recipe);
55. 55 }
56. 56 }
57. 57 }
58. 58 }
59. 59 }
60. 60 recipes.remove_revoked_skills(&revoked_skills);
61. 61 vak_delivery::PresentationPlanner { skills, recipes }
62. 62}
63. 63
64. 64/// Build a plugin-merged `SkillRegistry` for the worker's structured-fence
65. 65/// projection. Equivalent to `merged_presentation_planner(core).skills`
66. 66/// but avoids constructing a full planner.
67. 67fn merged_presentation_skills(core: &Core) -> vak_delivery::SkillRegistry {
68. 68 merged_presentation_planner(core).skills
69. 69}
70. 70
71. 71#[derive(Debug, Clone, Default, serde::Deserialize, serde::Serialize)]
72. 72#[serde(default, deny_unknown_fields)]
73. 73pub struct RequestedCapabilities {
74. 74 pub markup: Option<Markup>,
75. 75 pub max_chars: Option<usize>,
76. 76 pub supports_tables: Option<bool>,
77. 77 pub supports_code_blocks: Option<bool>,
78. 78 pub supports_links: Option<bool>,
79. 79 pub supports_actions: Option<bool>,
80. 80 /// The bridge sends files back to the chat: a turn's Office drafts are
81. 81 /// returned as documents (docs/design/72, P5).
82. 82 pub accepts_files: Option<bool>,
83. 83}
84. 84
85. 85#[async_trait]
86. 86trait ChannelAdapter: Send + Sync {
87. 87 fn scheme(&self) -> &'static str;
88. 88 fn profile(&self) -> DeliveryProfile;
89. 89 async fn send(&self, core: &Core, packet: &DeliveryPacket) -> Result<(), String>;
90. 90}
91. 91
92. 92struct AdapterRegistry {
93. 93 /// Legacy per-surface fallback, used for a two-part target
94. 94 /// (`surface:address`, no bot id) exactly as before multi-bot-per-
95. 95 /// channel existed.
96. 96 adapters: HashMap<&'static str, Arc<dyn ChannelAdapter>>,
97. 97 /// One adapter per configured bot, keyed by (surface, bot id) — used
98. 98 /// for a three-part target (`surface:address:bot_id`), so a reply goes
99. 99 /// out with *that* bot's own token rather than whichever token happens
100. 100 /// to be registered first for the surface (docs/design/34 Phase 5
101. 101 /// "known limitation", now fixed: the delivery target carries the bot
102. 102 /// id, so this map can pick the exact adapter instead of guessing).
103. 103 bot_adapters: HashMap<(String, String), Arc<dyn ChannelAdapter>>,
104. 104}
105. 105
106. 106impl AdapterRegistry {
107. 107 /// Every configured bot on `surface` with a token actually set, as
108. 108 /// (bot id, token) pairs. Each gets its own adapter: collapsing them
109. 109 /// into one per surface is how a reply goes out under the wrong bot's
110. 110 /// identity (AGENTS.md invariant 24).
111. 111 fn all_bot_tokens(sessions_home: &std::path::Path, surface: &str) -> Vec<(String, String)> {
112. 112 let Ok(raw) = std::fs::read_to_string(sessions_home.join("gateway").join("bots.json"))
113. 113 else {
114. 114 return Vec::new();
115. 115 };
116. 116 let Ok(file) = serde_json::from_str::<serde_json::Value>(&raw) else {
117. 117 return Vec::new();
118. 118 };
119. 119 file.get("bots")
120. 120 .and_then(|b| b.as_array())
121. 121 .map(|bots| {
122. 122 bots.iter()
123. 123 .filter_map(|b| {
124. 124 if b.get("surface")?.as_str()? != surface {
125. 125 return None;
126. 126 }
127. 127 let id = b.get("id")?.as_str()?.to_string();
128. 128 let env_var = b.get("token_env")?.as_str()?;
129. 129 let token = vak_config::get_var(env_var)?;
130. 130 Some((id, token))
131. 131 })
132. 132 .collect()
133. 133 })
134. 134 .unwrap_or_default()
135. 135 }
136. 136
137. 137 fn built_in(sessions_home: &std::path::Path) -> Self {
138. 138 let mut registry = Self {
139. 139 adapters: HashMap::new(),
140. 140 bot_adapters: HashMap::new(),
141. 141 };
142. 142 registry.register(LogAdapter);
143. 143 registry.register(WebhookAdapter);
144. 144 // No per-surface chat adapter is registered. A chat credential
145. 145 // belongs to a bot (invariant 23), so every chat surface resolves
146. 146 // through the (surface, bot_id) map below. Log and webhook stay
147. 147 // surface-level because neither is a bot.
148. 148 let telegram_api_base = vak_config::get_var("TELEGRAM_API_BASE")
149. 149 .unwrap_or_else(|| "https://api.telegram.org".into());
150. 150 for (id, bot_token) in Self::all_bot_tokens(sessions_home, "telegram") {
151. 151 registry.register_bot(
152. 152 "telegram",
153. 153 id,
154. 154 TelegramAdapter {
155. 155 bot_token,
156. 156 api_base: telegram_api_base.clone(),
157. 157 },
158. 158 );
159. 159 }
160. 160 let discord_api_base = vak_config::get_var("DISCORD_API_BASE")
161. 161 .unwrap_or_else(|| "https://discord.com/api/v10".into());
162. 162 for (id, bot_token) in Self::all_bot_tokens(sessions_home, "discord") {
163. 163 registry.register_bot(
164. 164 "discord",
165. 165 id,
166. 166 DiscordAdapter {
167. 167 bot_token,
168. 168 api_base: discord_api_base.clone(),
169. 169 },
170. 170 );
171. 171 }
172. 172 let slack_api_base =
173. 173 vak_config::get_var("SLACK_API_BASE").unwrap_or_else(|| "https://slack.com/api".into());
174. 174 for (id, bot_token) in Self::all_bot_tokens(sessions_home, "slack") {
175. 175 registry.register_bot(
176. 176 "slack",
177. 177 id,
178. 178 SlackAdapter {
179. 179 bot_token,
180. 180 api_base: slack_api_base.clone(),
181. 181 },
182. 182 );
183. 183 }
184. 184 registry
185. 185 }
186. 186
187. 187 fn register(&mut self, adapter: impl ChannelAdapter + 'static) {
188. 188 self.adapters.insert(adapter.scheme(), Arc::new(adapter));
189. 189 }
190. 190
191. 191 fn register_bot(
192. 192 &mut self,
193. 193 surface: &str,
194. 194 bot_id: String,
195. 195 adapter: impl ChannelAdapter + 'static,
196. 196 ) {
197. 197 self.bot_adapters
198. 198 .insert((surface.to_string(), bot_id), Arc::new(adapter));
199. 199 }
200. 200
201. 201 /// A delivery target is `surface:address:bot_id` for a chat surface,
202. 202 /// or `surface:address` for `log` and `webhook`, which are not bots.
203. 203 ///
204. 204 /// A chat target with no bot id is refused rather than resolved: there
205. 205 /// is no per-surface fallback any more, because picking "some bot on
206. 206 /// this surface" replies under an identity the chat was never bound to
207. 207 /// (AGENTS.md invariant 24).
208. 208 fn resolve(&self, target: &str) -> Result<(Arc<dyn ChannelAdapter>, String), String> {
209. 209 let mut parts = target.splitn(3, ':');
210. 210 let scheme = parts.next().filter(|s| !s.is_empty()).ok_or_else(|| {
211. 211 format!("invalid delivery target '{target}': expected '<surface>:<address>'")
212. 212 })?;
213. 213 let address = parts.next().ok_or_else(|| {
214. 214 format!("invalid delivery target '{target}': expected '<surface>:<address>'")
215. 215 })?;
216. 216 if address.trim().is_empty() {
217. 217 return Err(format!("delivery target '{target}' has an empty address"));
218. 218 }
219. 219 if let Some(bot_id) = parts.next().filter(|b| !b.trim().is_empty()) {
220. 220 return self
221. 221 .bot_adapters
222. 222 .get(&(scheme.to_string(), bot_id.to_string()))
223. 223 .cloned()
224. 224 .map(|adapter| (adapter, address.to_string()))
225. 225 // Named but not (yet) configured with a token: fail loudly
226. 226 // rather than silently falling back to a different bot's
227. 227 // token, which would reply under the wrong identity.
228. 228 .ok_or_else(|| {
229. 229 format!("delivery target '{target}' names bot '{bot_id}', which has no token configured")
230. 230 });
231. 231 }
232. 232 if vak_core::Core::is_surface(scheme) {
233. 233 return Err(format!(
234. 234 "delivery target '{target}' names no bot; a {scheme} target must be \
235. 235 '<surface>:<address>:<bot_id>'"
236. 236 ));
237. 237 }
238. 238 self.adapters
239. 239 .get(scheme)
240. 240 .cloned()
241. 241 .map(|adapter| (adapter, address.to_string()))
242. 242 .ok_or_else(|| format!("unsupported gateway surface '{scheme}'"))
243. 243 }
244. 244}
245. 245
246. 246struct DeliveryRuntime {
247. 247 outbox: Outbox,
248. 248 worker: Option<WorkerClient>,
249. 249 adapters: AdapterRegistry,
250. 250 serial: tokio::sync::Mutex<()>,
251. 251}
252. 252
253. 253impl DeliveryRuntime {
254. 254 fn new(core: &Core) -> Self {
255. 255 let worker = std::env::current_exe()
256. 256 .ok()
257. 257 .filter(|path| {
258. 258 path.parent()
259. 259 .and_then(|parent| parent.file_name())
260. 260 .is_none_or(|name| name != "deps")
261. 261 })
262. 262 .map(|path| WorkerClient::new(path, WORKER_TIMEOUT));
263. 263 Self {
264. 264 outbox: Outbox::new(core.shared_data_home().join("delivery").join("jobs")),
265. 265 worker,
266. 266 adapters: AdapterRegistry::built_in(&core.shared_data_home()),
267. 267 serial: tokio::sync::Mutex::new(()),
268. 268 }
269. 269 }
270. 270
271. 271 async fn render(&self, job: &DeliveryJob) -> Result<DeliveryPacket, String> {
272. 272 if let Some(worker) = &self.worker {
273. 273 match worker.render(job).await {
274. 274 Ok(packet) => return Ok(packet),
275. 275 Err(error) => {
276. 276 eprintln!(
277. 277 "[delivery] isolated renderer unavailable for {}: {error}; using safe fallback",
278. 278 job.job_id
279. 279 );
280. 280 }
281. 281 }
282. 282 }
283. 283 let mut packet = vak_delivery::render(job).map_err(|error| error.to_string())?;
284. 284 packet
285. 285 .diagnostics
286. 286 .push("isolated renderer unavailable; used deterministic in-process fallback".into());
287. 287 Ok(packet)
288. 288 }
289. 289
290. 290 async fn deliver_record(
291. 291 &self,
292. 292 core: &Core,
293. 293 record: OutboxRecord,
294. 294 ) -> Result<DeliveryPacket, String> {
295. 295 let mut packet = self.render(&record.job).await?;
296. 296 let (adapter, address) = self.adapters.resolve(&record.job.target)?;
297. 297 // Each adapter's own `send` re-derives its address from
298. 298 // `packet.target` via a plain `surface:address` split — normalize
299. 299 // away a three-part bot-scoped target (`surface:address:bot_id`)
300. 300 // here, once, rather than teaching every adapter about the bot id
301. 301 // segment it has no use for once the right adapter is already
302. 302 // picked.
303. 303 packet.target = format!("{}:{address}", adapter.scheme());
304. 304 adapter.send(core, &packet).await?;
305. 305 self.outbox
306. 306 .mark_delivered(&record.job.job_id, packet.clone())
307. 307 .map_err(|error| error.to_string())?;
308. 308 Ok(packet)
309. 309 }
310. 310}
311. 311
312. 312fn runtime(core: &Core) -> Arc<DeliveryRuntime> {
313. 313 static RUNTIMES: OnceLock<Mutex<BTreeMap<String, Arc<DeliveryRuntime>>>> = OnceLock::new();
314. 314 let key = core.sessions_home().to_string_lossy().into_owned();
315. 315 let runtimes = RUNTIMES.get_or_init(|| Mutex::new(BTreeMap::new()));
316. 316 let mut runtimes = runtimes
317. 317 .lock()
318. 318 .unwrap_or_else(std::sync::PoisonError::into_inner);
319. 319 if let Some(existing) = runtimes.get(&key) {
320. 320 return existing.clone();
321. 321 }
322. 322 let created = Arc::new(DeliveryRuntime::new(core));
323. 323 runtimes.insert(key, created.clone());
324. 324 created
325. 325}
326. 326
327. 327pub(crate) fn profile_for_surface(
328. 328 core: &Core,
329. 329 surface: &str,
330. 330 requested: Option<&RequestedCapabilities>,
331. 331) -> DeliveryProfile {
332. 332 let mut profile = built_in_surface_profile(surface);
333. 333 if let Some(requested) = requested {
334. 334 if let Some(markup) = requested.markup {
335. 335 profile.markup = markup;
336. 336 }
337. 337 if let Some(max_chars) = requested.max_chars.filter(|value| *value > 0) {
338. 338 profile.max_chars = Some(max_chars.min(MAX_CHANNEL_CHARS));
339. 339 }
340. 340 profile.supports_tables = requested.supports_tables.unwrap_or(profile.supports_tables);
341. 341 profile.supports_code_blocks = requested
342. 342 .supports_code_blocks
343. 343 .unwrap_or(profile.supports_code_blocks);
344. 344 profile.supports_links = requested.supports_links.unwrap_or(profile.supports_links);
345. 345 profile.supports_actions = requested
346. 346 .supports_actions
347. 347 .unwrap_or(profile.supports_actions);
348. 348 }
349. 349 apply_preferences(core, profile)
350. 350}
351. 351
352. 352fn built_in_surface_profile(surface: &str) -> DeliveryProfile {
353. 353 match surface {
354. 354 "telegram" => DeliveryProfile {
355. 355 surface: surface.into(),
356. 356 markup: Markup::TelegramHtml,
357. 357 max_chars: Some(4000),
358. 358 supports_tables: false,
359. 359 supports_code_blocks: true,
360. 360 supports_links: true,
361. 361 supports_actions: false,
362. 362 template: None,
363. 363 posture: vak_delivery::DeliveryPosture::default(),
364. 364 },
365. 365 "discord" => DeliveryProfile {
366. 366 surface: surface.into(),
367. 367 markup: Markup::DiscordMarkdown,
368. 368 max_chars: Some(1900),
369. 369 supports_tables: false,
370. 370 supports_code_blocks: true,
371. 371 supports_links: true,
372. 372 supports_actions: false,
373. 373 template: None,
374. 374 posture: vak_delivery::DeliveryPosture::default(),
375. 375 },
376. 376 "slack" => DeliveryProfile {
377. 377 surface: surface.into(),
378. 378 markup: Markup::SlackMrkdwn,
379. 379 max_chars: Some(3900),
380. 380 supports_tables: false,
381. 381 supports_code_blocks: true,
382. 382 supports_links: true,
383. 383 supports_actions: false,
384. 384 template: None,
385. 385 posture: vak_delivery::DeliveryPosture::default(),
386. 386 },
387. 387 "desktop" | "tui" => DeliveryProfile {
388. 388 surface: surface.into(),
389. 389 markup: Markup::Markdown,
390. 390 max_chars: None,
391. 391 supports_tables: true,
392. 392 supports_code_blocks: true,
393. 393 supports_links: true,
394. 394 supports_actions: true,
395. 395 template: None,
396. 396 posture: vak_delivery::DeliveryPosture::default(),
397. 397 },
398. 398 "background" => DeliveryProfile {
399. 399 surface: surface.into(),
400. 400 markup: Markup::Plain,
401. 401 max_chars: Some(4000),
402. 402 supports_tables: false,
403. 403 supports_code_blocks: false,
404. 404 supports_links: true,
405. 405 supports_actions: false,
406. 406 template: None,
407. 407 posture: vak_delivery::DeliveryPosture {
408. 408 cadence: vak_delivery::Cadence::Digest,
409. 409 urgency: vak_delivery::Urgency::Quiet,
410. 410 },
411. 411 },
412. 412 _ => DeliveryProfile {
413. 413 surface: surface.into(),
414. 414 markup: Markup::Plain,
415. 415 max_chars: Some(4000),
416. 416 supports_tables: false,
417. 417 supports_code_blocks: false,
418. 418 supports_links: false,
419. 419 supports_actions: false,
420. 420 template: None,
421. 421 posture: vak_delivery::DeliveryPosture::default(),
422. 422 },
423. 423 }
424. 424}
425. 425
426. 426fn apply_preferences(core: &Core, mut profile: DeliveryProfile) -> DeliveryProfile {
427. 427 let loaded = load_layers(
428. 428 &core.sessions_home().join("output.toml"),
429. 429 &core.cwd().join(".vak").join("output.toml"),
430. 430 core.project_config_trusted(),
431. 431 );
432. 432 for warning in loaded.warnings {
433. 433 eprintln!("[delivery] {warning}");
434. 434 }
435. 435 if let Some(preference) = loaded.channels.get(&profile.surface) {
436. 436 apply_preference(&mut profile, preference, &loaded.registry);
437. 437 }
438. 438 profile
439. 439}
440. 440
441. 441fn apply_preference(
442. 442 profile: &mut DeliveryProfile,
443. 443 preference: &ChannelPreference,
444. 444 templates: &vak_delivery::TemplateRegistry,
445. 445) {
446. 446 if let Some(max_chars) = preference.max_chars.filter(|value| *value > 0) {
447. 447 profile.max_chars = Some(
448. 448 profile
449. 449 .max_chars
450. 450 .map_or(max_chars, |hard_limit| hard_limit.min(max_chars)),
451. 451 );
452. 452 }
453. 453 if let Some(markup) = preference.markup
454. 454 && markup == profile.markup
455. 455 {
456. 456 profile.markup = markup;
457. 457 }
458. 458 if !matches!(profile.markup, Markup::Json)
459. 459 && let Some(template) = preference
460. 460 .template
461. 461 .as_deref()
462. 462 .and_then(|id| templates.resolve(id))
463. 463 {
464. 464 profile.template = Some(template.clone());
465. 465 }
466. 466}
467. 467
468. 468#[allow(clippy::too_many_arguments)]
469. 469pub(crate) async fn render_response(
470. 470 core: &Core,
471. 471 surface: &str,
472. 472 chat: &str,
473. 473 markdown: String,
474. 474 requested: Option<&RequestedCapabilities>,
475. 475 outcome_metadata: Option<std::collections::BTreeMap<String, String>>,
476. 476 provenance: Option<std::collections::BTreeMap<String, String>>,
477. 477 bot_id: Option<&str>,
478. 478 session_id: Option<&str>,
479. 479 intent_posture: Option<vak_intent::DeliveryPosture>,
480. 480) -> Result<DeliveryPacket, String> {
481. 481 let runtime = runtime(core);
482. 482 let mut profile = profile_for_surface(core, surface, requested);
483. 483 // The engagement decides *when* a packet goes out (docs/design/47,
484. 484 // delivery posture): an unattended run rolls up into the digest, an
485. 485 // irreversible step's confirmation breaks through.
486. 486 if let Some(posture) = intent_posture {
487. 487 profile.posture = vak_delivery::DeliveryPosture::from_intent_labels(
488. 488 posture.cadence.as_str(),
489. 489 posture.urgency.as_str(),
490. 490 );
491. 491 }
492. 492 let job = DeliveryJob {
493. 493 job_id: uuid::Uuid::now_v7().to_string(),
494. 494 target: bot_id
495. 495 .filter(|id| !id.trim().is_empty())
496. 496 .map(|id| format!("{surface}:{chat}:{id}"))
497. 497 .unwrap_or_else(|| format!("{surface}:{chat}")),
498. 498 kind: DeliveryKind::Assistant,
499. 499 content: DeliveryContent::Answer({
500. 500 let artifact_suffix = session_id
501. 501 .and_then(|id| {
502. 502 crate::projection::sandbox_artifact_markdown(&core.sessions_home(), id)
503. 503 })
504. 504 .unwrap_or_default();
505. 505 let cleaned_markdown = crate::projection::clean_scaffolding(&markdown);
506. 506 let mut answer =
507. 507 AnswerDraft::from_markdown(format!("{cleaned_markdown}{artifact_suffix}"));
508. 508 if let Some(metadata) = outcome_metadata {
509. 509 answer.metadata.extend(metadata.clone());
510. 510 answer.document.metadata.extend(metadata);
511. 511 }
512. 512 if let Some(provenance) = provenance {
513. 513 answer.metadata.extend(provenance.clone());
514. 514 answer.document.metadata.extend(provenance);
515. 515 }
516. 516 answer
517. 517 }),
518. 518 profile,
519. 519 skill_registry: Some(merged_presentation_skills(core)),
520. 520 };
521. 521 runtime.render(&job).await
522. 522}
523. 523
524. 524pub(crate) async fn deliver(
525. 525 core: &Core,
526. 526 target: &str,
527. 527 kind: DeliveryKind,
528. 528 content: DeliveryContent,
529. 529) -> Result<DeliveryPacket, String> {
530. 530 let runtime = runtime(core);
531. 531 let _serial = runtime.serial.lock().await;
532. 532 let (adapter, _) = runtime.adapters.resolve(target)?;
533. 533 let profile = apply_preferences(core, adapter.profile());
534. 534 let content = enrich_provenance(core, content);
535. 535 // Posture decides WHEN a packet goes out, never what it says.
536. 536 // Held packets (HoldUntilComplete / HoldForDigest) are enqueued to the
537. 537 // outbox and delivered later by the replay loop or turn-completion flush.
538. 538 let disposition = profile.posture.disposition(kind);
539. 539 let job = DeliveryJob {
540. 540 job_id: uuid::Uuid::now_v7().to_string(),
541. 541 target: target.into(),
542. 542 kind,
543. 543 content,
544. 544 profile: profile.clone(),
545. 545 skill_registry: Some(merged_presentation_skills(core)),
546. 546 };
547. 547 let record = runtime
548. 548 .outbox
549. 549 .enqueue(job)
550. 550 .map_err(|error| error.to_string())?;
551. 551 if disposition == vak_delivery::Disposition::Send {
552. 552 match runtime.deliver_record(core, record.clone()).await {
553. 553 Ok(packet) => Ok(packet),
554. 554 Err(error) => {
555. 555 let _ = runtime.outbox.mark_failed(&record.job.job_id, &error);
556. 556 Err(error)
557. 557 }
558. 558 }
559. 559 } else {
560. 560 Ok(vak_delivery::DeliveryPacket {
561. 561 schema_version: vak_delivery::DELIVERY_SCHEMA_VERSION,
562. 562 job_id: record.job.job_id.clone(),
563. 563 target: record.job.target.clone(),
564. 564 surface: record.job.profile.surface.clone(),
565. 565 kind: record.job.kind,
566. 566 payload: vak_delivery::DeliveryPayload::Text(String::new()),
567. 567 fallback_markdown: String::new(),
568. 568 chunks: Vec::new(),
569. 569 actions: Vec::new(),
570. 570 coverage: Vec::new(),
571. 571 diagnostics: vec![format!(
572. 572 "delivery held: disposition={:?}, will retry via outbox",
573. 573 disposition
574. 574 )],
575. 575 presentation: None,
576. 576 })
577. 577 }
578. 578}
579. 579
580. 580/// Attach the immutable ownership envelope before a generic task, schedule, or
581. 581/// approval packet enters the durable outbox. Gateway rendering supplies a
582. 582/// request id as well; this common path guarantees that packets emitted by
583. 583/// internal machinery still identify the Agent and authorized audience.
584. 584fn enrich_provenance(core: &Core, mut content: DeliveryContent) -> DeliveryContent {
585. 585 let DeliveryContent::Answer(answer) = &mut content else {
586. 586 return content;
587. 587 };
588. 588 if let Some(agent) = core.agent_identity() {
589. 589 for (key, value) in [
590. 590 ("agent_id", agent.id.clone()),
591. 591 ("agent_name", agent.name.clone()),
592. 592 ("agent_revision", agent.revision.to_string()),
593. 593 ("agent_character", agent.character.clone()),
594. 594 ("agent_animation", agent.animation.clone()),
595. 595 ("agent_voice", agent.voice.clone()),
596. 596 ] {
597. 597 answer.metadata.insert(key.into(), value.clone());
598. 598 answer.document.metadata.insert(key.into(), value);
599. 599 }
600. 600 }
601. 601 if let Some(context) = core.conversation_context() {
602. 602 for (key, value) in [
603. 603 ("audience_id", context.audience_id.clone()),
604. 604 ("conversation_id", context.conversation_id.clone()),
605. 605 ] {
606. 606 answer.metadata.insert(key.into(), value.clone());
607. 607 answer.document.metadata.insert(key.into(), value);
608. 608 }
609. 609 if let Some(origin) = &context.origin {
610. 610 answer
611. 611 .metadata
612. 612 .insert("origin_surface".into(), origin.surface.clone());
613. 613 answer
614. 614 .document
615. 615 .metadata
616. 616 .insert("origin_surface".into(), origin.surface.clone());
617. 617 answer
618. 618 .metadata
619. 619 .insert("origin_address".into(), origin.address.clone());
620. 620 answer
621. 621 .document
622. 622 .metadata
623. 623 .insert("origin_address".into(), origin.address.clone());
624. 624 if let Some(bot_id) = &origin.bot_id {
625. 625 answer.metadata.insert("bot_id".into(), bot_id.clone());
626. 626 answer
627. 627 .document
628. 628 .metadata
629. 629 .insert("bot_id".into(), bot_id.clone());
630. 630 }
631. 631 }
632. 632 }
633. 633 content
634. 634}
635. 635
636. 636pub(crate) async fn deliver_feed_intents(
637. 637 core: &Core,
638. 638 intents: &[serde_json::Value],
639. 639) -> Result<usize, String> {
640. 640 let mut delivered = 0;
641. 641 for intent in intents {
642. 642 let target = intent
643. 643 .get("deliver_to")
644. 644 .and_then(serde_json::Value::as_str)
645. 645 .filter(|value| !value.is_empty())
646. 646 .ok_or_else(|| "feed alert delivery intent has no target".to_string())?;
647. 647 let alert_name = intent
648. 648 .get("alert_name")
649. 649 .and_then(serde_json::Value::as_str)
650. 650 .unwrap_or("Feed alert");
651. 651 let item = intent.get("item").cloned().unwrap_or_default();
652. 652 let title = item.get("title").and_then(Value::as_str).unwrap_or("");
653. 653 let source = item
654. 654 .get("source_name")
655. 655 .and_then(Value::as_str)
656. 656 .unwrap_or("");
657. 657 let url = item.get("url").and_then(Value::as_str).unwrap_or("");
658. 658 let reasons = intent
659. 659 .get("match")
660. 660 .and_then(|value| value.get("reasons"))
661. 661 .and_then(Value::as_array)
662. 662 .map(|values| {
663. 663 values
664. 664 .iter()
665. 665 .filter_map(Value::as_str)
666. 666 .collect::<Vec<_>>()
667. 667 .join(", ")
668. 668 })
669. 669 .unwrap_or_default();
670. 670 let markdown = format!(
671. 671 "**Feed Alert: {alert_name}**\n\n**{title}**\nSource: {source}\nURL: {url}\nReasons: {reasons}"
672. 672 );
673. 673 deliver(
674. 674 core,
675. 675 target,
676. 676 DeliveryKind::Alert,
677. 677 DeliveryContent::Text { markdown },
678. 678 )
679. 679 .await?;
680. 680 delivered += 1;
681. 681 }
682. 682 Ok(delivered)
683. 683}
684. 684
685. 685pub(crate) fn start_replay(core: &Core) {
686. 686 let core = core.clone();
687. 687 tokio::spawn(async move {
688. 688 let runtime = runtime(&core);
689. 689 let mut interval = tokio::time::interval(Duration::from_secs(30));
690. 690 loop {
691. 691 interval.tick().await;
692. 692 let _serial = runtime.serial.lock().await;
693. 693 let records = match runtime.outbox.pending() {
694. 694 Ok(records) => records,
695. 695 Err(error) => {
696. 696 eprintln!("[delivery] outbox replay scan failed: {error}");
697. 697 continue;
698. 698 }
699. 699 };
700. 700 for record in records.into_iter().take(100) {
701. 701 if record.attempts >= 10 {
702. 702 let _ = runtime
703. 703 .outbox
704. 704 .mark_dead_letter(&record.job.job_id, "delivery retry budget exhausted");
705. 705 eprintln!(
706. 706 "[delivery] {} moved to dead letter after {} attempts",
707. 707 record.job.job_id, record.attempts
708. 708 );
709. 709 continue;
710. 710 }
711. 711 // Respect held postures: a packet held for completion or
712. 712 // digest stays in the outbox until its posture changes.
713. 713 let disposition = record.job.profile.posture.disposition(record.job.kind);
714. 714 if disposition != vak_delivery::Disposition::Send {
715. 715 eprintln!(
716. 716 "[delivery] {} held (disposition={:?}); skipping replay",
717. 717 record.job.job_id, disposition
718. 718 );
719. 719 continue;
720. 720 }
721. 721 if let Err(error) = runtime.deliver_record(&core, record.clone()).await {
722. 722 let _ = runtime.outbox.mark_failed(&record.job.job_id, &error);
723. 723 eprintln!("[delivery] replay {} failed: {error}", record.job.job_id);
724. 724 }
725. 725 }
726. 726 }
727. 727 });
728. 728}
729. 729
730. 730/// Read the durable outbox for the operations surfaces. Records are returned
731. 731/// as-is so the console can distinguish pending, delivered and dead-lettered
732. 732/// work without inventing a second status store.
733. 733pub(crate) fn outbox_records(core: &Core) -> Result<Vec<OutboxRecord>, String> {
734. 734 runtime(core)
735. 735 .outbox
736. 736 .list()
737. 737 .map_err(|error| error.to_string())
738. 738}
739. 739
740. 740/// Replay one pending or dead-lettered job through the same serialized
741. 741/// renderer/adapter path as the background worker. A missing job is surfaced
742. 742/// as an error; no new delivery target or capability is inferred here.
743. 743pub(crate) async fn replay_outbox_job(core: &Core, job_id: &str) -> Result<(), String> {
744. 744 let runtime = runtime(core);
745. 745 let _serial = runtime.serial.lock().await;
746. 746 let record = runtime
747. 747 .outbox
748. 748 .get(job_id)
749. 749 .map_err(|error| error.to_string())?;
750. 750 if record.state == vak_delivery::outbox::OutboxState::Delivered {
751. 751 return Err("delivery job is already delivered".into());
752. 752 }
753. 753 match runtime.deliver_record(core, record.clone()).await {
754. 754 Ok(_) => Ok(()),
755. 755 Err(error) => {
756. 756 let _ = runtime.outbox.mark_failed(job_id, &error);
757. 757 Err(error)
758. 758 }
759. 759 }
760. 760}
761. 761
762. 762struct LogAdapter;
763. 763
764. 764#[async_trait]
765. 765impl ChannelAdapter for LogAdapter {
766. 766 fn scheme(&self) -> &'static str {
767. 767 "log"
768. 768 }
769. 769
770. 770 fn profile(&self) -> DeliveryProfile {
771. 771 DeliveryProfile {
772. 772 surface: "log".into(),
773. 773 markup: Markup::Markdown,
774. 774 max_chars: None,
775. 775 supports_tables: true,
776. 776 supports_code_blocks: true,
777. 777 supports_links: true,
778. 778 supports_actions: false,
779. 779 template: None,
780. 780 posture: vak_delivery::DeliveryPosture::default(),
781. 781 }
782. 782 }
783. 783
784. 784 async fn send(&self, core: &Core, packet: &DeliveryPacket) -> Result<(), String> {
785. 785 let path = core
786. 786 .shared_data_home()
787. 787 .join("gateway")
788. 788 .join("deliveries.jsonl");
789. 789 if let Some(parent) = path.parent() {
790. 790 std::fs::create_dir_all(parent)
791. 791 .map_err(|error| format!("create deliveries directory: {error}"))?;
792. 792 }
793. 793 let line = serde_json::json!({
794. 794 "ts": chrono::Utc::now().to_rfc3339(),
795. 795 "target": packet.target,
796. 796 "text": packet.fallback_markdown,
797. 797 "job_id": packet.job_id,
798. 798 "delivery": packet,
799. 799 });
800. 800 let mut buffer = line.to_string();
801. 801 buffer.push('\n');
802. 802 use std::io::Write;
803. 803 let mut file = std::fs::OpenOptions::new()
804. 804 .create(true)
805. 805 .append(true)
806. 806 .open(path)
807. 807 .map_err(|error| format!("open deliveries log: {error}"))?;
808. 808 file.write_all(buffer.as_bytes())
809. 809 .map_err(|error| format!("append deliveries log: {error}"))
810. 810 }
811. 811}
812. 812
813. 813struct WebhookAdapter;
814. 814
815. 815#[async_trait]
816. 816impl ChannelAdapter for WebhookAdapter {
817. 817 fn scheme(&self) -> &'static str {
818. 818 "webhook"
819. 819 }
820. 820
821. 821 fn profile(&self) -> DeliveryProfile {
822. 822 DeliveryProfile {
823. 823 surface: "webhook".into(),
824. 824 markup: Markup::Json,
825. 825 max_chars: None,
826. 826 supports_tables: true,
827. 827 supports_code_blocks: true,
828. 828 supports_links: true,
829. 829 supports_actions: true,
830. 830 template: None,
831. 831 posture: vak_delivery::DeliveryPosture::default(),
832. 832 }
833. 833 }
834. 834
835. 835 async fn send(&self, core: &Core, packet: &DeliveryPacket) -> Result<(), String> {
836. 836 let (_, name) = packet
837. 837 .target
838. 838 .split_once(':')
839. 839 .ok_or_else(|| "webhook target has no name".to_string())?;
840. 840 super::gateway::deliver_webhook_packet(core, name, packet).await
841. 841 }
842. 842}
843. 843
844. 844/// Telegram's `reply_markup.inline_keyboard`: one row, one button per
845. 845/// action. `callback_data` is `"<verb>:<request_id>"` — the bridge's
846. 846/// `handle_callback` splits on the first `:` and maps `verb` straight onto
847. 847/// the same "yes"/"no" verdict text a typed chat reply would produce.
848. 848fn inline_keyboard_markup(actions: &[DeliveryAction]) -> serde_json::Value {
849. 849 let buttons: Vec<serde_json::Value> = actions
850. 850 .iter()
851. 851 .map(|action| {
852. 852 let request_id = action
853. 853 .data
854. 854 .get("request_id")
855. 855 .map(String::as_str)
856. 856 .unwrap_or_default();
857. 857 serde_json::json!({
858. 858 "text": action.label,
859. 859 "callback_data": format!("{}:{request_id}", action.verb),
860. 860 })
861. 861 })
862. 862 .collect();
863. 863 serde_json::json!({ "inline_keyboard": [buttons] })
864. 864}
865. 865
866. 866/// Proactive push to a Telegram chat: the async counterpart to the
867. 867/// bridge's own `sendMessage` reply. Used for anything that isn't a direct
868. 868/// reply to the message currently in flight — chiefly forwarded approval
869. 869/// gates (`[gateway] approver = "telegram:<chat>"`), which can open while
870. 870/// the approver chat isn't the one that triggered the turn.
871. 871struct TelegramAdapter {
872. 872 bot_token: String,
873. 873 api_base: String,
874. 874}
875. 875
876. 876#[async_trait]
877. 877impl ChannelAdapter for TelegramAdapter {
878. 878 fn scheme(&self) -> &'static str {
879. 879 "telegram"
880. 880 }
881. 881
882. 882 fn profile(&self) -> DeliveryProfile {
883. 883 DeliveryProfile {
884. 884 surface: "telegram".into(),
885. 885 markup: Markup::TelegramHtml,
886. 886 max_chars: Some(4000),
887. 887 supports_tables: false,
888. 888 supports_code_blocks: true,
889. 889 supports_links: true,
890. 890 // Unlike the synchronous per-turn reply profile, this adapter
891. 891 // renders `packet.actions` as inline-keyboard buttons below.
892. 892 supports_actions: true,
893. 893 template: None,
894. 894 posture: vak_delivery::DeliveryPosture::default(),
895. 895 }
896. 896 }
897. 897
898. 898 async fn send(&self, _core: &Core, packet: &DeliveryPacket) -> Result<(), String> {
899. 899 let (_, chat_id) = packet
900. 900 .target
901. 901 .split_once(':')
902. 902 .ok_or_else(|| "telegram target has no chat id".to_string())?;
903. 903 let chunks: Vec<&str> = if packet.chunks.is_empty() {
904. 904 vec![packet.fallback_markdown.as_str()]
905. 905 } else {
906. 906 packet.chunks.iter().map(String::as_str).collect()
907. 907 };
908. 908 let client = reqwest::Client::new();
909. 909 let last = chunks.len().saturating_sub(1);
910. 910 for (i, chunk) in chunks.iter().enumerate() {
911. 911 let mut body = serde_json::json!({
912. 912 "chat_id": chat_id,
913. 913 "text": chunk,
914. 914 "parse_mode": "HTML",
915. 915 "link_preview_options": { "is_disabled": true },
916. 916 });
917. 917 // Buttons ride on the last chunk so they land under the final
918. 918 // line of text, matching where a human reader expects them.
919. 919 if i == last && !packet.actions.is_empty() {
920. 920 body["reply_markup"] = inline_keyboard_markup(&packet.actions);
921. 921 }
922. 922 let resp = client
923. 923 .post(format!(
924. 924 "{}/bot{}/sendMessage",
925. 925 self.api_base, self.bot_token
926. 926 ))
927. 927 .json(&body)
928. 928 .send()
929. 929 .await
930. 930 .map_err(|error| format!("telegram sendMessage: {error}"))?;
931. 931 if !resp.status().is_success() {
932. 932 return Err(format!("telegram sendMessage returned {}", resp.status()));
933. 933 }
934. 934 }
935. 935 Ok(())
936. 936 }
937. 937}
938. 938
939. 939/// Forwarded-approval actions rendered as a typed-verdict prompt, for the
940. 940/// surfaces that do not (yet) get interactive components here. This is the
941. 941/// same fallback Telegram used before its inline keyboard: the reply text
942. 942/// it asks for is exactly what `parse_verdict` in `gateway.rs` already
943. 943/// understands, so approvals resolve through the one existing path.
944. 944fn typed_verdict_prompt(actions: &[DeliveryAction]) -> Option<String> {
945. 945 let request_id = actions
946. 946 .iter()
947. 947 .find_map(|action| action.data.get("request_id"))?;
948. 948 Some(format!(
949. 949 "\n\nReply `yes {request_id}` to approve or `no {request_id}` to deny."
950. 950 ))
951. 951}
952. 952
953. 953/// Proactive push to a Discord channel (docs/design/34 Phase 3): the async
954. 954/// counterpart to the bridge's own reply, chiefly forwarded approval
955. 955/// gates, which can open while the approver channel is not the one that
956. 956/// triggered the turn.
957. 957struct DiscordAdapter {
958. 958 bot_token: String,
959. 959 api_base: String,
960. 960}
961. 961
962. 962#[async_trait]
963. 963impl ChannelAdapter for DiscordAdapter {
964. 964 fn scheme(&self) -> &'static str {
965. 965 "discord"
966. 966 }
967. 967
968. 968 fn profile(&self) -> DeliveryProfile {
969. 969 built_in_surface_profile("discord")
970. 970 }
971. 971
972. 972 async fn send(&self, _core: &Core, packet: &DeliveryPacket) -> Result<(), String> {
973. 973 let (_, channel_id) = packet
974. 974 .target
975. 975 .split_once(':')
976. 976 .ok_or_else(|| "discord target has no channel id".to_string())?;
977. 977 post_chunks(
978. 978 packet,
979. 979 |chunk, last| {
980. 980 let mut content = chunk.to_string();
981. 981 if last && let Some(prompt) = typed_verdict_prompt(&packet.actions) {
982. 982 content.push_str(&prompt);
983. 983 }
984. 984 (
985. 985 format!("{}/channels/{channel_id}/messages", self.api_base),
986. 986 serde_json::json!({ "content": content }),
987. 987 )
988. 988 },
989. 989 |request| request.header("Authorization", format!("Bot {}", self.bot_token)),
990. 990 "discord createMessage",
991. 991 )
992. 992 .await
993. 993 }
994. 994}
995. 995
996. 996/// Proactive push to a Slack channel/DM via `chat.postMessage`.
997. 997struct SlackAdapter {
998. 998 bot_token: String,
999. 999 api_base: String,
1000. 1000}