mj_transcript/
turn_context.rs

1//! Process-local classification evidence using the shared transcript summary.
2use agent_client_protocol::schema::v1::{ContentBlock, SessionUpdate};
3use mj_core::activity::ActivityFacts;
4use mj_core::activity::verdict::*;
5use mj_core::config::HarnessKind;
6use std::sync::{Arc, Mutex};
7
8#[derive(Debug, Default)]
9struct TurnContextState {
10    generation: u64,
11    parent_activity: Option<std::time::Instant>,
12    decision_log: Option<mj_core::jev::DecisionLog>,
13    authorization: Option<mj_core::assessment::ContextHistory>,
14    user_prompt_tail: String,
15    message_id: Option<String>,
16    assistant_text_tail: String,
17    last_completed_message: String,
18    summary: crate::summary::TranscriptSummary,
19    background_commands: usize,
20    queued_commands: usize,
21    background_inventory: Vec<mj_core::relay::BackgroundCommand>,
22    native_agent_ids: Vec<String>,
23    session_id: String,
24}
25
26/// The prompt that became visible to the harness at this durable observation.
27/// A queued or unconfirmed steering prompt has not been delivered yet.
28pub fn delivered_prompt_command_id(observation: &mj_core::relay::RelayObservation) -> Option<&str> {
29    use mj_core::relay::{RelayCommandOutcome, RelayObservation};
30    match observation {
31        RelayObservation::CommandStarted { command_id, .. } => Some(command_id),
32        RelayObservation::CommandCompleted {
33            outcome: RelayCommandOutcome::Steered { queued_command_id },
34            ..
35        } => Some(queued_command_id),
36        RelayObservation::MailboxEventsDelivered {
37            steering: Some(steering),
38            ..
39        } => Some(&steering.queued_prompt_id),
40        _ => None,
41    }
42}
43
44/// The relay and runtime share this small process-local evidence accumulator.
45#[derive(Debug, Default, Clone)]
46pub struct TurnContext(Arc<Mutex<TurnContextState>>);
47
48impl TurnContext {
49    /// Process-local parent clock; child traffic and inventory updates never mark it.
50    pub fn mark_parent_activity(&self) {
51        self.mark_parent_activity_at(std::time::Instant::now());
52    }
53
54    /// Record the monotonic time at which parent activity was observed.
55    pub fn mark_parent_activity_at(&self, at: std::time::Instant) {
56        self.0
57            .lock()
58            .expect("turn context lock poisoned")
59            .parent_activity = Some(at);
60    }
61
62    pub fn parent_activity(&self) -> Option<std::time::Instant> {
63        self.0
64            .lock()
65            .expect("turn context lock poisoned")
66            .parent_activity
67    }
68
69    pub fn set_decision_log(&self, log: mj_core::jev::DecisionLog) {
70        self.0
71            .lock()
72            .expect("turn context lock poisoned")
73            .decision_log = Some(log);
74    }
75    pub fn decision_log(&self) -> Option<mj_core::jev::DecisionLog> {
76        self.0
77            .lock()
78            .expect("turn context lock poisoned")
79            .decision_log
80            .clone()
81    }
82    pub fn reset(&self, prompt: &str) {
83        let mut state = self.0.lock().expect("turn context lock poisoned");
84        let generation = state.generation.wrapping_add(1);
85        *state = TurnContextState {
86            generation,
87            parent_activity: Some(std::time::Instant::now()),
88            decision_log: state.decision_log.clone(),
89            authorization: state.authorization.clone(),
90            user_prompt_tail: tail(prompt, USER_PROMPT_BYTES),
91            background_commands: state.background_commands,
92            queued_commands: state.queued_commands,
93            background_inventory: std::mem::take(&mut state.background_inventory),
94            native_agent_ids: std::mem::take(&mut state.native_agent_ids),
95            session_id: std::mem::take(&mut state.session_id),
96            summary: std::mem::take(&mut state.summary),
97            ..Default::default()
98        };
99    }
100
101    /// Record durable transcript observations through the same path during live work and replay.
102    pub fn observe_relay(
103        &self,
104        observation: &mj_core::relay::RelayObservation,
105        prompt: Option<&str>,
106    ) {
107        use mj_core::relay::{RelayCommandOutcome, RelayObservation};
108        match observation {
109            RelayObservation::CommandStarted { .. }
110            | RelayObservation::CommandCompleted {
111                outcome: RelayCommandOutcome::Steered { .. },
112                ..
113            }
114            | RelayObservation::MailboxEventsDelivered {
115                steering: Some(_), ..
116            } => {
117                if let Some(prompt) = prompt {
118                    self.reset(prompt);
119                    self.0
120                        .lock()
121                        .expect("turn context lock poisoned")
122                        .summary
123                        .push_user(prompt);
124                }
125            }
126            RelayObservation::SessionUpdate { update } => self.observe(update),
127            RelayObservation::TerminalOutput {
128                terminal_id,
129                output,
130                truncated,
131                exit_code,
132                signal,
133            } => {
134                self.observe_terminal(&mj_core::transcript::TerminalOutputRecord {
135                    terminal_id: terminal_id.clone(),
136                    output: output.clone(),
137                    truncated: *truncated,
138                    exit_code: *exit_code,
139                    signal: signal.clone(),
140                });
141            }
142            RelayObservation::CommandCompleted {
143                outcome: RelayCommandOutcome::ContextCleared { .. },
144                ..
145            } => self.clear_history(),
146            _ => {}
147        }
148    }
149
150    /// Invalidate a pending verdict at a lifecycle boundary without discarding evidence.
151    pub fn invalidate(&self) {
152        let mut state = self.0.lock().expect("turn context lock poisoned");
153        state.generation = state.generation.wrapping_add(1);
154    }
155
156    pub fn generation(&self) -> u64 {
157        self.0
158            .lock()
159            .expect("turn context lock poisoned")
160            .generation
161    }
162
163    pub fn counts(&self) -> (usize, usize) {
164        let state = self.0.lock().expect("turn context lock poisoned");
165        (state.background_commands, state.queued_commands)
166    }
167
168    pub fn set_counts(&self, background_commands: usize, queued_commands: usize) {
169        let mut state = self.0.lock().expect("turn context lock poisoned");
170        if (state.background_commands, state.queued_commands)
171            != (background_commands, queued_commands)
172        {
173            state.generation = state.generation.wrapping_add(1);
174            state.background_commands = background_commands;
175            state.queued_commands = queued_commands;
176        }
177    }
178
179    pub fn set_session_id(&self, session_id: &str) {
180        self.0
181            .lock()
182            .expect("turn context lock poisoned")
183            .session_id = session_id.into();
184    }
185
186    pub fn session_id(&self) -> String {
187        self.0
188            .lock()
189            .expect("turn context lock poisoned")
190            .session_id
191            .clone()
192    }
193
194    /// Publish the relay owner's durable user-instruction history to the ACP
195    /// classifier. A change invalidates any request built from older context.
196    pub fn set_authorization_context(
197        &self,
198        authorization: Option<mj_core::assessment::ContextHistory>,
199    ) {
200        let mut state = self.0.lock().expect("turn context lock poisoned");
201        if state.authorization != authorization {
202            state.authorization = authorization;
203            state.generation = state.generation.wrapping_add(1);
204        }
205    }
206
207    pub fn authorization_context(&self) -> Option<mj_core::assessment::ContextHistory> {
208        self.0
209            .lock()
210            .expect("turn context lock poisoned")
211            .authorization
212            .clone()
213    }
214
215    /// Compare identities as well as counts; identical level reports are not activity.
216    pub fn set_background_inventory(
217        &self,
218        mut commands: Vec<mj_core::relay::BackgroundCommand>,
219        mut native_agent_ids: Vec<String>,
220    ) {
221        commands.sort_by(|a, b| a.id.cmp(&b.id));
222        native_agent_ids.sort();
223        let mut state = self.0.lock().expect("turn context lock poisoned");
224        if state.background_inventory != commands || state.native_agent_ids != native_agent_ids {
225            state.background_inventory = commands;
226            state.native_agent_ids = native_agent_ids;
227            state.generation = state.generation.wrapping_add(1);
228        }
229    }
230
231    pub fn observe(&self, update: &SessionUpdate) {
232        let mut state = self.0.lock().expect("turn context lock poisoned");
233        state.parent_activity = Some(std::time::Instant::now());
234        if matches!(
235            update,
236            SessionUpdate::AgentMessageChunk(_)
237                | SessionUpdate::AgentThoughtChunk(_)
238                | SessionUpdate::ToolCall(_)
239                | SessionUpdate::ToolCallUpdate(_)
240        ) {
241            state.generation = state.generation.wrapping_add(1);
242        }
243        state.summary.observe(update);
244        match update {
245            SessionUpdate::AgentMessageChunk(chunk) => {
246                let id = chunk.message_id.as_ref().map(ToString::to_string);
247                if id != state.message_id {
248                    complete_message(&mut state);
249                    state.message_id = id;
250                }
251                if let ContentBlock::Text(content) = &chunk.content {
252                    // Trim each chunk before appending, so a huge chunk never grows retained state.
253                    state
254                        .assistant_text_tail
255                        .push_str(&tail(&content.text, ASSISTANT_TEXT_BYTES));
256                    state.assistant_text_tail =
257                        tail(&state.assistant_text_tail, ASSISTANT_TEXT_BYTES);
258                }
259            }
260            SessionUpdate::ToolCall(_) => {
261                complete_message(&mut state);
262            }
263            SessionUpdate::ToolCallUpdate(_) | SessionUpdate::AgentThoughtChunk(_) => {
264                complete_message(&mut state)
265            }
266            _ => {}
267        }
268    }
269
270    pub fn observe_terminal(&self, record: &mj_core::transcript::TerminalOutputRecord) {
271        let mut state = self.0.lock().expect("turn context lock poisoned");
272        state.summary.observe_terminal(record);
273        state.generation = state.generation.wrapping_add(1);
274    }
275
276    pub fn mark_earlier_history_omitted(&self) {
277        self.0
278            .lock()
279            .expect("turn context lock poisoned")
280            .summary
281            .mark_earlier_history_omitted();
282    }
283
284    pub fn clear_history(&self) {
285        let mut state = self.0.lock().expect("turn context lock poisoned");
286        state.summary = Default::default();
287        state.user_prompt_tail.clear();
288        state.assistant_text_tail.clear();
289        state.last_completed_message.clear();
290        state.message_id = None;
291        state.authorization = None;
292        state.generation = state.generation.wrapping_add(1);
293    }
294
295    pub fn evidence(
296        &self,
297        harness: HarnessKind,
298        phase: TurnPhase,
299        facts: &ActivityFacts,
300        now_ms: i64,
301    ) -> TurnEvidence {
302        let state = self.0.lock().expect("turn context lock poisoned");
303        let summary = state.summary.latest_user_messages();
304        let last_assistant = state
305            .summary
306            .entries
307            .iter()
308            .rposition(|e| e.role == crate::summary::SummaryRole::Assistant);
309        let final_tool_calls = state
310            .summary
311            .entries
312            .iter()
313            .skip(last_assistant.map_or(0, |i| i + 1))
314            .filter(|e| e.role == crate::summary::SummaryRole::Tool)
315            .take(IN_FLIGHT_TOOLS)
316            .map(|e| mj_core::activity::verdict::ToolOutcome {
317                name: tail(
318                    e.tool
319                        .as_ref()
320                        .and_then(|t| t.get("name"))
321                        .and_then(|n| n.as_str())
322                        .unwrap_or(&e.text),
323                    TOOL_TITLE_BYTES,
324                ),
325                status: e
326                    .tool
327                    .as_ref()
328                    .and_then(|t| t.get("status"))
329                    .and_then(|s| s.as_str())
330                    .unwrap_or("unknown")
331                    .to_owned(),
332            })
333            .collect();
334        let mut evidence = TurnEvidence {
335            authorization: None,
336            final_tool_calls,
337            background: Vec::new(),
338            harness,
339            phase,
340            silent_for_s: facts
341                .last_acp_activity_at_ms
342                .map_or(0, |last| now_ms.saturating_sub(last).max(0) as u64 / 1000),
343            tools_in_flight: facts
344                .tools_in_flight
345                .iter()
346                .take(IN_FLIGHT_TOOLS)
347                .map(|tool| ToolEvidence {
348                    title: tail(
349                        &state
350                            .summary
351                            .entries
352                            .iter()
353                            .find(|e| e.id == tool.tool_call_id)
354                            .map(|e| e.text.clone())
355                            .unwrap_or_else(|| "unknown tool".into()),
356                        TOOL_TITLE_BYTES,
357                    ),
358                    running_s: now_ms.saturating_sub(tool.started_at_ms).max(0) as u64 / 1000,
359                })
360                .collect(),
361            transcript_summary: summary.render(48 * 1024),
362            background_commands: facts.background_commands,
363            queued_commands: facts.queued_commands,
364            user_prompt_tail: state.user_prompt_tail.clone(),
365            assistant_text_tail: if state.assistant_text_tail.is_empty() {
366                state.last_completed_message.clone()
367            } else {
368                state.assistant_text_tail.clone()
369            },
370            completion: None,
371        };
372        let mut limit = 48 * 1024;
373        while serde_json::to_vec(&evidence)
374            .expect("serialize evidence")
375            .len()
376            > 60 * 1024
377        {
378            limit /= 2;
379            evidence.transcript_summary = summary.render(limit);
380        }
381        evidence
382    }
383}
384
385fn complete_message(state: &mut TurnContextState) {
386    if !state.assistant_text_tail.is_empty() {
387        state.last_completed_message = std::mem::take(&mut state.assistant_text_tail);
388    }
389    state.message_id = None;
390}
391
392fn tail(text: &str, maximum_bytes: usize) -> String {
393    let start = text.floor_char_boundary(text.len().saturating_sub(maximum_bytes));
394    let mut value = text[start..].to_owned();
395    mj_core::transcript::truncate_string_start(&mut value, maximum_bytes);
396    value
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402    use mj_core::activity::InFlightToolCall;
403    use serde_json::json;
404    fn message(id: &str, text: &str) -> SessionUpdate {
405        serde_json::from_value(json!({"sessionUpdate":"agent_message_chunk","messageId":id,"content":{"type":"text","text":text}})).unwrap()
406    }
407
408    fn tool(title: &str) -> SessionUpdate {
409        serde_json::from_value(json!({"sessionUpdate":"tool_call","toolCallId":title,"title":title,"status":"in_progress"})).unwrap()
410    }
411
412    #[test]
413    fn diagnostic_logging_does_not_change_evidence_or_generation_and_survives_reset() {
414        let context = TurnContext::default();
415        context.reset("Implement the parser");
416        let generation = context.generation();
417        let before = serde_json::to_value(context.evidence(
418            HarnessKind::Codex,
419            TurnPhase::Running,
420            &Default::default(),
421            0,
422        ))
423        .unwrap();
424        let dir = tempfile::tempdir().unwrap();
425        let log = mj_core::jev::DecisionLog::open(dir.path().into()).unwrap();
426        context.set_decision_log(log.clone());
427        let attempt = log.start("s", "activity", "Who acts?", "Current request");
428        attempt.finish("unchanged", "Kept runtime facts");
429        assert_eq!(context.generation(), generation);
430        assert_eq!(
431            serde_json::to_value(context.evidence(
432                HarnessKind::Codex,
433                TurnPhase::Running,
434                &Default::default(),
435                0
436            ))
437            .unwrap(),
438            before
439        );
440        context.reset("New work");
441        assert!(context.decision_log().is_some());
442    }
443
444    #[test]
445    fn verdict_uses_latest_delivered_user_without_tool_history_but_keeps_live_facts() {
446        use mj_core::relay::RelayObservation;
447        let context = TurnContext::default();
448        let start = |id: &str| RelayObservation::CommandStarted {
449            command_id: id.into(),
450            started_at_ms: 0,
451        };
452        context.observe_relay(&start("old"), Some("OLD REQUEST"));
453        context.observe(&message("old", "OLD ANSWER"));
454        context.observe(&tool("active-build"));
455        context.observe_relay(&start("new"), Some("CURRENT REQUEST"));
456        let noisy: SessionUpdate = serde_json::from_value(json!({
457            "sessionUpdate":"tool_call", "toolCallId":"noisy", "title":"noisy", "status":"completed",
458            "rawInput":{"command":"TOOL_BODY".repeat(20_000)}
459        })).unwrap();
460        context.observe(&noisy);
461        context.observe(&message("new", "CURRENT ANSWER"));
462        let facts = ActivityFacts {
463            background_commands: 2,
464            queued_commands: 3,
465            tools_in_flight: vec![InFlightToolCall {
466                tool_call_id: "active-build".into(),
467                title: Some("active-build".into()),
468                status: agent_client_protocol::schema::v1::ToolCallStatus::InProgress,
469                started_at_ms: 1_000,
470            }],
471            ..Default::default()
472        };
473        let evidence = context.evidence(HarnessKind::Codex, TurnPhase::Running, &facts, 5_000);
474        assert!(evidence.transcript_summary.contains("CURRENT REQUEST"));
475        assert!(evidence.transcript_summary.contains("CURRENT ANSWER"));
476        for excluded in [
477            "OLD REQUEST",
478            "OLD ANSWER",
479            "TOOL_BODY",
480            "<tool",
481            "bytes omitted",
482        ] {
483            assert!(
484                !evidence.transcript_summary.contains(excluded),
485                "{excluded}"
486            );
487        }
488        assert_eq!(evidence.user_prompt_tail, "CURRENT REQUEST");
489        assert_eq!(evidence.assistant_text_tail, "CURRENT ANSWER");
490        assert_eq!(evidence.background_commands, 2);
491        assert_eq!(evidence.queued_commands, 3);
492        assert_eq!(evidence.tools_in_flight.len(), 1);
493        assert!(evidence.tools_in_flight[0].title.contains("active-build"));
494        assert_eq!(evidence.tools_in_flight[0].running_s, 4);
495        let full = context.0.lock().unwrap().summary.render(256 * 1024);
496        assert!(full.contains("OLD REQUEST") && full.contains("TOOL_BODY"));
497    }
498
499    #[test]
500    fn tool_calls_after_the_last_assistant_text_are_named_in_the_evidence() {
501        use mj_core::relay::RelayObservation;
502        let context = TurnContext::default();
503        context.observe_relay(
504            &RelayObservation::CommandStarted {
505                command_id: "audit".into(),
506                started_at_ms: 0,
507            },
508            Some("Audit the cache and hand back."),
509        );
510        context.observe(&message("m1", "I'm checking the update paths now."));
511        let handback: SessionUpdate = serde_json::from_value(json!({
512            "sessionUpdate":"tool_call", "toolCallId":"hb", "title":"mcp__mj-agents__handback",
513            "status":"completed", "rawInput":{"report":"done"}
514        }))
515        .unwrap();
516        context.observe(&handback);
517        let facts = ActivityFacts::default();
518        let evidence = context.evidence(HarnessKind::Codex, TurnPhase::Replied, &facts, 5_000);
519        assert_eq!(evidence.final_tool_calls.len(), 1);
520        assert!(evidence.final_tool_calls[0].name.contains("handback"));
521        assert_eq!(evidence.final_tool_calls[0].status, "completed");
522        // Text after the call closes the list again.
523        context.observe(&message("m2", "Handed back."));
524        let evidence = context.evidence(HarnessKind::Codex, TurnPhase::Replied, &facts, 5_000);
525        assert!(evidence.final_tool_calls.is_empty());
526    }
527
528    #[test]
529    fn evidence_caps_utf8_text_tools_and_durations() {
530        let context = TurnContext::default();
531        context.reset(&"é".repeat(20_000));
532        context.observe(&message("first", &"🦀".repeat(20_000)));
533        let first = context.evidence(
534            HarnessKind::Claude,
535            TurnPhase::Running,
536            &ActivityFacts::default(),
537            0,
538        );
539        assert_eq!(first.user_prompt_tail.len(), USER_PROMPT_BYTES);
540        assert_eq!(first.assistant_text_tail.len(), ASSISTANT_TEXT_BYTES);
541        for n in 0..20 {
542            context.observe(&tool(&format!("{n}{}", "é".repeat(300))));
543        }
544        let facts = ActivityFacts {
545            last_acp_activity_at_ms: Some(1_000),
546            background_commands: 3,
547            queued_commands: 2,
548            tools_in_flight: (0..100)
549                .map(|n| InFlightToolCall {
550                    tool_call_id: n.to_string(),
551                    title: Some("🦀".repeat(200)),
552                    status: agent_client_protocol::schema::v1::ToolCallStatus::InProgress,
553                    started_at_ms: 2_000,
554                })
555                .collect(),
556            ..Default::default()
557        };
558        let evidence = context.evidence(HarnessKind::Claude, TurnPhase::Running, &facts, 95_000);
559        assert_eq!(evidence.tools_in_flight.len(), IN_FLIGHT_TOOLS);
560        assert!(
561            evidence
562                .tools_in_flight
563                .iter()
564                .all(|tool| tool.title.len() <= TOOL_TITLE_BYTES && tool.running_s == 93)
565        );
566        assert!(evidence.transcript_summary.len() <= 48 * 1024);
567        assert_eq!(evidence.assistant_text_tail, first.assistant_text_tail);
568        assert_eq!(evidence.silent_for_s, 94);
569        let json = serde_json::to_value(evidence).unwrap();
570        assert_eq!(json["phase"], "running");
571        assert_eq!(json["harness"], "claude");
572        assert_eq!(json["background_commands"], 3);
573        assert_eq!(json["queued_commands"], 2);
574        assert!(serde_json::to_vec(&json).unwrap().len() < 64 * 1024);
575    }
576
577    #[test]
578    fn message_boundaries_and_prompt_reset_do_not_mix_answers() {
579        let context = TurnContext::default();
580        context.reset("First prompt");
581        context.observe(&message("one", "Old answer"));
582        context.observe(&message("two", "New "));
583        context.observe(&message("two", "answer?"));
584        let evidence = context.evidence(
585            HarnessKind::Claude,
586            TurnPhase::Replied,
587            &ActivityFacts::default(),
588            0,
589        );
590        assert_eq!(evidence.assistant_text_tail, "New answer?");
591        let generation = context.generation();
592        context.set_counts(1, 2);
593        assert_ne!(context.generation(), generation);
594        let generation = context.generation();
595        context.set_counts(1, 2);
596        assert_eq!(context.generation(), generation);
597        assert_eq!(context.counts(), (1, 2));
598        context.invalidate();
599        assert_ne!(context.generation(), generation);
600        assert_eq!(context.counts(), (1, 2));
601        let generation = context.generation();
602        context.reset("Second prompt");
603        assert_eq!(context.counts(), (1, 2));
604        assert_ne!(context.generation(), generation);
605        let evidence = context.evidence(
606            HarnessKind::Claude,
607            TurnPhase::Running,
608            &ActivityFacts::default(),
609            0,
610        );
611        assert_eq!(evidence.user_prompt_tail, "Second prompt");
612        assert!(evidence.assistant_text_tail.is_empty());
613        assert!(!evidence.transcript_summary.contains("<user>"));
614    }
615}
mj_transcript/turn_context.rs

mj_transcript/
turn_context.rs
