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

mj_transcript/
turn_context.rs
