harn_vm/agent_sessions/
transcript_lifecycle.rs

1use super::*;
2
3pub fn messages_json(id: &str) -> Vec<serde_json::Value> {
4    SESSIONS.with(|s| {
5        let map = s.borrow();
6        let Some(state) = map.get(id) else {
7            return Vec::new();
8        };
9        let Some(dict) = state.transcript.as_dict() else {
10            return Vec::new();
11        };
12        match dict.get("messages") {
13            Some(VmValue::List(list)) => list
14                .iter()
15                .map(crate::llm::helpers::vm_value_to_json)
16                .collect(),
17            _ => Vec::new(),
18        }
19    })
20}
21
22#[derive(Clone, Debug, Default)]
23pub struct SessionPromptState {
24    pub messages: Vec<serde_json::Value>,
25    pub summary: Option<String>,
26}
27
28fn summary_message_json(summary: &str) -> serde_json::Value {
29    serde_json::json!({
30        "role": "user",
31        "content": summary,
32    })
33}
34
35fn messages_begin_with_summary(messages: &[serde_json::Value], summary: &str) -> bool {
36    messages.first().is_some_and(|message| {
37        message.get("role").and_then(|value| value.as_str()) == Some("user")
38            && message.get("content").and_then(|value| value.as_str()) == Some(summary)
39    })
40}
41
42/// Prompt-surface resume state for a persisted session.
43///
44/// Returns the compacted/rehydratable message list plus the transcript's
45/// summary field. When the transcript carries a summary field but its
46/// message list does not already begin with the compacted summary
47/// message, this helper prepends one so session re-entry preserves the
48/// same prompt surface the previous loop was actually using.
49pub fn prompt_state_json(id: &str) -> SessionPromptState {
50    SESSIONS.with(|s| {
51        let map = s.borrow();
52        let Some(state) = map.get(id) else {
53            return SessionPromptState::default();
54        };
55        let Some(dict) = state.transcript.as_dict() else {
56            return SessionPromptState::default();
57        };
58        let mut messages = match dict.get("messages") {
59            Some(VmValue::List(list)) => list
60                .iter()
61                .map(crate::llm::helpers::vm_value_to_json)
62                .collect::<Vec<_>>(),
63            _ => Vec::new(),
64        };
65        let summary = dict.get("summary").and_then(|value| match value {
66            VmValue::String(text) if !text.trim().is_empty() => Some(text.to_string()),
67            _ => None,
68        });
69        if let Some(summary_text) = summary.as_deref() {
70            if !messages_begin_with_summary(&messages, summary_text) {
71                messages.insert(0, summary_message_json(summary_text));
72            }
73        }
74        SessionPromptState { messages, summary }
75    })
76}
77
78/// Overwrite the transcript for this session. Used by `agent_loop` on
79/// exit to persist the synthesized transcript.
80pub fn store_transcript(id: &str, transcript: VmValue) -> Result<(), String> {
81    SESSIONS.with(|s| {
82        let mut map = s.borrow_mut();
83        let Some(state) = map.get_mut(id) else {
84            return Err(format!(
85                "agent_session_store_transcript: unknown session id '{id}'"
86            ));
87        };
88        let transcript = transcript_with_session_metadata(transcript, state);
89        let text_tool_call_seq = next_text_tool_call_seq_from_transcript(&transcript);
90        apply_transcript_with_budget(state, transcript, "store_transcript")?;
91        state.text_tool_call_seq = state.text_tool_call_seq.max(text_tool_call_seq);
92        Ok(())
93    })
94}
95
96fn checkpoint_summary(checkpoint: &SessionTurnCheckpoint) -> SessionCheckpointSummary {
97    SessionCheckpointSummary {
98        checkpoint_id: checkpoint.checkpoint_id.clone(),
99        before_message_count: checkpoint.before_message_count,
100        after_message_count: checkpoint.after_message_count,
101        fs_snapshot_ids: checkpoint.fs_snapshot_ids.clone(),
102    }
103}
104
105fn checkpoint_error_status(error: SessionCheckpointError) -> &'static str {
106    match error {
107        SessionCheckpointError::UnknownSession => "unknown_session",
108        SessionCheckpointError::NoCheckpoint => "no_checkpoint",
109        SessionCheckpointError::NoRedo => "no_redo",
110    }
111}
112
113pub fn checkpoint_status_name(error: SessionCheckpointError) -> &'static str {
114    checkpoint_error_status(error)
115}
116
117/// Clear redo checkpoints after host-side workspace mutations that are not part
118/// of the redo flow. Returns whether any redo state was discarded.
119pub fn invalidate_redo(id: &str) -> bool {
120    SESSIONS.with(|s| {
121        let mut map = s.borrow_mut();
122        let Some(state) = map.get_mut(id) else {
123            return false;
124        };
125        let had_redo = !state.redo_stack.is_empty();
126        state.redo_stack.clear();
127        state.touch();
128        had_redo
129    })
130}
131
132/// Record a completed prompt turn boundary.
133///
134/// `before_transcript` must be captured immediately before the user turn
135/// starts. The current live transcript becomes the redo target, and optional
136/// `fs_snapshot_ids` name host-owned filesystem snapshots captured during the
137/// turn. Harn owns the transcript stack; hosts own concrete file restoration.
138pub fn record_completed_turn_checkpoint(
139    id: &str,
140    before_transcript: VmValue,
141    fs_snapshot_ids: Vec<String>,
142) -> Result<Option<SessionCheckpointSummary>, SessionCheckpointError> {
143    SESSIONS.with(|s| {
144        let mut map = s.borrow_mut();
145        let Some(state) = map.get_mut(id) else {
146            return Err(SessionCheckpointError::UnknownSession);
147        };
148        let after_transcript = transcript_with_session_metadata(state.transcript.clone(), state);
149        let before_message_count = transcript_message_count(&before_transcript);
150        let after_message_count = transcript_message_count(&after_transcript);
151        if crate::values_equal(&before_transcript, &after_transcript) && fs_snapshot_ids.is_empty()
152        {
153            return Ok(None);
154        }
155        let checkpoint = SessionTurnCheckpoint {
156            checkpoint_id: format!("turn_{}", uuid::Uuid::now_v7().simple()),
157            completed_at: crate::orchestration::now_unix_seconds_text(),
158            before_message_count,
159            after_message_count,
160            before_transcript,
161            after_transcript,
162            fs_snapshot_ids,
163        };
164        state.redo_stack.clear();
165        state.completed_turn_checkpoints.push(checkpoint.clone());
166        state.touch();
167        Ok(Some(checkpoint_summary(&checkpoint)))
168    })
169}
170
171pub fn rollback_plan(id: &str) -> Result<SessionCheckpointSummary, SessionCheckpointError> {
172    SESSIONS.with(|s| {
173        let map = s.borrow();
174        let Some(state) = map.get(id) else {
175            return Err(SessionCheckpointError::UnknownSession);
176        };
177        state
178            .completed_turn_checkpoints
179            .last()
180            .map(checkpoint_summary)
181            .ok_or(SessionCheckpointError::NoCheckpoint)
182    })
183}
184
185pub fn redo_plan(id: &str) -> Result<SessionCheckpointSummary, SessionCheckpointError> {
186    SESSIONS.with(|s| {
187        let map = s.borrow();
188        let Some(state) = map.get(id) else {
189            return Err(SessionCheckpointError::UnknownSession);
190        };
191        state
192            .redo_stack
193            .last()
194            .map(|entry| {
195                let mut summary = checkpoint_summary(&entry.checkpoint);
196                summary.fs_snapshot_ids = entry.redo_fs_snapshot_ids.clone();
197                summary
198            })
199            .ok_or(SessionCheckpointError::NoRedo)
200    })
201}
202
203pub fn rollback_last_completed_turn(
204    id: &str,
205    redo_fs_snapshot_ids: Vec<String>,
206) -> Result<SessionCheckpointOutcome, SessionCheckpointError> {
207    SESSIONS.with(|s| {
208        let mut map = s.borrow_mut();
209        let Some(state) = map.get_mut(id) else {
210            return Err(SessionCheckpointError::UnknownSession);
211        };
212        let Some(checkpoint) = state.completed_turn_checkpoints.pop() else {
213            return Err(SessionCheckpointError::NoCheckpoint);
214        };
215        state.transcript = checkpoint.before_transcript.clone();
216        state.redo_stack.push(SessionRedoEntry {
217            checkpoint: checkpoint.clone(),
218            redo_fs_snapshot_ids: redo_fs_snapshot_ids.clone(),
219        });
220        state.touch();
221        Ok(SessionCheckpointOutcome {
222            status: "rolled_back",
223            checkpoint: checkpoint_summary(&checkpoint),
224            redo_fs_snapshot_ids,
225        })
226    })
227}
228
229pub fn redo_last_rollback(id: &str) -> Result<SessionCheckpointOutcome, SessionCheckpointError> {
230    SESSIONS.with(|s| {
231        let mut map = s.borrow_mut();
232        let Some(state) = map.get_mut(id) else {
233            return Err(SessionCheckpointError::UnknownSession);
234        };
235        let Some(entry) = state.redo_stack.pop() else {
236            return Err(SessionCheckpointError::NoRedo);
237        };
238        let checkpoint = entry.checkpoint;
239        state.transcript = checkpoint.after_transcript.clone();
240        state.completed_turn_checkpoints.push(checkpoint.clone());
241        state.touch();
242        Ok(SessionCheckpointOutcome {
243            status: "redone",
244            checkpoint: checkpoint_summary(&checkpoint),
245            redo_fs_snapshot_ids: entry.redo_fs_snapshot_ids,
246        })
247    })
248}
249
250/// Remove malformed reminder events after their drop audit has been emitted.
251/// Pending-reminder rendering scans the transcript on every LLM call; pruning
252/// invalid entries makes the drop event one-shot instead of noisy per turn.
253pub fn prune_invalid_reminder_events(id: &str) -> usize {
254    SESSIONS.with(|s| {
255        let mut map = s.borrow_mut();
256        let Some(state) = map.get_mut(id) else {
257            return 0;
258        };
259        let Some(dict) = state.transcript.as_dict().cloned() else {
260            return 0;
261        };
262        let Some(VmValue::List(events)) = dict.get("events") else {
263            return 0;
264        };
265        let mut pruned = 0_usize;
266        let mut kept = Vec::with_capacity(events.len());
267        for event in events.iter().cloned() {
268            let is_reminder = event
269                .as_dict()
270                .and_then(|event| event.get("kind"))
271                .map(VmValue::display)
272                .as_deref()
273                == Some(crate::llm::helpers::SYSTEM_REMINDER_EVENT_KIND);
274            if !is_reminder {
275                kept.push(event);
276                continue;
277            }
278            let valid = crate::llm::helpers::reminder_from_event(&event)
279                .is_some_and(|reminder| !reminder.body.trim().is_empty());
280            if valid {
281                kept.push(event);
282            } else {
283                pruned += 1;
284            }
285        }
286        if pruned > 0 {
287            let mut next = dict;
288            next.insert(
289                crate::value::intern_key("events"),
290                VmValue::List(std::sync::Arc::new(kept)),
291            );
292            let _ = apply_transcript_with_budget(
293                state,
294                VmValue::dict(next),
295                "prune_invalid_reminder_events",
296            );
297            state.touch();
298        }
299        pruned
300    })
301}
302
303/// Apply the reminder TTL lifecycle that runs once per completed agent
304/// turn. Reminders with `ttl_turns = 1` expire and are removed; larger
305/// finite TTLs are decremented in place. Expiry audit events are emitted
306/// to the active EventLog when one is installed.
307pub fn apply_reminder_post_turn(id: &str, turn: i64) -> Result<serde_json::Value, String> {
308    let report = SESSIONS.with(|s| {
309        let mut map = s.borrow_mut();
310        let Some(state) = map.get_mut(id) else {
311            return Err(format!(
312                "agent_session_apply_reminder_post_turn: unknown session id '{id}'"
313            ));
314        };
315        let report = crate::llm::helpers::apply_reminder_post_turn(&state.transcript, turn);
316        if report.decremented_count > 0 || !report.expired.is_empty() {
317            if let Some(next) = report.transcript.clone() {
318                apply_transcript_with_budget(state, next, "apply_reminder_post_turn")?;
319            }
320            state.touch();
321        }
322        Ok(report)
323    })?;
324
325    for reminder in &report.expired {
326        let mut payload = crate::llm::helpers::reminder_lifecycle_payload(Some(id), reminder);
327        if let Some(obj) = payload.as_object_mut() {
328            obj.insert(
329                "transcript_id".to_string(),
330                serde_json::Value::String(id.to_string()),
331            );
332            obj.insert(
333                "reason".to_string(),
334                serde_json::Value::String("ttl".to_string()),
335            );
336            obj.insert(
337                "ttl_turns_before".to_string(),
338                serde_json::json!(&reminder.ttl_turns),
339            );
340            obj.insert("expired_at_turn".to_string(), serde_json::json!(turn));
341        }
342        crate::llm::helpers::emit_reminder_lifecycle_event(
343            crate::llm::helpers::REMINDER_EXPIRED_EVENT_KIND,
344            payload,
345        );
346    }
347
348    Ok(serde_json::json!({
349        "expired_count": report.expired.len(),
350        "decremented_count": report.decremented_count,
351        "remaining_count": report.remaining_count,
352    }))
353}
354
355/// Inject a typed system reminder into the session transcript's event
356/// stream. This mirrors `transcript.inject_reminder` for live sessions:
357/// reminders with the same `dedupe_key` are replaced before the new
358/// reminder event is appended.
359pub fn inject_reminder(
360    id: &str,
361    reminder: crate::llm::helpers::SystemReminder,
362) -> Result<ReminderInjectionReport, String> {
363    let reminder_id = reminder.id.clone();
364    let dedupe_key = reminder.dedupe_key.clone();
365    let mut deduped_reminder_ids = Vec::new();
366    SESSIONS.with(|s| {
367        let mut map = s.borrow_mut();
368        let Some(state) = map.get_mut(id) else {
369            return Err(format!(
370                "agent_session_inject_reminder: unknown session id '{id}'"
371            ));
372        };
373        let dict = state
374            .transcript
375            .as_dict()
376            .cloned()
377            .unwrap_or_else(crate::value::DictMap::new);
378        let mut events: Vec<VmValue> = match dict.get("events") {
379            Some(VmValue::List(list)) => list.iter().cloned().collect(),
380            _ => dict
381                .get("messages")
382                .and_then(|value| match value {
383                    VmValue::List(list) => Some(list.iter().cloned().collect::<Vec<_>>()),
384                    _ => None,
385                })
386                .map(|messages| crate::llm::helpers::transcript_events_from_messages(&messages))
387                .unwrap_or_default(),
388        };
389        if let Some(expected_key) = dedupe_key.as_deref() {
390            events.retain(|event| {
391                let Some(existing) = crate::llm::helpers::reminder_from_event(event) else {
392                    return true;
393                };
394                if existing.dedupe_key.as_deref() == Some(expected_key) {
395                    deduped_reminder_ids.push(existing.id);
396                    false
397                } else {
398                    true
399                }
400            });
401        }
402        events.push(crate::llm::helpers::transcript_reminder_event(&reminder));
403        let mut next = dict;
404        next.insert(
405            crate::value::intern_key("events"),
406            VmValue::List(std::sync::Arc::new(events)),
407        );
408        apply_transcript_with_budget(state, VmValue::dict(next), "inject_reminder")?;
409        state.touch();
410        Ok(())
411    })?;
412
413    if !deduped_reminder_ids.is_empty() {
414        let dropped_count = deduped_reminder_ids.len();
415        crate::llm::helpers::emit_reminder_lifecycle_event(
416            crate::llm::helpers::REMINDER_DEDUPED_EVENT_KIND,
417            serde_json::json!({
418                "session_id": id,
419                "transcript_id": id,
420                "reminder_id": &reminder_id,
421                "replacing_id": &reminder_id,
422                "replaced_id": deduped_reminder_ids.first(),
423                "replaced_ids": &deduped_reminder_ids,
424                "dedupe_key": &dedupe_key,
425                "dropped_reminder_ids": &deduped_reminder_ids,
426                "dropped_count": dropped_count,
427            }),
428        );
429    }
430
431    crate::llm::helpers::emit_reminder_lifecycle_event(
432        crate::llm::helpers::REMINDER_INJECTED_EVENT_KIND,
433        crate::llm::helpers::reminder_lifecycle_payload(Some(id), &reminder),
434    );
435
436    Ok(ReminderInjectionReport {
437        reminder_id,
438        deduped_count: deduped_reminder_ids.len(),
439    })
440}
harn_vm/agent_sessions/transcript_lifecycle.rs

harn_vm/agent_sessions/
transcript_lifecycle.rs
