@@ -9,13 +9,15 @@ use std::path::PathBuf;
99use std:: sync:: Arc ;
1010
1111use botloft_core:: ids:: BotId ;
12- use botloft_core:: protocol:: { PermissionMode , TokenUsage } ;
12+ use botloft_core:: protocol:: {
13+ ChatBody , NoticeCode , NoticeItem , NoticeLevel , PermissionMode , TokenUsage ,
14+ } ;
1315use serde_json:: { Value , json} ;
1416
1517use super :: OutputDecoder ;
1618use super :: codex_wire:: { Answer , Shared } ;
1719use crate :: approvals:: { self , Verdict } ;
18- use crate :: chat:: sink;
20+ use crate :: chat:: { items , sink} ;
1921use crate :: state:: Daemon ;
2022
2123/// Reads what the server prints and tells the daemon.
@@ -33,6 +35,10 @@ pub(super) struct CodexDecoder {
3335 /// command, or the files a change touches.
3436 commands : HashMap < String , String > ,
3537 changes : HashMap < String , Vec < ( String , bool ) > > ,
38+ /// When the usage limit resets, in Unix ms, from `account/rateLimits/updated`.
39+ limit_resets_at : Option < i64 > ,
40+ /// The running turn is a compaction: no line of its own for it.
41+ compacting : bool ,
3642}
3743
3844impl CodexDecoder {
@@ -46,6 +52,8 @@ impl CodexDecoder {
4652 base : ( 0 , 0 , 0 , 0 ) ,
4753 commands : HashMap :: new ( ) ,
4854 changes : HashMap :: new ( ) ,
55+ limit_resets_at : None ,
56+ compacting : false ,
4957 }
5058 }
5159}
@@ -109,6 +117,7 @@ impl CodexDecoder {
109117 Answer :: Failed ( message) => {
110118 tracing:: warn!( bot = %bot, "codex could not start a thread: {message:?}" ) ;
111119 }
120+ Answer :: SignedOut => self . signed_out ( daemon, bot, generation) ,
112121 Answer :: Lost | Answer :: None => { }
113122 }
114123 }
@@ -133,7 +142,7 @@ impl CodexDecoder {
133142 }
134143 "mcpServer/elicitation/request" => {
135144 let server = params[ "serverName" ] . as_str ( ) . unwrap_or ( "mcp" ) ;
136- let tool = params[ "_meta" ] [ "tool_name" ] . as_str ( ) . unwrap_or ( "tool" ) ;
145+ let tool = tool_in ( params[ "message" ] . as_str ( ) . unwrap_or_default ( ) ) ;
137146 (
138147 format ! ( "mcp__{server}__{tool}" ) ,
139148 params[ "_meta" ] [ "tool_params" ] . clone ( ) ,
@@ -206,6 +215,13 @@ impl CodexDecoder {
206215 "item/started" => self . item_started ( daemon, bot, generation, params) ,
207216 "item/completed" => self . item_completed ( daemon, bot, & params[ "item" ] ) ,
208217 "thread/tokenUsage/updated" => self . token_usage ( daemon, bot, & params[ "tokenUsage" ] ) ,
218+ "account/rateLimits/updated" => {
219+ // Seconds in Codex, milliseconds here.
220+ self . limit_resets_at = params[ "rateLimits" ] [ "primary" ] [ "resetsAt" ]
221+ . as_i64 ( )
222+ . map ( |seconds| seconds * 1000 )
223+ . or ( self . limit_resets_at ) ;
224+ }
209225 "turn/completed" => self . turn_completed ( daemon, bot, generation, & params[ "turn" ] ) ,
210226 _ => { }
211227 }
@@ -260,6 +276,10 @@ impl CodexDecoder {
260276 & json ! ( { "file_path" : change[ "path" ] } ) ,
261277 ) ;
262278 }
279+ Some ( "contextCompaction" ) => {
280+ self . compacting = true ;
281+ crate :: context:: status ( daemon, bot, & json ! ( { "status" : "compacting" } ) ) ;
282+ }
263283 Some ( "mcpToolCall" ) => {
264284 let name = format ! (
265285 "mcp__{}__{}" ,
@@ -276,6 +296,16 @@ impl CodexDecoder {
276296 let id = item[ "id" ] . as_str ( ) . unwrap_or_default ( ) ;
277297 let ok = item[ "status" ] . as_str ( ) == Some ( "completed" ) ;
278298 match item[ "type" ] . as_str ( ) {
299+ Some ( "contextCompaction" ) => {
300+ // The owner asked, or it was full and Codex did it by itself.
301+ let asked = daemon. contexts . get ( bot) . is_some_and ( |c| c. compacting ) ;
302+ let trigger = if asked { "manual" } else { "auto" } ;
303+ crate :: context:: compacted (
304+ daemon,
305+ bot,
306+ & json ! ( { "compact_metadata" : { "trigger" : trigger } } ) ,
307+ ) ;
308+ }
279309 Some ( "agentMessage" ) => {
280310 sink:: reply ( daemon, bot, item[ "text" ] . as_str ( ) . unwrap_or_default ( ) ) ;
281311 }
@@ -326,6 +356,26 @@ impl CodexDecoder {
326356
327357 fn turn_completed ( & mut self , daemon : & Daemon , bot : & BotId , generation : u64 , turn : & Value ) {
328358 let failed = turn[ "status" ] . as_str ( ) != Some ( "completed" ) ;
359+ // A compaction ends like a turn; its notice says what happened.
360+ if std:: mem:: take ( & mut self . compacting ) && !failed {
361+ self . usage_base ( ) ;
362+ self . shared . turn_done ( ) ;
363+ sink:: finish_turn ( daemon, bot, generation, false ) ;
364+ return ;
365+ }
366+ match turn[ "error" ] [ "codexErrorInfo" ] . as_str ( ) {
367+ Some ( "unauthorized" ) => {
368+ self . signed_out ( daemon, bot, generation) ;
369+ self . shared . turn_done ( ) ;
370+ return ;
371+ }
372+ Some ( "usageLimitExceeded" ) => {
373+ self . limit_reached ( daemon, bot, generation) ;
374+ self . shared . turn_done ( ) ;
375+ return ;
376+ }
377+ _ => { }
378+ }
329379 let error = failed. then ( || {
330380 turn[ "error" ] [ "message" ]
331381 . as_str ( )
@@ -353,6 +403,58 @@ impl CodexDecoder {
353403 }
354404}
355405
406+ impl CodexDecoder {
407+ /// What the thread used so far is the base of the next turn.
408+ fn usage_base ( & mut self ) {
409+ self . base = self . usage ;
410+ }
411+
412+ fn notice ( & self , daemon : & Daemon , bot : & BotId , text : & str ) {
413+ items:: add (
414+ daemon,
415+ bot,
416+ ChatBody :: Notice ( NoticeItem {
417+ level : NoticeLevel :: Error ,
418+ code : Some ( NoticeCode :: TurnFailed ) ,
419+ text : text. to_owned ( ) ,
420+ } ) ,
421+ ) ;
422+ }
423+
424+ /// Nobody is signed in to Codex: only the owner can fix that, so the bot
425+ /// stops until they restart it (spec 7.3).
426+ fn signed_out ( & self , daemon : & Daemon , bot : & BotId , generation : u64 ) {
427+ self . notice (
428+ daemon,
429+ bot,
430+ "Codex is not signed in. Sign in to Codex (codex login), then restart this bot." ,
431+ ) ;
432+ daemon. supervisor . signed_out ( bot, generation) ;
433+ }
434+
435+ /// The account reached its usage limit: messages wait until it resets.
436+ fn limit_reached ( & self , daemon : & Daemon , bot : & BotId , generation : u64 ) {
437+ self . notice (
438+ daemon,
439+ bot,
440+ "Codex reached its usage limit. Messages wait until it resets." ,
441+ ) ;
442+ let until = self
443+ . limit_resets_at
444+ . unwrap_or_else ( || daemon. clock . now_ms ( ) + 5 * 60 * 1000 ) ;
445+ daemon. supervisor . rate_limited ( bot, generation, until) ;
446+ }
447+ }
448+
449+ /// The tool a request of an MCP server is about: the name in quotes in its
450+ /// message ("Allow the x MCP server to run tool \"y\"?").
451+ fn tool_in ( message : & str ) -> & str {
452+ message
453+ . split_once ( "tool \" " )
454+ . and_then ( |( _, rest) | rest. split_once ( '"' ) )
455+ . map_or ( "tool" , |( name, _) | name)
456+ }
457+
356458/// Whether the request goes ahead: the bot's mode, a rule of the bot, or the
357459/// owner (spec 10.1).
358460async fn decide (
@@ -383,6 +485,15 @@ async fn decide(
383485mod tests {
384486 use super :: * ;
385487
488+ #[ test]
489+ fn the_tool_is_the_name_in_quotes_of_the_message ( ) {
490+ assert_eq ! (
491+ tool_in( "Allow the linkedin MCP server to run tool \" read_profile\" ?" ) ,
492+ "read_profile"
493+ ) ;
494+ assert_eq ! ( tool_in( "something else" ) , "tool" ) ;
495+ }
496+
386497 #[ test]
387498 fn a_text_that_names_a_fenced_folder_is_caught_whatever_the_slashes_and_case ( ) {
388499 let fenced = vec ! [ PathBuf :: from( r"C:\Users\Ana\Botloft\Other" ) ] ;
0 commit comments