Vakyartha documentation

1425 {
1426 out.push(vak_core::prompts::LayerInput::new(
1427 vak_core::prompts::PromptLayer::Bot,
1428 Some(format!("bot:{}", bot.id)),
1429 bot.prompt,
1430 ));
1431 }
1432 if !entry.prompt.is_empty() {
1433 out.push(vak_core::prompts::LayerInput::new(
1434 vak_core::prompts::PromptLayer::Chat,
1435 Some(format!("chat:{key}")),
1436 entry.prompt,
1437 ));
1438 }
1439 out
1440 }
1441
1442 /// The style directive for spoken replies on this chat.
1443 ///
1444 /// One source of truth (docs/design/45-prompt-layers.md): the bot/chat
1445 /// `identity` block *is* the persona. `VoiceConfig.persona` predates
1446 /// prompt layers and said the same thing in a second place; keeping both
1447 /// authoritative would let a bot's spoken and written selves drift apart
1448 /// within a release. The legacy field is still honoured when no prompt
1449 /// tier sets an identity, so existing configs keep working untouched.
1450 ///
1451 /// Only the *gateway tiers'* own identity text is used, never the
1452 /// assembled prompt — the seed identity and capability contract are
1453 /// meaningless as a text-to-speech style directive.
1454 pub(crate) fn resolve_persona(&self, key: &str) -> Option<String> {
1455 self.resolve_prompt_overlays(key)
1456 .into_iter()
1457 .rev()
1458 .find_map(|layer| layer.content.identity)
1459 .map(|text| text.trim().to_string())
1460 .filter(|text| !text.is_empty())
1461 .or_else(|| {
1462 self.resolve_voice(key)
1463 .and_then(|voice| voice.persona)
1464 .map(|p| p.trim().to_string())
1465 .filter(|p| !p.is_empty())
1466 })
1467 }
1468
1469 pub(crate) fn resolve_voice(&self, key: &str) -> Option<vak_config::VoiceConfig> {
1470 let entry = self
1471 .allowlist_get(key)
1472 .filter(|e| e.status == AllowlistStatus::Allowed);
1473 let parent = entry
1474 .as_ref()
1475 .filter(|e| e.inherit_bot_policy)
1476 .and_then(|e| e.bot_id.as_deref())
1477 .and_then(|id| self.bot_get(id))
1478 .and_then(|b| b.voice);
1479 entry
1480 .as_ref()
1481 .and_then(|e| e.voice.as_ref())
1482 .map(|v| vak_config::VoiceConfig::overlay(parent.as_ref(), v))
1483 .or(parent)
1484 }
1485
1486 /// Drop the cached route revision for `key` without dropping the
1487 /// session id — exactly what `set_route_override` does — so the next
1488 /// inbound message re-derives the effective route and rotates the
1489 /// frozen session if (and only if) it actually changed.
1490 pub(crate) fn invalidate_binding_revision(&self, core: &Core, key: &str) {
1491 let touched = {
1492 let mut bindings = self
1493 .bindings
1494 .lock()
1495 .unwrap_or_else(std::sync::PoisonError::into_inner);
1496 match bindings.get_mut(key) {
1497 Some(binding) => {
1498 binding.route_revision = None;
1499 true
1500 }
1501 None => false,
1502 }
1503 };
1504 if touched {
1505 persist_bindings(core, self);
1506 }
1507 }
1508
1509 pub(crate) fn invalidate_bindings_for_bot(&self, core: &Core, bot_id: &str) {
1510 let keys: Vec<String> = {
1511 let allowlist = self
1512 .allowlist
1513 .lock()
1514 .unwrap_or_else(std::sync::PoisonError::into_inner);
1515 allowlist
1516 .values()
1517 .filter(|entry| entry.bot_id.as_deref() == Some(bot_id))
1518 .map(|entry| entry.key.clone())
1519 .collect()
1520 };
1521 let touched = {
1522 let mut bindings = self
1523 .bindings
1524 .lock()
1525 .unwrap_or_else(std::sync::PoisonError::into_inner);
1526 let mut changed = false;
1527 for key in keys {
1528 if let Some(binding) = bindings.get_mut(&key) {
1529 binding.route_revision = None;
1530 changed = true;
1531 }
1532 }
1533 changed
1534 };
1535 if touched {
1536 persist_bindings(core, self);
1537 }
1538 }
1539
1540 /// Test seam: plant a bound session with a frozen route revision, the
1541 /// state dispatch leaves behind, without running a whole turn.
1542 #[cfg(test)]
1543 pub(crate) fn bind_for_test(&self, key: &str, session_id: &str, revision: &str) {
1544 let mut bindings = self
1545 .bindings
1546 .lock()
1547 .unwrap_or_else(std::sync::PoisonError::into_inner);
1548 let binding = bindings.entry(key.to_string()).or_default();
1549 binding.session_id = Some(session_id.to_string());
1550 binding.route_revision = Some(revision.to_string());
1551 }
1552
1553 #[cfg(test)]
1554 pub(crate) fn route_revision_for_test(&self, key: &str) -> Option<String> {
1555 self.bindings
1556 .lock()
1557 .unwrap_or_else(std::sync::PoisonError::into_inner)
1558 .get(key)
1559 .and_then(|b| b.route_revision.clone())
1560 }
1561
1562 #[cfg(test)]
1563 pub(crate) fn session_id_for_test(&self, key: &str) -> Option<String> {
1564 self.bindings
1565 .lock()
1566 .unwrap_or_else(std::sync::PoisonError::into_inner)
1567 .get(key)
1568 .and_then(|b| b.session_id.clone())
1569 }
1570
1571 /// Revoke an allowed entry: removes it from the store entirely (a
1572 /// future message from that key starts a fresh pending review, not a
1573 /// stale "denied" record masquerading as an audit trail).
1574 pub(crate) fn allowlist_revoke(&self, core: &Core, key: &str) -> bool {
1575 if self.allowlist_get(key).map(|e| e.status) != Some(AllowlistStatus::Allowed) {
1576 return false;
1577 }
1578 let removed = self
1579 .allowlist
1580 .lock()
1581 .unwrap_or_else(std::sync::PoisonError::into_inner)
1582 .remove(key)
1583 .is_some();
1584 if removed {
1585 persist_allowlist(core, self);
1586 }
1587 removed
1588 }
1589}
1590
1591/// What a channel's permission mode actually resolves to, for display and
1592/// for the approve/patch audit trail.
1593pub(crate) struct ResolvedPermission {
1594 /// The target workspace's own configured mode — the outermost ceiling.
1595 pub workspace_mode: vak_config::PermissionMode,
1596 /// The bot tier's pin, when the chat inherits from a bot that has one.
1597 /// `None` means no bot tier applies, not "the bot allows everything".
1598 pub bot_mode: Option<vak_config::PermissionMode>,
1599 /// What the entry asked for, if anything.
1600 pub requested: Option<vak_config::PermissionMode>,
1601 /// What the channel actually gets, folded the same way `core_for_entry`
1602 /// folds it: chat pin capped by bot pin, then capped by the workspace.
1603 pub effective: vak_config::PermissionMode,
1604}
1605
1606impl ResolvedPermission {
1607 /// True when a pin asked for more than the layers above it allow and
1608 /// was reduced. This is the condition worth an audit-log entry.
1609 ///
1610 /// It covers the bot tier too: a bot pinned narrower than its chat
1611 /// reduces that chat just as surely as the workspace does, and reporting
1612 /// only the workspace cap is what let the console show a chat as wider
1613 /// than it actually ran.
1614 pub fn was_capped(&self) -> bool {
1615 matches!(self.requested, Some(r) if r != self.effective)
1616 || matches!(
1617 (self.bot_mode, self.requested),
1618 (Some(bot), None) if bot != self.effective
1619 )
1620 }
1621}
1622
1623/// Read-only mirror of the cap dispatch enforces, for the admin surface.
1624/// Shares `PermissionMode::capped_by` with the pool so the number the
1625/// console shows is derived the same way as the one dispatch pins.
1626///
1627/// It must fold the SAME three tiers `GatewayState::core_for_entry` folds:
1628/// chat pin capped by bot pin, then capped by the workspace. Leaving the
1629/// bot tier out — as this did — meant a bot pinned to `read-only` under a
1630/// chat pinned to `workspace-write` ran read-only and was reported as
1631/// workspace-write, and a chat with no pin under a bot that had one was
1632/// reported as the workspace's mode instead of the bot's. Showing a channel
1633/// as wider than it runs is the one direction of error that matters here.
1634///
1635/// A workspace whose config fails to load falls back to the compiled
1636/// default (`WorkspaceWrite`), matching `vak_config`'s own layering; the
1637/// pool remains the authority at dispatch either way.
1638pub(crate) fn resolve_channel_permission(
1639 workspace: &std::path::Path,
1640 requested: Option<vak_config::PermissionMode>,
1641 bot_mode: Option<vak_config::PermissionMode>,
1642) -> ResolvedPermission {
1643 // Same trust the pool will use at dispatch. Reading with `true` here
1644 // while the pool read the marker store would put the console back to
1645 // reporting a mode no run would get.
1646 let workspace_mode =
1647 vak_config::load_with_trust(workspace, vak_core::trust::is_trusted(workspace))
1648 .map(|c| c.permission_mode)
1649 .unwrap_or_default();
1650 // Exactly `core_for_entry`'s fold, then the pool's workspace cap.
1651 let pinned = match (requested, bot_mode) {
1652 (Some(chat), Some(bot)) => Some(chat.capped_by(bot)),
1653 (Some(chat), None) => Some(chat),
1654 (None, bot) => bot,
1655 };
1656 let effective = match pinned {
1657 Some(mode) => mode.capped_by(workspace_mode),
1658 None => workspace_mode,
1659 };
1660 ResolvedPermission {
1661 workspace_mode,
1662 bot_mode,
1663 requested,
1664 effective,
1665 }
1666}
1667
1668/// True when `added_at` parses as an RFC3339 stamp strictly older than
1669/// `cutoff`. An unparseable stamp is never treated as expired: a corrupt
1670/// timestamp must not silently auto-deny a live channel.
1671fn parse_added_at_before(added_at: &str, cutoff: chrono::DateTime<chrono::Utc>) -> bool {
1672 chrono::DateTime::parse_from_rfc3339(added_at)
1673 .map(|ts| ts.with_timezone(&chrono::Utc) < cutoff)
1674 .unwrap_or(false)
1675}
1676
1677fn persist_bindings(core: &Core, gw: &GatewayState) {
1678 let bindings = gw
1679 .bindings
1680 .lock()
1681 .unwrap_or_else(std::sync::PoisonError::into_inner)
1682 .clone();
1683 let path = bindings_path(&core.shared_data_home());
1684 if let Some(parent) = path.parent() {
1685 let _ = std::fs::create_dir_all(parent);
1686 }
1687 let file = BindingsFile {
1688 version: 2,
1689 bindings,
1690 };
1691 if let Ok(json) = serde_json::to_string_pretty(&file) {
1692 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1693 if std::fs::write(&temp, json).is_ok() {
1694 let _ = std::fs::rename(temp, path);
1695 }
1696 }
1697}
1698
1699fn persist_allowlist(core: &Core, gw: &GatewayState) {
1700 let entries = gw
1701 .allowlist
1702 .lock()
1703 .unwrap_or_else(std::sync::PoisonError::into_inner)
1704 .clone();
1705 let path = allowlist_path(&core.shared_data_home());
1706 write_allowlist_file(&path, &entries);
1707}
1708
1709fn persist_bots(core: &Core, gw: &GatewayState) {
1710 let bots = gw
1711 .bots
1712 .lock()
1713 .unwrap_or_else(std::sync::PoisonError::into_inner)
1714 .clone();
1715 persist_bots_map(&bots_path(&core.shared_data_home()), &bots);
1716}
1717
1718/// Atomic temp-file+rename write, same pattern as `write_allowlist_file`.
1719/// Free function (not a `GatewayState` method) so the one-time migration in
1720/// `GatewayState::load` can call it before a `GatewayState` exists.
1721fn persist_bots_map(path: &std::path::Path, bots: &HashMap<String, Bot>) {
1722 if let Some(parent) = path.parent() {
1723 let _ = std::fs::create_dir_all(parent);
1724 }
1725 let mut sorted: Vec<Bot> = bots.values().cloned().collect();
1726 sorted.sort_by(|a, b| a.id.cmp(&b.id));
1727 let file = BotsFile {
1728 schema: 1,
1729 bots: sorted,
1730 };
1731 if let Ok(json) = serde_json::to_string_pretty(&file) {
1732 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1733 if std::fs::write(&temp, json).is_ok() {
1734 let _ = std::fs::rename(temp, path);
1735 }
1736 }
1737}
1738
1739/// Atomic temp-file+rename write, same pattern as `persist_bindings`.
1740fn write_allowlist_file(path: &std::path::Path, entries: &HashMap<String, AllowlistEntry>) {
1741 if let Some(parent) = path.parent() {
1742 let _ = std::fs::create_dir_all(parent);
1743 }
1744 let mut sorted: Vec<AllowlistEntry> = entries.values().cloned().collect();
1745 sorted.sort_by(|a, b| a.key.cmp(&b.key));
1746 let file = AllowlistFile {
1747 schema: 1,
1748 entries: sorted,
1749 };
1750 if let Ok(json) = serde_json::to_string_pretty(&file) {
1751 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1752 if std::fs::write(&temp, json).is_ok() {
1753 let _ = std::fs::rename(temp, path);
1754 }
1755 }
1756}
1757
1758pub fn routes() -> Router<AppState> {
1759 Router::new()
1760 .route(
1761 "/gateway/inbound",
1762 axum::routing::post(gateway_inbound)
1763 .layer(axum::extract::DefaultBodyLimit::max(INBOUND_BODY_MAX_BYTES)),
1764 )
1765 .route("/gateway/status", axum::routing::get(gateway_status))
1766 .route(
1767 "/gateway/bindings/{key}",
1768 axum::routing::delete(gateway_unbind),
1769 )
1770}
1771
1772// ---- Inbound ----------------------------------------------------------------
1773
1774#[derive(serde::Deserialize)]
1775struct InboundBody {
1776 surface: String,
1777 chat: String,
1778 /// Who sent this, when the adapter knows (a Telegram @user, a webhook
1779 /// identity). Typed since G1 groundwork: it is recorded on deliveries
1780 /// and attributed on queued turns instead of being silently dropped.
1781 #[serde(default)]
1782 sender: Option<String>,
1783 text: String,
1784 #[serde(default)]
1785 wait: bool,
1786 /// Base64 images appended to the prompt as vision content.
1787 #[serde(default)]
1788 attachments: Vec<InboundAttachment>,
1789 #[serde(default)]
1790 capabilities: Option<crate::delivery::RequestedCapabilities>,
1791 /// See [`InboundRequest::bot_id`].
1792 #[serde(default)]
1793 bot_id: Option<String>,
1794 /// Caller-provided idempotency key. Repeating it returns the original
1795 /// admission without dispatching a second model turn.
1796 #[serde(default)]
1797 request_id: Option<String>,
1798}
1799
1800#[derive(serde::Deserialize)]
1801struct InboundAttachment {
1802 /// MIME type; defaults to image/png for Telegram-style senders.
1803 #[serde(default = "default_image_mime")]
1804 mime: String,
1805 data: String,
1806 /// Original filename, when the channel knows it (documents only).
1807 #[serde(default)]
1808 filename: Option<String>,
1809 /// "image" (default, vision content) or "document": small text is
1810 /// inlined, anything else is saved to the workspace inbox and named.
1811 #[serde(default = "default_attachment_kind")]
1812 kind: String,
1813 #[serde(default)]
1814 error: Option<String>,
1815}
1816
1817fn default_image_mime() -> String {
1818 "image/png".into()
1819}
1820
1821fn default_attachment_kind() -> String {
1822 "image".into()
1823}
1824
1825/// The largest document a channel may hand the gateway, shared by every
1826/// bridge so a file is never downloaded by one side and dropped by the
1827/// other. The gateway's JSON body limit (2 MiB, base64 inflates by a third)
1828/// bounds it.
1829pub(crate) const INBOUND_DOCUMENT_MAX_BYTES: usize = 20 * 1024 * 1024;
1830
1831/// `/gateway/inbound`'s body limit: a document at the limit, base64-encoded,
1832/// with room for the rest of the request.
1833const INBOUND_BODY_MAX_BYTES: usize = INBOUND_DOCUMENT_MAX_BYTES / 3 * 4 + 1024 * 1024;
1834
1835/// Text documents at or under this size are inlined in the prompt; larger
1836/// text, and every non-text file, is saved to the workspace inbox instead.
1837const DOCUMENT_INLINE_MAX_BYTES: usize = 64 * 1024;
1838
1839/// What a channel turn answers: its text, and each workspace file the turn
1840/// drafted with `office_apply`, for the channel to get back.
1841pub(crate) struct ChatReply {
1842 pub(crate) text: String,
1843 pub(crate) drafts: Vec<TurnDraft>,
1844}
1845
1846/// The latest draft this turn made of one workspace file.
1847pub(crate) struct TurnDraft {
1848 /// Workspace-relative, as the call named it.
1849 path: String,
1850 draft: std::path::PathBuf,
1851}
1852
1853/// Each workspace file this turn's successful `office_apply` calls drafted,
1854/// with the last draft of it: an `office_apply` call's draft is at
1855/// `.vak/scratch/<agent>/<call id>/<path>`, the convention Review's lineage
1856/// relies on too.
1857fn turn_drafts(log: &vak_session::SessionLog, workspace: &std::path::Path) -> Vec<TurnDraft> {
1858 let Some(directive) = log.latest_directive_entry_id() else {
1859 return Vec::new();
1860 };
1861 let agent = log
1862 .header()
1863 .and_then(|header| header.agent.as_ref().map(|agent| agent.id.clone()))
1864 .unwrap_or_else(|| "vak".into());
1865 let mut in_turn = false;
1866 let mut calls: Vec<(String, String)> = Vec::new();
1867 let mut succeeded = std::collections::HashSet::new();
1868 for (entry_id, message) in log.message_chain() {
1869 in_turn |= entry_id == directive;
1870 if !in_turn {
1871 continue;
1872 }
1873 for block in &message.content {
1874 match block {
1875 vak_llm::ContentBlock::ToolUse { id, name, input } if name == "office_apply" => {
1876 if let Some(path) = input.get("path").and_then(serde_json::Value::as_str) {
1877 calls.push((id.clone(), path.trim().to_string()));
1878 }
1879 }
1880 vak_llm::ContentBlock::ToolResult {
1881 tool_use_id,
1882 is_error: false,
1883 ..
1884 } => {
1885 succeeded.insert(tool_use_id.clone());
1886 }
1887 _ => {}
1888 }
1889 }
1890 }
1891 let mut drafts: Vec<TurnDraft> = Vec::new();
1892 for (id, path) in calls.into_iter().filter(|(id, _)| succeeded.contains(id)) {
1893 let draft = workspace
1894 .join(".vak")
1895 .join("scratch")
1896 .join(&agent)
1897 .join(&id)
1898 .join(&path);
1899 if !draft.is_file() {
1900 continue;
1901 }
1902 drafts.retain(|earlier| earlier.path != path);
1903 drafts.push(TurnDraft { path, draft });
1904 }
1905 drafts
1906}
1907
1908/// The largest draft sent back on a channel; Telegram's bots may send up to
1909/// 50 MB, and a document this large is better opened in Vakyartha.
1910const RETURN_FILE_MAX_BYTES: u64 = 20 * 1024 * 1024;
1911
1912/// Each draft the turn made, for a channel that takes files: its bytes and a
1913/// caption saying what changed (from the worker's semantic diff), under the
1914/// name the person knows it by. A draft that carries a sensitivity label is
1915/// not sent (labels only narrow where a file goes); a channel that takes no
1916/// files, or a draft too large, gets a line saying where the file is.
1917async fn return_drafts(
1918 core: &Core,
1919 mut text: String,
1920 drafts: &[TurnDraft],
1921 accepts_files: bool,
1922) -> (String, Vec<serde_json::Value>) {
1923 use base64::Engine as _;
1924 let worker = core.tool_worker_exe();
1925 let mut files = Vec::new();
1926 let mut notes = Vec::new();
1927 for draft in drafts {
1928 let name = inbox::display_name(
1929 std::path::Path::new(&draft.path)
1930 .file_name()
1931 .and_then(|name| name.to_str())
1932 .unwrap_or(&draft.path),
1933 );
1934 let facts = vak_tools::broker::office_project(
1935 &worker,
1936 &draft.draft,
1937 vak_tools::broker::OfficeView::Facts,
1938 )
1939 .await;
1940 let labels: Vec<String> = facts
1941 .as_ref()
1942 .ok()
1943 .and_then(|facts| facts.get("sensitivity_labels"))
1944 .and_then(|labels| serde_json::from_value(labels.clone()).ok())
1945 .unwrap_or_default();
1946 if !labels.is_empty() {
1947 notes.push(format!(
1948 "{name} carries the sensitivity label {}, so it is not sent on this channel; review the draft in Vakyartha.",
1949 labels.join(", ")
1950 ));
1951 continue;
1952 }
1953 let size = std::fs::metadata(&draft.draft)
1954 .map(|metadata| metadata.len())
1955 .unwrap_or(u64::MAX);
1956 if !accepts_files || size > RETURN_FILE_MAX_BYTES {
1957 notes.push(format!(
1958 "The updated {name} is ready in Vakyartha for review; this channel does not receive it."
1959 ));
1960 continue;
1961 }
1962 let current = core.cwd().join(&draft.path);
1963 let review = vak_tools::broker::office_review(
1964 &worker,
1965 current.is_file().then_some(current.as_path()),
1966 &draft.draft,
1967 None,
1968 )
1969 .await;
1970 let summary = review
1971 .as_ref()
1972 .ok()
1973 .and_then(|review| review.get("summary"))
1974 .and_then(|summary| serde_json::from_value::<Vec<String>>(summary.clone()).ok())
1975 .filter(|summary| !summary.is_empty())
1976 .map(|summary| summary.join("; "))
1977 .unwrap_or_else(|| "no visible change".into());
1978 let Ok(bytes) = std::fs::read(&draft.draft) else {
1979 notes.push(format!("The updated {name} could not be read to send it."));
1980 continue;
1981 };
1982 files.push(serde_json::json!({
1983 "name": name,
1984 "mime": office_mime(&name),
1985 "data": base64::engine::general_purpose::STANDARD.encode(bytes),
1986 "caption": format!("Updated {name}: {summary}"),
1987 }));
1988 }
1989 if !notes.is_empty() {
1990 text = format!("{}\n\n{}", text.trim_end(), notes.join("\n"));
1991 }
1992 (text, files)
1993}
1994
1995fn office_mime(name: &str) -> &'static str {
1996 match name
1997 .rsplit('.')
1998 .next()
1999 .map(str::to_ascii_lowercase)
2000 .as_deref()
2001 {
2002 Some("docx") => "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
2003 Some("xlsx") => "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
2004 Some("pptx") => "application/vnd.openxmlformats-officedocument.presentationml.presentation",
2005 _ => "application/octet-stream",
2006 }
2007}
2008
2009/// Compose the prompt message: text, vision blocks, inlined text documents,
2010/// and a note for each document saved to the inbox. A document's bytes
2011/// never enter the prompt unless they are text (docs/design/72, F1). The
2012/// ledger stores exactly what the model will see (invariant 1).
2013fn compose_prompt(
2014 text: &str,
2015 attachments: &[InboundAttachment],
2016 workspace: &std::path::Path,
2017) -> vak_llm::Message {
2018 let mut blocks = Vec::new();
2019 if !text.is_empty() {
2020 blocks.push(vak_llm::ContentBlock::text(text));
2021 }
2022 for a in attachments {
2023 if a.data.trim().is_empty() {
2024 continue;
2025 }
2026 if a.kind == "document" {
2027 blocks.push(vak_llm::ContentBlock::text(document_block(a, workspace)));
2028 continue;
2029 }
2030 // A voice note reaches the model as its transcript (or the reason
2031 // there is none) in the message text, never as an attachment block.
2032 if a.kind == "audio" {
2033 continue;
2034 }
2035 blocks.push(vak_llm::ContentBlock::image_base64(
2036 a.mime.clone(),
2037 a.data.trim().to_string(),
2038 ));
2039 }
2040 vak_llm::Message {
2041 role: vak_llm::Role::User,
2042 content: blocks,
2043 }
2044}
2045
2046fn document_block(attachment: &InboundAttachment, workspace: &std::path::Path) -> String {
2047 use base64::Engine as _;
2048 let filename = attachment.filename.as_deref().unwrap_or("file");
2049 let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(attachment.data.trim()) else {
2050 return format!("[attached file '{filename}' could not be decoded; not included]");
2051 };
2052 if bytes.len() > INBOUND_DOCUMENT_MAX_BYTES {
2053 return format!(
2054 "[attached file '{filename}' ({} KiB) exceeds the {} KiB channel limit; not received]",
2055 bytes.len() / 1024,
2056 INBOUND_DOCUMENT_MAX_BYTES / 1024
2057 );
2058 }
2059 let text = std::str::from_utf8(&bytes)
2060 .ok()
2061 .filter(|text| !text.contains('\0'));
2062 if let Some(content) = text
2063 && bytes.len() <= DOCUMENT_INLINE_MAX_BYTES
2064 {
2065 return format!("Attached file `{filename}`:\n```\n{content}\n```");
2066 }
2067 match save_to_inbox(workspace, filename, &bytes) {
2068 Ok(saved) => inbox::note(filename, &saved, &bytes),
2069 Err(error) => {
2070 format!("[attached file '{filename}' could not be saved ({error}); not included]")
2071 }
2072 }
2073}
2074
2075pub(crate) fn compose_voice_prompt(text: &str) -> vak_llm::Message {
2076 compose_prompt(text, &[], std::path::Path::new("."))
2077}
2078
2079// ---- Approval forwarding (G2) ----------------------------------------------
2080
2081/// Approver for gateway-driven turns. In `deny` mode this behaves like
2082/// `AutoDeny`. In `forward` mode the gate is announced on the approver
2083/// surface and resolved by a yes/no reply; timeout or silence fails closed.
2084struct GatewayApprover {
2085 events_tx: crate::events::EventBus,
2086 state: Arc<GatewayState>,
2087 core: Core,
2088 session_id: String,
2089}
2090
2091#[async_trait::async_trait]
2092impl vak_agent::Approver for GatewayApprover {
2093 /// A forwarded gate reaches the configured approver chat; without
2094 /// forward mode nothing is listening, and the reachability preflight
2095 /// must see that before the prompt advertises a gated capability.
2096 fn answerable(&self) -> bool {
2097 self.state.forward_mode()
2098 }
2099
2100 async fn approve(&self, tool: &str, args_json: &str, reason: &str) -> bool {
2101 if !self.state.forward_mode() {
2102 return false;
2103 }
2104 let id = uuid::Uuid::now_v7().to_string();
2105 let rx = self.state.register_gate(&id, &self.session_id);
2106 let _ = self.events_tx.send(AgentEvent::ApprovalRequested {
2107 id: id.clone(),
2108 tool: tool.to_string(),
2109 args_json: args_json.to_string(),
2110 reason: reason.to_string(),
2111 });
2112 let short = &id[..8];
2113 let announce = format!(
2114 "Approval requested [{short}]\nTool: {tool}\nArgs: {args_json}\nReason: {reason}\nReply 'yes' or 'no' to decide."
2115 );
2116 if let Err(e) = deliver_approval_and_record(
2117 &self.core,
2118 self.state.approver_target().unwrap_or_default().as_str(),
2119 ApprovalPayload {
2120 request_id: id.clone(),
2121 title: format!("Approval requested [{short}]"),
2122 detail: announce.clone(),
2123 expires_at: Some(
2124 (chrono::Utc::now()
2125 + chrono::Duration::from_std(self.state.approval_timeout())
2126 .unwrap_or_default())
2127 .to_rfc3339(),
2128 ),
2129 actions: vec![
2130 delivery_action("approve", "Approve", "approve", &id),
2131 delivery_action("deny", "Deny", "deny", &id),
2132 ],
2133 },
2134 vak_core::inbox::Kind::ApprovalPending,
2135 format!("Approval requested [{short}]"),
2136 Some(&self.session_id),
2137 None,
2138 )
2139 .await
2140 {
2141 eprintln!("[gateway] approval announcement failed: {e}");
2142 let _ = self.state.resolve_gate(false, Some(&id));
2143 return false;
2144 }
2145 match tokio::time::timeout(self.state.approval_timeout(), rx).await {
2146 Ok(Ok(v)) => v,
2147 Ok(Err(_)) => false, // gate dropped (run cancelled)
2148 Err(_) => {
2149 // Timed out: remove our own entry so a late reply resolves
2150 // nothing.
2151 self.state
2152 .pending_approvals
2153 .lock()
2154 .unwrap_or_else(std::sync::PoisonError::into_inner)
2155 .remove(&id);
2156 eprintln!("[gateway] approval {short} timed out; denied");
2157 false
2158 }
2159 }
2160 }
2161}
2162
2163/// "yes"/"no" vocabulary for chat replies, optionally addressed to one
2164/// gate: "yes ab12cd34". Deliberately small and strict — casual chatter
2165/// from the approver chat must not resolve gates. Returns the verdict and
2166/// the gate-id prefix when one was supplied.
2167fn parse_verdict(text: &str) -> Option<(bool, Option<String>)> {
2168 let mut tokens = text.split_whitespace();
2169 let head = tokens.next()?.to_lowercase();
2170 let verdict = match head.as_str() {
2171 "y" | "yes" | "approve" | "approved" | "ok" | "allow" => true,
2172 "n" | "no" | "deny" | "denied" | "block" => false,
2173 _ => return None,
2174 };
2175 // Extra prose after a bare verdict is ignored; exactly one short token
2176 // is treated as a gate id.
2177 let id = match tokens.next() {
2178 Some(t)
2179 if tokens.next().is_none()
2180 && t.len() >= 4
2181 && t.chars().all(|c| c.is_ascii_alphanumeric()) =>
2182 {
2183 Some(t.to_lowercase())
2184 }
2185 _ => None,
2186 };
2187 Some((verdict, id))
2188}
2189
2190async fn gateway_inbound(
2191 State(state): State<AppState>,
2192 Json(body): Json<InboundBody>,
2193) -> axum::response::Response {
2194 crate::refresh_control_plane(&state);
2195 if !state.gateway.enabled {
2196 return (
2197 StatusCode::CONFLICT,
2198 Json(serde_json::json!({
2199 "error": "gateway disabled: set [gateway] enabled = true (trusted config) or pass serve --gateway"
2200 })),
2201 )
2202 .into_response();
2203 }
2204 let text = body.text.trim().to_string();
2205 let has_attachment = body.attachments.iter().any(|a| !a.data.trim().is_empty());
2206 if body.surface.trim().is_empty()
2207 || body.chat.trim().is_empty()
2208 || (text.is_empty() && !has_attachment)
2209 {
2210 return (
2211 StatusCode::BAD_REQUEST,
2212 Json(serde_json::json!({"error": "surface, chat and text are required"})),
2213 )
2214 .into_response();
2215 }
2216 // Multi-bot-per-channel: a chat's key is scoped to the bot that
2217 // delivered the message whenever the bridge knows its own bot id
2218 // (`--bot-id`), so the same physical chat served by several bots gets
2219 // one independent allowlist entry, session, and policy per bot instead
2220 // of all of them colliding onto one shared conversation. Legacy/
2221 // single-bot bridges (no bot id) keep the original two-part key
2222 // unchanged. See `legacy_key_for` for the one-time migration this
2223 // implies for an already-approved chat or a `chat_allowlist` row.
2224 let key = match body
2225 .bot_id
2226 .as_deref()
2227 .map(str::trim)
2228 .filter(|b| !b.is_empty())
2229 {
2230 Some(bot_id) => format!("{}:{}:{bot_id}", body.surface.trim(), body.chat.trim()),
2231 None => format!("{}:{}", body.surface.trim(), body.chat.trim()),
2232 };
2233 // 0c-01/0c-02/docs/design/34: chat allowlist — reject messages from
2234 // unknown chats, but record a reviewable *pending* entry instead of a
2235 // flat rejection so the operator has a forward path to "let it
2236 // through" that isn't a hand-edited config file + process restart.
2237 // `chat_allowlist_open = true` still bypasses the store entirely.
2238 if !state.gateway.chat_allowlist_open() {
2239 let decision = state.gateway.allowlist_resolve_inbound(
2240 &state.core,
2241 &key,
2242 &text,
2243 body.bot_id.as_deref(),
2244 );
2245 match decision {
2246 AllowlistDecision::Allowed => {}
2247 AllowlistDecision::Denied => {
2248 vak_core::security_events::record(
2249 &state.core.sessions_home(),
2250 vak_core::security_events::EventKind::ChatDenied,
2251 "chat_denied",
2252 &format!("key={key}"),
2253 None,
2254 );
2255 return (
2256 StatusCode::FORBIDDEN,
2257 Json(serde_json::json!({
2258 "error": format!("chat '{key}' rejected: denied by operator"),
2259 "state": "denied",
2260 })),
2261 )
2262 .into_response();
2263 }
2264 AllowlistDecision::NewlyPending => {
2265 vak_core::security_events::record(
2266 &state.core.sessions_home(),
2267 vak_core::security_events::EventKind::ChatPending,
2268 "chat_pending",
2269 &format!("key={key}"),
2270 None,
2271 );
2272 return (
2273 StatusCode::FORBIDDEN,
2274 Json(serde_json::json!({
2275 "error": format!(
2276 "chat '{key}' rejected: awaiting operator approval in the admin console"
2277 ),
2278 "state": "pending",
2279 })),
2280 )
2281 .into_response();
2282 }
2283 AllowlistDecision::StillPending => {
2284 // A lighter, non-security-event log line: this is expected
2285 // repeat traffic from an already-reviewable key, not a
2286 // fresh incident to append to the audit trail each time.
2287 eprintln!("[gateway] chat '{key}' still pending operator review");
2288 return (
2289 StatusCode::FORBIDDEN,
2290 Json(serde_json::json!({
2291 "error": format!(
2292 "chat '{key}' rejected: still awaiting operator approval"
2293 ),
2294 "state": "pending",
2295 })),
2296 )
2297 .into_response();
2298 }
2299 }
2300 }
2301
2302 // Approval replies from the designated approver surface resolve the
2303 // addressed gate (or the oldest one) instead of becoming conversation
2304 // input. Any non-verdict text from that chat falls through to normal
2305 // routing.
2306 if state.gateway.forward_mode()
2307 && state.gateway.approver_target().as_deref() == Some(key.as_str())
2308 && let Some((verdict, gate_id)) = parse_verdict(&text)
2309 {
2310 return match state.gateway.resolve_gate(verdict, gate_id.as_deref()) {
2311 Ok(resolved) => {
2312 // A chat "no" is observable here and nowhere else, so the
2313 // durable record of the denial is written at the same beat.
2314 if !verdict {
2315 let short = resolved.id.get(..8).unwrap_or(resolved.id.as_str());
2316 let _ = vak_core::inbox::record(
2317 &state.core.shared_data_home(),
2318 vak_core::inbox::Kind::ApprovalDenied,
2319 &format!("approval denied [{short}]"),
2320 &format!(
2321 "session {} denied forwarded gate {} ({} pending)",
2322 resolved.session_id, resolved.id, resolved.remaining
2323 ),
2324 Some(&resolved.session_id),
2325 None,
2326 );
2327 }
2328 (
2329 StatusCode::OK,
2330 Json(serde_json::json!({
2331 "state": "approval_resolved",
2332 "approved": verdict,
2333 "gate": resolved.id,
2334 "session_id": resolved.session_id,
2335 "remaining": resolved.remaining,
2336 })),
2337 )
2338 .into_response()
2339 }
2340 Err(()) => (
2341 StatusCode::OK,
2342 Json(serde_json::json!({ "state": "no_pending_approvals" })),
2343 )
2344 .into_response(),
2345 };
2346 }
2347
2348 // docs/design/34 Phase 2: run this key's entry through its own
2349 // workspace's Core (sandbox, permission mode, session ledger) — not
2350 // just its provider/model — when the entry names a workspace other
2351 // than the gateway's own. Falls back to the gateway's default Core
2352 // when the entry has no workspace override, exactly as before.
2353 let core = match state.gateway.core_for_entry(&state.core, &key) {
2354 Ok(core) => core,
2355 Err(e) => {
2356 return (
2357 StatusCode::INTERNAL_SERVER_ERROR,
2358 Json(serde_json::json!({"error": format!("workspace core unavailable: {e}")})),
2359 )
2360 .into_response();
2361 }
2362 };
2363
2364 // Admission owns the conversation context. Stamp it before creating or
2365 // reopening the bound session so the ledger, prompt contract, and every
2366 // later delivery can identify the authorized audience and originating
2367 // transport without reverse-engineering mutable gateway state.
2368 let conversation_context = vak_session::ConversationContext {
2369 conversation_id: key.clone(),
2370 audience_id: key.clone(),
2371 origin: Some(vak_session::ConversationOrigin {
2372 surface: body.surface.trim().to_string(),
2373 address: body.chat.trim().to_string(),
2374 bot_id: body.bot_id.clone(),
2375 }),
2376 };
2377 let core = core.with_conversation_context(Some(conversation_context));
2378
2379 let handle = match resolve_session(&state, &core, &key).await {
2380 Ok(h) => h,
2381 Err(e) => {
2382 return (
2383 StatusCode::INTERNAL_SERVER_ERROR,
2384 Json(serde_json::json!({"error": e})),
2385 )
2386 .into_response();
2387 }
2388 };
2389
2390 let request_id = body
2391 .request_id
2392 .as_deref()
2393 .map(str::trim)
2394 .filter(|value| !value.is_empty())
2395 .map(ToOwned::to_owned)
2396 .unwrap_or_else(|| format!("gateway-{}", uuid::Uuid::now_v7()));
2397 let already_admitted = handle
2398 .admissions
2399 .lock()
2400 .unwrap_or_else(std::sync::PoisonError::into_inner)
2401 .contains(&request_id)
2402 || handle
2403 .session
2404 .lock()
2405 .ok()
2406 .and_then(|guard| {
2407 guard
2408 .as_ref()
2409 .map(|log| log.has_request_admission(&request_id))
2410 })
2411 .unwrap_or(false);
2412 if already_admitted {
2413 return (
2414 StatusCode::ACCEPTED,
2415 Json(serde_json::json!({
2416 "request_id": request_id,
2417 "state": "already_admitted",
2418 "decision": "duplicate",
2419 "session_id": binding_session(&state, &key),
2420 })),
2421 )
2422 .into_response();
2423 }
2424 handle

1425 {
1426 out.push(vak_core::prompts::LayerInput::new(
1427 vak_core::prompts::PromptLayer::Bot,
1428 Some(format!("bot:{}", bot.id)),
1429 bot.prompt,
1430 ));
1431 }
1432 if !entry.prompt.is_empty() {
1433 out.push(vak_core::prompts::LayerInput::new(
1434 vak_core::prompts::PromptLayer::Chat,
1435 Some(format!("chat:{key}")),
1436 entry.prompt,
1437 ));
1438 }
1439 out
1440 }
1441
1442 /// The style directive for spoken replies on this chat.
1443 ///
1444 /// One source of truth (docs/design/45-prompt-layers.md): the bot/chat
1445 /// `identity` block *is* the persona. `VoiceConfig.persona` predates
1446 /// prompt layers and said the same thing in a second place; keeping both
1447 /// authoritative would let a bot's spoken and written selves drift apart
1448 /// within a release. The legacy field is still honoured when no prompt
1449 /// tier sets an identity, so existing configs keep working untouched.
1450 ///
1451 /// Only the *gateway tiers'* own identity text is used, never the
1452 /// assembled prompt — the seed identity and capability contract are
1453 /// meaningless as a text-to-speech style directive.
1454 pub(crate) fn resolve_persona(&self, key: &str) -> Option<String> {
1455 self.resolve_prompt_overlays(key)
1456 .into_iter()
1457 .rev()
1458 .find_map(|layer| layer.content.identity)
1459 .map(|text| text.trim().to_string())
1460 .filter(|text| !text.is_empty())
1461 .or_else(|| {
1462 self.resolve_voice(key)
1463 .and_then(|voice| voice.persona)
1464 .map(|p| p.trim().to_string())
1465 .filter(|p| !p.is_empty())
1466 })
1467 }
1468
1469 pub(crate) fn resolve_voice(&self, key: &str) -> Option<vak_config::VoiceConfig> {
1470 let entry = self
1471 .allowlist_get(key)
1472 .filter(|e| e.status == AllowlistStatus::Allowed);
1473 let parent = entry
1474 .as_ref()
1475 .filter(|e| e.inherit_bot_policy)
1476 .and_then(|e| e.bot_id.as_deref())
1477 .and_then(|id| self.bot_get(id))
1478 .and_then(|b| b.voice);
1479 entry
1480 .as_ref()
1481 .and_then(|e| e.voice.as_ref())
1482 .map(|v| vak_config::VoiceConfig::overlay(parent.as_ref(), v))
1483 .or(parent)
1484 }
1485
1486 /// Drop the cached route revision for `key` without dropping the
1487 /// session id — exactly what `set_route_override` does — so the next
1488 /// inbound message re-derives the effective route and rotates the
1489 /// frozen session if (and only if) it actually changed.
1490 pub(crate) fn invalidate_binding_revision(&self, core: &Core, key: &str) {
1491 let touched = {
1492 let mut bindings = self
1493 .bindings
1494 .lock()
1495 .unwrap_or_else(std::sync::PoisonError::into_inner);
1496 match bindings.get_mut(key) {
1497 Some(binding) => {
1498 binding.route_revision = None;
1499 true
1500 }
1501 None => false,
1502 }
1503 };
1504 if touched {
1505 persist_bindings(core, self);
1506 }
1507 }
1508
1509 pub(crate) fn invalidate_bindings_for_bot(&self, core: &Core, bot_id: &str) {
1510 let keys: Vec<String> = {
1511 let allowlist = self
1512 .allowlist
1513 .lock()
1514 .unwrap_or_else(std::sync::PoisonError::into_inner);
1515 allowlist
1516 .values()
1517 .filter(|entry| entry.bot_id.as_deref() == Some(bot_id))
1518 .map(|entry| entry.key.clone())
1519 .collect()
1520 };
1521 let touched = {
1522 let mut bindings = self
1523 .bindings
1524 .lock()
1525 .unwrap_or_else(std::sync::PoisonError::into_inner);
1526 let mut changed = false;
1527 for key in keys {
1528 if let Some(binding) = bindings.get_mut(&key) {
1529 binding.route_revision = None;
1530 changed = true;
1531 }
1532 }
1533 changed
1534 };
1535 if touched {
1536 persist_bindings(core, self);
1537 }
1538 }
1539
1540 /// Test seam: plant a bound session with a frozen route revision, the
1541 /// state dispatch leaves behind, without running a whole turn.
1542 #[cfg(test)]
1543 pub(crate) fn bind_for_test(&self, key: &str, session_id: &str, revision: &str) {
1544 let mut bindings = self
1545 .bindings
1546 .lock()
1547 .unwrap_or_else(std::sync::PoisonError::into_inner);
1548 let binding = bindings.entry(key.to_string()).or_default();
1549 binding.session_id = Some(session_id.to_string());
1550 binding.route_revision = Some(revision.to_string());
1551 }
1552
1553 #[cfg(test)]
1554 pub(crate) fn route_revision_for_test(&self, key: &str) -> Option<String> {
1555 self.bindings
1556 .lock()
1557 .unwrap_or_else(std::sync::PoisonError::into_inner)
1558 .get(key)
1559 .and_then(|b| b.route_revision.clone())
1560 }
1561
1562 #[cfg(test)]
1563 pub(crate) fn session_id_for_test(&self, key: &str) -> Option<String> {
1564 self.bindings
1565 .lock()
1566 .unwrap_or_else(std::sync::PoisonError::into_inner)
1567 .get(key)
1568 .and_then(|b| b.session_id.clone())
1569 }
1570
1571 /// Revoke an allowed entry: removes it from the store entirely (a
1572 /// future message from that key starts a fresh pending review, not a
1573 /// stale "denied" record masquerading as an audit trail).
1574 pub(crate) fn allowlist_revoke(&self, core: &Core, key: &str) -> bool {
1575 if self.allowlist_get(key).map(|e| e.status) != Some(AllowlistStatus::Allowed) {
1576 return false;
1577 }
1578 let removed = self
1579 .allowlist
1580 .lock()
1581 .unwrap_or_else(std::sync::PoisonError::into_inner)
1582 .remove(key)
1583 .is_some();
1584 if removed {
1585 persist_allowlist(core, self);
1586 }
1587 removed
1588 }
1589}
1590
1591/// What a channel's permission mode actually resolves to, for display and
1592/// for the approve/patch audit trail.
1593pub(crate) struct ResolvedPermission {
1594 /// The target workspace's own configured mode — the outermost ceiling.
1595 pub workspace_mode: vak_config::PermissionMode,
1596 /// The bot tier's pin, when the chat inherits from a bot that has one.
1597 /// `None` means no bot tier applies, not "the bot allows everything".
1598 pub bot_mode: Option<vak_config::PermissionMode>,
1599 /// What the entry asked for, if anything.
1600 pub requested: Option<vak_config::PermissionMode>,
1601 /// What the channel actually gets, folded the same way `core_for_entry`
1602 /// folds it: chat pin capped by bot pin, then capped by the workspace.
1603 pub effective: vak_config::PermissionMode,
1604}
1605
1606impl ResolvedPermission {
1607 /// True when a pin asked for more than the layers above it allow and
1608 /// was reduced. This is the condition worth an audit-log entry.
1609 ///
1610 /// It covers the bot tier too: a bot pinned narrower than its chat
1611 /// reduces that chat just as surely as the workspace does, and reporting
1612 /// only the workspace cap is what let the console show a chat as wider
1613 /// than it actually ran.
1614 pub fn was_capped(&self) -> bool {
1615 matches!(self.requested, Some(r) if r != self.effective)
1616 || matches!(
1617 (self.bot_mode, self.requested),
1618 (Some(bot), None) if bot != self.effective
1619 )
1620 }
1621}
1622
1623/// Read-only mirror of the cap dispatch enforces, for the admin surface.
1624/// Shares `PermissionMode::capped_by` with the pool so the number the
1625/// console shows is derived the same way as the one dispatch pins.
1626///
1627/// It must fold the SAME three tiers `GatewayState::core_for_entry` folds:
1628/// chat pin capped by bot pin, then capped by the workspace. Leaving the
1629/// bot tier out — as this did — meant a bot pinned to `read-only` under a
1630/// chat pinned to `workspace-write` ran read-only and was reported as
1631/// workspace-write, and a chat with no pin under a bot that had one was
1632/// reported as the workspace's mode instead of the bot's. Showing a channel
1633/// as wider than it runs is the one direction of error that matters here.
1634///
1635/// A workspace whose config fails to load falls back to the compiled
1636/// default (`WorkspaceWrite`), matching `vak_config`'s own layering; the
1637/// pool remains the authority at dispatch either way.
1638pub(crate) fn resolve_channel_permission(
1639 workspace: &std::path::Path,
1640 requested: Option<vak_config::PermissionMode>,
1641 bot_mode: Option<vak_config::PermissionMode>,
1642) -> ResolvedPermission {
1643 // Same trust the pool will use at dispatch. Reading with `true` here
1644 // while the pool read the marker store would put the console back to
1645 // reporting a mode no run would get.
1646 let workspace_mode =
1647 vak_config::load_with_trust(workspace, vak_core::trust::is_trusted(workspace))
1648 .map(|c| c.permission_mode)
1649 .unwrap_or_default();
1650 // Exactly `core_for_entry`'s fold, then the pool's workspace cap.
1651 let pinned = match (requested, bot_mode) {
1652 (Some(chat), Some(bot)) => Some(chat.capped_by(bot)),
1653 (Some(chat), None) => Some(chat),
1654 (None, bot) => bot,
1655 };
1656 let effective = match pinned {
1657 Some(mode) => mode.capped_by(workspace_mode),
1658 None => workspace_mode,
1659 };
1660 ResolvedPermission {
1661 workspace_mode,
1662 bot_mode,
1663 requested,
1664 effective,
1665 }
1666}
1667
1668/// True when `added_at` parses as an RFC3339 stamp strictly older than
1669/// `cutoff`. An unparseable stamp is never treated as expired: a corrupt
1670/// timestamp must not silently auto-deny a live channel.
1671fn parse_added_at_before(added_at: &str, cutoff: chrono::DateTime<chrono::Utc>) -> bool {
1672 chrono::DateTime::parse_from_rfc3339(added_at)
1673 .map(|ts| ts.with_timezone(&chrono::Utc) < cutoff)
1674 .unwrap_or(false)
1675}
1676
1677fn persist_bindings(core: &Core, gw: &GatewayState) {
1678 let bindings = gw
1679 .bindings
1680 .lock()
1681 .unwrap_or_else(std::sync::PoisonError::into_inner)
1682 .clone();
1683 let path = bindings_path(&core.shared_data_home());
1684 if let Some(parent) = path.parent() {
1685 let _ = std::fs::create_dir_all(parent);
1686 }
1687 let file = BindingsFile {
1688 version: 2,
1689 bindings,
1690 };
1691 if let Ok(json) = serde_json::to_string_pretty(&file) {
1692 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1693 if std::fs::write(&temp, json).is_ok() {
1694 let _ = std::fs::rename(temp, path);
1695 }
1696 }
1697}
1698
1699fn persist_allowlist(core: &Core, gw: &GatewayState) {
1700 let entries = gw
1701 .allowlist
1702 .lock()
1703 .unwrap_or_else(std::sync::PoisonError::into_inner)
1704 .clone();
1705 let path = allowlist_path(&core.shared_data_home());
1706 write_allowlist_file(&path, &entries);
1707}
1708
1709fn persist_bots(core: &Core, gw: &GatewayState) {
1710 let bots = gw
1711 .bots
1712 .lock()
1713 .unwrap_or_else(std::sync::PoisonError::into_inner)
1714 .clone();
1715 persist_bots_map(&bots_path(&core.shared_data_home()), &bots);
1716}
1717
1718/// Atomic temp-file+rename write, same pattern as `write_allowlist_file`.
1719/// Free function (not a `GatewayState` method) so the one-time migration in
1720/// `GatewayState::load` can call it before a `GatewayState` exists.
1721fn persist_bots_map(path: &std::path::Path, bots: &HashMap<String, Bot>) {
1722 if let Some(parent) = path.parent() {
1723 let _ = std::fs::create_dir_all(parent);
1724 }
1725 let mut sorted: Vec<Bot> = bots.values().cloned().collect();
1726 sorted.sort_by(|a, b| a.id.cmp(&b.id));
1727 let file = BotsFile {
1728 schema: 1,
1729 bots: sorted,
1730 };
1731 if let Ok(json) = serde_json::to_string_pretty(&file) {
1732 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1733 if std::fs::write(&temp, json).is_ok() {
1734 let _ = std::fs::rename(temp, path);
1735 }
1736 }
1737}
1738
1739/// Atomic temp-file+rename write, same pattern as `persist_bindings`.
1740fn write_allowlist_file(path: &std::path::Path, entries: &HashMap<String, AllowlistEntry>) {
1741 if let Some(parent) = path.parent() {
1742 let _ = std::fs::create_dir_all(parent);
1743 }
1744 let mut sorted: Vec<AllowlistEntry> = entries.values().cloned().collect();
1745 sorted.sort_by(|a, b| a.key.cmp(&b.key));
1746 let file = AllowlistFile {
1747 schema: 1,
1748 entries: sorted,
1749 };
1750 if let Ok(json) = serde_json::to_string_pretty(&file) {
1751 let temp = path.with_extension(format!("json.{}.tmp", std::process::id()));
1752 if std::fs::write(&temp, json).is_ok() {
1753 let _ = std::fs::rename(temp, path);
1754 }
1755 }
1756}
1757
1758pub fn routes() -> Router<AppState> {
1759 Router::new()
1760 .route(
1761 "/gateway/inbound",
1762 axum::routing::post(gateway_inbound)
1763 .layer(axum::extract::DefaultBodyLimit::max(INBOUND_BODY_MAX_BYTES)),
1764 )
1765 .route("/gateway/status", axum::routing::get(gateway_status))
1766 .route(
1767 "/gateway/bindings/{key}",
1768 axum::routing::delete(gateway_unbind),
1769 )
1770}
1771
1772// ---- Inbound ----------------------------------------------------------------
1773
1774#[derive(serde::Deserialize)]
1775struct InboundBody {
1776 surface: String,
1777 chat: String,
1778 /// Who sent this, when the adapter knows (a Telegram @user, a webhook
1779 /// identity). Typed since G1 groundwork: it is recorded on deliveries
1780 /// and attributed on queued turns instead of being silently dropped.
1781 #[serde(default)]
1782 sender: Option<String>,
1783 text: String,
1784 #[serde(default)]
1785 wait: bool,
1786 /// Base64 images appended to the prompt as vision content.
1787 #[serde(default)]
1788 attachments: Vec<InboundAttachment>,
1789 #[serde(default)]
1790 capabilities: Option<crate::delivery::RequestedCapabilities>,
1791 /// See [`InboundRequest::bot_id`].
1792 #[serde(default)]
1793 bot_id: Option<String>,
1794 /// Caller-provided idempotency key. Repeating it returns the original
1795 /// admission without dispatching a second model turn.
1796 #[serde(default)]
1797 request_id: Option<String>,
1798}
1799
1800#[derive(serde::Deserialize)]
1801struct InboundAttachment {
1802 /// MIME type; defaults to image/png for Telegram-style senders.
1803 #[serde(default = "default_image_mime")]
1804 mime: String,
1805 data: String,
1806 /// Original filename, when the channel knows it (documents only).
1807 #[serde(default)]
1808 filename: Option<String>,
1809 /// "image" (default, vision content) or "document": small text is
1810 /// inlined, anything else is saved to the workspace inbox and named.
1811 #[serde(default = "default_attachment_kind")]
1812 kind: String,
1813 #[serde(default)]
1814 error: Option<String>,
1815}
1816
1817fn default_image_mime() -> String {
1818 "image/png".into()
1819}
1820
1821fn default_attachment_kind() -> String {
1822 "image".into()
1823}
1824
1825/// The largest document a channel may hand the gateway, shared by every
1826/// bridge so a file is never downloaded by one side and dropped by the
1827/// other. The gateway's JSON body limit (2 MiB, base64 inflates by a third)
1828/// bounds it.
1829pub(crate) const INBOUND_DOCUMENT_MAX_BYTES: usize = 20 * 1024 * 1024;
1830
1831/// `/gateway/inbound`'s body limit: a document at the limit, base64-encoded,
1832/// with room for the rest of the request.
1833const INBOUND_BODY_MAX_BYTES: usize = INBOUND_DOCUMENT_MAX_BYTES / 3 * 4 + 1024 * 1024;
1834
1835/// Text documents at or under this size are inlined in the prompt; larger
1836/// text, and every non-text file, is saved to the workspace inbox instead.
1837const DOCUMENT_INLINE_MAX_BYTES: usize = 64 * 1024;
1838
1839/// What a channel turn answers: its text, and each workspace file the turn
1840/// drafted with `office_apply`, for the channel to get back.
1841pub(crate) struct ChatReply {
1842 pub(crate) text: String,
1843 pub(crate) drafts: Vec<TurnDraft>,
1844}
1845
1846/// The latest draft this turn made of one workspace file.
1847pub(crate) struct TurnDraft {
1848 /// Workspace-relative, as the call named it.
1849 path: String,
1850 draft: std::path::PathBuf,
1851}
1852
1853/// Each workspace file this turn's successful `office_apply` calls drafted,
1854/// with the last draft of it: an `office_apply` call's draft is at
1855/// `.vak/scratch/<agent>/<call id>/<path>`, the convention Review's lineage
1856/// relies on too.
1857fn turn_drafts(log: &vak_session::SessionLog, workspace: &std::path::Path) -> Vec<TurnDraft> {
1858 let Some(directive) = log.latest_directive_entry_id() else {
1859 return Vec::new();
1860 };
1861 let agent = log
1862 .header()
1863 .and_then(|header| header.agent.as_ref().map(|agent| agent.id.clone()))
1864 .unwrap_or_else(|| "vak".into());
1865 let mut in_turn = false;
1866 let mut calls: Vec<(String, String)> = Vec::new();
1867 let mut succeeded = std::collections::HashSet::new();
1868 for (entry_id, message) in log.message_chain() {
1869 in_turn |= entry_id == directive;
1870 if !in_turn {
1871 continue;
1872 }
1873 for block in &message.content {
1874 match block {
1875 vak_llm::ContentBlock::ToolUse { id, name, input } if name == "office_apply" => {
1876 if let Some(path) = input.get("path").and_then(serde_json::Value::as_str) {
1877 calls.push((id.clone(), path.trim().to_string()));
1878 }
1879 }
1880 vak_llm::ContentBlock::ToolResult {
1881 tool_use_id,
1882 is_error: false,
1883 ..
1884 } => {
1885 succeeded.insert(tool_use_id.clone());
1886 }
1887 _ => {}
1888 }
1889 }
1890 }
1891 let mut drafts: Vec<TurnDraft> = Vec::new();
1892 for (id, path) in calls.into_iter().filter(|(id, _)| succeeded.contains(id)) {
1893 let draft = workspace
1894 .join(".vak")
1895 .join("scratch")
1896 .join(&agent)
1897 .join(&id)
1898 .join(&path);
1899 if !draft.is_file() {
1900 continue;
1901 }
1902 drafts.retain(|earlier| earlier.path != path);
1903 drafts.push(TurnDraft { path, draft });
1904 }
1905 drafts
1906}
1907
1908/// The largest draft sent back on a channel; Telegram's bots may send up to
1909/// 50 MB, and a document this large is better opened in Vakyartha.
1910const RETURN_FILE_MAX_BYTES: u64 = 20 * 1024 * 1024;
1911
1912/// Each draft the turn made, for a channel that takes files: its bytes and a
1913/// caption saying what changed (from the worker's semantic diff), under the
1914/// name the person knows it by. A draft that carries a sensitivity label is
1915/// not sent (labels only narrow where a file goes); a channel that takes no
1916/// files, or a draft too large, gets a line saying where the file is.
1917async fn return_drafts(
1918 core: &Core,
1919 mut text: String,
1920 drafts: &[TurnDraft],
1921 accepts_files: bool,
1922) -> (String, Vec<serde_json::Value>) {
1923 use base64::Engine as _;
1924 let worker = core.tool_worker_exe();
1925 let mut files = Vec::new();
1926 let mut notes = Vec::new();
1927 for draft in drafts {
1928 let name = inbox::display_name(
1929 std::path::Path::new(&draft.path)
1930 .file_name()
1931 .and_then(|name| name.to_str())
1932 .unwrap_or(&draft.path),
1933 );
1934 let facts = vak_tools::broker::office_project(
1935 &worker,
1936 &draft.draft,
1937 vak_tools::broker::OfficeView::Facts,
1938 )
1939 .await;
1940 let labels: Vec<String> = facts
1941 .as_ref()
1942 .ok()
1943 .and_then(|facts| facts.get("sensitivity_labels"))
1944 .and_then(|labels| serde_json::from_value(labels.clone()).ok())
1945 .unwrap_or_default();
1946 if !labels.is_empty() {
1947 notes.push(format!(
1948 "{name} carries the sensitivity label {}, so it is not sent on this channel; review the draft in Vakyartha.",
1949 labels.join(", ")
1950 ));
1951 continue;
1952 }
1953 let size = std::fs::metadata(&draft.draft)
1954 .map(|metadata| metadata.len())
1955 .unwrap_or(u64::MAX);
1956 if !accepts_files || size > RETURN_FILE_MAX_BYTES {
1957 notes.push(format!(
1958 "The updated {name} is ready in Vakyartha for review; this channel does not receive it."
1959 ));
1960 continue;
1961 }
1962 let current = core.cwd().join(&draft.path);
1963 let review = vak_tools::broker::office_review(
1964 &worker,
1965 current.is_file().then_some(current.as_path()),
1966 &draft.draft,
1967 None,
1968 )
1969 .await;
1970 let summary = review
1971 .as_ref()
1972 .ok()
1973 .and_then(|review| review.get("summary"))
1974 .and_then(|summary| serde_json::from_value::<Vec<String>>(summary.clone()).ok())
1975 .filter(|summary| !summary.is_empty())
1976 .map(|summary| summary.join("; "))
1977 .unwrap_or_else(|| "no visible change".into());
1978 let Ok(bytes) = std::fs::read(&draft.draft) else {
1979 notes.push(format!("The updated {name} could not be read to send it."));
1980 continue;
1981 };
1982 files.push(serde_json::json!({
1983 "name": name,
1984 "mime": office_mime(&name),
1985 "data": base64::engine::general_purpose::STANDARD.encode(bytes),
1986 "caption": format!("Updated {name}: {summary}"),
1987 }));
1988 }
1989 if !notes.is_empty() {
1990 text = format!("{}\n\n{}", text.trim_end(), notes.join("\n"));
1991 }
1992 (text, files)
1993}
1994
1995fn office_mime(name: &str) -> &'static str {
1996 match name
1997 .rsplit('.')
1998 .next()
1999 .map(str::to_ascii_lowercase)
2000 .as_deref()
2001 {
2002 Some("docx") => "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
2003 Some("xlsx") => "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
2004 Some("pptx") => "application/vnd.openxmlformats-officedocument.presentationml.presentation",
2005 _ => "application/octet-stream",
2006 }
2007}
2008
2009/// Compose the prompt message: text, vision blocks, inlined text documents,
2010/// and a note for each document saved to the inbox. A document's bytes
2011/// never enter the prompt unless they are text (docs/design/72, F1). The
2012/// ledger stores exactly what the model will see (invariant 1).
2013fn compose_prompt(
2014 text: &str,
2015 attachments: &[InboundAttachment],
2016 workspace: &std::path::Path,
2017) -> vak_llm::Message {
2018 let mut blocks = Vec::new();
2019 if !text.is_empty() {
2020 blocks.push(vak_llm::ContentBlock::text(text));
2021 }
2022 for a in attachments {
2023 if a.data.trim().is_empty() {
2024 continue;
2025 }
2026 if a.kind == "document" {
2027 blocks.push(vak_llm::ContentBlock::text(document_block(a, workspace)));
2028 continue;
2029 }
2030 // A voice note reaches the model as its transcript (or the reason
2031 // there is none) in the message text, never as an attachment block.
2032 if a.kind == "audio" {
2033 continue;
2034 }
2035 blocks.push(vak_llm::ContentBlock::image_base64(
2036 a.mime.clone(),
2037 a.data.trim().to_string(),
2038 ));
2039 }
2040 vak_llm::Message {
2041 role: vak_llm::Role::User,
2042 content: blocks,
2043 }
2044}
2045
2046fn document_block(attachment: &InboundAttachment, workspace: &std::path::Path) -> String {
2047 use base64::Engine as _;
2048 let filename = attachment.filename.as_deref().unwrap_or("file");
2049 let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(attachment.data.trim()) else {
2050 return format!("[attached file '{filename}' could not be decoded; not included]");
2051 };
2052 if bytes.len() > INBOUND_DOCUMENT_MAX_BYTES {
2053 return format!(
2054 "[attached file '{filename}' ({} KiB) exceeds the {} KiB channel limit; not received]",
2055 bytes.len() / 1024,
2056 INBOUND_DOCUMENT_MAX_BYTES / 1024
2057 );
2058 }
2059 let text = std::str::from_utf8(&bytes)
2060 .ok()
2061 .filter(|text| !text.contains('\0'));
2062 if let Some(content) = text
2063 && bytes.len() <= DOCUMENT_INLINE_MAX_BYTES
2064 {
2065 return format!("Attached file `{filename}`:\n```\n{content}\n```");
2066 }
2067 match save_to_inbox(workspace, filename, &bytes) {
2068 Ok(saved) => inbox::note(filename, &saved, &bytes),
2069 Err(error) => {
2070 format!("[attached file '{filename}' could not be saved ({error}); not included]")
2071 }
2072 }
2073}
2074
2075pub(crate) fn compose_voice_prompt(text: &str) -> vak_llm::Message {
2076 compose_prompt(text, &[], std::path::Path::new("."))
2077}
2078
2079// ---- Approval forwarding (G2) ----------------------------------------------
2080
2081/// Approver for gateway-driven turns. In `deny` mode this behaves like
2082/// `AutoDeny`. In `forward` mode the gate is announced on the approver
2083/// surface and resolved by a yes/no reply; timeout or silence fails closed.
2084struct GatewayApprover {
2085 events_tx: crate::events::EventBus,
2086 state: Arc<GatewayState>,
2087 core: Core,
2088 session_id: String,
2089}
2090
2091#[async_trait::async_trait]
2092impl vak_agent::Approver for GatewayApprover {
2093 /// A forwarded gate reaches the configured approver chat; without
2094 /// forward mode nothing is listening, and the reachability preflight
2095 /// must see that before the prompt advertises a gated capability.
2096 fn answerable(&self) -> bool {
2097 self.state.forward_mode()
2098 }
2099
2100 async fn approve(&self, tool: &str, args_json: &str, reason: &str) -> bool {
2101 if !self.state.forward_mode() {
2102 return false;
2103 }
2104 let id = uuid::Uuid::now_v7().to_string();
2105 let rx = self.state.register_gate(&id, &self.session_id);
2106 let _ = self.events_tx.send(AgentEvent::ApprovalRequested {
2107 id: id.clone(),
2108 tool: tool.to_string(),
2109 args_json: args_json.to_string(),
2110 reason: reason.to_string(),
2111 });
2112 let short = &id[..8];
2113 let announce = format!(
2114 "Approval requested [{short}]\nTool: {tool}\nArgs: {args_json}\nReason: {reason}\nReply 'yes' or 'no' to decide."
2115 );
2116 if let Err(e) = deliver_approval_and_record(
2117 &self.core,
2118 self.state.approver_target().unwrap_or_default().as_str(),
2119 ApprovalPayload {
2120 request_id: id.clone(),
2121 title: format!("Approval requested [{short}]"),
2122 detail: announce.clone(),
2123 expires_at: Some(
2124 (chrono::Utc::now()
2125 + chrono::Duration::from_std(self.state.approval_timeout())
2126 .unwrap_or_default())
2127 .to_rfc3339(),
2128 ),
2129 actions: vec![
2130 delivery_action("approve", "Approve", "approve", &id),
2131 delivery_action("deny", "Deny", "deny", &id),
2132 ],
2133 },
2134 vak_core::inbox::Kind::ApprovalPending,
2135 format!("Approval requested [{short}]"),
2136 Some(&self.session_id),
2137 None,
2138 )
2139 .await
2140 {
2141 eprintln!("[gateway] approval announcement failed: {e}");
2142 let _ = self.state.resolve_gate(false, Some(&id));
2143 return false;
2144 }
2145 match tokio::time::timeout(self.state.approval_timeout(), rx).await {
2146 Ok(Ok(v)) => v,
2147 Ok(Err(_)) => false, // gate dropped (run cancelled)
2148 Err(_) => {
2149 // Timed out: remove our own entry so a late reply resolves
2150 // nothing.
2151 self.state
2152 .pending_approvals
2153 .lock()
2154 .unwrap_or_else(std::sync::PoisonError::into_inner)
2155 .remove(&id);
2156 eprintln!("[gateway] approval {short} timed out; denied");
2157 false
2158 }
2159 }
2160 }
2161}
2162
2163/// "yes"/"no" vocabulary for chat replies, optionally addressed to one
2164/// gate: "yes ab12cd34". Deliberately small and strict — casual chatter
2165/// from the approver chat must not resolve gates. Returns the verdict and
2166/// the gate-id prefix when one was supplied.
2167fn parse_verdict(text: &str) -> Option<(bool, Option<String>)> {
2168 let mut tokens = text.split_whitespace();
2169 let head = tokens.next()?.to_lowercase();
2170 let verdict = match head.as_str() {
2171 "y" | "yes" | "approve" | "approved" | "ok" | "allow" => true,
2172 "n" | "no" | "deny" | "denied" | "block" => false,
2173 _ => return None,
2174 };
2175 // Extra prose after a bare verdict is ignored; exactly one short token
2176 // is treated as a gate id.
2177 let id = match tokens.next() {
2178 Some(t)
2179 if tokens.next().is_none()
2180 && t.len() >= 4
2181 && t.chars().all(|c| c.is_ascii_alphanumeric()) =>
2182 {
2183 Some(t.to_lowercase())
2184 }
2185 _ => None,
2186 };
2187 Some((verdict, id))
2188}
2189
2190async fn gateway_inbound(
2191 State(state): State<AppState>,
2192 Json(body): Json<InboundBody>,
2193) -> axum::response::Response {
2194 crate::refresh_control_plane(&state);
2195 if !state.gateway.enabled {
2196 return (
2197 StatusCode::CONFLICT,
2198 Json(serde_json::json!({
2199 "error": "gateway disabled: set [gateway] enabled = true (trusted config) or pass serve --gateway"
2200 })),
2201 )
2202 .into_response();
2203 }
2204 let text = body.text.trim().to_string();
2205 let has_attachment = body.attachments.iter().any(|a| !a.data.trim().is_empty());
2206 if body.surface.trim().is_empty()
2207 || body.chat.trim().is_empty()
2208 || (text.is_empty() && !has_attachment)
2209 {
2210 return (
2211 StatusCode::BAD_REQUEST,
2212 Json(serde_json::json!({"error": "surface, chat and text are required"})),
2213 )
2214 .into_response();
2215 }
2216 // Multi-bot-per-channel: a chat's key is scoped to the bot that
2217 // delivered the message whenever the bridge knows its own bot id
2218 // (`--bot-id`), so the same physical chat served by several bots gets
2219 // one independent allowlist entry, session, and policy per bot instead
2220 // of all of them colliding onto one shared conversation. Legacy/
2221 // single-bot bridges (no bot id) keep the original two-part key
2222 // unchanged. See `legacy_key_for` for the one-time migration this
2223 // implies for an already-approved chat or a `chat_allowlist` row.
2224 let key = match body
2225 .bot_id
2226 .as_deref()
2227 .map(str::trim)
2228 .filter(|b| !b.is_empty())
2229 {
2230 Some(bot_id) => format!("{}:{}:{bot_id}", body.surface.trim(), body.chat.trim()),
2231 None => format!("{}:{}", body.surface.trim(), body.chat.trim()),
2232 };
2233 // 0c-01/0c-02/docs/design/34: chat allowlist — reject messages from
2234 // unknown chats, but record a reviewable *pending* entry instead of a
2235 // flat rejection so the operator has a forward path to "let it
2236 // through" that isn't a hand-edited config file + process restart.
2237 // `chat_allowlist_open = true` still bypasses the store entirely.
2238 if !state.gateway.chat_allowlist_open() {
2239 let decision = state.gateway.allowlist_resolve_inbound(
2240 &state.core,
2241 &key,
2242 &text,
2243 body.bot_id.as_deref(),
2244 );
2245 match decision {
2246 AllowlistDecision::Allowed => {}
2247 AllowlistDecision::Denied => {
2248 vak_core::security_events::record(
2249 &state.core.sessions_home(),
2250 vak_core::security_events::EventKind::ChatDenied,
2251 "chat_denied",
2252 &format!("key={key}"),
2253 None,
2254 );
2255 return (
2256 StatusCode::FORBIDDEN,
2257 Json(serde_json::json!({
2258 "error": format!("chat '{key}' rejected: denied by operator"),
2259 "state": "denied",
2260 })),
2261 )
2262 .into_response();
2263 }
2264 AllowlistDecision::NewlyPending => {
2265 vak_core::security_events::record(
2266 &state.core.sessions_home(),
2267 vak_core::security_events::EventKind::ChatPending,
2268 "chat_pending",
2269 &format!("key={key}"),
2270 None,
2271 );
2272 return (
2273 StatusCode::FORBIDDEN,
2274 Json(serde_json::json!({
2275 "error": format!(
2276 "chat '{key}' rejected: awaiting operator approval in the admin console"
2277 ),
2278 "state": "pending",
2279 })),
2280 )
2281 .into_response();
2282 }
2283 AllowlistDecision::StillPending => {
2284 // A lighter, non-security-event log line: this is expected
2285 // repeat traffic from an already-reviewable key, not a
2286 // fresh incident to append to the audit trail each time.
2287 eprintln!("[gateway] chat '{key}' still pending operator review");
2288 return (
2289 StatusCode::FORBIDDEN,
2290 Json(serde_json::json!({
2291 "error": format!(
2292 "chat '{key}' rejected: still awaiting operator approval"
2293 ),
2294 "state": "pending",
2295 })),
2296 )
2297 .into_response();
2298 }
2299 }
2300 }
2301
2302 // Approval replies from the designated approver surface resolve the
2303 // addressed gate (or the oldest one) instead of becoming conversation
2304 // input. Any non-verdict text from that chat falls through to normal
2305 // routing.
2306 if state.gateway.forward_mode()
2307 && state.gateway.approver_target().as_deref() == Some(key.as_str())
2308 && let Some((verdict, gate_id)) = parse_verdict(&text)
2309 {
2310 return match state.gateway.resolve_gate(verdict, gate_id.as_deref()) {
2311 Ok(resolved) => {
2312 // A chat "no" is observable here and nowhere else, so the
2313 // durable record of the denial is written at the same beat.
2314 if !verdict {
2315 let short = resolved.id.get(..8).unwrap_or(resolved.id.as_str());
2316 let _ = vak_core::inbox::record(
2317 &state.core.shared_data_home(),
2318 vak_core::inbox::Kind::ApprovalDenied,
2319 &format!("approval denied [{short}]"),
2320 &format!(
2321 "session {} denied forwarded gate {} ({} pending)",
2322 resolved.session_id, resolved.id, resolved.remaining
2323 ),
2324 Some(&resolved.session_id),
2325 None,
2326 );
2327 }
2328 (
2329 StatusCode::OK,
2330 Json(serde_json::json!({
2331 "state": "approval_resolved",
2332 "approved": verdict,
2333 "gate": resolved.id,
2334 "session_id": resolved.session_id,
2335 "remaining": resolved.remaining,
2336 })),
2337 )
2338 .into_response()
2339 }
2340 Err(()) => (
2341 StatusCode::OK,
2342 Json(serde_json::json!({ "state": "no_pending_approvals" })),
2343 )
2344 .into_response(),
2345 };
2346 }
2347
2348 // docs/design/34 Phase 2: run this key's entry through its own
2349 // workspace's Core (sandbox, permission mode, session ledger) — not
2350 // just its provider/model — when the entry names a workspace other
2351 // than the gateway's own. Falls back to the gateway's default Core
2352 // when the entry has no workspace override, exactly as before.
2353 let core = match state.gateway.core_for_entry(&state.core, &key) {
2354 Ok(core) => core,
2355 Err(e) => {
2356 return (
2357 StatusCode::INTERNAL_SERVER_ERROR,
2358 Json(serde_json::json!({"error": format!("workspace core unavailable: {e}")})),
2359 )
2360 .into_response();
2361 }
2362 };
2363
2364 // Admission owns the conversation context. Stamp it before creating or
2365 // reopening the bound session so the ledger, prompt contract, and every
2366 // later delivery can identify the authorized audience and originating
2367 // transport without reverse-engineering mutable gateway state.
2368 let conversation_context = vak_session::ConversationContext {
2369 conversation_id: key.clone(),
2370 audience_id: key.clone(),
2371 origin: Some(vak_session::ConversationOrigin {
2372 surface: body.surface.trim().to_string(),
2373 address: body.chat.trim().to_string(),
2374 bot_id: body.bot_id.clone(),
2375 }),
2376 };
2377 let core = core.with_conversation_context(Some(conversation_context));
2378
2379 let handle = match resolve_session(&state, &core, &key).await {
2380 Ok(h) => h,
2381 Err(e) => {
2382 return (
2383 StatusCode::INTERNAL_SERVER_ERROR,
2384 Json(serde_json::json!({"error": e})),
2385 )
2386 .into_response();
2387 }
2388 };
2389
2390 let request_id = body
2391 .request_id
2392 .as_deref()
2393 .map(str::trim)
2394 .filter(|value| !value.is_empty())
2395 .map(ToOwned::to_owned)
2396 .unwrap_or_else(|| format!("gateway-{}", uuid::Uuid::now_v7()));
2397 let already_admitted = handle
2398 .admissions
2399 .lock()
2400 .unwrap_or_else(std::sync::PoisonError::into_inner)
2401 .contains(&request_id)
2402 || handle
2403 .session
2404 .lock()
2405 .ok()
2406 .and_then(|guard| {
2407 guard
2408 .as_ref()
2409 .map(|log| log.has_request_admission(&request_id))
2410 })
2411 .unwrap_or(false);
2412 if already_admitted {
2413 return (
2414 StatusCode::ACCEPTED,
2415 Json(serde_json::json!({
2416 "request_id": request_id,
2417 "state": "already_admitted",
2418 "decision": "duplicate",
2419 "session_id": binding_session(&state, &key),
2420 })),
2421 )
2422 .into_response();
2423 }
2424 handle

Indexing the workspace…