Vakyartha documentation

1//! Deterministic subject namespace algebra and role-based ACL evaluator.
2//!
3//! NATS subject format:
4//! `vak.<plane>.<workspace_id>.<entity>.<target_or_session>.<verb_or_type>`
5
6use serde::{Deserialize, Serialize};
7use thiserror::Error;
8
9#[derive(Debug, Error)]
10pub enum SubjectError {
11 #[error("empty subject string")]
12 Empty,
13 #[error("subject must start with 'vak.': {0}")]
14 InvalidPrefix(String),
15 #[error("invalid subject structure: {0}")]
16 InvalidStructure(String),
17}
18
19/// Typed subject definitions across the ephemeral event and durable work planes.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub enum Subject {
22 /// Ephemeral streaming assistant & thinking tokens
23 EventsTokens {
24 workspace_id: String,
25 session_id: String,
26 },
27 /// Ephemeral streaming ANSI stdout/stderr chunks
28 EventsTerminal {
29 workspace_id: String,
30 session_id: String,
31 },
32 /// Ephemeral streaming process telemetry (500ms probe)
33 EventsTelemetry {
34 workspace_id: String,
35 session_id: String,
36 },
37 /// Durable JetStream work queue for worker task claiming
38 WorkTask { workspace_id: String, role: String },
39 /// Durable point-to-point agent steering & direct inbox
40 AgentInbox {
41 workspace_id: String,
42 agent_id: String,
43 },
44 /// Durable verified work receipts and execution evidence
45 Receipts {
46 workspace_id: String,
47 session_id: String,
48 },
49 /// Durable elevated permission approval requests & verdicts
50 Approvals {
51 workspace_id: String,
52 session_id: String,
53 },
54 /// Durable Dead-Letter Queue (DLQ) for poison tasks
55 DeadLetter {
56 workspace_id: String,
57 agent_id: String,
58 },
59 /// Generic or custom subject string
60 Custom(String),
61}
62
63impl Subject {
64 /// Render subject to canonical NATS wire string.
65 pub fn to_subject_string(&self) -> String {
66 match self {
67 Self::EventsTokens {
68 workspace_id,
69 session_id,
70 } => format!("vak.events.{workspace_id}.{session_id}.tokens"),
71 Self::EventsTerminal {
72 workspace_id,
73 session_id,
74 } => format!("vak.events.{workspace_id}.{session_id}.terminal"),
75 Self::EventsTelemetry {
76 workspace_id,
77 session_id,
78 } => format!("vak.events.{workspace_id}.{session_id}.telemetry"),
79 Self::WorkTask { workspace_id, role } => {
80 format!("vak.work.{workspace_id}.{role}.task")
81 }
82 Self::AgentInbox {
83 workspace_id,
84 agent_id,
85 } => format!("vak.agent.{workspace_id}.{agent_id}.inbox"),
86 Self::Receipts {
87 workspace_id,
88 session_id,
89 } => format!("vak.receipts.{workspace_id}.{session_id}.completed"),
90 Self::Approvals {
91 workspace_id,
92 session_id,
93 } => format!("vak.approvals.{workspace_id}.{session_id}.request"),
94 Self::DeadLetter {
95 workspace_id,
96 agent_id,
97 } => format!("vak.dlq.{workspace_id}.{agent_id}.failed"),
98 Self::Custom(s) => s.clone(),
99 }
100 }
101
102 /// Parse a wire subject string into a typed Subject.
103 pub fn parse(s: &str) -> Result<Self, SubjectError> {
104 let parts: Vec<&str> = s.split('.').collect();
105 if parts.is_empty() {
106 return Err(SubjectError::Empty);
107 }
108 if parts[0] != "vak" {
109 return Err(SubjectError::InvalidPrefix(s.to_string()));
110 }
111
112 if parts.len() < 4 {
113 return Ok(Self::Custom(s.to_string()));
114 }
115
116 match (parts[1], parts.len()) {
117 ("events", 5) if parts[4] == "tokens" => Ok(Self::EventsTokens {
118 workspace_id: parts[2].to_string(),
119 session_id: parts[3].to_string(),
120 }),
121 ("events", 5) if parts[4] == "terminal" => Ok(Self::EventsTerminal {
122 workspace_id: parts[2].to_string(),
123 session_id: parts[3].to_string(),
124 }),
125 ("events", 5) if parts[4] == "telemetry" => Ok(Self::EventsTelemetry {
126 workspace_id: parts[2].to_string(),
127 session_id: parts[3].to_string(),
128 }),
129 ("work", 5) if parts[4] == "task" => Ok(Self::WorkTask {
130 workspace_id: parts[2].to_string(),
131 role: parts[3].to_string(),
132 }),
133 ("agent", 5) if parts[4] == "inbox" => Ok(Self::AgentInbox {
134 workspace_id: parts[2].to_string(),
135 agent_id: parts[3].to_string(),
136 }),
137 ("receipts", 5) if parts[4] == "completed" => Ok(Self::Receipts {
138 workspace_id: parts[2].to_string(),
139 session_id: parts[3].to_string(),
140 }),
141 ("approvals", 5) if parts[4] == "request" => Ok(Self::Approvals {
142 workspace_id: parts[2].to_string(),
143 session_id: parts[3].to_string(),
144 }),
145 ("dlq", 5) if parts[4] == "failed" => Ok(Self::DeadLetter {
146 workspace_id: parts[2].to_string(),
147 agent_id: parts[3].to_string(),
148 }),
149 _ => Ok(Self::Custom(s.to_string())),
150 }
151 }
152}
153
154/// Role-based subject ACL evaluator enforcing least-privilege topic isolation.
155#[derive(Debug, Clone, Default, Serialize, Deserialize)]
156pub struct AclPolicy {
157 pub allow_publish: Vec<String>,
158 pub deny_publish: Vec<String>,
159 pub allow_subscribe: Vec<String>,
160 pub deny_subscribe: Vec<String>,
161}
162
163impl AclPolicy {
164 /// Construct a least-privilege policy for a specific worker agent.
165 pub fn for_worker(workspace_id: &str, session_id: &str, agent_id: &str, role: &str) -> Self {
166 Self {
167 allow_publish: vec![
168 format!("vak.events.{workspace_id}.{session_id}.>"),
169 format!("vak.receipts.{workspace_id}.{session_id}.*"),
170 format!("vak.approvals.{workspace_id}.{session_id}.*"),
171 format!("vak.dlq.{workspace_id}.{agent_id}.*"),
172 ],
173 deny_publish: vec![
174 format!("vak.work.{workspace_id}.>"), // Workers cannot forge work items
175 ],
176 allow_subscribe: vec![
177 format!("vak.agent.{workspace_id}.{agent_id}.inbox"),
178 format!("vak.work.{workspace_id}.{role}.task"),
179 ],
180 deny_subscribe: vec![
181 format!("vak.approvals.{workspace_id}.>"), // Workers cannot snoop on approval channels
182 ],
183 }
184 }
185
186 /// Check whether publishing to `subject` is permitted.
187 pub fn can_publish(&self, subject: &str) -> bool {
188 // Deny rules take strict precedence
189 for deny in &self.deny_publish {
190 if matches_pattern(deny, subject) {
191 return false;
192 }
193 }
194 for allow in &self.allow_publish {
195 if matches_pattern(allow, subject) {
196 return true;
197 }
198 }
199 false
200 }
201
202 /// Check whether subscribing to `subject` is permitted.
203 pub fn can_subscribe(&self, subject: &str) -> bool {
204 // Deny rules take strict precedence
205 for deny in &self.deny_subscribe {
206 if matches_pattern(deny, subject) {
207 return false;
208 }
209 }
210 for allow in &self.allow_subscribe {
211 if matches_pattern(allow, subject) {
212 return true;
213 }
214 }
215 false
216 }
217}
218
219/// Matches a NATS subject pattern containing `*` (single token) or `>` (multi-token tail).
220pub fn matches_pattern(pattern: &str, subject: &str) -> bool {
221 let p_tokens: Vec<&str> = pattern.split('.').collect();
222 let s_tokens: Vec<&str> = subject.split('.').collect();
223
224 let mut i = 0;
225 while i < p_tokens.len() {
226 if p_tokens[i] == ">" {
227 // '>' matches everything from here to the end
228 return true;
229 }
230 if i >= s_tokens.len() {
231 return false;
232 }
233 if p_tokens[i] != "*" && p_tokens[i] != s_tokens[i] {
234 return false;
235 }
236 i += 1;
237 }
238
239 i == s_tokens.len()
240}
241
242#[cfg(test)]
243mod tests {
244 use super::*;
245
246 #[test]
247 fn subject_pattern_matching() {
248 assert!(matches_pattern(
249 "vak.events.ws1.>",
250 "vak.events.ws1.sess1.tokens"
251 ));
252 assert!(matches_pattern(
253 "vak.events.ws1.sess1.*",
254 "vak.events.ws1.sess1.tokens"
255 ));
256 assert!(!matches_pattern(
257 "vak.events.ws1.sess1.*",
258 "vak.events.ws1.sess1.tokens.extra"
259 ));
260 assert!(!matches_pattern(
261 "vak.events.ws2.>",
262 "vak.events.ws1.sess1.tokens"
263 ));
264 }
265
266 #[test]
267 fn acl_worker_enforcement() {
268 let policy = AclPolicy::for_worker("ws_alpha", "sess_001", "agent_coder", "coder");
269
270 // Allowed publish
271 assert!(policy.can_publish("vak.events.ws_alpha.sess_001.tokens"));
272 assert!(policy.can_publish("vak.events.ws_alpha.sess_001.terminal"));
273 assert!(policy.can_publish("vak.receipts.ws_alpha.sess_001.completed"));
274
275 // Denied publish: cannot publish to work queue (denied) or other session
276 assert!(!policy.can_publish("vak.work.ws_alpha.coder.task"));
277 assert!(!policy.can_publish("vak.events.ws_alpha.sess_002.tokens"));
278
279 // Allowed subscribe
280 assert!(policy.can_subscribe("vak.agent.ws_alpha.agent_coder.inbox"));
281 assert!(policy.can_subscribe("vak.work.ws_alpha.coder.task"));
282
283 // Denied subscribe: cannot snoop other agent inboxes
284 assert!(!policy.can_subscribe("vak.agent.ws_alpha.agent_other.inbox"));
285 }
286}
287

1//! Deterministic subject namespace algebra and role-based ACL evaluator.
2//!
3//! NATS subject format:
4//! `vak.<plane>.<workspace_id>.<entity>.<target_or_session>.<verb_or_type>`
5
6use serde::{Deserialize, Serialize};
7use thiserror::Error;
8
9#[derive(Debug, Error)]
10pub enum SubjectError {
11 #[error("empty subject string")]
12 Empty,
13 #[error("subject must start with 'vak.': {0}")]
14 InvalidPrefix(String),
15 #[error("invalid subject structure: {0}")]
16 InvalidStructure(String),
17}
18
19/// Typed subject definitions across the ephemeral event and durable work planes.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub enum Subject {
22 /// Ephemeral streaming assistant & thinking tokens
23 EventsTokens {
24 workspace_id: String,
25 session_id: String,
26 },
27 /// Ephemeral streaming ANSI stdout/stderr chunks
28 EventsTerminal {
29 workspace_id: String,
30 session_id: String,
31 },
32 /// Ephemeral streaming process telemetry (500ms probe)
33 EventsTelemetry {
34 workspace_id: String,
35 session_id: String,
36 },
37 /// Durable JetStream work queue for worker task claiming
38 WorkTask { workspace_id: String, role: String },
39 /// Durable point-to-point agent steering & direct inbox
40 AgentInbox {
41 workspace_id: String,
42 agent_id: String,
43 },
44 /// Durable verified work receipts and execution evidence
45 Receipts {
46 workspace_id: String,
47 session_id: String,
48 },
49 /// Durable elevated permission approval requests & verdicts
50 Approvals {
51 workspace_id: String,
52 session_id: String,
53 },
54 /// Durable Dead-Letter Queue (DLQ) for poison tasks
55 DeadLetter {
56 workspace_id: String,
57 agent_id: String,
58 },
59 /// Generic or custom subject string
60 Custom(String),
61}
62
63impl Subject {
64 /// Render subject to canonical NATS wire string.
65 pub fn to_subject_string(&self) -> String {
66 match self {
67 Self::EventsTokens {
68 workspace_id,
69 session_id,
70 } => format!("vak.events.{workspace_id}.{session_id}.tokens"),
71 Self::EventsTerminal {
72 workspace_id,
73 session_id,
74 } => format!("vak.events.{workspace_id}.{session_id}.terminal"),
75 Self::EventsTelemetry {
76 workspace_id,
77 session_id,
78 } => format!("vak.events.{workspace_id}.{session_id}.telemetry"),
79 Self::WorkTask { workspace_id, role } => {
80 format!("vak.work.{workspace_id}.{role}.task")
81 }
82 Self::AgentInbox {
83 workspace_id,
84 agent_id,
85 } => format!("vak.agent.{workspace_id}.{agent_id}.inbox"),
86 Self::Receipts {
87 workspace_id,
88 session_id,
89 } => format!("vak.receipts.{workspace_id}.{session_id}.completed"),
90 Self::Approvals {
91 workspace_id,
92 session_id,
93 } => format!("vak.approvals.{workspace_id}.{session_id}.request"),
94 Self::DeadLetter {
95 workspace_id,
96 agent_id,
97 } => format!("vak.dlq.{workspace_id}.{agent_id}.failed"),
98 Self::Custom(s) => s.clone(),
99 }
100 }
101
102 /// Parse a wire subject string into a typed Subject.
103 pub fn parse(s: &str) -> Result<Self, SubjectError> {
104 let parts: Vec<&str> = s.split('.').collect();
105 if parts.is_empty() {
106 return Err(SubjectError::Empty);
107 }
108 if parts[0] != "vak" {
109 return Err(SubjectError::InvalidPrefix(s.to_string()));
110 }
111
112 if parts.len() < 4 {
113 return Ok(Self::Custom(s.to_string()));
114 }
115
116 match (parts[1], parts.len()) {
117 ("events", 5) if parts[4] == "tokens" => Ok(Self::EventsTokens {
118 workspace_id: parts[2].to_string(),
119 session_id: parts[3].to_string(),
120 }),
121 ("events", 5) if parts[4] == "terminal" => Ok(Self::EventsTerminal {
122 workspace_id: parts[2].to_string(),
123 session_id: parts[3].to_string(),
124 }),
125 ("events", 5) if parts[4] == "telemetry" => Ok(Self::EventsTelemetry {
126 workspace_id: parts[2].to_string(),
127 session_id: parts[3].to_string(),
128 }),
129 ("work", 5) if parts[4] == "task" => Ok(Self::WorkTask {
130 workspace_id: parts[2].to_string(),
131 role: parts[3].to_string(),
132 }),
133 ("agent", 5) if parts[4] == "inbox" => Ok(Self::AgentInbox {
134 workspace_id: parts[2].to_string(),
135 agent_id: parts[3].to_string(),
136 }),
137 ("receipts", 5) if parts[4] == "completed" => Ok(Self::Receipts {
138 workspace_id: parts[2].to_string(),
139 session_id: parts[3].to_string(),
140 }),
141 ("approvals", 5) if parts[4] == "request" => Ok(Self::Approvals {
142 workspace_id: parts[2].to_string(),
143 session_id: parts[3].to_string(),
144 }),
145 ("dlq", 5) if parts[4] == "failed" => Ok(Self::DeadLetter {
146 workspace_id: parts[2].to_string(),
147 agent_id: parts[3].to_string(),
148 }),
149 _ => Ok(Self::Custom(s.to_string())),
150 }
151 }
152}
153
154/// Role-based subject ACL evaluator enforcing least-privilege topic isolation.
155#[derive(Debug, Clone, Default, Serialize, Deserialize)]
156pub struct AclPolicy {
157 pub allow_publish: Vec<String>,
158 pub deny_publish: Vec<String>,
159 pub allow_subscribe: Vec<String>,
160 pub deny_subscribe: Vec<String>,
161}
162
163impl AclPolicy {
164 /// Construct a least-privilege policy for a specific worker agent.
165 pub fn for_worker(workspace_id: &str, session_id: &str, agent_id: &str, role: &str) -> Self {
166 Self {
167 allow_publish: vec![
168 format!("vak.events.{workspace_id}.{session_id}.>"),
169 format!("vak.receipts.{workspace_id}.{session_id}.*"),
170 format!("vak.approvals.{workspace_id}.{session_id}.*"),
171 format!("vak.dlq.{workspace_id}.{agent_id}.*"),
172 ],
173 deny_publish: vec![
174 format!("vak.work.{workspace_id}.>"), // Workers cannot forge work items
175 ],
176 allow_subscribe: vec![
177 format!("vak.agent.{workspace_id}.{agent_id}.inbox"),
178 format!("vak.work.{workspace_id}.{role}.task"),
179 ],
180 deny_subscribe: vec![
181 format!("vak.approvals.{workspace_id}.>"), // Workers cannot snoop on approval channels
182 ],
183 }
184 }
185
186 /// Check whether publishing to `subject` is permitted.
187 pub fn can_publish(&self, subject: &str) -> bool {
188 // Deny rules take strict precedence
189 for deny in &self.deny_publish {
190 if matches_pattern(deny, subject) {
191 return false;
192 }
193 }
194 for allow in &self.allow_publish {
195 if matches_pattern(allow, subject) {
196 return true;
197 }
198 }
199 false
200 }
201
202 /// Check whether subscribing to `subject` is permitted.
203 pub fn can_subscribe(&self, subject: &str) -> bool {
204 // Deny rules take strict precedence
205 for deny in &self.deny_subscribe {
206 if matches_pattern(deny, subject) {
207 return false;
208 }
209 }
210 for allow in &self.allow_subscribe {
211 if matches_pattern(allow, subject) {
212 return true;
213 }
214 }
215 false
216 }
217}
218
219/// Matches a NATS subject pattern containing `*` (single token) or `>` (multi-token tail).
220pub fn matches_pattern(pattern: &str, subject: &str) -> bool {
221 let p_tokens: Vec<&str> = pattern.split('.').collect();
222 let s_tokens: Vec<&str> = subject.split('.').collect();
223
224 let mut i = 0;
225 while i < p_tokens.len() {
226 if p_tokens[i] == ">" {
227 // '>' matches everything from here to the end
228 return true;
229 }
230 if i >= s_tokens.len() {
231 return false;
232 }
233 if p_tokens[i] != "*" && p_tokens[i] != s_tokens[i] {
234 return false;
235 }
236 i += 1;
237 }
238
239 i == s_tokens.len()
240}
241
242#[cfg(test)]
243mod tests {
244 use super::*;
245
246 #[test]
247 fn subject_pattern_matching() {
248 assert!(matches_pattern(
249 "vak.events.ws1.>",
250 "vak.events.ws1.sess1.tokens"
251 ));
252 assert!(matches_pattern(
253 "vak.events.ws1.sess1.*",
254 "vak.events.ws1.sess1.tokens"
255 ));
256 assert!(!matches_pattern(
257 "vak.events.ws1.sess1.*",
258 "vak.events.ws1.sess1.tokens.extra"
259 ));
260 assert!(!matches_pattern(
261 "vak.events.ws2.>",
262 "vak.events.ws1.sess1.tokens"
263 ));
264 }
265
266 #[test]
267 fn acl_worker_enforcement() {
268 let policy = AclPolicy::for_worker("ws_alpha", "sess_001", "agent_coder", "coder");
269
270 // Allowed publish
271 assert!(policy.can_publish("vak.events.ws_alpha.sess_001.tokens"));
272 assert!(policy.can_publish("vak.events.ws_alpha.sess_001.terminal"));
273 assert!(policy.can_publish("vak.receipts.ws_alpha.sess_001.completed"));
274
275 // Denied publish: cannot publish to work queue (denied) or other session
276 assert!(!policy.can_publish("vak.work.ws_alpha.coder.task"));
277 assert!(!policy.can_publish("vak.events.ws_alpha.sess_002.tokens"));
278
279 // Allowed subscribe
280 assert!(policy.can_subscribe("vak.agent.ws_alpha.agent_coder.inbox"));
281 assert!(policy.can_subscribe("vak.work.ws_alpha.coder.task"));
282
283 // Denied subscribe: cannot snoop other agent inboxes
284 assert!(!policy.can_subscribe("vak.agent.ws_alpha.agent_other.inbox"));
285 }
286}
287

Subjects

Indexing the workspace…

Subjects