1. 1//! Gateway: route always-on surfaces (chat adapters, webhooks, cron) to
2. 2//! persistent agent sessions. See docs/design/22-gateway.md.
3. 3//!
4. 4//! A surface speaks HTTP: `POST /gateway/inbound` carries `{surface, chat,
5. 5//! sender, text}` and either returns immediately or long-polls the final
6. 6//! assistant text (`wait`). Bindings persist `(surface:chat) -> session_id`
7. 7//! under `<home>/gateway/bindings.json` so conversations survive restarts.
8. 8//!
9. 9//! Unattended turns fail closed: approval gates are auto-denied and the
10. 10//! denial is fed back to the model as a tool error. Messages arriving while
11. 11//! a turn runs are queued as logged steering input and consumed between
12. 12//! model steps — nothing typed while busy is ever dropped.
13. 13
14. 14use std::collections::HashMap;
15. 15use std::path::PathBuf;
16. 16use std::sync::{Arc, Mutex};
17. 17use std::time::Duration;
18. 18
19. 19use crate::inbox::{self, save_to_inbox};
20. 20
21. 21use axum::extract::{Path, State};
22. 22use axum::http::StatusCode;
23. 23use axum::response::IntoResponse;
24. 24use axum::{Json, Router};
25. 25use tokio::sync::oneshot;
26. 26
27. 27use vak_agent::{AgentEvent, AutoDeny};
28. 28use vak_core::Core;
29. 29use vak_delivery::{
30. 30 AnswerDraft, ApprovalPayload, DeliveryAction, DeliveryContent, DeliveryKind, DeliveryPacket,
31. 31};
32. 32
33. 33use crate::{AppState, SessionHandle};
34. 34
35. 35/// One-shot guard so the forward-mode-but-FullAccess warning does not spam
36. 36/// stderr on every inbound turn.
37. 37static FORWARD_FULLACCESS_WARNED: std::sync::atomic::AtomicBool =
38. 38 std::sync::atomic::AtomicBool::new(false);
39. 39
40. 40/// Contract every inbound channel bridge (Telegram today; Slack, Discord,
41. 41/// ... later) must satisfy before calling `POST /gateway/inbound` (0c-03).
42. 42/// `gateway.chat_allowlist` and the per-conversation session binding are
43. 43/// only as strong as `chat`/`sender` being the real remote identity — a
44. 44/// bridge that reuses one fixed value for every user would silently merge
45. 45/// every stranger into one session and defeat the allowlist outright.
46. 46/// Route new bridges through `InboundRequest::new` rather than hand-rolling
47. 47/// the JSON body so that mistake fails loudly instead of shipping quietly.
48. 48pub trait InboundChannel {
49. 49 /// Stable lowercase surface name ("telegram", "slack", ...) — the
50. 50 /// first half of the `surface:chat` allowlist key.
51. 51 fn surface(&self) -> &'static str;
52. 52}
53. 53
54. 54/// Validated `{surface, chat, sender, text}` payload for one inbound
55. 55/// message, built via [`InboundRequest::new`].
56. 56#[derive(Debug, Clone, serde::Serialize)]
57. 57pub struct InboundRequest {
58. 58 pub surface: String,
59. 59 pub chat: String,
60. 60 pub sender: String,
61. 61 pub text: String,
62. 62 #[serde(default)]
63. 63 pub attachments: Vec<serde_json::Value>,
64. 64 #[serde(default)]
65. 65 pub wait: bool,
66. 66 /// Which configured bot this bridge process is running as
67. 67 /// (multi-bot-per-channel, docs/design/34), when it knows — set via
68. 68 /// `--bot-id` on the CLI bridge. Lets a chat's first-sight pending
69. 69 /// entry record the bot that actually delivered it, instead of
70. 70 /// forcing the operator to pick one by hand for a fact the bridge
71. 71 /// already had.
72. 72 #[serde(default)]
73. 73 pub bot_id: Option<String>,
74. 74 #[serde(default)]
75. 75 pub request_id: Option<String>,
76. 76 /// What the bridge can show and send, when it says.
77. 77 #[serde(default, skip_serializing_if = "Option::is_none")]
78. 78 pub capabilities: Option<crate::delivery::RequestedCapabilities>,
79. 79}
80. 80
81. 81impl InboundRequest {
82. 82 /// Rejects the two shapes a careless bridge tends to produce before it
83. 83 /// has wired up real per-user identity: an empty `chat`/`sender`, or
84. 84 /// one that is literally the surface name (a copy-pasted placeholder).
85. 85 pub fn new(
86. 86 channel: &impl InboundChannel,
87. 87 chat: impl Into<String>,
88. 88 sender: impl Into<String>,
89. 89 text: impl Into<String>,
90. 90 ) -> Result<Self, String> {
91. 91 let surface = channel.surface().to_string();
92. 92 let chat = chat.into();
93. 93 let sender = sender.into();
94. 94 if chat.trim().is_empty() {
95. 95 return Err(format!("{surface} bridge: chat key must not be empty"));
96. 96 }
97. 97 if sender.trim().is_empty() {
98. 98 return Err(format!("{surface} bridge: sender id must not be empty"));
99. 99 }
100. 100 if chat.trim() == surface || sender.trim() == surface {
101. 101 return Err(format!(
102. 102 "{surface} bridge: chat/sender must be the remote identity, not the surface name itself"
103. 103 ));
104. 104 }
105. 105 Ok(Self {
106. 106 surface,
107. 107 chat,
108. 108 sender,
109. 109 text: text.into(),
110. 110 attachments: Vec::new(),
111. 111 wait: false,
112. 112 bot_id: None,
113. 113 request_id: None,
114. 114 capabilities: None,
115. 115 })
116. 116 }
117. 117
118. 118 pub fn with_attachments(mut self, attachments: Vec<serde_json::Value>) -> Self {
119. 119 const MAX_AUDIO_ATTACHMENT_BYTES: usize = 16 * 1024 * 1024;
120. 120 self.attachments = attachments
121. 121 .into_iter()
122. 122 .map(|mut attachment| {
123. 123 if attachment.get("kind").and_then(|v| v.as_str()) == Some("audio") {
124. 124 let encoded = attachment
125. 125 .get("data")
126. 126 .and_then(|v| v.as_str())
127. 127 .unwrap_or_default();
128. 128 if encoded.len() > MAX_AUDIO_ATTACHMENT_BYTES.saturating_mul(4) / 3 {
129. 129 attachment["data"] = serde_json::Value::String(String::new());
130. 130 attachment["error"] =
131. 131 serde_json::Value::String("audio attachment exceeds 16 MiB".into());
132. 132 }
133. 133 }
134. 134 attachment
135. 135 })
136. 136 .collect();
137. 137 self
138. 138 }
139. 139
140. 140 pub fn waiting(mut self) -> Self {
141. 141 self.wait = true;
142. 142 self
143. 143 }
144. 144
145. 145 /// The bridge sends files to the chat, so a turn's Office drafts come
146. 146 /// back in the reply's `files` rather than as a note.
147. 147 pub fn accepting_files(mut self) -> Self {
148. 148 self.capabilities
149. 149 .get_or_insert_with(Default::default)
150. 150 .accepts_files = Some(true);
151. 151 self
152. 152 }
153. 153
154. 154 /// Tag this request with the bot identity the bridge is running as, if
155. 155 /// any (see the `bot_id` field doc).
156. 156 pub fn with_bot_id(mut self, bot_id: Option<String>) -> Self {
157. 157 self.bot_id = bot_id;
158. 158 self
159. 159 }
160. 160
161. 161 /// Attach the bridge's durable idempotency key when the upstream
162. 162 /// transport provides one (Telegram update id, webhook event id, etc.).
163. 163 pub fn with_request_id(mut self, request_id: Option<String>) -> Self {
164. 164 self.request_id = request_id;
165. 165 self
166. 166 }
167. 167}
168. 168
169. 169/// Long-poll ceiling for `wait: true` inbound messages.
170. 170const WAIT_TIMEOUT: Duration = Duration::from_secs(240);
171. 171
172. 172/// Upper bound on one background reflection pass (docs/design/29 P1) so a
173. 173/// stuck auxiliary stream cannot hold the session ledger indefinitely.
174. 174const REFLECTION_CALL_TIMEOUT: Duration = Duration::from_secs(120);
175. 175
176. 176fn bindings_path(home: &std::path::Path) -> PathBuf {
177. 177 home.join("gateway").join("bindings.json")
178. 178}
179. 179
180. 180fn allowlist_path(home: &std::path::Path) -> PathBuf {
181. 181 home.join("gateway").join("allowlist.json")
182. 182}
183. 183
184. 184fn bots_path(home: &std::path::Path) -> PathBuf {
185. 185 home.join("gateway").join("bots.json")
186. 186}
187. 187
188. 188/// Multi-bot-per-channel (docs/design/34 Phase 5 follow-up): an allowlist
189. 189/// key is `surface:chat` for a legacy/single-bot chat, or
190. 190/// `surface:chat:bot_id` once a specific bot is scoped into it — the bot id
191. 191/// is the third segment precisely so [`legacy_key_for`] can strip it back
192. 192/// off. Returns `None` for a key that is already legacy-shaped (nothing to
193. 193/// strip) or malformed.
194. 194fn legacy_key_for(key: &str) -> Option<String> {
195. 195 let mut parts = key.splitn(3, ':');
196. 196 let surface = parts.next()?;
197. 197 let chat = parts.next()?;
198. 198 parts.next()?; // only a genuinely 3-part (bot-scoped) key has a legacy form
199. 199 Some(format!("{surface}:{chat}"))
200. 200}
201. 201
202. 202/// Read-only lookup of one bot's token env var by id, straight from
203. 203/// `bots.json`, without needing a running `GatewayState` — the `vak
204. 204/// telegram/discord/slack --bot-id` CLI bridges are separate short-lived
205. 205/// processes that never construct one, but still need to resolve which env
206. 206/// var holds their token (docs/design/34, multi-bot).
207. 207pub fn bot_token_env_for_id(sessions_home: &std::path::Path, id: &str) -> Option<String> {
208. 208 let raw = std::fs::read_to_string(bots_path(sessions_home)).ok()?;
209. 209 let file: BotsFile = serde_json::from_str(&raw).ok()?;
210. 210 file.bots
211. 211 .into_iter()
212. 212 .find(|b| b.id == id)
213. 213 .map(|b| b.token_env)
214. 214}
215. 215
216. 216/// Truncation cap for `first_seen_text` on a freshly pending entry — kept
217. 217/// only for operator review, never used as agent input.
218. 218const FIRST_SEEN_TEXT_MAX_CHARS: usize = 500;
219. 219
220. 220#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
221. 221#[serde(rename_all = "snake_case")]
222. 222pub enum AllowlistStatus {
223. 223 Pending,
224. 224 Allowed,
225. 225 Denied,
226. 226}
227. 227
228. 228#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
229. 229pub struct AllowlistRoute {
230. 230 pub provider: String,
231. 231 pub model: String,
232. 232}
233. 233
234. 234#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
235. 235pub struct AllowlistEntry {
236. 236 pub key: String,
237. 237 pub status: AllowlistStatus,
238. 238 #[serde(default, skip_serializing_if = "Option::is_none")]
239. 239 pub workspace: Option<PathBuf>,
240. 240 /// Agent selected for this endpoint. Missing values are normalized to the
241. 241 /// reserved Vakyartha identity when loading older allowlist rows.
242. 242 #[serde(default, skip_serializing_if = "Option::is_none")]
243. 243 pub agent_id: Option<String>,
244. 244 #[serde(default, skip_serializing_if = "Option::is_none")]
245. 245 pub route: Option<AllowlistRoute>,
246. 246 /// Voice/persona override for this chat. `None` inherits the bound
247. 247 /// bot's (or workspace default's) voice, gated the same as `route` by
248. 248 /// `inherit_bot_policy`. See `GatewayState::core_for_entry`.
249. 249 #[serde(default, skip_serializing_if = "Option::is_none")]
250. 250 pub voice: Option<vak_config::VoiceConfig>,
251. 251 /// Per-channel permission mode (docs/design/34 "Per-channel permission
252. 252 /// mode"). `None` inherits the target workspace's own configured mode,
253. 253 /// which is the pre-existing behavior and stays the default. `Some(m)`
254. 254 /// pins this channel to `m` — but only ever as a *reduction*: the pool
255. 255 /// caps it to the workspace's own resolved mode, so an override can
256. 256 /// never grant more than a local `vak` run in that workspace has.
257. 257 #[serde(default, skip_serializing_if = "Option::is_none")]
258. 258 pub permission_mode: Option<vak_config::PermissionMode>,
259. 259 /// Per-channel capability restrictions. Each `None` field inherits the
260. 260 /// selected workspace; an explicit empty allow list denies that class.
261. 261 #[serde(default, skip_serializing_if = "is_default_channel_policy")]
262. 262 pub policy: vak_config::ChannelPolicy,
263. 263 /// Which `Bot` this chat is bound to, when the surface has more than
264. 264 /// one. `None` keeps today's behavior (surface's sole/legacy bot).
265. 265 #[serde(default, skip_serializing_if = "Option::is_none")]
266. 266 pub bot_id: Option<String>,
267. 267 /// Whether this chat inherits its bot's policy/permission_mode/route as
268. 268 /// a tier below its own (default) or resolves purely against the
269. 269 /// workspace, ignoring the bot entirely — the explicit "break
270. 270 /// inheritance" switch. Meaningless when `bot_id` is `None`.
271. 271 #[serde(default = "default_true")]
272. 272 pub inherit_bot_policy: bool,
273. 273 pub added_at: String,
274. 274 pub added_by: String,
275. 275 /// Only meaningful while `status == Pending` — the first message text
276. 276 /// that triggered this entry, truncated for operator review.
277. 277 #[serde(default, skip_serializing_if = "Option::is_none")]
278. 278 pub first_seen_text: Option<String>,
279. 279 /// Prompt tier for this chat, the narrowest gateway layer
280. 280 /// (docs/design/45). Gated by `inherit_bot_policy` the same way `voice`
281. 281 /// and `route` are.
282. 282 #[serde(
283. 283 default,
284. 284 skip_serializing_if = "vak_core::prompts::LayerContent::is_empty"
285. 285 )]
286. 286 pub prompt: vak_core::prompts::LayerContent,
287. 287}
288. 288
289. 289pub(crate) fn default_true() -> bool {
290. 290 true
291. 291}
292. 292
293. 293/// Deserializer for a PATCH field shaped `Option<Option<T>>`, where the
294. 294/// three JSON states must stay distinguishable: the key absent ("leave
295. 295/// this alone"), the key present as `null` ("clear it"), and the key
296. 296/// present with a value ("set it"). A plain `Option<Option<T>>` field
297. 297/// cannot do this on its own — serde's derived `deserialize_option` maps
298. 298/// JSON `null` to the *outer* `None`, identical to the key being absent,
299. 299/// so "explicit null clears it" silently never fires
300. 300/// (<https://github.com/serde-rs/serde/issues/984>). Pair with
301. 301/// `#[serde(default, deserialize_with = "deserialize_present")]`: the
302. 302/// `default` only ever applies when the key is missing entirely (serde
303. 303/// skips `deserialize_with` in that case), and this function itself
304. 304/// wraps whatever it sees — including a `null` that becomes `Some(None)`
305. 305/// — in the outer `Some`.
306. 306pub(crate) fn deserialize_present<'de, T, D>(deserializer: D) -> Result<Option<T>, D::Error>
307. 307where
308. 308 T: serde::Deserialize<'de>,
309. 309 D: serde::Deserializer<'de>,
310. 310{
311. 311 T::deserialize(deserializer).map(Some)
312. 312}
313. 313
314. 314fn is_default_channel_policy(policy: &vak_config::ChannelPolicy) -> bool {
315. 315 policy == &vak_config::ChannelPolicy::default()
316. 316}
317. 317
318. 318#[derive(serde::Serialize, serde::Deserialize)]
319. 319struct AllowlistFile {
320. 320 schema: u32,
321. 321 entries: Vec<AllowlistEntry>,
322. 322}
323. 323
324. 324/// A gateway bot identity: one credential/token slot, independently
325. 325/// addressable even when it shares a `surface` with other bots. Sits
326. 326/// between the workspace and a chat's `AllowlistEntry` in the
327. 327/// policy/permission/route resolution chain (`core_for_entry`,
328. 328/// `resolve_channel_permission`) — see `vak_config::ChannelPolicy::merge`
329. 329/// and `vak_config::PermissionMode::capped_by`.
330. 330#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
331. 331pub struct Bot {
332. 332 /// Stable slug, e.g. "telegram-support". Chosen at creation, immutable.
333. 333 pub id: String,
334. 334 /// "telegram" | "discord" | "slack".
335. 335 pub surface: String,
336. 336 /// Operator-facing name shown in the admin console.
337. 337 pub label: String,
338. 338 /// Name of the env var holding this bot's token. The token value
339. 339 /// itself is never stored here or returned by the admin API.
340. 340 pub token_env: String,
341. 341 /// Default agent identity this bot binds to (e.g. "support", "researcher").
342. 342 /// If unset, resolves to "vak" (built-in default agent).
343. 343 #[serde(default, skip_serializing_if = "Option::is_none")]
344. 344 pub agent_id: Option<String>,
345. 345 #[serde(default, skip_serializing_if = "is_default_channel_policy")]
346. 346 pub policy: vak_config::ChannelPolicy,
347. 347 #[serde(default, skip_serializing_if = "Option::is_none")]
348. 348 pub permission_mode: Option<vak_config::PermissionMode>,
349. 349 #[serde(default, skip_serializing_if = "Option::is_none")]
350. 350 pub route: Option<AllowlistRoute>,
351. 351 #[serde(default, skip_serializing_if = "Option::is_none")]
352. 352 pub workspace: Option<PathBuf>,
353. 353 /// Voice/persona override for this bot's spoken replies. `None`
354. 354 /// inherits the workspace default (no voice); `Some` sets this bot's
355. 355 /// tier for any chat that inherits it.
356. 356 #[serde(default, skip_serializing_if = "Option::is_none")]
357. 357 pub voice: Option<vak_config::VoiceConfig>,
358. 358 /// Prompt tier for this bot (docs/design/45). Identity and rules fall
359. 359 /// through to the chat tier below; guardrails concatenate and cannot be
360. 360 /// removed by anything narrower.
361. 361 #[serde(
362. 362 default,
363. 363 skip_serializing_if = "vak_core::prompts::LayerContent::is_empty"
364. 364 )]
365. 365 pub prompt: vak_core::prompts::LayerContent,
366. 366}
367. 367
368. 368#[derive(serde::Serialize, serde::Deserialize, Default)]
369. 369struct BotsFile {
370. 370 schema: u32,
371. 371 bots: Vec<Bot>,
372. 372}
373. 373
374. 374/// Outcome of resolving an inbound key against the allowlist store, so the
375. 375/// caller can distinguish "just became pending" from "still pending" from
376. 376/// a flat denial without re-deriving it from mutable state.
377. 377pub(crate) enum AllowlistDecision {
378. 378 Allowed,
379. 379 Denied,
380. 380 NewlyPending,
381. 381 StillPending,
382. 382}
383. 383
384. 384/// The resolved approval policy (docs/design/22-gateway.md G2), held as one
385. 385/// value so the three fields can never be observed mid-update.
386. 386///
387. 387/// This is behind a lock rather than being plain fields because the policy
388. 388/// is now settable at runtime: it decides whether an `Ask` on a chat
389. 389/// surface reaches a human at all, and an operator who changes it must see
390. 390/// the next inbound message honour the change without restarting the
391. 391/// process. `forward` without a target is not representable — the
392. 392/// constructor and the setter both collapse that case to `deny`, which is
393. 393/// the same rule `vak_config`'s loader applies.
394. 394#[derive(Debug, Clone, PartialEq, Eq)]
395. 395pub(crate) struct ApprovalPolicy {
396. 396 pub(crate) approvals: String,
397. 397 pub(crate) approver: Option<String>,
398. 398 pub(crate) timeout: Duration,
399. 399}
400. 400
401. 401impl ApprovalPolicy {
402. 402 /// Build a policy, collapsing an unbacked `forward` to `deny`.
403. 403 /// A target must carry a `<surface>:<chat>` separator to count.
404. 404 pub(crate) fn resolve(
405. 405 approvals: &str,
406. 406 approver: Option<&str>,
407. 407 timeout: Duration,
408. 408 ) -> ApprovalPolicy {
409. 409 let target = approver
410. 410 .map(str::trim)
411. 411 .filter(|t| !t.is_empty() && t.contains(':'));
412. 412 let forward = approvals == "forward" && target.is_some();
413. 413 ApprovalPolicy {
414. 414 approvals: if forward { "forward" } else { "deny" }.into(),
415. 415 approver: forward.then(|| target.unwrap_or_default().to_string()),
416. 416 timeout,
417. 417 }
418. 418 }
419. 419}
420. 420
421. 421pub struct GatewayState {
422. 422 pub enabled: bool,
423. 423 bindings: Mutex<HashMap<String, ChannelBinding>>,
424. 424 /// Resolved approval policy (docs/design/22-gateway.md G2). Mutable at
425. 425 /// runtime through [`GatewayState::set_approval_policy`].
426. 426 approval_policy: Mutex<ApprovalPolicy>,
427. 427 /// Forwarded gates awaiting a yes/no from the approver surface,
428. 428 /// oldest first (uuidv7 keys sort by insertion time).
429. 429 pending_approvals: Mutex<std::collections::BTreeMap<String, PendingGate>>,
430. 430 chat_allowlist_open: bool,
431. 431 /// Live, schema-versioned allowlist store (docs/design/34). Authoritative
432. 432 /// once it exists on disk; seeded once from `chat_allowlist` otherwise.
433. 433 allowlist: Mutex<HashMap<String, AllowlistEntry>>,
434. 434 /// Bot identities (docs/design/34, multi-bot). Keyed by `Bot::id`.
435. 435 /// Independent of `allowlist`/`bindings` on purpose: several chats can
436. 436 /// share a bot, and a bot can exist with no chats bound to it yet.
437. 437 bots: Mutex<HashMap<String, Bot>>,
438. 438 /// Multi-tenant Core pool (docs/design/34 Phase 2). The gateway's own
439. 439 /// default workspace is the pool's permanent entry; every other
440. 440 /// workspace an allowlist entry names is lazily started here.
441. 441 pub(crate) core_pool: crate::core_pool::CorePool,
442. 442}
443. 443
444. 444#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
445. 445pub(crate) struct ChannelBinding {
446. 446 #[serde(default, skip_serializing_if = "Option::is_none")]
447. 447 pub session_id: Option<String>,
448. 448 #[serde(default, skip_serializing_if = "Option::is_none")]
449. 449 pub provider: Option<String>,
450. 450 #[serde(default, skip_serializing_if = "Option::is_none")]
451. 451 pub model: Option<String>,
452. 452 #[serde(default, skip_serializing_if = "Option::is_none")]
453. 453 pub workspace: Option<PathBuf>,
454. 454 #[serde(default, skip_serializing_if = "Option::is_none")]
455. 455 pub route_revision: Option<String>,
456. 456}
457. 457
458. 458#[derive(serde::Serialize, serde::Deserialize)]
459. 459struct BindingsFile {
460. 460 version: u32,
461. 461 bindings: HashMap<String, ChannelBinding>,
462. 462}
463. 463
464. 464#[derive(serde::Deserialize)]
465. 465#[serde(untagged)]
466. 466enum StoredBindings {
467. 467 Versioned(BindingsFile),
468. 468 Legacy(HashMap<String, String>),
469. 469}
470. 470
471. 471struct PendingGate {
472. 472 session_id: String,
473. 473 tx: oneshot::Sender<bool>,
474. 474}
475. 475
476. 476/// What a resolved gate was, so a bare yes/no is never silent about which
477. 477/// session's tool run it just decided.
478. 478pub(crate) struct ResolvedGate {
479. 479 pub id: String,
480. 480 pub session_id: String,
481. 481 pub remaining: usize,
482. 482}
483. 483
484. 484impl GatewayState {
485. 485 /// Load persisted bindings; `force` overrides the config gate
486. 486 /// (`serve --gateway`).
487. 487 pub fn load(core: &Core, force: bool) -> Self {
488. 488 let mut bindings = HashMap::new();
489. 489 if let Ok(raw) = std::fs::read_to_string(bindings_path(&core.shared_data_home()))
490. 490 && let Ok(stored) = serde_json::from_str::<StoredBindings>(&raw)
491. 491 {
492. 492 bindings = match stored {
493. 493 StoredBindings::Versioned(file) => file.bindings,
494. 494 StoredBindings::Legacy(map) => map
495. 495 .into_iter()
496. 496 .map(|(key, session_id)| {
497. 497 (
498. 498 key,
499. 499 ChannelBinding {
500. 500 session_id: Some(session_id),
501. 501 ..ChannelBinding::default()
502. 502 },
503. 503 )
504. 504 })
505. 505 .collect(),
506. 506 };
507. 507 }
508. 508 let gw = &core.config().gateway;
509. 509
510. 510 // Allowlist store: authoritative once allowlist.json exists; a
511. 511 // one-time import from config.toml's `chat_allowlist` seeds it the
512. 512 // first time a process ever loads (same relationship bindings.json
513. 513 // already has to route overrides — config.toml itself is untouched).
514. 514 let path = allowlist_path(&core.shared_data_home());
515. 515 let allowlist: HashMap<String, AllowlistEntry> = match std::fs::read_to_string(&path) {
516. 516 Ok(raw) => serde_json::from_str::<AllowlistFile>(&raw)
517. 517 .map(|file| {
518. 518 file.entries
519. 519 .into_iter()
520. 520 .map(|e| (e.key.clone(), e))
521. 521 .collect()
522. 522 })
523. 523 .unwrap_or_default(),
524. 524 Err(_) => {
525. 525 let now = chrono::Utc::now().to_rfc3339();
526. 526 let seeded: HashMap<String, AllowlistEntry> = gw
527. 527 .chat_allowlist
528. 528 .iter()
529. 529 .map(|key| {
530. 530 (
531. 531 key.clone(),
532. 532 AllowlistEntry {
533. 533 key: key.clone(),
534. 534 status: AllowlistStatus::Allowed,
535. 535 workspace: None,
536. 536 agent_id: Some("vak".into()),
537. 537 route: None,
538. 538 voice: None,
539. 539 permission_mode: None,
540. 540 policy: vak_config::ChannelPolicy::default(),
541. 541 added_at: now.clone(),
542. 542 added_by: "config_import".into(),
543. 543 first_seen_text: None,
544. 544 prompt: Default::default(),
545. 545 bot_id: None,
546. 546 inherit_bot_policy: true,
547. 547 },
548. 548 )
549. 549 })
550. 550 .collect();
551. 551 if !seeded.is_empty() {
552. 552 write_allowlist_file(&path, &seeded);
553. 553 }
554. 554 seeded
555. 555 }
556. 556 };
557. 557
558. 558 // Bot store. There is no migration from a per-surface token slot:
559. 559 // those are deleted (AGENTS.md invariants 23 and 29), and
560. 560 // synthesizing a bot from one would be exactly the pre-baseline
561. 561 // fold-forward the baseline forbids. A bot is created explicitly,
562. 562 // through setup or the admin console, and owns its own token env.
563. 563 let bots_file_path = bots_path(&core.shared_data_home());
564. 564 let bots: HashMap<String, Bot> = match std::fs::read_to_string(&bots_file_path) {
565. 565 Ok(raw) => serde_json::from_str::<BotsFile>(&raw)
566. 566 .map(|file| file.bots.into_iter().map(|b| (b.id.clone(), b)).collect())
567. 567 .unwrap_or_default(),
568. 568 // No bots.json yet means no bots. Not an error.
569. 569 Err(_) => HashMap::new(),
570. 570 };
571. 571
572. 572 let state = GatewayState {
573. 573 enabled: force || gw.enabled,
574. 574 bindings: Mutex::new(bindings),
575. 575 bots: Mutex::new(bots),
576. 576 approval_policy: Mutex::new(ApprovalPolicy::resolve(
577. 577 &gw.approvals,
578. 578 gw.approver.as_deref(),
579. 579 Duration::from_secs(gw.approval_timeout_secs),
580. 580 )),
581. 581 pending_approvals: Mutex::new(std::collections::BTreeMap::new()),
582. 582 chat_allowlist_open: gw.chat_allowlist_open,
583. 583 allowlist: Mutex::new(allowlist),
584. 584 core_pool: crate::core_pool::CorePool::new(
585. 585 core.clone(),
586. 586 gw.core_pool_max,
587. 587 Duration::from_secs(gw.core_pool_idle_secs),
588. 588 ),
589. 589 };
590. 590 // docs/design/34: a pending request nobody acted on inside the
591. 591 // expiry window auto-denies (visibly, `added_by = "expiry"`).
592. 592 // `vak doctor --repair` does the same thing offline against the
593. 593 // store; doing it here too means a restarted gateway self-heals
594. 594 // and the two paths converge on the same state.
595. 595 let expired = state
596. 596 .allowlist_expire_pending(core, chrono::Duration::days(gw.pending_expiry_days as i64));
597. 597 for key in expired {
598. 598 vak_core::security_events::record(
599. 599 &core.sessions_home(),
600. 600 vak_core::security_events::EventKind::ChatDenied,
601. 601 "chat_denied",
602. 602 &format!("key={key} reason=expiry"),
603. 603 None,
604. 604 );
605. 605 }
606. 606 state
607. 607 }
608. 608
609. 609 /// Resolve the `Core` a channel's entry should actually run through:
610. 610 /// the pool's default entry when the entry has no workspace override or
611. 611 /// names the gateway's own workspace, otherwise the (lazily started)
612. 612 /// pooled `Core` for that workspace. This is the Phase 2 seam that
613. 613 /// makes an allowlist entry's `workspace` field actually run that
614. 614 /// workspace's own sandbox/permission/session state, not just pick its
615. 615 /// provider/model.
616. 616 pub(crate) fn core_for_entry(&self, default_core: &Core, key: &str) -> Result<Core, String> {
617. 617 let entry = self.allowlist_get(key);
618. 618 let allowed_entry = entry
619. 619 .as_ref()
620. 620 .filter(|e| e.status == AllowlistStatus::Allowed);
621. 621 // Bot tier: only consulted when the chat both names a bot and has
622. 622 // not opted out of inheriting from it (`inherit_bot_policy`). A
623. 623 // dangling `bot_id` (removed bot) resolves as "no bot tier", same
624. 624 // as an unset one — never a hard failure at dispatch.
625. 625 let bot = allowed_entry
626. 626 .filter(|e| e.inherit_bot_policy)
627. 627 .and_then(|e| e.bot_id.as_deref())
628. 628 .and_then(|id| self.bot_get(id));
629. 629 let workspace = allowed_entry
630. 630 .and_then(|e| e.workspace.clone())
631. 631 .or_else(|| bot.as_ref().and_then(|b| b.workspace.clone()))
632. 632 .unwrap_or_else(|| default_core.cwd().clone());
633. 633
634. 634 // Policy: bot policy (lower tier) folded under the chat's own
635. 635 // (higher tier) via the same restrictive-only merge used to
636. 636 // reconcile any two policy layers.
637. 637 let chat_policy = allowed_entry.map(|e| e.policy.clone()).unwrap_or_default();
638. 638 let policy = match &bot {
639. 639 Some(b) => vak_config::ChannelPolicy::merge(&b.policy, &chat_policy),
640. 640 None => chat_policy,
641. 641 };
642. 642
643. 643 // Permission mode: chat pin capped by bot mode (itself already
644. 644 // capped by the workspace inside `resolve_at_with_policy`) so a bot
645. 645 // can narrow but never widen what the workspace allows, and a chat
646. 646 // can narrow but never widen what its bot allows.
647. 647 let permission_override = match (allowed_entry.and_then(|e| e.permission_mode), &bot) {
648. 648 (Some(chat_mode), Some(b)) => Some(match b.permission_mode {
649. 649 Some(bot_mode) => chat_mode.capped_by(bot_mode),
650. 650 None => chat_mode,
651. 651 }),
652. 652 (Some(chat_mode), None) => Some(chat_mode),
653. 653 (None, Some(b)) => b.permission_mode,
654. 654 (None, None) => None,
655. 655 };
656. 656
657. 657 let resolved = self.core_pool.resolve_at_with_policy(
658. 658 &workspace,
659. 659 permission_override,
660. 660 policy,
661. 661 std::time::Instant::now(),
662. 662 )?;
663. 663 let selected_agent = allowed_entry
664. 664 .and_then(|entry| {
665. 665 let entry_agent = entry.agent_id.as_deref();
666. 666 let bot_agent = bot.as_ref().and_then(|b| b.agent_id.as_deref());
667. 667 if entry.inherit_bot_policy
668. 668 && (entry_agent.is_none() || entry_agent == Some("vak"))
669. 669 && bot_agent.is_some()
670. 670 {
671. 671 return bot_agent;
672. 672 }
673. 673 entry_agent
674. 674 })
675. 675 .or_else(|| bot.as_ref().and_then(|b| b.agent_id.as_deref()))
676. 676 .unwrap_or("vak");
677. 677 let identity = if selected_agent == "vak" {
678. 678 vak_core::vak_agent_identity()
679. 679 } else {
680. 680 let profiles = crate::agents::effective(&resolved)
681. 681 .map_err(|error| format!("agent catalog unavailable: {error}"))?;
682. 682 profiles
683. 683 .into_iter()
684. 684 .find(|profile| profile.id == selected_agent)
685. 685 .filter(|profile| profile.is_admissible())
686. 686 .map(|profile| profile.identity())
687. 687 .ok_or_else(|| format!("configured Agent '{selected_agent}' is unavailable"))?
688. 688 };
689. 689 Ok(resolved.with_agent_identity(Some(identity)))
690. 690 }
691. 691
692. 692 pub(crate) fn workspace_for_entry(&self, default_core: &Core, key: &str) -> PathBuf {
693. 693 let entry = self.allowlist_get(key);
694. 694 let allowed = entry
695. 695 .as_ref()
696. 696 .filter(|e| e.status == AllowlistStatus::Allowed);
697. 697 let bot = allowed
698. 698 .filter(|e| e.inherit_bot_policy)
699. 699 .and_then(|e| e.bot_id.as_deref())
700. 700 .and_then(|id| self.bot_get(id));
701. 701 allowed
702. 702 .and_then(|e| e.workspace.clone())
703. 703 .or_else(|| bot.and_then(|b| b.workspace))
704. 704 .unwrap_or_else(|| default_core.cwd().to_path_buf())
705. 705 }
706. 706
707. 707 pub(crate) fn workspace_override_for_entry(&self, key: &str) -> Option<PathBuf> {
708. 708 let entry = self.allowlist_get(key)?;
709. 709 if entry.status != AllowlistStatus::Allowed {
710. 710 return None;
711. 711 }
712. 712 let bot = entry
713. 713 .inherit_bot_policy
714. 714 .then_some(entry.bot_id.as_deref())
715. 715 .flatten()
716. 716 .and_then(|id| self.bot_get(id));
717. 717 entry.workspace.or_else(|| bot.and_then(|b| b.workspace))
718. 718 }
719. 719
720. 720 pub(crate) fn set_enabled(&mut self, enabled: bool) {
721. 721 self.enabled = enabled;
722. 722 }
723. 723
724. 724 /// Snapshot of current surface bindings (key→value).
725. 725 pub(crate) fn bindings_snapshot(&self) -> Vec<(String, ChannelBinding)> {
726. 726 self.bindings
727. 727 .lock()
728. 728 .unwrap_or_else(|p| p.into_inner())
729. 729 .iter()
730. 730 .map(|(k, v)| (k.clone(), v.clone()))
731. 731 .collect()
732. 732 }
733. 733
734. 734 /// True when forwarded gates are active.
735. 735 fn approval_policy(&self) -> ApprovalPolicy {
736. 736 self.approval_policy
737. 737 .lock()
738. 738 .unwrap_or_else(std::sync::PoisonError::into_inner)
739. 739 .clone()
740. 740 }
741. 741
742. 742 pub(crate) fn forward_mode(&self) -> bool {
743. 743 let policy = self.approval_policy();
744. 744 self.enabled && policy.approvals == "forward" && policy.approver.is_some()
745. 745 }
746. 746
747. 747 /// The chat that answers forwarded gates. Returns an owned `String`
748. 748 /// rather than a borrow because the policy now lives behind a lock —
749. 749 /// handing out a reference into it would either hold the lock across
750. 750 /// an await or dangle.
751. 751 pub(crate) fn approver_target(&self) -> Option<String> {
752. 752 self.approval_policy().approver
753. 753 }
754. 754
755. 755 pub(crate) fn approval_timeout(&self) -> Duration {
756. 756 self.approval_policy().timeout
757. 757 }
758. 758
759. 759 pub(crate) fn approvals_mode(&self) -> String {
760. 760 self.approval_policy().approvals
761. 761 }
762. 762
763. 763 /// Chats that could serve as the forwarded-approval target, as
764. 764 /// `<surface>:<chat>` delivery addresses.
765. 765 ///
766. 766 /// An allowlist key may be bot-scoped (`telegram:12345:vakyartha`);
767. 767 /// that third segment identifies the bot the message arrived through,
768. 768 /// not a place a reply can be delivered. `deliver_to` addresses are
769. 769 /// two-part, so the key is truncated here rather than at every reader.
770. 770 /// Only `Allowed` entries are offered: forwarding a gate to a pending
771. 771 /// or denied chat would announce it somewhere the operator has
772. 772 /// explicitly not admitted.
773. 773 pub(crate) fn approver_candidates(&self) -> Vec<String> {
774. 774 let mut out: Vec<String> = self
775. 775 .allowlist
776. 776 .lock()
777. 777 .unwrap_or_else(std::sync::PoisonError::into_inner)
778. 778 .values()
779. 779 .filter(|entry| entry.status == AllowlistStatus::Allowed)
780. 780 .filter_map(|entry| {
781. 781 let mut parts = entry.key.splitn(3, ':');
782. 782 match (parts.next(), parts.next()) {
783. 783 (Some(surface), Some(chat)) if !surface.is_empty() && !chat.is_empty() => {
784. 784 Some(format!("{surface}:{chat}"))
785. 785 }
786. 786 _ => None,
787. 787 }
788. 788 })
789. 789 .collect();
790. 790 out.sort();
791. 791 out.dedup();
792. 792 out
793. 793 }
794. 794
795. 795 /// Replace the live approval policy. Returns the policy actually
796. 796 /// installed, which is [`ApprovalPolicy::resolve`]'s answer — asking
797. 797 /// for `forward` with no usable target installs `deny`, so a caller
798. 798 /// can compare and tell the operator their request was reduced instead
799. 799 /// of reporting a success that did not happen.
800. 800 ///
801. 801 /// In-flight forwarded gates are NOT resolved here. They were raised
802. 802 /// under the old policy and already have an announcement sitting in the
803. 803 /// approver's chat; cancelling them would strand a run that a human is
804. 804 /// actively about to answer. Narrowing to `deny` stops the NEXT gate,
805. 805 /// which is the guarantee that matters (nothing new reaches a chat that
806. 806 /// should no longer be asked).
807. 807 pub(crate) fn set_approval_policy(&self, next: ApprovalPolicy) -> ApprovalPolicy {
808. 808 let resolved =
809. 809 ApprovalPolicy::resolve(&next.approvals, next.approver.as_deref(), next.timeout);
810. 810 *self
811. 811 .approval_policy
812. 812 .lock()
813. 813 .unwrap_or_else(std::sync::PoisonError::into_inner) = resolved.clone();
814. 814 resolved
815. 815 }
816. 816
817. 817 /// True when an empty `chat_allowlist` was explicitly opted into
818. 818 /// staying open. Defaults to false: fail closed (0c-02).
819. 819 pub(crate) fn chat_allowlist_open(&self) -> bool {
820. 820 self.chat_allowlist_open
821. 821 }
822. 822
823. 823 pub(crate) fn pending_approval_count(&self) -> usize {
824. 824 self.pending_approvals
825. 825 .lock()
826. 826 .unwrap_or_else(std::sync::PoisonError::into_inner)
827. 827 .len()
828. 828 }
829. 829
830. 830 /// Register a gate for `session_id` and hand back the reply receiver.
831. 831 /// The sender must be stored before the request is announced so an
832. 832 /// instant reply cannot race a missing entry.
833. 833 pub(crate) fn register_gate(&self, id: &str, session_id: &str) -> oneshot::Receiver<bool> {
834. 834 let (tx, rx) = oneshot::channel();
835. 835 self.pending_approvals
836. 836 .lock()
837. 837 .unwrap_or_else(std::sync::PoisonError::into_inner)
838. 838 .insert(
839. 839 id.to_string(),
840. 840 PendingGate {
841. 841 session_id: session_id.to_string(),
842. 842 tx,
843. 843 },
844. 844 );
845. 845 rx
846. 846 }
847. 847
848. 848 /// Resolve a gate. With an id prefix, only that exact gate resolves —
849. 849 /// a reply meant for one session can never approve another's tool run.
850. 850 /// Without one, the globally oldest gate resolves and is reported so
851. 851 /// the approver surface can see what their bare yes/no did.
852. 852 pub(crate) fn resolve_gate(
853. 853 &self,
854. 854 approve: bool,
855. 855 id_prefix: Option<&str>,
856. 856 ) -> Result<ResolvedGate, ()> {
857. 857 let mut map = self
858. 858 .pending_approvals
859. 859 .lock()
860. 860 .unwrap_or_else(std::sync::PoisonError::into_inner);
861. 861 let key = match id_prefix {
862. 862 Some(prefix) => match map.keys().find(|k| k.starts_with(prefix)).cloned() {
863. 863 Some(k) => k,
864. 864 None => return Err(()),
865. 865 },
866. 866 None => map.keys().next().cloned().ok_or(())?,
867. 867 };
868. 868 let (_, gate) = map.remove_entry(&key).ok_or(())?;
869. 869 let _ = gate.tx.send(approve);
870. 870 Ok(ResolvedGate {
871. 871 id: key,
872. 872 session_id: gate.session_id,
873. 873 remaining: map.len(),
874. 874 })
875. 875 }
876. 876
877. 877 /// Reject forwarded approval gates belonging to a revoked session. A
878. 878 /// late reply then finds no gate and cannot authorize stale work.
879. 879 pub(crate) fn deny_pending_for_session(&self, session_id: &str) -> usize {
880. 880 let mut pending = self
881. 881 .pending_approvals
882. 882 .lock()
883. 883 .unwrap_or_else(std::sync::PoisonError::into_inner);
884. 884 let keys: Vec<String> = pending
885. 885 .iter()
886. 886 .filter(|(_, gate)| gate.session_id == session_id)
887. 887 .map(|(id, _)| id.clone())
888. 888 .collect();
889. 889 let mut denied = 0;
890. 890 for key in keys {
891. 891 if let Some(gate) = pending.remove(&key) {
892. 892 let _ = gate.tx.send(false);
893. 893 denied += 1;
894. 894 }
895. 895 }
896. 896 denied
897. 897 }
898. 898
899. 899 pub(crate) fn snapshot(&self) -> Vec<(String, ChannelBinding)> {
900. 900 let mut pairs: Vec<(String, ChannelBinding)> = self
901. 901 .bindings
902. 902 .lock()
903. 903 .unwrap_or_else(std::sync::PoisonError::into_inner)
904. 904 .iter()
905. 905 .map(|(k, v)| (k.clone(), v.clone()))
906. 906 .collect();
907. 907 pairs.sort_by(|a, b| a.0.cmp(&b.0));
908. 908 pairs
909. 909 }
910. 910
911. 911 fn bind(&self, core: &Core, key: String, session_id: String, revision: String) {
912. 912 let mut bindings = self
913. 913 .bindings
914. 914 .lock()
915. 915 .unwrap_or_else(std::sync::PoisonError::into_inner);
916. 916 let binding = bindings.entry(key).or_default();
917. 917 binding.session_id = Some(session_id);
918. 918 binding.workspace = Some(core.cwd().clone());
919. 919 binding.route_revision = Some(revision);
920. 920 drop(bindings);
921. 921 persist_bindings(core, self);
922. 922 }
923. 923
924. 924 pub(crate) fn set_route_override(
925. 925 &self,
926. 926 core: &Core,
927. 927 key: String,
928. 928 route: Option<(String, String)>,
929. 929 ) {
930. 930 let mut bindings = self
931. 931 .bindings
932. 932 .lock()
933. 933 .unwrap_or_else(std::sync::PoisonError::into_inner);
934. 934 let binding = bindings.entry(key).or_default();
935. 935 match route {
936. 936 Some((provider, model)) => {
937. 937 binding.provider = Some(provider);
938. 938 binding.model = Some(model);
939. 939 }
940. 940 None => {
941. 941 binding.provider = None;
942. 942 binding.model = None;
943. 943 }
944. 944 }
945. 945 binding.route_revision = None;
946. 946 drop(bindings);
947. 947 persist_bindings(core, self);
948. 948 }
949. 949
950. 950 pub(crate) fn rotate(&self, core: &Core, key: &str) -> bool {
951. 951 let mut bindings = self
952. 952 .bindings
953. 953 .lock()
954. 954 .unwrap_or_else(std::sync::PoisonError::into_inner);
955. 955 let Some(binding) = bindings.get_mut(key) else {
956. 956 return false;
957. 957 };
958. 958 binding.session_id = None;
959. 959 binding.route_revision = None;
960. 960 drop(bindings);
961. 961 persist_bindings(core, self);
962. 962 true
963. 963 }
964. 964
965. 965 pub(crate) fn unbind(&self, core: &Core, key: &str) -> bool {
966. 966 let removed = self
967. 967 .bindings
968. 968 .lock()
969. 969 .unwrap_or_else(std::sync::PoisonError::into_inner)
970. 970 .remove(key)
971. 971 .is_some();
972. 972 if removed {
973. 973 persist_bindings(core, self);
974. 974 }
975. 975 removed
976. 976 }
977. 977
978. 978 // ---- Bot store (multi-bot-per-channel) ---------------------------------
979. 979
980. 980 /// Snapshot of all bots, sorted by id. Secrets never included — a `Bot`
981. 981 /// row only ever holds the env var *name*, not the token value.
982. 982 pub(crate) fn bots_snapshot(&self) -> Vec<Bot> {
983. 983 let mut bots: Vec<Bot> = self
984. 984 .bots
985. 985 .lock()
986. 986 .unwrap_or_else(std::sync::PoisonError::into_inner)
987. 987 .values()
988. 988 .cloned()
989. 989 .collect();
990. 990 bots.sort_by(|a, b| a.id.cmp(&b.id));
991. 991 bots
992. 992 }
993. 993
994. 994 pub(crate) fn bot_get(&self, id: &str) -> Option<Bot> {
995. 995 self.bots
996. 996 .lock()
997. 997 .unwrap_or_else(std::sync::PoisonError::into_inner)
998. 998 .get(id)
999. 999 .cloned()
1000. 1000 }