Skip to content

Commit ee54284

Browse files
owenlin0owenlin0
committed
feat(core): emit canonical turn item lifecycle events
1 parent 945ab43 commit ee54284

15 files changed

Lines changed: 705 additions & 341 deletions

File tree

‎codex-rs/app-server/src/bespoke_event_handling.rs‎

Lines changed: 284 additions & 34 deletions
Large diffs are not rendered by default.

‎codex-rs/app-server/src/thread_state.rs‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,13 +67,56 @@ pub(crate) enum ThreadListenerCommand {
6767
}
6868

6969
/// Per-conversation accumulation of the latest states e.g. error message while a turn runs.
70+
#[derive(Clone, Copy, PartialEq, Eq)]
71+
pub(crate) enum LegacyItemLifecycle {
72+
Started,
73+
Completed,
74+
}
75+
76+
#[derive(Clone)]
77+
struct PendingLegacyItemSuppression {
78+
lifecycle: LegacyItemLifecycle,
79+
item_id: String,
80+
}
81+
7082
#[derive(Default, Clone)]
7183
pub(crate) struct TurnSummary {
7284
pub(crate) started_at: Option<i64>,
7385
pub(crate) command_execution_started: HashSet<String>,
86+
pending_legacy_item_suppression: Option<PendingLegacyItemSuppression>,
7487
pub(crate) last_error: Option<TurnError>,
7588
}
7689

90+
impl TurnSummary {
91+
pub(crate) fn suppress_next_legacy_item(
92+
&mut self,
93+
lifecycle: LegacyItemLifecycle,
94+
item_id: &str,
95+
) {
96+
self.pending_legacy_item_suppression = Some(PendingLegacyItemSuppression {
97+
lifecycle,
98+
item_id: item_id.to_string(),
99+
});
100+
}
101+
102+
pub(crate) fn take_legacy_item_suppression(
103+
&mut self,
104+
lifecycle: LegacyItemLifecycle,
105+
item_id: &str,
106+
) -> bool {
107+
if self
108+
.pending_legacy_item_suppression
109+
.as_ref()
110+
.is_some_and(|pending| pending.lifecycle == lifecycle && pending.item_id == item_id)
111+
{
112+
self.pending_legacy_item_suppression = None;
113+
true
114+
} else {
115+
false
116+
}
117+
}
118+
}
119+
77120
#[derive(Default)]
78121
pub(crate) struct ThreadState {
79122
pub(crate) pending_interrupts: PendingInterruptQueue,

‎codex-rs/core/src/tasks/user_shell.rs‎

Lines changed: 88 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ use crate::turn_timing::now_unix_timestamp_ms;
2929
use crate::user_shell_command::user_shell_command_record_item;
3030
use codex_protocol::exec_output::ExecToolCallOutput;
3131
use codex_protocol::exec_output::StreamOutput;
32+
use codex_protocol::items::CommandExecutionItem;
33+
use codex_protocol::items::TurnItem;
3234
use codex_protocol::protocol::ErrorEvent;
3335
use codex_protocol::protocol::EventMsg;
3436
use codex_protocol::protocol::ExecCommandBeginEvent;
@@ -181,19 +183,21 @@ pub(crate) async fn execute_user_shell_command(
181183

182184
let parsed_cmd = parse_command(&display_command);
183185
session
184-
.send_event(
186+
.emit_turn_item_started(
185187
turn_context.as_ref(),
186-
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
187-
call_id: call_id.clone(),
188-
process_id: None,
189-
turn_id: turn_context.sub_id.clone(),
190-
started_at_ms: now_unix_timestamp_ms(),
191-
command: display_command.clone(),
192-
cwd: cwd.clone().into(),
193-
parsed_cmd: parsed_cmd.clone(),
194-
source: ExecCommandSource::UserShell,
195-
interaction_input: None,
196-
}),
188+
&TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_begin_event(
189+
ExecCommandBeginEvent {
190+
call_id: call_id.clone(),
191+
process_id: None,
192+
turn_id: turn_context.sub_id.clone(),
193+
started_at_ms: now_unix_timestamp_ms(),
194+
command: display_command.clone(),
195+
cwd: cwd.clone().into(),
196+
parsed_cmd: parsed_cmd.clone(),
197+
source: ExecCommandSource::UserShell,
198+
interaction_input: None,
199+
},
200+
)),
197201
)
198202
.await;
199203

@@ -259,58 +263,62 @@ pub(crate) async fn execute_user_shell_command(
259263
)
260264
.await;
261265
session
262-
.send_event(
266+
.emit_turn_item_completed(
263267
turn_context.as_ref(),
264-
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
265-
call_id,
266-
process_id: None,
267-
turn_id: turn_context.sub_id.clone(),
268-
completed_at_ms: now_unix_timestamp_ms(),
269-
command: display_command.clone(),
270-
cwd: cwd.clone().into(),
271-
parsed_cmd: parsed_cmd.clone(),
272-
source: ExecCommandSource::UserShell,
273-
interaction_input: None,
274-
stdout: String::new(),
275-
stderr: aborted_message.clone(),
276-
aggregated_output: aborted_message.clone(),
277-
exit_code: -1,
278-
duration: Duration::ZERO,
279-
formatted_output: aborted_message,
280-
status: ExecCommandStatus::Failed,
281-
}),
268+
TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_end_event(
269+
ExecCommandEndEvent {
270+
call_id,
271+
process_id: None,
272+
turn_id: turn_context.sub_id.clone(),
273+
completed_at_ms: now_unix_timestamp_ms(),
274+
command: display_command.clone(),
275+
cwd: cwd.clone().into(),
276+
parsed_cmd: parsed_cmd.clone(),
277+
source: ExecCommandSource::UserShell,
278+
interaction_input: None,
279+
stdout: String::new(),
280+
stderr: aborted_message.clone(),
281+
aggregated_output: aborted_message.clone(),
282+
exit_code: -1,
283+
duration: Duration::ZERO,
284+
formatted_output: aborted_message,
285+
status: ExecCommandStatus::Failed,
286+
},
287+
)),
282288
)
283289
.await;
284290
}
285291
Ok(Ok(output)) => {
286292
session
287-
.send_event(
293+
.emit_turn_item_completed(
288294
turn_context.as_ref(),
289-
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
290-
call_id: call_id.clone(),
291-
process_id: None,
292-
turn_id: turn_context.sub_id.clone(),
293-
completed_at_ms: now_unix_timestamp_ms(),
294-
command: display_command.clone(),
295-
cwd: cwd.clone().into(),
296-
parsed_cmd: parsed_cmd.clone(),
297-
source: ExecCommandSource::UserShell,
298-
interaction_input: None,
299-
stdout: output.stdout.text.clone(),
300-
stderr: output.stderr.text.clone(),
301-
aggregated_output: output.aggregated_output.text.clone(),
302-
exit_code: output.exit_code,
303-
duration: output.duration,
304-
formatted_output: format_exec_output_str(
305-
&output,
306-
turn_context.model_info.truncation_policy.into(),
307-
),
308-
status: if output.exit_code == 0 {
309-
ExecCommandStatus::Completed
310-
} else {
311-
ExecCommandStatus::Failed
295+
TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_end_event(
296+
ExecCommandEndEvent {
297+
call_id: call_id.clone(),
298+
process_id: None,
299+
turn_id: turn_context.sub_id.clone(),
300+
completed_at_ms: now_unix_timestamp_ms(),
301+
command: display_command.clone(),
302+
cwd: cwd.clone().into(),
303+
parsed_cmd: parsed_cmd.clone(),
304+
source: ExecCommandSource::UserShell,
305+
interaction_input: None,
306+
stdout: output.stdout.text.clone(),
307+
stderr: output.stderr.text.clone(),
308+
aggregated_output: output.aggregated_output.text.clone(),
309+
exit_code: output.exit_code,
310+
duration: output.duration,
311+
formatted_output: format_exec_output_str(
312+
&output,
313+
turn_context.model_info.truncation_policy.into(),
314+
),
315+
status: if output.exit_code == 0 {
316+
ExecCommandStatus::Completed
317+
} else {
318+
ExecCommandStatus::Failed
319+
},
312320
},
313-
}),
321+
)),
314322
)
315323
.await;
316324

@@ -329,29 +337,31 @@ pub(crate) async fn execute_user_shell_command(
329337
timed_out: false,
330338
};
331339
session
332-
.send_event(
340+
.emit_turn_item_completed(
333341
turn_context.as_ref(),
334-
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
335-
call_id,
336-
process_id: None,
337-
turn_id: turn_context.sub_id.clone(),
338-
completed_at_ms: now_unix_timestamp_ms(),
339-
command: display_command,
340-
cwd: cwd.into(),
341-
parsed_cmd,
342-
source: ExecCommandSource::UserShell,
343-
interaction_input: None,
344-
stdout: exec_output.stdout.text.clone(),
345-
stderr: exec_output.stderr.text.clone(),
346-
aggregated_output: exec_output.aggregated_output.text.clone(),
347-
exit_code: exec_output.exit_code,
348-
duration: exec_output.duration,
349-
formatted_output: format_exec_output_str(
350-
&exec_output,
351-
turn_context.model_info.truncation_policy.into(),
352-
),
353-
status: ExecCommandStatus::Failed,
354-
}),
342+
TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_end_event(
343+
ExecCommandEndEvent {
344+
call_id,
345+
process_id: None,
346+
turn_id: turn_context.sub_id.clone(),
347+
completed_at_ms: now_unix_timestamp_ms(),
348+
command: display_command,
349+
cwd: cwd.into(),
350+
parsed_cmd,
351+
source: ExecCommandSource::UserShell,
352+
interaction_input: None,
353+
stdout: exec_output.stdout.text.clone(),
354+
stderr: exec_output.stderr.text.clone(),
355+
aggregated_output: exec_output.aggregated_output.text.clone(),
356+
exit_code: exec_output.exit_code,
357+
duration: exec_output.duration,
358+
formatted_output: format_exec_output_str(
359+
&exec_output,
360+
turn_context.model_info.truncation_policy.into(),
361+
),
362+
status: ExecCommandStatus::Failed,
363+
},
364+
)),
355365
)
356366
.await;
357367
persist_user_shell_output(

‎codex-rs/core/src/tools/events.rs‎

Lines changed: 46 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ use codex_apply_patch::AppliedPatchDelta;
88
use codex_protocol::error::CodexErr;
99
use codex_protocol::error::SandboxErr;
1010
use codex_protocol::exec_output::ExecToolCallOutput;
11+
use codex_protocol::items::CommandExecutionItem;
1112
use codex_protocol::items::FileChangeItem;
1213
use codex_protocol::items::TurnItem;
1314
use codex_protocol::parse_command::ParsedCommand;
@@ -102,20 +103,27 @@ pub(crate) async fn emit_exec_command_begin(
102103
interaction_input: Option<String>,
103104
process_id: Option<&str>,
104105
) {
106+
let event = ExecCommandBeginEvent {
107+
call_id: ctx.call_id.to_string(),
108+
process_id: process_id.map(str::to_owned),
109+
turn_id: ctx.turn.sub_id.clone(),
110+
started_at_ms: now_unix_timestamp_ms(),
111+
command: command.to_vec(),
112+
cwd: cwd.clone(),
113+
parsed_cmd: parsed_cmd.to_vec(),
114+
source,
115+
interaction_input,
116+
};
117+
if matches!(source, ExecCommandSource::UnifiedExecInteraction) {
118+
ctx.session
119+
.send_event(ctx.turn, EventMsg::ExecCommandBegin(event))
120+
.await;
121+
return;
122+
}
105123
ctx.session
106-
.send_event(
124+
.emit_turn_item_started(
107125
ctx.turn,
108-
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
109-
call_id: ctx.call_id.to_string(),
110-
process_id: process_id.map(str::to_owned),
111-
turn_id: ctx.turn.sub_id.clone(),
112-
started_at_ms: now_unix_timestamp_ms(),
113-
command: command.to_vec(),
114-
cwd: cwd.clone(),
115-
parsed_cmd: parsed_cmd.to_vec(),
116-
source,
117-
interaction_input,
118-
}),
126+
&TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_begin_event(event)),
119127
)
120128
.await;
121129
}
@@ -542,27 +550,34 @@ async fn emit_exec_end(
542550
exec_input: ExecCommandInput<'_>,
543551
exec_result: ExecCommandResult,
544552
) {
553+
let event = ExecCommandEndEvent {
554+
call_id: ctx.call_id.to_string(),
555+
process_id: exec_input.process_id.map(str::to_owned),
556+
turn_id: ctx.turn.sub_id.clone(),
557+
completed_at_ms: now_unix_timestamp_ms(),
558+
command: exec_input.command.to_vec(),
559+
cwd: exec_input.cwd.clone(),
560+
parsed_cmd: exec_input.parsed_cmd.to_vec(),
561+
source: exec_input.source,
562+
interaction_input: exec_input.interaction_input.map(str::to_owned),
563+
stdout: exec_result.stdout,
564+
stderr: exec_result.stderr,
565+
aggregated_output: exec_result.aggregated_output,
566+
exit_code: exec_result.exit_code,
567+
duration: exec_result.duration,
568+
formatted_output: exec_result.formatted_output,
569+
status: exec_result.status,
570+
};
571+
if matches!(exec_input.source, ExecCommandSource::UnifiedExecInteraction) {
572+
ctx.session
573+
.send_event(ctx.turn, EventMsg::ExecCommandEnd(event))
574+
.await;
575+
return;
576+
}
545577
ctx.session
546-
.send_event(
578+
.emit_turn_item_completed(
547579
ctx.turn,
548-
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
549-
call_id: ctx.call_id.to_string(),
550-
process_id: exec_input.process_id.map(str::to_owned),
551-
turn_id: ctx.turn.sub_id.clone(),
552-
completed_at_ms: now_unix_timestamp_ms(),
553-
command: exec_input.command.to_vec(),
554-
cwd: exec_input.cwd.clone(),
555-
parsed_cmd: exec_input.parsed_cmd.to_vec(),
556-
source: exec_input.source,
557-
interaction_input: exec_input.interaction_input.map(str::to_owned),
558-
stdout: exec_result.stdout,
559-
stderr: exec_result.stderr,
560-
aggregated_output: exec_result.aggregated_output,
561-
exit_code: exec_result.exit_code,
562-
duration: exec_result.duration,
563-
formatted_output: exec_result.formatted_output,
564-
status: exec_result.status,
565-
}),
580+
TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_end_event(event)),
566581
)
567582
.await;
568583
}

0 commit comments

Comments
 (0)