72. 72 matches!(&entry.payload, EntryPayload::Activity(activity)
73. 73 if activity.data.get("request_id").map(String::as_str) == Some(request_id))
74. 74 })
75. 75 }
76. 76 pub fn create(path: PathBuf, header: SessionHeader) -> Result<Self, SessionError> {
77. 77 if let Some(parent) = path.parent() {
78. 78 std::fs::create_dir_all(parent)?;
79. 79 }
80. 80 let file = OpenOptions::new()
81. 81 .create(true)
82. 82 .append(true)
83. 83 .truncate(false)
84. 84 .open(&path)?;
85. 85 // Cross-process safety: an exclusive lock for the lifetime of the
86. 86 // handle keeps two processes from interleaving appends.
87. 87 let file = LedgerFile::lock(file, &path)?;
88. 88 if file.file.metadata()?.len() > 0 {
89. 89 return Err(SessionError::Exists(path));
90. 90 }
91. 91 let mut log = SessionLog {
92. 92 path,
93. 93 file,
94. 94 entries: Vec::new(),
95. 95 by_id: HashMap::new(),
96. 96 tail_id: None,
97. 97 tail_hash: None,
98. 98 warnings: Vec::new(),
99. 99 };
100. 100 log.append(Entry::new(None, EntryPayload::Header(header)))?;
101. 101 Ok(log)
102. 102 }
103. 103}
104. 104
105. 105struct ParsedEntries {
106. 106 entries: Vec<Entry>,
107. 107 by_id: HashMap<String, usize>,
108. 108 tail_id: Option<String>,
109. 109 tail_hash: Option<String>,
110. 110 warnings: Vec<String>,
111. 111}
112. 112
113. 113impl SessionLog {
114. 114 fn parse_entries(path: &Path, reader: BufReader<File>) -> Result<ParsedEntries, SessionError> {
115. 115 let mut entries = Vec::new();
116. 116 let mut by_id = HashMap::new();
117. 117 let mut warnings = Vec::new();
118. 118 let mut expected_prev: Option<String> = None;
119. 119 let mut tail_hash: Option<String> = None;
120. 120 let mut unchained = 0usize;
121. 121 let mut broken = Vec::new();
122. 122 for (i, line) in reader.lines().enumerate() {
123. 123 let line = line.map_err(|e| SessionError::Corrupt {
124. 124 line: i + 1,
125. 125 message: e.to_string(),
126. 126 })?;
127. 127 if line.trim().is_empty() {
128. 128 continue;
129. 129 }
130. 130 // A torn final line (crash mid-append) or a damaged interior
131. 131 // line must not make the whole session unresumable; skip it
132. 132 // and surface a warning. The append-only ledger on disk is
133. 133 // never rewritten.
134. 134 let Ok(entry) = serde_json::from_str::<Entry>(&line) else {
135. 135 warnings.push(format!(
136. 136 "skipped unparseable entry at line {} of {}",
137. 137 i + 1,
138. 138 path.display()
139. 139 ));
140. 140 expected_prev = None;
141. 141 continue;
142. 142 };
143. 143 match (&entry.prev_hash, &expected_prev) {
144. 144 // An entry that carries a link must match it. A mismatch means
145. 145 // the ledger was edited after the fact, and that is reported
146. 146 // rather than raised: the record is evidence, and refusing to
147. 147 // open it would destroy the only copy of what happened.
148. 148 (Some(found), Some(want)) if found != want => broken.push(i + 1),
149. 149 // Absent where a predecessor exists: written before chaining.
150. 150 // The very first entry legitimately has no link.
151. 151 (None, Some(_)) => unchained += 1,
152. 152 _ => {}
153. 153 }
154. 154 let digest = crate::types::line_digest(&line);
155. 155 expected_prev = Some(digest.clone());
156. 156 tail_hash = Some(digest);
157. 157 by_id.insert(entry.id.clone(), entries.len());
158. 158 entries.push(entry);
159. 159 }
160. 160 if !broken.is_empty() {
161. 161 warnings.push(format!(
162. 162 "ledger {} has a broken hash chain at line(s) {}: \
163. 163 entries before that point were modified after they were written",
164. 164 path.display(),
165. 165 broken
166. 166 .iter()
167. 167 .map(usize::to_string)
168. 168 .collect::<Vec<_>>()
169. 169 .join(", ")
170. 170 ));
171. 171 }
172. 172 if unchained > 0 {
173. 173 warnings.push(format!(
174. 174 "ledger {} has {unchained} entry/entries written before hash chaining; \
175. 175 those cannot be verified",
176. 176 path.display()
177. 177 ));
178. 178 }
179. 179 let tail_id = entries.last().map(|e| e.id.clone());
180. 180 Ok(ParsedEntries {
181. 181 entries,
182. 182 by_id,
183. 183 tail_id,
184. 184 tail_hash,
185. 185 warnings,
186. 186 })
187. 187 }
188. 188
189. 189 pub fn open(path: PathBuf) -> Result<Self, SessionError> {
190. 190 let file = LedgerFile::lock(OpenOptions::new().append(true).open(&path)?, &path)?;
191. 191 let reader = BufReader::new(File::open(&path)?);
192. 192 let parsed = Self::parse_entries(&path, reader)?;
193. 193 Ok(SessionLog {
194. 194 path,
195. 195 file,
196. 196 entries: parsed.entries,
197. 197 by_id: parsed.by_id,
198. 198 tail_id: parsed.tail_id,
199. 199 tail_hash: parsed.tail_hash,
200. 200 warnings: parsed.warnings,
201. 201 })
202. 202 }
203. 203
204. 204 /// Open an existing session for reading and inspection without acquiring an
205. 205 /// exclusive write lock. Allows web clients, exports, and inspectors to
206. 206 /// read and rehydrate sessions that are currently active in another process.
207. 207 pub fn open_read_only(path: PathBuf) -> Result<Self, SessionError> {
208. 208 let file = LedgerFile {
209. 209 file: File::open(&path)?,
210. 210 locked: false,
211. 211 };
212. 212 let reader = BufReader::new(File::open(&path)?);
213. 213 let parsed = Self::parse_entries(&path, reader)?;
214. 214 Ok(SessionLog {
215. 215 path,
216. 216 file,
217. 217 entries: parsed.entries,
218. 218 by_id: parsed.by_id,
219. 219 tail_id: parsed.tail_id,
220. 220 tail_hash: parsed.tail_hash,
221. 221 warnings: parsed.warnings,
222. 222 })
223. 223 }
224. 224
225. 225 /// Returns whether this SessionLog was opened read-only.
226. 226 pub fn is_read_only(&self) -> bool {
227. 227 !self.file.locked
228. 228 }
229. 229
230. 230 /// Non-fatal problems seen while opening the ledger.
231. 231 pub fn warnings(&self) -> &[String] {
232. 232 &self.warnings
233. 233 }
234. 234
235. 235 pub fn append(&mut self, entry: Entry) -> Result<Entry, SessionError> {
236. 236 if !self.file.locked {
237. 237 return Err(SessionError::Locked(self.path.clone()));
238. 238 }
239. 239 if let Some(pid) = &entry.parent_id
240. 240 && !self.by_id.contains_key(pid)
241. 241 {
242. 242 return Err(SessionError::Corrupt {
243. 243 line: 0,
244. 244 message: format!("parent entry {pid} not found"),
245. 245 });
246. 246 }
247. 247 let mut entry = entry;
248. 248 entry.prev_hash = self.tail_hash.clone();
249. 249 let line = serde_json::to_string(&entry).map_err(|e| SessionError::Corrupt {
250. 250 line: 0,
251. 251 message: e.to_string(),
252. 252 })?;
253. 253 writeln!(self.file.file, "{line}")?;
254. 254 // `File::flush` is a no-op — `std::fs::File` has no userspace buffer,
255. 255 // so its `Write::flush` returns Ok without a syscall. That is what
256. 256 // this used to call, which meant the "durable, reconstructable"
257. 257 // ledger had no write barrier at all and lost its tail on power loss.
258. 258 // `sync_data` skips the metadata flush `sync_all` forces; the file
259. 259 // length is data for an append-only log.
260. 260 self.file.file.sync_data()?;
261. 261 self.tail_hash = Some(crate::types::line_digest(&line));
262. 262 self.by_id.insert(entry.id.clone(), self.entries.len());
263. 263 self.tail_id = Some(entry.id.clone());
264. 264 self.entries.push(entry.clone());
265. 265 Ok(entry)
266. 266 }
267. 267
268. 268 pub fn append_message(&mut self, record: MessageRecord) -> Result<Entry, SessionError> {
269. 269 let parent = self.tail_id.clone();
270. 270 self.append(Entry::new(parent, EntryPayload::Message(record)))
271. 271 }
272. 272
273. 273 /// Appends a work receipt (audit entry; never model-visible).
274. 274 pub fn append_receipt(&mut self, receipt: vak_llm::WorkReceipt) -> Result<Entry, SessionError> {
275. 275 let parent = self.tail_id.clone();
276. 276 self.append(Entry::new(parent, EntryPayload::Receipt(receipt)))
277. 277 }
278. 278
279. 279 /// Appends a goal-lifecycle entry (audit; never model-visible).
280. 280 pub fn append_goal(&mut self, goal: crate::types::GoalEntry) -> Result<Entry, SessionError> {
281. 281 let parent = self.tail_id.clone();
282. 282 self.append(Entry::new(parent, EntryPayload::Goal(goal)))
283. 283 }
284. 284
285. 285 pub fn append_goal_update(
286. 286 &mut self,
287. 287 update: vak_intent::GoalUpdate,
288. 288 ) -> Result<Entry, SessionError> {
289. 289 let parent = self.tail_id.clone();
290. 290 self.append(Entry::new(parent, EntryPayload::GoalUpdate(update)))
291. 291 }
292. 292
293. 293 /// The latest goal update on the active branch. An update on a branch
294. 294 /// the conversation left behind is not part of its goal.
295. 295 pub fn latest_goal_update(&self) -> Option<vak_intent::GoalUpdate> {
296. 296 self.chain_to_root()
297. 297 .into_iter()
298. 298 .rev()
299. 299 .find_map(|entry| match &entry.payload {
300. 300 EntryPayload::GoalUpdate(update) => Some(update.clone()),
301. 301 _ => None,
302. 302 })
303. 303 }
304. 304
305. 305 /// How `text` relates to this conversation's goal, as the next update to
306. 306 /// append (`vak_intent::next_goal_update`).
307. 307 pub fn next_goal_update(&self, text: &str) -> vak_intent::GoalUpdate {
308. 308 vak_intent::next_goal_update(
309. 309 text,
310. 310 self.goal_state().as_ref(),
311. 311 self.latest_goal_update().map(|update| update.revision),
312. 312 )
313. 313 }
314. 314
315. 315 pub fn goal_state(&self) -> Option<vak_intent::GoalState> {
316. 316 vak_intent::GoalState::from_updates(self.chain_to_root().iter().filter_map(|entry| {
317. 317 if let EntryPayload::GoalUpdate(update) = &entry.payload {
318. 318 Some(update.clone())
319. 319 } else {
320. 320 None
321. 321 }
322. 322 }))
323. 323 }
324. 324
325. 325 /// Appends a presentation/audit lifecycle fact. It is deliberately
326. 326 /// excluded from `derive_messages`.
327. 327 pub fn append_activity(
328. 328 &mut self,
329. 329 activity: crate::types::ActivityRecord,
330. 330 ) -> Result<Entry, SessionError> {
331. 331 let parent = self.tail_id.clone();
332. 332 self.append(Entry::new(parent, EntryPayload::Activity(activity)))
333. 333 }
334. 334
335. 335 /// Appends a validated presentation (docs/design/68-context-engine.md
336. 336 /// §10). Hash-linked like every entry; never rewritten.
337. 337 pub fn append_presentation(
338. 338 &mut self,
339. 339 record: PresentationRecord,
340. 340 ) -> Result<Entry, SessionError> {
341. 341 let parent = self.tail_id.clone();
342. 342 self.append(Entry::new(parent, EntryPayload::Presentation(record)))
343. 343 }
344. 344
345. 345 /// Records the whole result behind a windowed `ToolResult` block
346. 346 /// (docs/design/68-context-engine.md §3), so `recall` and the closed-turn
347. 347 /// digests read what the tool returned, not what the request carried.
348. 348 pub fn append_evidence_body(
349. 349 &mut self,
350. 350 tool_use_id: &str,
351. 351 content: String,
352. 352 ) -> Result<Entry, SessionError> {
353. 353 let parent = self.tail_id.clone();
354. 354 self.append(Entry::new(
355. 355 parent,
356. 356 EntryPayload::EvidenceBody(crate::types::EvidenceBodyRecord {
357. 357 tool_use_id: tool_use_id.to_string(),
358. 358 content,
359. 359 }),
360. 360 ))
361. 361 }
362. 362
363. 363 /// The most recently recorded capacity profile matching `key`,
364. 364 /// reconstructed from the ledger's `CapacityProbe`/`CapacityFeedback`
365. 365 /// activities (docs/design/68-context-engine.md §1). Generic over the
366. 366 /// caller's key/profile types (vak-agent's `ProfileKey`/`CapacityProfile`)
367. 367 /// because vak-session cannot depend on vak-agent — that dependency runs
368. 368 /// the other way — so this crate stores and retrieves the profile as the
369. 369 /// JSON the caller already serialized, never interpreting its shape.
370. 370 pub fn latest_capacity_profile<K, P>(&self, key: &K) -> Option<P>
371. 371 where
372. 372 K: serde::Serialize,
373. 373 P: serde::de::DeserializeOwned,
374. 374 {
375. 375 let key_json = serde_json::to_value(key).ok()?;
376. 376 self.chain_to_root().into_iter().rev().find_map(|entry| {
377. 377 let EntryPayload::Activity(activity) = &entry.payload else {
378. 378 return None;
379. 379 };
380. 380 if !matches!(
381. 381 activity.kind,
382. 382 crate::types::ActivityKind::CapacityProbe
383. 383 | crate::types::ActivityKind::CapacityFeedback
384. 384 ) {
385. 385 return None;
386. 386 }
387. 387 let stored_key: serde_json::Value =
388. 388 serde_json::from_str(activity.data.get("key")?).ok()?;
389. 389 if stored_key != key_json {
390. 390 return None;
391. 391 }
392. 392 serde_json::from_str(activity.data.get("profile")?).ok()
393. 393 })
394. 394 }
395. 395
396. 396 /// Appends a turn's closing card (docs/design/68-context-engine.md §10).
397. 397 /// Written once, at turn close, and never rewritten.
398. 398 pub fn append_turn_card(&mut self, record: TurnCardRecord) -> Result<Entry, SessionError> {
399. 399 let parent = self.tail_id.clone();
400. 400 self.append(Entry::new(parent, EntryPayload::TurnCard(record)))
401. 401 }
402. 402
403. 403 /// `(turn_id, card)` for every `TurnCard` entry along the active chain,
404. 404 /// in the order they were written.
405. 405 pub fn turn_cards(&self) -> Vec<(String, TurnCard)> {
406. 406 self.chain_to_root()
407. 407 .into_iter()
408. 408 .filter_map(|entry| match &entry.payload {
409. 409 EntryPayload::TurnCard(record) => {
410. 410 Some((record.turn_id.clone(), record.card.clone()))
411. 411 }
412. 412 _ => None,
413. 413 })
414. 414 .collect()
415. 415 }
416. 416
417. 417 /// Resolves a tool_use_id to its full `Evidence` — the tool name, its
418. 418 /// call arguments, its whole result content (the evidence body when the
419. 419 /// request carried a window of it), and whether it failed — by
420. 420 /// scanning the active chain for the matching `ToolUse`/`ToolResult`
421. 421 /// pair. Works for evidence in any turn, open or closed, which is what
422. 422 /// lets `recall({ id })` reopen a result from a turn now reduced to a
423. 423 /// card (docs/design/68-context-engine.md §3).
424. 424 pub fn evidence(&self, tool_use_id: &str) -> Option<Evidence> {
425. 425 let chain = self.chain_to_root();
426. 426 let mut tool: Option<String> = None;
427. 427 let mut input = serde_json::Value::Null;
428. 428 for entry in &chain {
429. 429 let EntryPayload::Message(record) = &entry.payload else {
430. 430 continue;
431. 431 };
432. 432 for block in &record.message.content {
433. 433 if let vak_llm::ContentBlock::ToolUse {
434. 434 id,
435. 435 name,
436. 436 input: call_input,
437. 437 } = block
438. 438 && id == tool_use_id
439. 439 {
440. 440 tool = Some(name.clone());
441. 441 input = call_input.clone();
442. 442 }
443. 443 }
444. 444 }
445. 445 let tool = tool?;
446. 446 let body = chain.iter().find_map(|entry| match &entry.payload {
447. 447 EntryPayload::EvidenceBody(body) if body.tool_use_id == tool_use_id => {
448. 448 Some(body.content.clone())
449. 449 }
450. 450 _ => None,
451. 451 });
452. 452 for entry in &chain {
453. 453 let EntryPayload::Message(record) = &entry.payload else {
454. 454 continue;
455. 455 };
456. 456 for block in &record.message.content {
457. 457 if let vak_llm::ContentBlock::ToolResult {
458. 458 tool_use_id: id,
459. 459 content,
460. 460 is_error,
461. 461 } = block
462. 462 && id == tool_use_id
463. 463 {
464. 464 return Some(Evidence {
465. 465 tool,
466. 466 input,
467. 467 content: body.unwrap_or_else(|| content.clone()),
468. 468 is_error: *is_error,
469. 469 });
470. 470 }
471. 471 }
472. 472 }
473. 473 None
474. 474 }
475. 475
476. 476 /// Presentation entries along the active path, root→leaf, with their
477. 477 /// entry ids — the single source both the model-visible history and the
478. 478 /// display channel read.
479. 479 pub fn presentations(&self) -> Vec<(String, &PresentationRecord)> {
480. 480 self.chain_to_root()
481. 481 .into_iter()
482. 482 .filter_map(|e| match &e.payload {
483. 483 EntryPayload::Presentation(record) => Some((e.id.clone(), record)),
484. 484 _ => None,
485. 485 })
486. 486 .collect()
487. 487 }
488. 488
489. 489 /// Whether a presentation with this exact canonical payload already
490. 490 /// exists for this turn — the duplicate rule that drops a repeated
491. 491 /// fence from the model-visible projection (docs/design/68-context-engine.md
492. 492 /// §10: "a second card in the same turn with the same `payload_digest`
493. 493 /// is not written").
494. 494 pub fn has_presentation(&self, turn_id: &str, payload_digest: &str) -> bool {
495. 495 self.presentations()
496. 496 .iter()
497. 497 .any(|(_, record)| record.turn_id == turn_id && record.payload_digest == payload_digest)
498. 498 }
499. 499
500. 500 /// The id of the most recent non-control user message entry in the
501. 501 /// active chain — the directive the current turn answers. `None` before
502. 502 /// any real user message exists.
503. 503 pub fn latest_directive_entry_id(&self) -> Option<String> {
504. 504 self.chain_to_root()
505. 505 .into_iter()
506. 506 .rev()
507. 507 .find_map(|entry| match &entry.payload {
508. 508 EntryPayload::Message(record)
509. 509 if record.message.role == vak_llm::Role::User
510. 510 && record.control_kind().is_none()
511. 511 && record
512. 512 .message
513. 513 .content
514. 514 .iter()
515. 515 .any(|b| matches!(b, vak_llm::ContentBlock::Text { .. }))
516. 516 && !record
517. 517 .message
518. 518 .content
519. 519 .iter()
520. 520 .any(|b| matches!(b, vak_llm::ContentBlock::ToolResult { .. })) =>
521. 521 {
522. 522 Some(entry.id.clone())
523. 523 }
524. 524 _ => None,
525. 525 })
526. 526 }
527. 527
528. 528 /// Ids of every non-card tool result already committed to the ledger
529. 529 /// strictly after `turn_id`, in chain order — the evidence a card built
530. 530 /// after them was derived from (`PresentationRecord::derived_from`).
531. 531 /// `is_card_tool` is supplied by the caller so this crate never needs to
532. 532 /// know what a "card" tool is (that knowledge lives in
533. 533 /// `vak-core::presentation_tools`).
534. 534 pub fn non_card_evidence_since(
535. 535 &self,
536. 536 turn_id: &str,
537. 537 is_card_tool: impl Fn(&str) -> bool,
538. 538 ) -> Vec<String> {
539. 539 let chain = self.chain_to_root();
540. 540 let start = chain
541. 541 .iter()
542. 542 .position(|entry| entry.id == turn_id)
543. 543 .map(|idx| idx + 1)
544. 544 .unwrap_or(0);
545. 545 let mut tool_names: HashMap<String, String> = HashMap::new();
546. 546 let mut out = Vec::new();
547. 547 for entry in &chain[start..] {
548. 548 let EntryPayload::Message(record) = &entry.payload else {
549. 549 continue;
550. 550 };
551. 551 for block in &record.message.content {
552. 552 match block {
553. 553 vak_llm::ContentBlock::ToolUse { id, name, .. } => {
554. 554 tool_names.insert(id.clone(), name.clone());
555. 555 }
556. 556 vak_llm::ContentBlock::ToolResult { tool_use_id, .. } => {
557. 557 let is_card = tool_names
558. 558 .get(tool_use_id)
559. 559 .map(|name| is_card_tool(name))
560. 560 .unwrap_or(false);
561. 561 if !is_card && !out.iter().any(|seen| seen == tool_use_id) {
562. 562 out.push(tool_use_id.clone());
563. 563 }
564. 564 }
565. 565 _ => {}
566. 566 }
567. 567 }
568. 568 }
569. 569 out
570. 570 }
571. 571
572. 572 /// Append a finalized or provisional voice transcript. The transcript
573. 573 /// is an audit projection; callers must append a normal Message entry
574. 574 /// separately when the utterance is committed as model input.
575. 575 pub fn append_voice_transcript(
576. 576 &mut self,
577. 577 activity_id: impl Into<String>,
578. 578 text: impl Into<String>,
579. 579 finalized: bool,
580. 580 ) -> Result<Entry, SessionError> {
581. 581 let mut data = std::collections::BTreeMap::new();
582. 582 data.insert("text".into(), text.into());
583. 583 data.insert("finalized".into(), finalized.to_string());
584. 584 self.append_activity(crate::types::ActivityRecord {
585. 585 activity_id: activity_id.into(),
586. 586 turn: None,
587. 587 kind: crate::types::ActivityKind::VoiceTranscript,
588. 588 status: if finalized {
589. 589 crate::types::ActivityStatus::Succeeded
590. 590 } else {
591. 591 crate::types::ActivityStatus::Running
592. 592 },
593. 593 label: "Voice transcript".into(),
594. 594 detail: None,
595. 595 data,
596. 596 })
597. 597 }
598. 598
599. 599 /// Append an auditable presentation selection without making it part of
600. 600 /// model context. The original result and fallback remain authoritative.
601. 601 pub fn append_presentation_selection(
602. 602 &mut self,
603. 603 activity_id: impl Into<String>,
604. 604 semantic_type: impl Into<String>,
605. 605 spec_id: impl Into<String>,
606. 606 revision: u64,
607. 607 mode: impl Into<String>,
608. 608 fallback_used: bool,
609. 609 ) -> Result<Entry, SessionError> {
610. 610 let mut data = std::collections::BTreeMap::new();
611. 611 data.insert("semantic_type".into(), semantic_type.into());
612. 612 data.insert("spec_id".into(), spec_id.into());
613. 613 data.insert("revision".into(), revision.to_string());
614. 614 data.insert("mode".into(), mode.into());
615. 615 data.insert("fallback_used".into(), fallback_used.to_string());
616. 616 self.append_activity(crate::types::ActivityRecord {
617. 617 activity_id: activity_id.into(),
618. 618 turn: None,
619. 619 kind: crate::types::ActivityKind::PresentationSelection,
620. 620 status: crate::types::ActivityStatus::Succeeded,
621. 621 label: "Presentation selected".into(),
622. 622 detail: None,
623. 623 data,
624. 624 })
625. 625 }
626. 626
627. 627 /// Append a projection choice or correction. If the text is sent to the
628. 628 /// model, callers must also append a normal Message entry so derivation
629. 629 /// remains complete.
630. 630 pub fn append_presentation_feedback(
631. 631 &mut self,
632. 632 activity_id: impl Into<String>,
633. 633 choice: impl Into<String>,
634. 634 feedback: Option<String>,
635. 635 ) -> Result<Entry, SessionError> {
636. 636 let mut data = std::collections::BTreeMap::new();
637. 637 data.insert("choice".into(), choice.into());
638. 638 if let Some(feedback) = feedback {
639. 639 data.insert("feedback".into(), feedback);
640. 640 }
641. 641 self.append_activity(crate::types::ActivityRecord {
642. 642 activity_id: activity_id.into(),
643. 643 turn: None,
644. 644 kind: crate::types::ActivityKind::PresentationFeedback,
645. 645 status: crate::types::ActivityStatus::Succeeded,
646. 646 label: "Presentation feedback".into(),
647. 647 detail: None,
648. 648 data,
649. 649 })
650. 650 }
651. 651
652. 652 /// Append what the playback client reports it emitted. This is distinct
653. 653 /// from provider output because buffering and interruption can prevent
654. 654 /// the user from hearing the complete synthesis.
655. 655 pub fn append_voice_playback(
656. 656 &mut self,
657. 657 activity_id: impl Into<String>,
658. 658 emitted_ms: u64,
659. 659 interrupted: bool,
660. 660 ) -> Result<Entry, SessionError> {
661. 661 let mut data = std::collections::BTreeMap::new();
662. 662 data.insert("emitted_ms".into(), emitted_ms.to_string());
663. 663 data.insert("interrupted".into(), interrupted.to_string());
664. 664 self.append_activity(crate::types::ActivityRecord {
665. 665 activity_id: activity_id.into(),
666. 666 turn: None,
667. 667 kind: crate::types::ActivityKind::VoicePlayback,
668. 668 status: if interrupted {
669. 669 crate::types::ActivityStatus::Partial
670. 670 } else {
671. 671 crate::types::ActivityStatus::Succeeded
672. 672 },
673. 673 label: "Voice playback".into(),
674. 674 detail: None,
675. 675 data,
676. 676 })
677. 677 }
678. 678
679. 679 /// Record this turn's resolved intent.
680. 680 ///
681. 681 /// Appended before the turn dispatches, so the note it carries is in the
682. 682 /// projection the model actually sees — the entry *is* the record of what
683. 683 /// was said, not a description of it (invariant 1).
684. 684 pub fn append_intent(
685. 685 &mut self,
686. 686 record: crate::types::IntentRecord,
687. 687 ) -> Result<Entry, SessionError> {
688. 688 let parent = self.tail_id.clone();
689. 689 self.append(Entry::new(parent, EntryPayload::Intent(Box::new(record))))
690. 690 }
691. 691
692. 692 pub fn append_child_run_status(
693. 693 &mut self,
694. 694 status: crate::types::ChildRunStatus,
695. 695 ) -> Result<Entry, SessionError> {
696. 696 self.append_child_run_result(status, None)
697. 697 }
698. 698
699. 699 pub fn append_child_run_result(
700. 700 &mut self,
701. 701 status: crate::types::ChildRunStatus,
702. 702 outcome: Option<vak_intent::OutcomeSpec>,
703. 703 ) -> Result<Entry, SessionError> {
704. 704 let parent = self.tail_id.clone();
705. 705 self.append(Entry::new(
706. 706 parent,
707. 707 EntryPayload::ChildRun { status, outcome },
708. 708 ))
709. 709 }
710. 710
711. 711 pub fn child_run_status(&self) -> Option<crate::types::ChildRunStatus> {
712. 712 self.chain_to_root()
713. 713 .iter()
714. 714 .rev()
715. 715 .find_map(|entry| match &entry.payload {
716. 716 EntryPayload::ChildRun { status, .. } => Some(status.clone()),
717. 717 _ => None,
718. 718 })
719. 719 }
720. 720
721. 721 pub fn child_run_outcome(&self) -> Option<vak_intent::OutcomeSpec> {
722. 722 self.chain_to_root()
723. 723 .iter()
724. 724 .rev()
725. 725 .find_map(|entry| match &entry.payload {
726. 726 EntryPayload::ChildRun { outcome, .. } => outcome.clone(),
727. 727 _ => None,
728. 728 })
729. 729 }
730. 730
731. 731 /// Records the interface a turn was bound to: in full when it differs
732. 732 /// from the last one this conversation bound, otherwise as a
733. 733 /// `TurnCapabilitiesRef` to that entry.
734. 734 pub fn append_turn_capabilities(
735. 735 &mut self,
736. 736 bound: crate::types::TurnCapabilitiesBound,
737. 737 ) -> Result<Entry, SessionError> {
738. 738 let digest = bound.digest();
739. 739 let previous =
740. 740 self.chain_to_root()
741. 741 .into_iter()
742. 742 .rev()
743. 743 .find_map(|entry| match &entry.payload {
744. 744 EntryPayload::TurnCapabilitiesBound(earlier) => {
745. 745 Some((entry.id.clone(), earlier.digest()))
746. 746 }
747. 747 _ => None,
748. 748 });
749. 749 let parent = self.tail_id.clone();
750. 750 let payload = match previous {
751. 751 Some((entry, earlier)) if earlier == digest => {
752. 752 EntryPayload::TurnCapabilitiesRef(crate::types::TurnCapabilitiesRef {
753. 753 entry,
754. 754 digest,
755. 755 epoch: bound.epoch,
756. 756 })
757. 757 }
758. 758 _ => EntryPayload::TurnCapabilitiesBound(bound),
759. 759 };
760. 760 self.append(Entry::new(parent, payload))
761. 761 }
762. 762
763. 763 pub fn append_work(&mut self, event: WorkEvent) -> Result<Entry, SessionError> {
764. 764 let parent = self.tail_id.clone();
765. 765 let candidate = Entry::new(parent, EntryPayload::Work(event));
766. 766 let mut chain = self.chain_to_root();
767. 767 chain.push(&candidate);
768. 768 crate::work::project_work(&chain).map_err(|error| SessionError::Corrupt {
769. 769 line: 0,
770. 770 message: format!("invalid work event: {error}"),
771. 771 })?;
772. 772 let appended = self.append(candidate)?;
773. 773 self.promote_ready_work_items()?;
774. 774 Ok(appended)
775. 775 }
776. 776
777. 777 fn promote_ready_work_items(&mut self) -> Result<(), SessionError> {
778. 778 let Some(projection) = self
779. 779 .work_projection()
780. 780 .map_err(|error| SessionError::Corrupt {
781. 781 line: 0,
782. 782 message: error.to_string(),
783. 783 })?
784. 784 else {
785. 785 return Ok(());
786. 786 };
787. 787 if projection.status != crate::types::WorkContractStatus::Active {
788. 788 return Ok(());
789. 789 }
790. 790 let ready: Vec<String> = projection
791. 791 .contract
792. 792 .items
793. 793 .iter()
794. 794 .filter(|definition| {
795. 795 projection
796. 796 .items
797. 797 .get(&definition.item_id)
798. 798 .is_some_and(|state| state.status == crate::types::WorkItemStatus::Proposed)
799. 799 && definition.dependencies.iter().all(|dependency| {
800. 800 projection.items.get(dependency).is_some_and(|state| {
801. 801 matches!(
802. 802 state.status,
803. 803 crate::types::WorkItemStatus::Succeeded
804. 804 | crate::types::WorkItemStatus::Skipped
805. 805 )
806. 806 })
807. 807 })
808. 808 })
809. 809 .map(|definition| definition.item_id.clone())
810. 810 .collect();
811. 811 for item_id in ready {
812. 812 let current = self
813. 813 .work_projection()
814. 814 .map_err(|error| SessionError::Corrupt {
815. 815 line: 0,
816. 816 message: error.to_string(),
817. 817 })?;
818. 818 let Some(current) = current else { break };
819. 819 self.append_work(crate::types::WorkEvent {
820. 820 contract_id: current.contract.contract_id,
821. 821 revision: current.contract.revision,
822. 822 kind: crate::types::WorkEventKind::ItemStatusChanged {
823. 823 item_id,
824. 824 from: crate::types::WorkItemStatus::Proposed,
825. 825 to: crate::types::WorkItemStatus::Ready,
826. 826 attempt: 0,
827. 827 reason: "dependencies satisfied".into(),
828. 828 },
829. 829 })?;
830. 830 }
831. 831 Ok(())
832. 832 }
833. 833
834. 834 pub fn work_projection(
835. 835 &self,
836. 836 ) -> Result<Option<crate::work::WorkProjection>, crate::work::WorkError> {
837. 837 crate::work::project_work(&self.chain_to_root())
838. 838 }
839. 839
840. 840 pub fn evidence_exists(&self, evidence: &crate::types::EvidenceRef) -> bool {
841. 841 let session_id = self.header().map(|header| header.session_id.as_str());
842. 842 self.chain_to_root().iter().any(|entry| match evidence {
843. 843 crate::types::EvidenceRef::LedgerEntry {
844. 844 session_id: evidence_session,
845. 845 entry_id,
846. 846 }
847. 847 | crate::types::EvidenceRef::Receipt {
848. 848 session_id: evidence_session,
849. 849 entry_id,
850. 850 } => session_id == Some(evidence_session.as_str()) && entry.id == *entry_id,
851. 851 crate::types::EvidenceRef::ToolResult {
852. 852 session_id: evidence_session,
853. 853 tool_use_id,
854. 854 } => {
855. 855 session_id == Some(evidence_session.as_str())
856. 856 && matches!(
857. 857 &entry.payload,
858. 858 crate::types::EntryPayload::Message(record)
859. 859 if record.message.content.iter().any(|block| matches!(
860. 860 block,
861. 861 vak_llm::ContentBlock::ToolResult { tool_use_id: id, .. }
862. 862 if id == tool_use_id
863. 863 ))
864. 864 )
865. 865 }
866. 866 crate::types::EvidenceRef::CheckpointDiff { .. }
867. 867 | crate::types::EvidenceRef::FlowNode { .. }
868. 868 | crate::types::EvidenceRef::ChildSession { .. }
869. 869 | crate::types::EvidenceRef::ExternalOperation { .. } => false,
870. 870 })
871. 871 }
872. 872
873. 873 /// Reconcile managed items left running by a process restart. Only child
874. 874 /// sessions explicitly reported as live may remain running; every other
875. 875 /// running item is conservatively interrupted and made retry-eligible.
876. 876 /// Possible side effects are never replayed automatically.
877. 877 pub fn reconcile_running_work(
878. 878 &mut self,
879. 879 live_child_sessions: &std::collections::HashSet<String>,
880. 880 ) -> Result<usize, SessionError> {
881. 881 self.reconcile_running_work_with_child_ledgers(live_child_sessions, None)
882. 882 }
883. 883
884. 884 pub fn reconcile_running_work_with_child_ledgers(
885. 885 &mut self,
886. 886 live_child_sessions: &std::collections::HashSet<String>,
887. 887 sessions_home: Option<&Path>,
888. 888 ) -> Result<usize, SessionError> {
889. 889 let Some(projection) = self
890. 890 .work_projection()
891. 891 .map_err(|error| SessionError::Corrupt {
892. 892 line: 0,
893. 893 message: error.to_string(),
894. 894 })?
895. 895 else {
896. 896 return Ok(0);
897. 897 };
898. 898 let stale: Vec<(String, u32, Option<String>)> = projection
899. 899 .items
900. 900 .values()
901. 901 .filter(|item| {
902. 902 item.status == crate::types::WorkItemStatus::Running
903. 903 && !item
904. 904 .child_session_id
905. 905 .as_ref()
906. 906 .is_some_and(|id| live_child_sessions.contains(id))
907. 907 })
908. 908 .map(|item| {
909. 909 (
910. 910 item.item_id.clone(),
911. 911 item.attempt,
912. 912 item.child_session_id.clone(),
913. 913 )
914. 914 })
915. 915 .collect();
916. 916 let mut reconciled = 0;
917. 917 for (item_id, attempt, child_id) in stale {
918. 918 let Some(current) = self
919. 919 .work_projection()
920. 920 .map_err(|error| SessionError::Corrupt {
921. 921 line: 0,
922. 922 message: error.to_string(),
923. 923 })?
924. 924 else {
925. 925 break;
926. 926 };
927. 927 let child_status = child_id.as_deref().and_then(|id| {
928. 928 let home = sessions_home?;
929. 929 let cwd = self.header().map(|h| h.contract_cwd())?;
930. 930 let path = SessionPath::new_session_file(home, &cwd, id);
931. 931 SessionLog::open(path)
932. 932 .ok()
933. 933 .and_then(|child| child.child_run_status())
934. 934 });
935. 935 if matches!(child_status, Some(crate::types::ChildRunStatus::Completed)) {
936. 936 self.append_work(WorkEvent {
937. 937 contract_id: current.contract.contract_id.clone(),
938. 938 revision: current.contract.revision,
939. 939 kind: crate::types::WorkEventKind::EvidenceAttached {
940. 940 item_id: item_id.clone(),
941. 941 evidence: crate::types::EvidenceRef::ChildSession {
942. 942 session_id: child_id.clone().unwrap_or_default(),
943. 943 },
944. 944 },
945. 945 })?;
946. 946 self.append_work(WorkEvent {
947. 947 contract_id: current.contract.contract_id.clone(),
948. 948 revision: current.contract.revision,
949. 949 kind: crate::types::WorkEventKind::ItemStatusChanged {
950. 950 item_id,
951. 951 from: crate::types::WorkItemStatus::Running,
952. 952 to: crate::types::WorkItemStatus::ReadyForVerification,
953. 953 attempt,
954. 954 reason: "recovered completed child; verify its durable evidence".into(),
955. 955 },
956. 956 })?;
957. 957 reconciled += 1;
958. 958 continue;
959. 959 }
960. 960 self.append_work(WorkEvent {
961. 961 contract_id: current.contract.contract_id.clone(),
962. 962 revision: current.contract.revision,
963. 963 kind: crate::types::WorkEventKind::ItemStatusChanged {
964. 964 item_id,
965. 965 from: crate::types::WorkItemStatus::Running,
966. 966 to: crate::types::WorkItemStatus::Interrupted,
967. 967 attempt,
968. 968 reason: match child_status {
969. 969 Some(crate::types::ChildRunStatus::Failed) => {
970. 970 "child failed before restart; review before retry"
971. 971 }
972. 972 Some(crate::types::ChildRunStatus::Aborted) => {
973. 973 "child was aborted before restart; review before retry"
974. 974 }
975. 975 Some(crate::types::ChildRunStatus::MaxTurns) => {
976. 976 "child hit its turn limit; review before retry"
977. 977 }
978. 978 _ => "recovered after process restart; review before retry",
979. 979 }
980. 980 .into(),
981. 981 },
982. 982 })?;
983. 983 reconciled += 1;
984. 984 }
985. 985 Ok(reconciled)
986. 986 }
987. 987
988. 988 pub fn activities(
989. 989 &self,
990. 990 ) -> Vec<(
991. 991 String,
992. 992 chrono::DateTime<chrono::Utc>,
993. 993 crate::types::ActivityRecord,
994. 994 )> {
995. 995 self.chain_to_root()
996. 996 .into_iter()
997. 997 .filter_map(|entry| match &entry.payload {
998. 998 EntryPayload::Activity(activity) => {
999. 999 Some((entry.id.clone(), entry.ts, activity.clone()))
1000. 1000 }
1001. 1001 _ => None,
1002. 1002 })
1003. 1003 .collect()
1004. 1004 }
1005. 1005
1006. 1006 /// Successful tool-result receipts with the ledger timestamp at which
1007. 1007 /// the result was recorded. This is a replay-safe source for evidence
1008. 1008 /// freshness; callers choose the domain-specific validity window.
1009. 1009 pub fn successful_tool_receipts(&self) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1010. 1010 let mut known_calls = std::collections::HashSet::new();
1011. 1011 let mut receipts = Vec::new();
1012. 1012 for entry in self.chain_to_root() {
1013. 1013 if let EntryPayload::Message(record) = &entry.payload {
1014. 1014 for block in &record.message.content {
1015. 1015 match block {
1016. 1016 vak_llm::ContentBlock::ToolUse { id, .. } => {
1017. 1017 known_calls.insert(id.clone());
1018. 1018 }
1019. 1019 vak_llm::ContentBlock::ToolResult {
1020. 1020 tool_use_id,
1021. 1021 is_error: false,
1022. 1022 ..
1023. 1023 } if known_calls.contains(tool_use_id) => {
1024. 1024 receipts.push((tool_use_id.clone(), entry.ts));
1025. 1025 }
1026. 1026 _ => {}
1027. 1027 }
1028. 1028 }
1029. 1029 }
1030. 1030 }
1031. 1031 receipts
1032. 1032 }
1033. 1033
1034. 1034 /// Successful receipts on the active, latest user turn only. Earlier
1035. 1035 /// turns are deliberately excluded so a stale unrelated command cannot
1036. 1036 /// establish evidence for the current result.
1037. 1037 pub fn successful_tool_receipts_for_latest_turn(
1038. 1038 &self,
1039. 1039 ) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1040. 1040 let mut known_calls = std::collections::HashSet::new();
1041. 1041 let mut receipts = Vec::new();
1042. 1042 for entry in self.chain_to_root() {
1043. 1043 if let EntryPayload::Message(record) = &entry.payload {
1044. 1044 if record.message.role == vak_llm::Role::User
1045. 1045 && record.control_kind().is_none()
1046. 1046 && record
1047. 1047 .message
1048. 1048 .content
1049. 1049 .iter()
1050. 1050 .any(|block| matches!(block, vak_llm::ContentBlock::Text { .. }))
1051. 1051 {
1052. 1052 known_calls.clear();
1053. 1053 receipts.clear();
1054. 1054 }
1055. 1055 for block in &record.message.content {
1056. 1056 match block {
1057. 1057 vak_llm::ContentBlock::ToolUse { id, .. } => {
1058. 1058 known_calls.insert(id.clone());
1059. 1059 }
1060. 1060 vak_llm::ContentBlock::ToolResult {
1061. 1061 tool_use_id,
1062. 1062 is_error: false,
1063. 1063 ..
1064. 1064 } if known_calls.contains(tool_use_id) => {
1065. 1065 receipts.push((tool_use_id.clone(), entry.ts));
1066. 1066 }
1067. 1067 _ => {}
1068. 1068 }
1069. 1069 }
1070. 1070 }
1071. 1071 }