11. 11use async_trait::async_trait;
12. 12use serde_json::Value;
13. 13use sha2::{Digest, Sha256};
14. 14use tokio_util::sync::CancellationToken;
15. 15use vak_llm::Provider;
16. 16use vak_permission::{Mode, PermissionEngine};
17. 17use vak_session::types::{CapabilityDescriptor, CapabilityKind, FrozenContract, SessionHeader};
18. 18use vak_session::{SessionLog, SessionPath};
19. 19use vak_tools::sandbox::Sandbox;
20. 20use vak_tools::{Tool, ToolContext, ToolOutput};
21. 21
22. 22use crate::{
23. 23 Agent, AgentConfig, ApprovalMode, Approver, InputNormalizer, McpToolIndex, SteeringQueues,
24. 24};
25. 25
26. 26fn vak_core_identity() -> vak_session::types::AgentIdentity {
27. 27 vak_session::types::AgentIdentity {
28. 28 id: "vak".into(),
29. 29 revision: 1,
30. 30 name: "Vakyartha".into(),
31. 31 character: "vak".into(),
32. 32 personality: String::new(),
33. 33 animation: "subtle".into(),
34. 34 voice: "default".into(),
35. 35 behaviour: String::new(),
36. 36 responsibilities: String::new(),
37. 37 instructions: String::new(),
38. 38 }
39. 39}
40. 40
41. 41pub struct TaskDeps {
42. 42 /// Resolved parent Agent identity; inherited by default-delegated children.
43. 43 pub parent_agent_identity: Option<vak_session::types::AgentIdentity>,
44. 44 /// Parent outcome context carried into the child for alignment only.
45. 45 pub outcome_objective: Option<String>,
46. 46 /// The parent's admitted outcome, narrowed for this child at dispatch.
47. 47 pub outcome: Option<vak_intent::OutcomeSpec>,
48. 48 /// Prompts for named roles, admitted up front by the host exactly like
49. 49 /// capabilities are. A child can only ever run under a role that was
50. 50 /// resolvable when the parent session was admitted, so an unknown or
51. 51 /// injected role name cannot conjure new instructions mid-run.
52. 52 pub role_prompts: std::collections::BTreeMap<String, String>,
53. 53 pub provider: Arc<dyn Provider>,
54. 54 pub system_prompt: String,
55. 55 /// The parent turn's tail (clock instant + epistemic stance,
56. 56 /// docs/design/68-context-engine.md §6/§10). Children get the same
57. 57 /// turn context block as the parent rather than an empty one, since a
58. 58 /// worker dispatched mid-turn is still answering as of that turn's
59. 59 /// instant and stance.
60. 60 pub tail: crate::TailInput,
61. 61 pub model: String,
62. 62 pub tools: Vec<Arc<dyn Tool>>,
63. 63 pub capabilities: Vec<CapabilityDescriptor>,
64. 64 pub hooks: Option<Arc<Vec<vak_hooks::HookDef>>>,
65. 65 pub revocation_check: Option<crate::RevocationCheck>,
66. 66 /// Records a worker's cards in its own ledger as they validate, so they
67. 67 /// can be handed to the delegating conversation when the worker ends.
68. 68 pub presentation_rebuild: Option<crate::PresentationRebuild>,
69. 69 pub mcp_tool_index: Option<McpToolIndex>,
70. 70 pub input_normalizer: Option<InputNormalizer>,
71. 71 /// Read-only subset (read/glob/grep) used when a task declares
72. 72 /// `readonly: true`; children get these plus ReadOnly permission mode.
73. 73 pub read_only_tools: Vec<Arc<dyn Tool>>,
74. 74 pub max_turns: usize,
75. 75 pub max_retries: u32,
76. 76 pub retry_base_backoff_ms: u64,
77. 77 pub request_timeout: Option<std::time::Duration>,
78. 78 pub circuit_breaker: Option<Arc<crate::CircuitBreaker>>,
79. 79 pub run_retry_attempts: u32,
80. 80 pub run_retry_base_backoff_ms: u64,
81. 81 pub dispatch_ceiling: u32,
82. 82 pub spend_gate: Option<Arc<dyn crate::SpendGate>>,
83. 83 pub permission: Option<Arc<PermissionEngine>>,
84. 84 pub mode: Mode,
85. 85 pub approval_mode: ApprovalMode,
86. 86 pub approver: Option<Arc<dyn Approver>>,
87. 87 pub sandbox: Option<Arc<dyn Sandbox>>,
88. 88 pub cwd: PathBuf,
89. 89 pub sessions_home: PathBuf,
90. 90 pub parent_session_id: String,
91. 91 pub contract_id: Option<String>,
92. 92 pub work_item_id: Option<String>,
93. 93 pub work_item_ids: Vec<String>,
94. 94 /// Parent-loop event channel so worker lifecycles surface in the UI.
95. 95 pub events: Option<tokio::sync::mpsc::Sender<crate::AgentEvent>>,
96. 96 /// Shared registry of live children. None disables attach/steer (the
97. 97 /// child still runs normally).
98. 98 pub registry: Option<Arc<WorkerRegistry>>,
99. 99}
100. 100
101. 101pub struct TaskTool {
102. 102 deps: Arc<TaskDeps>,
103. 103}
104. 104
105. 105#[derive(serde::Deserialize)]
106. 106struct AgentDefinition {
107. 107 id: String,
108. 108 #[serde(default = "default_agent_revision")]
109. 109 revision: u64,
110. 110 #[serde(default = "default_agent_lifecycle")]
111. 111 lifecycle: String,
112. 112 name: String,
113. 113 character: String,
114. 114 personality: String,
115. 115 behaviour: String,
116. 116 #[serde(default)]
117. 117 responsibilities: String,
118. 118 #[serde(default)]
119. 119 instructions: String,
120. 120}
121. 121
122. 122fn default_agent_revision() -> u64 {
123. 123 1
124. 124}
125. 125
126. 126fn default_agent_lifecycle() -> String {
127. 127 "active".into()
128. 128}
129. 129
130. 130fn load_agent(cwd: &std::path::Path, requested: &str) -> Result<Option<AgentDefinition>, String> {
131. 131 // Delegated Agents resolve the same effective Shared → trusted project
132. 132 // layers as a top-level Agent. Previously this delegated helper read only the
133. 133 // project file, so a user-level Agent worked from the sidebar and
134. 134 // scheduler but was invisible to `task(agent=...)`.
135. 135 let shared = vak_config::paths::default_workspace();
136. 136 let mut profiles = read_agents(&shared)?;
137. 137 if cwd != shared {
138. 138 for profile in read_agents(cwd)? {
139. 139 profiles.retain(|candidate| candidate.id != profile.id);
140. 140 profiles.push(profile);
141. 141 }
142. 142 }
143. 143 if let Some(profile) = profiles.iter().find(|profile| profile.id == requested) {
144. 144 if profile.lifecycle != "active" {
145. 145 return Err(format!(
146. 146 "Agent '{}' is {} and cannot be selected for delegated work",
147. 147 profile.id, profile.lifecycle
148. 148 ));
149. 149 }
150. 150 return Ok(Some(AgentDefinition {
151. 151 id: profile.id.clone(),
152. 152 revision: profile.revision,
153. 153 lifecycle: profile.lifecycle.clone(),
154. 154 name: profile.name.clone(),
155. 155 character: profile.character.clone(),
156. 156 personality: profile.personality.clone(),
157. 157 behaviour: profile.behaviour.clone(),
158. 158 responsibilities: profile.responsibilities.clone(),
159. 159 instructions: profile.instructions.clone(),
160. 160 }));
161. 161 }
162. 162 let matches = profiles
163. 163 .into_iter()
164. 164 .filter(|profile| profile.name.eq_ignore_ascii_case(requested))
165. 165 .collect::<Vec<_>>();
166. 166 match matches.len() {
167. 167 0 => Ok(None),
168. 168 1 => {
169. 169 let agent = matches.into_iter().next();
170. 170 if let Some(agent) = &agent
171. 171 && agent.lifecycle != "active"
172. 172 {
173. 173 return Err(format!(
174. 174 "Agent '{}' is {} and cannot be selected for delegated work",
175. 175 agent.id, agent.lifecycle
176. 176 ));
177. 177 }
178. 178 Ok(agent)
179. 179 }
180. 180 _ => Err(format!(
181. 181 "Agent name '{}' is ambiguous; choose an Agent by its exact name or id",
182. 182 requested
183. 183 )),
184. 184 }
185. 185}
186. 186
187. 187fn read_agents(cwd: &std::path::Path) -> Result<Vec<AgentDefinition>, String> {
188. 188 let path = cwd.join(".vak/agents.json");
189. 189 let Ok(raw) = std::fs::read_to_string(path) else {
190. 190 return Ok(Vec::new());
191. 191 };
192. 192 serde_json::from_str::<Vec<AgentDefinition>>(&raw)
193. 193 .map_err(|_| "saved Agent definitions are invalid".to_string())
194. 194}
195. 195
196. 196struct RegistryGuard {
197. 197 registry: Arc<WorkerRegistry>,
198. 198 id: String,
199. 199}
200. 200
201. 201impl Drop for RegistryGuard {
202. 202 fn drop(&mut self) {
203. 203 self.registry.unregister(&self.id);
204. 204 }
205. 205}
206. 206
207. 207/// A live child agent: its steering queues and cancellation token, so an
208. 208/// attached UI can push steering or stop it while the parent's task tool
209. 209/// call is still blocking.
210. 210#[derive(Debug)]
211. 211pub struct WorkerHandle {
212. 212 pub label: String,
213. 213 pub agent_id: Option<String>,
214. 214 pub agent_revision: Option<u64>,
215. 215 pub started_at: std::time::Instant,
216. 216 pub steering: Arc<SteeringQueues>,
217. 217 pub cancel: CancellationToken,
218. 218 /// Session that spawned this child; routes registry lookups to the
219. 219 /// owning surface's endpoint scope.
220. 220 pub parent_session_id: String,
221. 221}
222. 222
223. 223#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
224. 224pub struct ActiveWorker {
225. 225 pub id: String,
226. 226 pub label: String,
227. 227 pub agent_id: Option<String>,
228. 228 pub agent_revision: Option<u64>,
229. 229 pub elapsed_secs: u64,
230. 230 pub parent_session_id: String,
231. 231}
232. 232
233. 233/// Registry of currently-running workers, keyed by unique child session
234. 234/// id. Interior-mutable: the UI holds a shared reference across runs.
235. 235#[derive(Debug, Default)]
236. 236pub struct WorkerRegistry {
237. 237 inner: Mutex<BTreeMap<String, WorkerHandle>>,
238. 238}
239. 239
240. 240impl WorkerRegistry {
241. 241 pub fn new() -> Self {
242. 242 Self::default()
243. 243 }
244. 244
245. 245 fn register(&self, id: String, handle: WorkerHandle) {
246. 246 if let Ok(mut map) = self.inner.lock() {
247. 247 map.insert(id, handle);
248. 248 }
249. 249 }
250. 250
251. 251 fn unregister(&self, id: &str) {
252. 252 if let Ok(mut map) = self.inner.lock() {
253. 253 map.remove(id);
254. 254 }
255. 255 }
256. 256
257. 257 pub fn active(&self) -> Vec<ActiveWorker> {
258. 258 let Ok(map) = self.inner.lock() else {
259. 259 return Vec::new();
260. 260 };
261. 261 map.iter()
262. 262 .map(|(id, h)| ActiveWorker {
263. 263 id: id.clone(),
264. 264 label: h.label.clone(),
265. 265 agent_id: h.agent_id.clone(),
266. 266 agent_revision: h.agent_revision,
267. 267 elapsed_secs: h.started_at.elapsed().as_secs(),
268. 268 parent_session_id: h.parent_session_id.clone(),
269. 269 })
270. 270 .collect()
271. 271 }
272. 272
273. 273 /// Live children spawned by `parent`, oldest first.
274. 274 pub fn active_for(&self, parent: &str) -> Vec<ActiveWorker> {
275. 275 let Ok(map) = self.inner.lock() else {
276. 276 return Vec::new();
277. 277 };
278. 278 map.iter()
279. 279 .filter(|(_, h)| h.parent_session_id == parent)
280. 280 .map(|(id, h)| ActiveWorker {
281. 281 id: id.clone(),
282. 282 label: h.label.clone(),
283. 283 agent_id: h.agent_id.clone(),
284. 284 agent_revision: h.agent_revision,
285. 285 elapsed_secs: h.started_at.elapsed().as_secs(),
286. 286 parent_session_id: h.parent_session_id.clone(),
287. 287 })
288. 288 .collect()
289. 289 }
290. 290
291. 291 /// Owning session of a live child, for endpoint-scope checks.
292. 292 pub fn parent_of(&self, id: &str) -> Option<String> {
293. 293 let map = self.inner.lock().ok()?;
294. 294 map.get(id).map(|h| h.parent_session_id.clone())
295. 295 }
296. 296
297. 297 /// Queues steering text for the child. Returns false when no such
298. 298 /// child is live.
299. 299 pub fn steer(&self, id: &str, text: &str) -> bool {
300. 300 let Ok(map) = self.inner.lock() else {
301. 301 return false;
302. 302 };
303. 303 match map.get(id) {
304. 304 Some(h) => {
305. 305 h.steering.push_steering(text.to_string());
306. 306 true
307. 307 }
308. 308 None => false,
309. 309 }
310. 310 }
311. 311
312. 312 /// Queues a follow-up turn for the child (runs after its natural stop).
313. 313 pub fn queue_follow_up(&self, id: &str, text: &str) -> bool {
314. 314 let Ok(map) = self.inner.lock() else {
315. 315 return false;
316. 316 };
317. 317 match map.get(id) {
318. 318 Some(h) => {
319. 319 h.steering.push_follow_up(text.to_string());
320. 320 true
321. 321 }
322. 322 None => false,
323. 323 }
324. 324 }
325. 325
326. 326 /// Cancels the child. Returns false when no such child is live.
327. 327 pub fn stop(&self, id: &str) -> bool {
328. 328 let Ok(map) = self.inner.lock() else {
329. 329 return false;
330. 330 };
331. 331 match map.get(id) {
332. 332 Some(h) => {
333. 333 h.cancel.cancel();
334. 334 true
335. 335 }
336. 336 None => false,
337. 337 }
338. 338 }
339. 339}
340. 340
341. 341impl TaskTool {
342. 342 /// Declared as a constant so capability declarations can read it
343. 343 /// without constructing the tool's dependencies.
344. 344 pub const SERVES: &'static [&'static str] = &["orchestration"];
345. 345
346. 346 /// Constant so the prompt's tool catalogue can list the tool before its
347. 347 /// turn-bound dependencies exist.
348. 348 pub const DESCRIPTION: &'static str = "Delegate a self-contained subtask to a worker with its own context window and transcript. Use for focused research or exploration whose details you do not need in your own context. Optionally select a saved Agent so its identity and working style are applied; this never changes permissions. The worker cannot spawn further workers.";
349. 349
350. 350 pub fn new(deps: TaskDeps) -> Self {
351. 351 TaskTool {
352. 352 deps: Arc::new(deps),
353. 353 }
354. 354 }
355. 355}
356. 356
357. 357#[async_trait]
358. 358impl Tool for TaskTool {
359. 359 fn name(&self) -> &str {
360. 360 "task"
361. 361 }
362. 362
363. 363 fn serves(&self) -> &'static [&'static str] {
364. 364 Self::SERVES
365. 365 }
366. 366
367. 367 fn description(&self) -> &str {
368. 368 Self::DESCRIPTION
369. 369 }
370. 370
371. 371 fn schema(&self) -> Value {
372. 372 // The admitted roles go in the schema as an `enum` rather than in
373. 373 // prose: it is the only form the provider will actually constrain
374. 374 // the model against, and an unadmitted name is refused at call time
375. 375 // anyway.
376. 376 let roles: Vec<&str> = self.deps.role_prompts.keys().map(String::as_str).collect();
377. 377 let mut role_property = serde_json::json!({
378. 378 "type": "string",
379. 379 "description": "Named role whose instructions this child runs under. Omit to use the default worker instructions."
380. 380 });
381. 381 if !roles.is_empty() {
382. 382 role_property["enum"] = serde_json::json!(roles);
383. 383 }
384. 384 serde_json::json!({
385. 385 "type": "object",
386. 386 "properties": {
387. 387 "prompt": {"type": "string", "description": "Complete, self-contained instructions for the worker"},
388. 388 "role": role_property,
389. 389 "agent": {"type": "string", "description": "Optional saved Agent name or id. Applies its identity and working style without changing permissions."},
390. 390 "label": {"type": "string", "description": "Short label shown in the UI"},
391. 391 "readonly": {"type": "boolean", "description": "If true, the worker gets only read/glob/grep and may run concurrently with other tasks", "default": false},
392. 392 "paths": {"type": "array", "items": {"type": "string"}, "description": "Path scopes (globs) this task will write to; tasks with disjoint scopes run in parallel, overlapping scopes are serialized"},
393. 393 "contract_id": {"type": "string", "description": "Managed contract this child is executing"},
394. 394 "work_item_id": {"type": "string", "description": "Managed work item assigned to this child"}
395. 395 },
396. 396 "required": ["prompt"]
397. 397 })
398. 398 }
399. 399
400. 400 fn claims(&self, args: &Value) -> vak_tools::ResourceClaims {
401. 401 let readonly = args
402. 402 .get("readonly")
403. 403 .and_then(|r| r.as_bool())
404. 404 .unwrap_or(false);
405. 405 if readonly {
406. 406 return vak_tools::ResourceClaims {
407. 407 exclusive: false,
408. 408 read_only: true,
409. 409 paths: Vec::new(),
410. 410 };
411. 411 }
412. 412 let paths: Vec<String> = args
413. 413 .get("paths")
414. 414 .and_then(|p| p.as_array())
415. 415 .map(|a| {
416. 416 a.iter()
417. 417 .filter_map(|v| v.as_str().map(String::from))
418. 418 .collect()
419. 419 })
420. 420 .unwrap_or_default();
421. 421 vak_tools::ResourceClaims {
422. 422 exclusive: paths.is_empty(),
423. 423 read_only: false,
424. 424 paths,
425. 425 }
426. 426 }
427. 427
428. 428 async fn execute(&self, args: &Value, ctx: &ToolContext) -> ToolOutput {
429. 429 self.execute_inner(args, ctx).await
430. 430 }
431. 431}
432. 432
433. 433impl TaskTool {
434. 434 async fn execute_inner(&self, args: &Value, ctx: &ToolContext) -> ToolOutput {
435. 435 let Some(prompt) = args.get("prompt").and_then(|p| p.as_str()) else {
436. 436 return ToolOutput::error("missing required parameter: prompt");
437. 437 };
438. 438 let requested_contract = args.get("contract_id").and_then(|value| value.as_str());
439. 439 let requested_item = args.get("work_item_id").and_then(|value| value.as_str());
440. 440 if requested_contract.is_some() != requested_item.is_some() {
441. 441 return ToolOutput::error("contract_id and work_item_id must be supplied together");
442. 442 }
443. 443 if let Some(contract_id) = requested_contract
444. 444 && self
445. 445 .deps
446. 446 .contract_id
447. 447 .as_deref()
448. 448 .is_some_and(|known| known != contract_id)
449. 449 {
450. 450 return ToolOutput::error(
451. 451 "child contract does not match the parent's managed contract",
452. 452 );
453. 453 }
454. 454 if let Some(item_id) = requested_item
455. 455 && !self.deps.work_item_ids.is_empty()
456. 456 && !self.deps.work_item_ids.iter().any(|known| known == item_id)
457. 457 {
458. 458 return ToolOutput::error(
459. 459 "child work item does not exist in the parent's managed contract",
460. 460 );
461. 461 }
462. 462 let session_id = next_child_session_id();
463. 463 let readonly = args
464. 464 .get("readonly")
465. 465 .and_then(|r| r.as_bool())
466. 466 .unwrap_or(false);
467. 467 let child_outcome = child_outcome(prompt, readonly, self.deps.outcome.as_ref());
468. 468 let child_tools: Vec<Arc<dyn Tool>> = if readonly {
469. 469 self.deps.read_only_tools.clone()
470. 470 } else {
471. 471 self.deps.tools.clone()
472. 472 }
473. 473 .into_iter()
474. 474 .filter(|tool| tool.name() != "flow")
475. 475 .collect();
476. 476 let child_mode = if readonly {
477. 477 Mode::ReadOnly
478. 478 } else {
479. 479 self.deps.mode
480. 480 };
481. 481 // An unknown role is refused rather than quietly ignored: a child
482. 482 // that silently ran under the default prompt when a role was asked
483. 483 // for would be the hardest kind of misconfiguration to notice.
484. 484 let mut child_system_prompt = match args.get("role").and_then(|r| r.as_str()) {
485. 485 Some(role) if !role.trim().is_empty() => {
486. 486 match self.deps.role_prompts.get(role.trim()) {
487. 487 Some(prompt) => prompt.clone(),
488. 488 None => {
489. 489 let known = self
490. 490 .deps
491. 491 .role_prompts
492. 492 .keys()
493. 493 .cloned()
494. 494 .collect::<Vec<_>>()
495. 495 .join(", ");
496. 496 return ToolOutput::error(if known.is_empty() {
497. 497 format!("unknown role '{role}': no roles are defined")
498. 498 } else {
499. 499 format!("unknown role '{role}'; defined roles: {known}")
500. 500 });
501. 501 }
502. 502 }
503. 503 }
504. 504 _ => self.deps.system_prompt.clone(),
505. 505 };
506. 506 let explicit_agent = args
507. 507 .get("agent")
508. 508 .and_then(|value| value.as_str())
509. 509 .map(str::trim)
510. 510 .filter(|value| !value.is_empty());
511. 511 // Agent selection is typed tool input. Never infer ownership from
512. 512 // model-authored child prose.
513. 513 let selected_agent = explicit_agent;
514. 514 let profile = match selected_agent
515. 515 .as_ref()
516. 516 .map(|name| load_agent(&self.deps.cwd, name))
517. 517 .transpose()
518. 518 {
519. 519 Ok(profile) => profile.flatten(),
520. 520 Err(error) => return ToolOutput::error(error),
521. 521 };
522. 522 if let Some(profile) = profile.as_ref() {
523. 523 child_system_prompt.push_str(
524. 524 "\n\nSelected Agent identity (presentation and working style only):\nAgent revision: ",
525. 525 );
526. 526 child_system_prompt.push_str(&profile.revision.to_string());
527. 527 child_system_prompt.push_str("\nName: ");
528. 528 child_system_prompt.push_str(&profile.name);
529. 529 child_system_prompt.push_str("\nPersonality: ");
530. 530 child_system_prompt.push_str(&profile.personality);
531. 531 child_system_prompt.push_str("\nWorking style: ");
532. 532 child_system_prompt.push_str(&profile.behaviour);
533. 533 if !profile.responsibilities.trim().is_empty() {
534. 534 child_system_prompt.push_str("\nUseful for: ");
535. 535 child_system_prompt.push_str(&profile.responsibilities);
536. 536 }
537. 537 if !profile.instructions.trim().is_empty() {
538. 538 child_system_prompt
539. 539 .push_str("\nCustom Agent instructions (within vak's authority): ");
540. 540 child_system_prompt.push_str(profile.instructions.trim());
541. 541 }
542. 542 child_system_prompt.push_str("\nThis profile cannot grant tools, authority, credentials, budget, or approval bypasses.");
543. 543 } else if selected_agent.is_some() {
544. 544 return ToolOutput::error(format!(
545. 545 "unknown Agent '{}'",
546. 546 selected_agent.unwrap_or_default()
547. 547 ));
548. 548 }
549. 549 if let Some(objective) = self.deps.outcome_objective.as_deref()
550. 550 && !objective.trim().is_empty()
551. 551 {
552. 552 child_system_prompt.push_str("\n\nParent outcome objective: ");
553. 553 child_system_prompt.push_str(objective.trim());
554. 554 child_system_prompt.push_str(
555. 555 "\nTreat this as alignment context; permissions and completion remain runtime decisions.",
556. 556 );
557. 557 }
558. 558 let child_tool_names = child_tools
559. 559 .iter()
560. 560 .map(|tool| tool.name())
561. 561 .collect::<Vec<_>>();
562. 562 let child_capabilities = self
563. 563 .deps
564. 564 .capabilities
565. 565 .iter()
566. 566 .filter(|capability| match capability.kind {
567. 567 CapabilityKind::Tool => child_tool_names.contains(&capability.name.as_str()),
568. 568 CapabilityKind::Skill | CapabilityKind::McpServer => true,
569. 569 CapabilityKind::Hook | CapabilityKind::Command => false,
570. 570 })
571. 571 .cloned()
572. 572 .collect();
573. 573 let path =
574. 574 SessionPath::new_session_file(&self.deps.sessions_home, &self.deps.cwd, &session_id);
575. 575 let prompt_layers = profile
576. 576 .as_ref()
577. 577 .map(|profile| vak_session::types::AgentIdentity {
578. 578 id: profile.id.clone(),
579. 579 revision: profile.revision,
580. 580 name: profile.name.clone(),
581. 581 character: profile.character.clone(),
582. 582 personality: profile.personality.clone(),
583. 583 animation: "subtle".into(),
584. 584 voice: "default".into(),
585. 585 behaviour: profile.behaviour.clone(),
586. 586 responsibilities: profile.responsibilities.clone(),
587. 587 instructions: profile.instructions.clone(),
588. 588 })
589. 589 .or_else(|| self.deps.parent_agent_identity.clone())
590. 590 .as_ref()
591. 591 .map(|identity| {
592. 592 let text = format!(
593. 593 "{}\n{}\n{}\n{}\n{}",
594. 594 identity.name,
595. 595 identity.personality,
596. 596 identity.behaviour,
597. 597 identity.responsibilities,
598. 598 identity.instructions
599. 599 );
600. 600 let digest = format!("{:x}", Sha256::digest(text.as_bytes()));
601. 601 vec![vak_session::types::PromptLayerDescriptor {
602. 602 block: "identity".into(),
603. 603 layer: "agent".into(),
604. 604 source: Some(identity.id.clone()),
605. 605 digest,
606. 606 bytes: text.len(),
607. 607 }]
608. 608 })
609. 609 .unwrap_or_default();
610. 610 let header = SessionHeader {
611. 611 agent: profile
612. 612 .as_ref()
613. 613 .map(|profile| vak_session::types::AgentIdentity {
614. 614 id: profile.id.clone(),
615. 615 revision: profile.revision,
616. 616 name: profile.name.clone(),
617. 617 character: profile.character.clone(),
618. 618 personality: profile.personality.clone(),
619. 619 animation: "subtle".into(),
620. 620 voice: "default".into(),
621. 621 behaviour: profile.behaviour.clone(),
622. 622 responsibilities: profile.responsibilities.clone(),
623. 623 instructions: profile.instructions.clone(),
624. 624 })
625. 625 .or_else(|| self.deps.parent_agent_identity.clone())
626. 626 .or_else(|| Some(vak_core_identity())),
627. 627 session_id: session_id.clone(),
628. 628 created_at: chrono::Utc::now(),
629. 629 cwd: self.deps.cwd.clone(),
630. 630 parent_session_id: Some(self.deps.parent_session_id.clone()),
631. 631 contract_id: args
632. 632 .get("contract_id")
633. 633 .and_then(|value| value.as_str())
634. 634 .map(str::to_string)
635. 635 .or_else(|| self.deps.contract_id.clone()),
636. 636 work_item_id: args
637. 637 .get("work_item_id")
638. 638 .and_then(|value| value.as_str())
639. 639 .map(str::to_string)
640. 640 .or_else(|| self.deps.work_item_id.clone()),
641. 641 conversation: Some(vak_session::ConversationContext::local(
642. 642 &session_id,
643. 643 "worker",
644. 644 )),
645. 645 contract: FrozenContract {
646. 646 app_version: env!("CARGO_PKG_VERSION").into(),
647. 647 provider: self.deps.provider.name().into(),
648. 648 model: self.deps.model.clone(),
649. 649 route_ladder: Vec::new(),
650. 650 route_objective: String::new(),
651. 651 route_annotations: Vec::new(),
652. 652 system_prompt: child_system_prompt.clone(),
653. 653 permission_mode: match child_mode {
654. 654 Mode::ReadOnly => "read-only",
655. 655 Mode::WorkspaceWrite => "workspace-write",
656. 656 Mode::FullAccess => "full-access",
657. 657 }
658. 658 .into(),
659. 659 capabilities: child_capabilities,
660. 660 prompt_layers,
661. 661 },
662. 662 };
663. 663 let log = match SessionLog::create(path, header) {
664. 664 Ok(l) => l,
665. 665 Err(e) => return ToolOutput::error(format!("cannot create child session: {e}")),
666. 666 };
667. 667
668. 668 let mut cfg = AgentConfig::new(child_system_prompt.clone());
669. 669 cfg.model = self.deps.model.clone();
670. 670 cfg.tail = self.deps.tail.clone();
671. 671 cfg.tool_definitions = Some(vak_tools::definitions(&child_tools));
672. 672 cfg.tools = child_tools;
673. 673 cfg.hooks = self.deps.hooks.clone();
674. 674 cfg.revocation_check = self.deps.revocation_check.clone();
675. 675 cfg.presentation_rebuild = self.deps.presentation_rebuild.clone();
676. 676 cfg.mcp_tool_index = self.deps.mcp_tool_index.clone().unwrap_or_default();
677. 677 cfg.input_normalizer = self.deps.input_normalizer.clone();
678. 678 cfg.max_turns = self.deps.max_turns;
679. 679 cfg.max_retries = self.deps.max_retries;
680. 680 cfg.retry_base_backoff_ms = self.deps.retry_base_backoff_ms;
681. 681 cfg.request_timeout = self.deps.request_timeout;
682. 682 cfg.circuit_breaker = self.deps.circuit_breaker.clone();
683. 683 cfg.run_retry_attempts = self.deps.run_retry_attempts;
684. 684 cfg.run_retry_base_backoff_ms = self.deps.run_retry_base_backoff_ms;
685. 685 cfg.dispatch_ceiling = self.deps.dispatch_ceiling;
686. 686 cfg.spend_gate = self.deps.spend_gate.clone();
687. 687 cfg.outcome = child_outcome.clone();
688. 688 cfg.parallel_tools = true;
689. 689 cfg.permission = self.deps.permission.clone();
690. 690 cfg.mode = child_mode;
691. 691 cfg.approval_mode = self.deps.approval_mode;
692. 692 cfg.approver = self.deps.approver.clone();
693. 693 cfg.sandbox = self.deps.sandbox.clone();
694. 694
695. 695 let mut agent = Agent::new(self.deps.provider.clone(), log, cfg);
696. 696 let steering = Arc::new(SteeringQueues::new());
697. 697 let cancel = ctx.cancel.child_token();
698. 698 let label = args
699. 699 .get("label")
700. 700 .and_then(|l| l.as_str())
701. 701 .map(String::from)
702. 702 .or_else(|| profile.as_ref().map(|profile| profile.name.clone()))
703. 703 .unwrap_or_else(|| prompt.chars().take(48).collect());
704. 704 let _registry_guard = if let Some(registry) = &self.deps.registry {
705. 705 registry.register(
706. 706 session_id.clone(),
707. 707 WorkerHandle {
708. 708 label: label.clone(),
709. 709 agent_id: profile.as_ref().map(|profile| profile.id.clone()),
710. 710 agent_revision: profile.as_ref().map(|profile| profile.revision),
711. 711 started_at: std::time::Instant::now(),
712. 712 steering: steering.clone(),
713. 713 cancel: cancel.clone(),
714. 714 parent_session_id: self.deps.parent_session_id.clone(),
715. 715 },
716. 716 );
717. 717 Some(RegistryGuard {
718. 718 registry: registry.clone(),
719. 719 id: session_id.clone(),
720. 720 })
721. 721 } else {
722. 722 None
723. 723 };
724. 724 let (ev_tx, mut ev_rx) = tokio::sync::mpsc::channel::<crate::AgentEvent>(256);
725. 725 // Always drain the child stream (a full channel would deadlock the
726. 726 // child loop); tool calls are additionally forwarded to the parent
727. 727 // event stream so parallel workers are visible in the UI.
728. 728 let parent = self.deps.events.clone();
729. 729 let fwd_label = label.clone();
730. 730 let pump = tokio::spawn(async move {
731. 731 while let Some(ev) = ev_rx.recv().await {
732. 732 match ev {
733. 733 crate::AgentEvent::ToolCallEnd { name, is_error, .. } => {
734. 734 if let Some(parent) = &parent {
735. 735 let _ = parent
736. 736 .send(crate::AgentEvent::WorkerToolCall {
737. 737 label: fwd_label.clone(),
738. 738 name,
739. 739 is_error,
740. 740 })
741. 741 .await;
742. 742 }
743. 743 }
744. 744 crate::AgentEvent::TurnEnd { usage } => {
745. 745 if let Some(parent) = &parent {
746. 746 let _ = parent
747. 747 .send(crate::AgentEvent::WorkerUsage {
748. 748 label: fwd_label.clone(),
749. 749 input_tokens: usage.input_tokens,
750. 750 output_tokens: usage.output_tokens,
751. 751 })
752. 752 .await;
753. 753 }
754. 754 }
755. 755 crate::AgentEvent::Sandbox(event) => {
756. 756 // Sandbox output belongs to the parent surface too:
757. 757 // worker tool calls execute through the same
758. 758 // broker and must remain visible and rehydratable in
759. 759 // Workbench with their own execution identity.
760. 760 if let Some(parent) = &parent {
761. 761 let _ = parent.send(crate::AgentEvent::Sandbox(event)).await;
762. 762 }
763. 763 }
764. 764 _ => {}
765. 765 }
766. 766 }
767. 767 });
768. 768 if let Some(events) = &self.deps.events {
769. 769 let _ = events
770. 770 .send(crate::AgentEvent::WorkerStarted {
771. 771 label: label.clone(),
772. 772 })
773. 773 .await;
774. 774 }
775. 775 let started = std::time::Instant::now();
776. 776 let outcome = agent.run(prompt, &steering, cancel, ev_tx).await;
777. 777 let child_status = match &outcome {
778. 778 crate::TurnOutcome::Completed { .. } => vak_session::types::ChildRunStatus::Completed,
779. 779 crate::TurnOutcome::Failed { .. } => vak_session::types::ChildRunStatus::Failed,
780. 780 crate::TurnOutcome::Aborted { .. } => vak_session::types::ChildRunStatus::Aborted,
781. 781 crate::TurnOutcome::MaxTurnsReached => vak_session::types::ChildRunStatus::MaxTurns,
782. 782 };
783. 783 // Persist the terminal marker before notifying the parent. Recovery
784. 784 // must never observe a finished child without a durable status.
785. 785 let _ = agent
786. 786 .session
787. 787 .lock()
788. 788 .await
789. 789 .append_child_run_result(child_status, child_outcome);
790. 790 if let Some(events) = &self.deps.events {
791. 791 let _ = events
792. 792 .send(crate::AgentEvent::WorkerFinished {
793. 793 label: label.clone(),
794. 794 is_error: !matches!(outcome, crate::TurnOutcome::Completed { .. }),
795. 795 elapsed_ms: started.elapsed().as_millis() as u64,
796. 796 })
797. 797 .await;
798. 798 }
799. 799 let _ = pump.await;
800. 800 let cards: Vec<vak_tools::PresentationCard> = agent
801. 801 .session
802. 802 .lock()
803. 803 .await
804. 804 .presentations()
805. 805 .into_iter()
806. 806 .map(|(_, record)| vak_tools::PresentationCard {
807. 807 semantic_type: record.semantic_type.clone(),
808. 808 skill_id: record.skill_id.clone(),
809. 809 skill_version: record.skill_version.clone(),
810. 810 schema_version: record.schema_version,
811. 811 payload: record.payload.clone(),
812. 812 title: record.title.clone(),
813. 813 identity_digest: record.identity_digest.clone(),
814. 814 })
815. 815 .collect();
816. 816 let mut output = worker_output(&session_id, outcome, requested_contract.is_some());
817. 817 if !cards.is_empty() {
818. 818 output.delegated = Some(vak_tools::DelegatedCards {
819. 819 session_id: session_id.clone(),
820. 820 cards,
821. 821 });
822. 822 }
823. 823 output
824. 824 }
825. 825}
826. 826
827. 827/// The `task` result for a finished worker: its final text, or why there is
828. 828/// none. The worker's cards travel beside it (`ToolOutput::delegated`).
829. 829fn worker_output(session_id: &str, outcome: crate::TurnOutcome, contracted: bool) -> ToolOutput {
830. 830 match outcome {
831. 831 crate::TurnOutcome::Completed { response } => {
832. 832 let text = response.text_content();
833. 833 if text.is_empty() {
834. 834 ToolOutput::ok(format!("worker '{session_id}' completed without output"))
835. 835 } else {
836. 836 if contracted {
837. 837 ToolOutput::ok(format!("worker '{session_id}' completed:\n{text}"))
838. 838 } else {
839. 839 ToolOutput::ok(text)
840. 840 }
841. 841 }
842. 842 }
843. 843 crate::TurnOutcome::Aborted { partial } => {
844. 844 let text = partial.map(|p| p.text_content()).unwrap_or_default();
845. 845 ToolOutput::error(format!(
846. 846 "worker '{session_id}' was cancelled. Partial output:\n{text}"
847. 847 ))
848. 848 }
849. 849 crate::TurnOutcome::Failed { error } => {
850. 850 ToolOutput::error(format!("worker '{session_id}' failed: {error}"))
851. 851 }
852. 852 crate::TurnOutcome::MaxTurnsReached => ToolOutput::error(format!(
853. 853 "worker '{session_id}' hit its turn limit before finishing"
854. 854 )),
855. 855 }
856. 856}
857. 857
858. 858/// A child session's id, unique by construction.
859. 859///
860. 860/// The contract a worker is held to: what its own prompt asks for, not the
861. 861/// parent's whole request. A research worker spawned from "fix the bug and
862. 862/// run the tests" was held to the parent's execution requirement and sent
863. 863/// back to act on files its task never asked it to touch. A read-only worker
864. 864/// is clamped further: its contract may not demand an effect or a file its
865. 865/// tools cannot produce. Read with the same tier-1 reader as every turn; the
866. 866/// parent's evidence freshness and turn budget carry over.
867. 867fn child_outcome(
868. 868 prompt: &str,
869. 869 readonly: bool,
870. 870 parent: Option<&vak_intent::OutcomeSpec>,
871. 871) -> Option<vak_intent::OutcomeSpec> {
872. 872 let parent = parent?;
873. 873 let request = vak_intent::Request {
874. 874 text: prompt,
875. 875 surface: vak_intent::Surface::Worker,
876. 876 ..vak_intent::Request::default()
877. 877 };
878. 878 let intent = vak_intent::resolve(
879. 879 &request,
880. 880 &vak_intent::Declared::default(),
881. 881 &vak_intent::Authority::default(),
882. 882 &vak_intent::ResolverConfig::default(),
883. 883 )
884. 884 .intent();
885. 885 let mut spec = vak_intent::OutcomeSpec::from_intent(prompt, &intent);
886. 886 spec.evidence_max_age_secs = parent.evidence_max_age_secs;
887. 887 spec.max_turns = parent.max_turns;
888. 888 if readonly {
889. 889 spec.acts
890. 890 .retain(|act| !act.requires_execution() && !matches!(act, vak_intent::Act::Author));
891. 891 if spec.acts.is_empty() {
892. 892 spec.acts.insert(vak_intent::Act::Analyze);
893. 893 }
894. 894 if spec.stop.rank() > vak_intent::StopProfile::Inspection.rank() {
895. 895 spec.stop = vak_intent::StopProfile::Inspection;
896. 896 }
897. 897 }
898. 898 Some(spec)
899. 899}
900. 900
901. 901/// The id names the child's ledger file, and the file is exclusively locked, so
902. 902/// two children with the same id cannot both exist. The id used to be the
903. 903/// clock's nanoseconds alone; tasks launched in the same wave can read the same
904. 904/// value (clock resolution is coarser than the launch rate), and the second
905. 905/// then failed with "session is locked by another process" — a spurious error
906. 906/// the stop gate turned into an extra parent turn. A process-wide counter makes
907. 907/// a collision impossible whatever the clock does.
908. 908fn next_child_session_id() -> String {
909. 909 static SEQUENCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
910. 910 let nanos = std::time::SystemTime::now()
911. 911 .duration_since(std::time::UNIX_EPOCH)
912. 912 .unwrap_or_default()
913. 913 .as_nanos();
914. 914 let sequence = SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
915. 915 format!("child-{nanos}-{sequence}")
916. 916}
917. 917
918. 918#[cfg(test)]
919. 919#[allow(clippy::unwrap_used, clippy::expect_used)]
920. 920mod registry_tests {
921. 921 use super::*;
922. 922
923. 923 fn parent_contract(text: &str) -> vak_intent::OutcomeSpec {
924. 924 let request = vak_intent::Request {
925. 925 text,
926. 926 ..vak_intent::Request::default()
927. 927 };
928. 928 let intent = vak_intent::resolve(
929. 929 &request,
930. 930 &vak_intent::Declared::default(),
931. 931 &vak_intent::Authority::default(),
932. 932 &vak_intent::ResolverConfig::default(),
933. 933 )
934. 934 .intent();
935. 935 vak_intent::OutcomeSpec::from_intent(text, &intent)
936. 936 }
937. 937
938. 938 /// A worker answers for its own task. A research child of an effectful
939. 939 /// request is not held to the parent's execution requirement.
940. 940 #[test]
941. 941 fn a_worker_is_held_to_its_own_task_not_the_parents() {
942. 942 let parent = parent_contract("fix the failing test and run the suite");
943. 943 assert!(parent.requires_execution());
944. 944 let child = child_outcome(
945. 945 "find where the parser handles empty input",
946. 946 false,
947. 947 Some(&parent),
948. 948 )
949. 949 .expect("a parent contract yields a child contract");
950. 950 assert!(!child.requires_execution(), "acts={:?}", child.acts);
951. 951 assert_eq!(child.objective, "find where the parser handles empty input");
952. 952 }
953. 953
954. 954 /// A read-only worker cannot be asked for what its tools cannot do,
955. 955 /// whatever its prompt says.
956. 956 #[test]
957. 957 fn a_read_only_worker_is_never_owed_an_effect() {
958. 958 let parent = parent_contract("fix the failing test");
959. 959 let child = child_outcome(
960. 960 "update the parser and save the notes as notes.md",
961. 961 true,
962. 962 Some(&parent),
963. 963 )
964. 964 .expect("child contract");
965. 965 assert!(!child.requires_execution(), "acts={:?}", child.acts);
966. 966 assert!(child.stop.rank() <= vak_intent::StopProfile::Inspection.rank());
967. 967 // No parent contract, no child contract: nothing to be held to.
968. 968 assert!(child_outcome("anything", false, None).is_none());
969. 969 }
970. 970
971. 971 #[test]
972. 972 fn duplicate_agent_names_fail_closed() {
973. 973 vak_config::paths::isolate_home_for_tests();
974. 974 let dir = tempfile::tempdir().expect("Agent workspace");
975. 975 std::fs::create_dir_all(dir.path().join(".vak")).expect("profile directory");
976. 976 std::fs::write(
977. 977 dir.path().join(".vak/agents.json"),
978. 978 r#"[{"id":"one","revision":1,"name":"Pip","character":"pip","personality":"","behaviour":""},{"id":"two","revision":1,"name":"Pip","character":"pip","personality":"","behaviour":""}]"#,
979. 979 )
980. 980 .expect("profiles");
981. 981 let result = load_agent(dir.path(), "Pip");
982. 982 assert!(matches!(result, Err(error) if error.contains("ambiguous")));
983. 983 }
984. 984
985. 985 #[test]
986. 986 fn register_steer_stop_lifecycle() {
987. 987 let reg = WorkerRegistry::new();
988. 988 assert!(reg.active().is_empty());
989. 989 assert!(!reg.steer("child-1", "go"));
990. 990 assert!(!reg.stop("child-1"));
991. 991
992. 992 let cancel = CancellationToken::new();
993. 993 reg.register(
994. 994 "child-1".into(),
995. 995 WorkerHandle {
996. 996 label: "explore".into(),
997. 997 agent_id: None,
998. 998 agent_revision: None,
999. 999 started_at: std::time::Instant::now(),
1000. 1000 steering: Arc::new(SteeringQueues::new()),
1001. 1001 cancel: cancel.clone(),
1002. 1002 parent_session_id: "parent-a".into(),
1003. 1003 },
1004. 1004 );
1005. 1005 assert!(reg.steer("child-1", "look left"));
1006. 1006 assert!(reg.queue_follow_up("child-1", "then right"));
1007. 1007 let active = reg.active();
1008. 1008 assert_eq!(active.len(), 1);
1009. 1009 assert_eq!(active[0].id, "child-1");
1010. 1010 assert_eq!(active[0].label, "explore");