Skip to content

Commit f691f27

Browse files
owenlin0owenlin0
committed
refactor: simplify canonical lifecycle emission
1 parent a6f98df commit f691f27

18 files changed

Lines changed: 608 additions & 1041 deletions

File tree

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

Lines changed: 143 additions & 317 deletions
Large diffs are not rendered by default.

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

Lines changed: 0 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -67,56 +67,13 @@ 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-
8270
#[derive(Default, Clone)]
8371
pub(crate) struct TurnSummary {
8472
pub(crate) started_at: Option<i64>,
8573
pub(crate) command_execution_started: HashSet<String>,
86-
pending_legacy_item_suppression: Option<PendingLegacyItemSuppression>,
8774
pub(crate) last_error: Option<TurnError>,
8875
}
8976

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-
12077
#[derive(Default)]
12178
pub(crate) struct ThreadState {
12279
pub(crate) pending_interrupts: PendingInterruptQueue,

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

Lines changed: 74 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -25,18 +25,15 @@ use crate::tools::runtimes::RuntimePathPrepends;
2525
use crate::tools::runtimes::apply_package_path_prepend;
2626
use crate::tools::runtimes::maybe_wrap_shell_lc_with_snapshot;
2727
use crate::tools::runtimes::strip_managed_proxy_env;
28-
use crate::turn_timing::now_unix_timestamp_ms;
2928
use crate::user_shell_command::user_shell_command_record_item;
3029
use codex_protocol::exec_output::ExecToolCallOutput;
3130
use codex_protocol::exec_output::StreamOutput;
3231
use codex_protocol::items::CommandExecutionItem;
32+
use codex_protocol::items::CommandExecutionStatus;
3333
use codex_protocol::items::TurnItem;
3434
use codex_protocol::protocol::ErrorEvent;
3535
use codex_protocol::protocol::EventMsg;
36-
use codex_protocol::protocol::ExecCommandBeginEvent;
37-
use codex_protocol::protocol::ExecCommandEndEvent;
3836
use codex_protocol::protocol::ExecCommandSource;
39-
use codex_protocol::protocol::ExecCommandStatus;
4037
use codex_protocol::protocol::TurnStartedEvent;
4138
use codex_sandboxing::SandboxType;
4239
use codex_shell_command::parse_command::parse_command;
@@ -185,19 +182,22 @@ pub(crate) async fn execute_user_shell_command(
185182
session
186183
.emit_turn_item_started(
187184
turn_context.as_ref(),
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-
)),
185+
&TurnItem::CommandExecution(CommandExecutionItem {
186+
id: call_id.clone(),
187+
process_id: None,
188+
command: display_command.clone(),
189+
cwd: cwd.clone().into(),
190+
parsed_cmd: parsed_cmd.clone(),
191+
source: ExecCommandSource::UserShell,
192+
interaction_input: None,
193+
status: CommandExecutionStatus::InProgress,
194+
stdout: None,
195+
stderr: None,
196+
aggregated_output: None,
197+
exit_code: None,
198+
duration: None,
199+
formatted_output: None,
200+
}),
201201
)
202202
.await;
203203

@@ -265,60 +265,52 @@ pub(crate) async fn execute_user_shell_command(
265265
session
266266
.emit_turn_item_completed(
267267
turn_context.as_ref(),
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-
)),
268+
TurnItem::CommandExecution(CommandExecutionItem {
269+
id: call_id,
270+
process_id: None,
271+
command: display_command.clone(),
272+
cwd: cwd.clone().into(),
273+
parsed_cmd: parsed_cmd.clone(),
274+
source: ExecCommandSource::UserShell,
275+
interaction_input: None,
276+
status: CommandExecutionStatus::Failed,
277+
stdout: Some(String::new()),
278+
stderr: Some(aborted_message.clone()),
279+
aggregated_output: Some(aborted_message.clone()),
280+
exit_code: Some(-1),
281+
duration: Some(Duration::ZERO),
282+
formatted_output: Some(aborted_message),
283+
}),
288284
)
289285
.await;
290286
}
291287
Ok(Ok(output)) => {
292288
session
293289
.emit_turn_item_completed(
294290
turn_context.as_ref(),
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-
},
291+
TurnItem::CommandExecution(CommandExecutionItem {
292+
id: call_id.clone(),
293+
process_id: None,
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+
status: if output.exit_code == 0 {
300+
CommandExecutionStatus::Completed
301+
} else {
302+
CommandExecutionStatus::Failed
320303
},
321-
)),
304+
stdout: Some(output.stdout.text.clone()),
305+
stderr: Some(output.stderr.text.clone()),
306+
aggregated_output: Some(output.aggregated_output.text.clone()),
307+
exit_code: Some(output.exit_code),
308+
duration: Some(output.duration),
309+
formatted_output: Some(format_exec_output_str(
310+
&output,
311+
turn_context.model_info.truncation_policy.into(),
312+
)),
313+
}),
322314
)
323315
.await;
324316

@@ -339,29 +331,25 @@ pub(crate) async fn execute_user_shell_command(
339331
session
340332
.emit_turn_item_completed(
341333
turn_context.as_ref(),
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-
)),
334+
TurnItem::CommandExecution(CommandExecutionItem {
335+
id: call_id,
336+
process_id: None,
337+
command: display_command,
338+
cwd: cwd.into(),
339+
parsed_cmd,
340+
source: ExecCommandSource::UserShell,
341+
interaction_input: None,
342+
status: CommandExecutionStatus::Failed,
343+
stdout: Some(exec_output.stdout.text.clone()),
344+
stderr: Some(exec_output.stderr.text.clone()),
345+
aggregated_output: Some(exec_output.aggregated_output.text.clone()),
346+
exit_code: Some(exec_output.exit_code),
347+
duration: Some(exec_output.duration),
348+
formatted_output: Some(format_exec_output_str(
349+
&exec_output,
350+
turn_context.model_info.truncation_policy.into(),
351+
)),
352+
}),
365353
)
366354
.await;
367355
persist_user_shell_output(

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

Lines changed: 68 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ use codex_protocol::error::CodexErr;
99
use codex_protocol::error::SandboxErr;
1010
use codex_protocol::exec_output::ExecToolCallOutput;
1111
use codex_protocol::items::CommandExecutionItem;
12+
use codex_protocol::items::CommandExecutionStatus;
1213
use codex_protocol::items::FileChangeItem;
1314
use codex_protocol::items::TurnItem;
1415
use codex_protocol::parse_command::ParsedCommand;
@@ -103,27 +104,44 @@ pub(crate) async fn emit_exec_command_begin(
103104
interaction_input: Option<String>,
104105
process_id: Option<&str>,
105106
) {
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-
};
117107
if matches!(source, ExecCommandSource::UnifiedExecInteraction) {
118108
ctx.session
119-
.send_event(ctx.turn, EventMsg::ExecCommandBegin(event))
109+
.send_event(
110+
ctx.turn,
111+
EventMsg::ExecCommandBegin(ExecCommandBeginEvent {
112+
call_id: ctx.call_id.to_string(),
113+
process_id: process_id.map(str::to_owned),
114+
turn_id: ctx.turn.sub_id.clone(),
115+
started_at_ms: now_unix_timestamp_ms(),
116+
command: command.to_vec(),
117+
cwd: cwd.clone(),
118+
parsed_cmd: parsed_cmd.to_vec(),
119+
source,
120+
interaction_input,
121+
}),
122+
)
120123
.await;
121124
return;
122125
}
123126
ctx.session
124127
.emit_turn_item_started(
125128
ctx.turn,
126-
&TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_begin_event(event)),
129+
&TurnItem::CommandExecution(CommandExecutionItem {
130+
id: ctx.call_id.to_string(),
131+
process_id: process_id.map(str::to_owned),
132+
command: command.to_vec(),
133+
cwd: cwd.clone(),
134+
parsed_cmd: parsed_cmd.to_vec(),
135+
source,
136+
interaction_input,
137+
status: CommandExecutionStatus::InProgress,
138+
stdout: None,
139+
stderr: None,
140+
aggregated_output: None,
141+
exit_code: None,
142+
duration: None,
143+
formatted_output: None,
144+
}),
127145
)
128146
.await;
129147
}
@@ -550,34 +568,51 @@ async fn emit_exec_end(
550568
exec_input: ExecCommandInput<'_>,
551569
exec_result: ExecCommandResult,
552570
) {
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-
};
571571
if matches!(exec_input.source, ExecCommandSource::UnifiedExecInteraction) {
572572
ctx.session
573-
.send_event(ctx.turn, EventMsg::ExecCommandEnd(event))
573+
.send_event(
574+
ctx.turn,
575+
EventMsg::ExecCommandEnd(ExecCommandEndEvent {
576+
call_id: ctx.call_id.to_string(),
577+
process_id: exec_input.process_id.map(str::to_owned),
578+
turn_id: ctx.turn.sub_id.clone(),
579+
completed_at_ms: now_unix_timestamp_ms(),
580+
command: exec_input.command.to_vec(),
581+
cwd: exec_input.cwd.clone(),
582+
parsed_cmd: exec_input.parsed_cmd.to_vec(),
583+
source: exec_input.source,
584+
interaction_input: exec_input.interaction_input.map(str::to_owned),
585+
stdout: exec_result.stdout,
586+
stderr: exec_result.stderr,
587+
aggregated_output: exec_result.aggregated_output,
588+
exit_code: exec_result.exit_code,
589+
duration: exec_result.duration,
590+
formatted_output: exec_result.formatted_output,
591+
status: exec_result.status,
592+
}),
593+
)
574594
.await;
575595
return;
576596
}
577597
ctx.session
578598
.emit_turn_item_completed(
579599
ctx.turn,
580-
TurnItem::CommandExecution(CommandExecutionItem::from_exec_command_end_event(event)),
600+
TurnItem::CommandExecution(CommandExecutionItem {
601+
id: ctx.call_id.to_string(),
602+
process_id: exec_input.process_id.map(str::to_owned),
603+
command: exec_input.command.to_vec(),
604+
cwd: exec_input.cwd.clone(),
605+
parsed_cmd: exec_input.parsed_cmd.to_vec(),
606+
source: exec_input.source,
607+
interaction_input: exec_input.interaction_input.map(str::to_owned),
608+
status: exec_result.status.into(),
609+
stdout: Some(exec_result.stdout),
610+
stderr: Some(exec_result.stderr),
611+
aggregated_output: Some(exec_result.aggregated_output),
612+
exit_code: Some(exec_result.exit_code),
613+
duration: Some(exec_result.duration),
614+
formatted_output: Some(exec_result.formatted_output),
615+
}),
581616
)
582617
.await;
583618
}

0 commit comments

Comments
 (0)