From 16ff14c266179e6a762dc8081e9dab73a96683e0 Mon Sep 17 00:00:00 2001 From: felixxia-oai Date: Mon, 7 Sep 2026 13:49:24 +0000 Subject: [PATCH] Retain inherited Guardian instructions in standalone forks (#43478) ## Why A standalone root forked from a worker needs its inherited user instructions for Guardian review, including after compaction. Parent input counters belong to a different thread, and omitted parent answers may have restricted an inherited authorization. ## What changed - Adopt inherited user instructions into thread-owned retained context for standalone roots, recovering surviving checkpoint messages when needed. - Keep inherited instructions before local evidence without sharing local acceptance counters, and preserve that distinction during reconciliation and rollback. - Mark parent verified answers as incomplete and treat recovered checkpoint instructions as incomplete excerpts. ## Testing Add coverage for standalone forks from copied history and truncated checkpoints, retention through compaction and resume, inherited/local ordering collisions, and rollback scope. GitOrigin-RevId: bf716e224a1d29d6c528ef0dd7cdb118a413a08c --- .../app-server-exports-stable.json.zst | Bin 148308 -> 148316 bytes .../src/agent/control/user_authorization.rs | 7 +- codex-rs/core/src/context_manager/history.rs | 79 ++----- .../core/src/context_manager/history_tests.rs | 5 +- .../history_user_authorization.rs | 105 +++++++++ .../src/session/rollout_reconstruction.rs | 5 +- codex-rs/core/src/session/session.rs | 5 +- codex-rs/core/src/session/tests.rs | 7 +- codex-rs/core/src/state/session.rs | 6 +- .../tests/suite/guardian_retained_context.rs | 222 ++++++++++++++++++ .../suite/guardian_subagent_authorization.rs | 48 ++++ .../src/retained_instructions_tests.rs | 5 +- codex-rs/history/src/lib.rs | 2 + .../src/reconciled_retained_context.rs | 12 +- .../src/reconciled_retained_context_tests.rs | 30 ++- codex-rs/history/src/retained_context.rs | 122 ++++++++-- .../history/src/retained_context_tests.rs | 97 +++++++- codex-rs/rollout/src/lib.rs | 1 + .../local/rollout_migration/rollback_plan.rs | 19 +- 19 files changed, 646 insertions(+), 131 deletions(-) create mode 100644 codex-rs/core/src/context_manager/history_user_authorization.rs diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 10d00f3a35194480bc6b308fe50216a3e759929d..08ebc8e755bfa404bf0f4b66f3db86754fef2ba8 100644 GIT binary patch delta 10088 zcmV-uCzsgNh6&t;34nwFv;rp>0sOZp7y@zu63nRc-BwLfXbv+V5P-lyfdU1WtQi6` ze>CwtvC~6<^<^NkquCH7dU(mA+z)OC1uR|W1?)^ea407IQzktclS9VB@P5WbTMqeQ zrD0cTf=;4EvzL=&skxSg*DkoNTZTy>?9t(Uq#}tXFbeDwJRl z&uy_+0+{@M!P>$OhECYygRn01gh1YK)r8^+;f*!+y*xN(-%HIud6a$cr3u>ijxkvK z-pLQQ?{$syb?IsQ-e!;Pd$VYF-@B$o?|TKrdEX0@@_q07`}e)Lk_Eq)Dr|`SfBr-# z^k@t;b#+gvKW_!xe!WWp<`BP^{R`-P+|jc~TO2-r^DjJq+uHuAjo~Z#&8NNKjX3uo z6LwyN*u)v;KuJgE`l)BJRZ&inLuPh~NJu`AxHLY~gE1=4HH5Hf39@n`3bZuKAq;6q=ETdv%n8EcINU!!P84j!hyQsY7HCoV#|`+(@NoZ7h~SR7NQ zaUoKi%7a`7J~5nvhz3o`x`K18f)u%ZNzCrKl&y5g>UP4=9_u7zq)?g_e~pVXk`@y+ zn42(Zng1|0+cS6%{@6lfLj465#Ng~;8w8fo$Ak4#B3WbjKm@>gOz^d_yP%=ZpKFRikvzh#3h}{5 zU4^7mn(_UZ3c03UffBAM?_fcp;q7z2$ z?KsLo48p6HcdX*@DYTl(NEkFKB?=J9WV2?+Ij-G*7I1O@ZW4|KC}{@?FKK)}@V4pH z?M_^h#exyp`Kn@E=xQ{L*@B{98v%kUNJn_WT^b%T#yo;4VkZOutJJqDoQFsy)i zAF)nwxNcWp^r6<+OmH}HYo?(eQ5_Y3co;rUn{~KySYV*1-Z{p4{Z&<3b}KvAL5gx# zjP5LoXO5{zNHQ?~!%hW1??;{vo^YjWN-D}Zd{TRg7nPm)e@UDc@O+6Qw$?wRAOnO9 z`}}a>1gtNgG_we}x^=h>k-O@JgVPWOjy6jc1W| zZ8wv3VS2ZRP|6=osqxjU{aD|S8D!3+aa&M#s~tRwi5AJxn=n;s&HXRFzE@C>H&3J2 zw*#~(g^O*P7bE0>bNnVx)MgY5_iMeV%Y6(Lm+*=Ek4OI`q*1FUulYqX4L#O-L`FzG zY!zNxf3i6uzrDNs0yEOOGIO&WHz=9F$c3R`F}!oHu8i*C)F6=&25JgZafV@(5(V9kb$F5Xid_Kz)_ zQyiwe^kdF8#o=|D`UFqei`yd@SDkS6!3GtMe}LIQGjh5&-X^nlaY^uH&=f5VCm+#8 z@9F>X$7U~2aSg@WURG|4fo}32w`m(TV2j96I8L4s7h5=)hNYXMAVmW_lhV;1cmH53 zAoA?+-R*O;p4GrW_d%sG7SgK64wRajw!sRw2iA(e1!XXGxUkb<%vYe| z^?oqilna!*>qX$>X9#S_^f!cTTCl#~H?&Jf?07fvNWIIHDoPz6P& zr_t*ow9h+6ET(ar@2sN}!Rn@uP+H3JCSpK~MBuIuRsr~@W@SR#>z9KRhZSUqC?d#$ zTm^XxPHDHlW}RH$B*NB~Hw!aKI`qI|Ua&A~ zLxu<)L6jex*jp{@IiJ_^6IdL-e--HdxP@s$tv7M|9o%mC*nLZ>PsV8bzs?G9w3`2N z?7I+5zQ8>28JLEBAo76(KpuC%O$~hj>V{??$fK%=h^i@9uDVHc?$$%6mJ#>c`J2D@ z=Ea-8VNalUX-VM#h5~U_lRx77AzTLHA30-I`5~NBQ2Y>9=mZ*vU9^O=f0|f5t9}#-WVE%1M)PI1n(B zad=xZmdH4)O9L5)*##(@Mk2;YwC*q(tSCleInPD_^nlcb2!EH1T>zuo0RY94FcOs{ z3`XK8qx%tCY3X(H>t^qgf0!u^I8jND5HjhZ3Y<@w&3=Kr?GKqUM>NxxpO)Y9JcY_i zomvyyhvE~x;_6uf`32PSy4a2)%1chKLrofu)ZZF|*eCSWlDPKVbh08aO$k?1lh;L$ zdKS|$q!jN=9-iF{+F(BnbIFTpEV=<$DX$v~E(3we<;~hkYv1w#?^X8 zl9fKM*bQ+D$EwYqHnWk6AA%-^ClC$fR2sw^hkzR(A(&@WmYZ&Jtyp0pXO1v4)0=S! zcYHRMBzS&ND-|Za4S$YTUA---xZsn*zX|)69YDfG>rO4`tL<4vW{f7uNkNpDu=rK+ zJ+17Dd%&JD6*vrBe?tYfP)RthSx_S_;v^=Rs2BMK=^ic2c~G3i+$<$rh<<=YQrQ3X zn_v@;?a$A_Z8WW11j)&kJ}O`U+$&j5E-@T7!syHY_|!LwS$#UyqAkB;!#Qk)6pfUn z6OUFmyfS%kb?A8raS(LwC9`}VzCzqcM4@>7cpkdpsRTb?f6lPJs_2I2_bhvRQLtmY zxBC#$1O)N)g&{l4hNk9DRV`zbZyVk4dFiHzDmsqmfE$l)7#@|{-=rJnb%~@KHqFbX zr5h%2IvJqphR4d8Za7evB-6t}iNa-aGjD=y_;kZQsHYoN_m3s{!CG2TwL%?*&#WZ& zE+V*u=_aeHe;W1j&9R?)x8Xv_t%P_RCXuZfPMNpilZBJ5-i8Sb%-*}V;S=4X59I30 z>;}WKI_obZHAeC;kWV^%v(u9`omKINMc-L@mh*8_^188s|AMX?i%75hk65Yt1Qz$ zY&&bobg2l;-o!hW_w+74EBE~(@HFC#W`mM)zSz@Q(-Yuh6;caG13scsJJj9_ zuM)v&e+(7{6m8cXEY;sPyRKJ+3Lj}sF-}sBKAf`?r*IGn>#>=S`R+!Fn2%5+39NBZ zrnfX1;|&R6!qcNf-v|r)^^gB7PEHPU6A~ykN`XC%K zLo2tH5{hk>SP}jEDXp=Ir>+@u)S^pPT%)3~f2pE(`65(>#XjEQra0WZL@(X*E3_(@ znP~SQ!q%-$!K3&o>govY8WlV4?EeuU`XirJvn3sA6I>eg#VWiZD6;&`E}#Vs$krGy ze>Mt6Ooo{5n)6&Hq9vUnR??44f-x?`xzU+h{rL5N`ei)mYa1T$j^z7yO*wI_bARN@b&K;7~p>t zKkrp#q}=i*KCW(RI;&J3QvTyEptlKwW!MY~D%Y2eCfxcLn-1R4701{mh(K#-e`du! z@0I-o5D;V=4U+=CtP2+fx~1|B@HrB#3{{Pb!g+!+L;0j3k<_XoC4Fs>!l&IHuuJ>L z%M2O^iAup;_1(e1Px6zQX+MvmZE>=j%zQQ3P0n^&`YX$OM2y>@PkU+Dv?Hl~H$$Z` zZZ|9OsAMT3!NO3--L4c1^3{D{AQ z1E@HYFGsZGltD0*X>D=4z^);r;sUUXzifdMJWw&DchdqtN*K7`yRAIvPPhORY@z=2 zT7V<72~`$Q3l`i*zX8d(F;O-2$UK_3FS}eTCPpz19~7h8K!uN)H?Z#+f7|yJCwk?o zq4F<$?6QNPK(mw=o7tIQCop29-7t{vd{i)Q-U22#9jgV6)+|qDYjz20F?$Oywg}*! zWFx)2EbECXt1eH5TIp{GQ$3>L?~Q3O)wyeqM?`#jv_e!Ucr%RR7jAz(mf<}k=;cAR zF7_>DB?WKv(i;$ZPzC_Oe~FVmuy7wvl_5}k;D8Sa^A8-Wl=c&l8}-^VzC=BtasP4N zr8yw$FVeF@mep;8XQXk&Y#!+FU$_!9fW`9Zl*o}`oqIM27ssa6aSg8nFh2O!Mmx6y zfdB)|IDj08mT>&|f0PS({$}&X@aj^(EpcwrP_79}zTfIC2<@?zf7p8;uWL}4CE7jx z<&8yj3H%mV(id{{3x=vZl9mu}#uHRNLyQkUGttR)d>kSx%{UR!UaSNg(L_K^1>h&_ zZ=*%`52F6pMfgrQYC;a1;%yBw!%ly_V^$$FM6*En)*{Yziy&g?M$6UdtFZMFH7X!c zDnaKlXm3p1JsPD-f22#$^k8xd8#*`JKQ`VUcfMtwK;-Acmb>M^1=;qKK@$?^0CN1hWatPR*}PxCw#~e~bHyIAzVW=g)KSXLYDC z)zJ0kWJo=F>+LVmnwmx zao8YpzM~C6lH2Vh#!H3nyF-ig_2TUZr&7bc$NJ60y_QGJ?m!V!N6Eow{z8Di>GflE-rvs zoq2$rf9_A()*40a+a5dGc0rtGtLUUP*JweJLJ4e=sUB$KC8%k}QV%8PZb$DgrGAKmL5JtK;(niJS4_Lk(|Y6kL|V($n1 z0EA;3hkHk_`l77b@>Sc3lz2k0~K2_GOs^8)P~Fb#2^h{cbFK*PnMF*AQdnV9$j ze+E0UT&`hI?^aGjSMOL{nBQWnKZ=2pUtr&>7~o*~k70>FagkLA6979@!o+z6zJc|Y zxJ-O=P}wPk)Z07cf2kha z%iIv9IsyrZqV`I>v7DJ%Ep#;f82k{L;U*6>dW+9CQSd~g3@8}!;c9p9AjGD##y1F` z6HKPP4Lbj!VEYhftqg&Zb%TYXCniadFE90*2(3kfm#5D)0@E~MZ?O`_aX{hV6$LUd z<~Cxeqvs5xZj-K<*{F?kW zrzyX{Lwc9IFajulbODhR=85n~i?@^#!bZCm7ixo^7-Q2oR=S(>>$>k)%z63Tr&+y@ zc6($M*u!+#DRoGD)yw?y=U!U0|MYwerjni?91~A~T%!jO1nrsm5x9*-Tp| zOcO*eIo^ESBjLeCdZ)yoRq&Q*JD#Lkz!{+J_=5w#$8z>o(_n1})F*8MEpaZsFO)s$Ojdg?EK^XI~ zOSmv?6gEsX{(p$T=FVU@xWI;VDkm*U48aJ=oKc}^g`B#Yk?j@uT#W?j_-RX1Mk))h zcmf*Rac^!%ughTxAl{x%XU6)<+et9h1%K7RGhUM{ZRPm^mIj#l>x`)||0$VZrxjcd z0H4_EEdtJN;@(>?KG~IY;}LO?ClIA!(1qW!I167&1)8fkSvdnbOOLD?m{-YNUx!Yn zF}w4b#Rro*;g_=$P5~8+YO8*pbql~OBXkbmuWYirF9|gdpdId4#}OmH>ve)j&VQmj zsBD(PKkPthtmWMeHuCuo-tO^ESH7Lk(^j$t@$0>_l<%Uw;==BX2Quoumd3SMh*@;c z)3%7{WCi=w!*e-Fey=<4=Og9*Fvs1jfK2Fcm)k@uWQ{Ivcdig!!Lvt-yh;5pn;(oi zusgJNvu*&QG7wftBw(LS(9{A!9e-$08_+g%6phh(-S@iGF!gjlLyVo7#93w}4COls zQy{E}hBb{6+Q#!18~$-6dmfQs^mQ1IAaB;5_EMEpeBmKO&4BJwCC8m5)L7t8XaaEx zs(*-fE-pS1zl9%uwW*>}aZ}>_F9vtNr}eZN*&Jd|lt1%TMb6XApPBJNaet)-&!1V3 z7;PzYokc7P@v_1E%+Rsn0hOog)xy>TKE0&a3lQyr`gzWtEYl^?rVJ-3u_HjpXNF$f z63`Cw(B_qw7(KjjQJEisPg@NzG`2rb6Nj06!PQ}gE+(d<$Fk=7_Ki__Xwk)?Y?U8+ zgd{;Vk<2cz?$HmKQe}DlAV=o#*neLPBI`wK8OnoVnS3Lu7K&xUy}Mq6(Y&3Dphcxw zEmG25?a1X;np+pzUV{`eY(1O-PR9mB57mD zv>#4L1Nx*^Ro@E211h91-&r~{jTe6+-`TfiddD;I4I&$Z3z#ZIrOoeK*C1RBPSw)r z8{z7xTPImp1Z0LCV;7<_IvQJ{``^0^XGIw}=7*E4IEV#V#GTFgYD{}a*$HmOt1^o) z)ZD1HCYVuWf@}KzUd2;k7#J*bD{r9nCqpdB+`1_BY0Py)rp}g&xh;5+x#oW+?Wq9+ z(pX*~J!ZL74G`;!sq7V_0d)q8eSUWFa)jcYY7x`>LqOgBu)$MJp##=vDy3^Q(1}BY z^gTl|f^{5Gf(U5Jmhz8QW)f|B11^lX16ZNdVyRQ8cLG7oX60_MBn%jpzJD@QT;;5ZT0<3 zyi`8^=m7I^x@yU?r}8PZf)Ak#)A3Rw+8@4N8plYg;JyTo=s3PuE0F(FP)iUDg1dDh zxpSXhPgyjUbM%=(t-%R~Eb&~G*)aAVRdiJ*dO>D2?n%H2GT^-s)5 z>!m#d&3iriS^K(#6_D#M5%%UI1D7Bt$74g=bn^uNb~ zDY6`oT3W{8#XHtoE?IwuOThw!O`)A^Wcl;nt?NdXU6V|OQghJWG~1s%d!C5UM{`2C zN`!v#J8f-pjdj^tC{MmAQo6Z@BFerdI4BPxn zbb-ZG> zLab)`Q$F34SUePqsLmvH&{3>_wdBKUn=R5TT^40zKm|cV*OE+fj8=%C!({f;*UHx! zzwl@J2ssN5O{h1^KuN_!QY?x_)yAh|aJbNqlB7ICCpXz!Y|kz&f5-tVRcy0NA^?^u zWo4$jq5!e74W@sr94zxJCXHqA8u!ECP4CiJ_AW#%a*MOi)f5gX^(Yo{S6p9!V-{`k zSC~Fx;aBfOjG?6%(r6+5tA@`kG1r&vn7*f@mP(D^P!^LO%KM)j23hXK`B# z^fom5`$+S8!YhCd93agj@UqHaTsglABf3P|l3OL?F6B0omZU{pH{f_^@>hq@Y5Fad z(n`k(v(#N+c039ToZ$Qlezlb}1)GH#0PTXjI{NkPLtDWbYnP=t0v~@7AtNz_uTW?iCE~%2CKqyw*resHVuYx z3~Ro2jC`Kj4h9Y04a$F*{SP!+tmW+W`W%?`^5M~`4n5U0^QpagdEnn`I0o{@Drj_9 zJ8LMp5o>MzlZIGZp#1}`wDvZx7Dv2i5u=c1DV+7xu9Qa{*tjT_+yq&@1xcInR5$S8 z7RT$T;_ihCjz{OXs&IszkqpeDh>I4>%<7Kh~;wbJS~<%W|%mm_ia?sMM|sF2xNKUNuFAsPGe;CSzhhe-XM2sc_|nQZ?q7jC`W994A#*$*k91+m|c)LT(irx*ciK*6nCn*9FD< zsFMqtM4lcoS#?=noo66I0~3m)a$Z7l5Wm@cM#{P+6sJ#nTTvo^c0{dY7O_BZVa9!; z_3T~-p&5X6FjXgusD$c&BM^<<>7pk+<;GPIh_Sop!`P9_;~QN8BT}0CB%g5HB!aT6 z42AfFvlZBG%(8VYNlf+hw{dnQ79m&tL^L=>CZyf+vv$Y~KubOu(gDb1SvNHY9Wy~a zj5w0AQ_`qH3Qg62K?|@65Ob)Lf~xCB#|Jh9rJ@l|W;~q~(#UadfOMo?kf|qZjF=VN!03FV zK>l?2Dzn2KYk*c%scITN_fC{s~4C^w;>=vsM4qNcVgZI{jFQ`BNCwPx)y0zBrGnibB6Q z9!zj{alQi{DIt9Xo3A0ZZ<0);%rOw|%P7M|^biG@7B1bw13U=1HNWva`dt#NIK=HC zrQz^mScbq23|H$rp9+q7WNx;`h?={AVnV@{rURRQNZW^CZw{~no*+mVeu4ChnH-1D z%_wIqBvlxe|6dk~wXf|PEOd0!uP25A`~KBh>&YSC%~cVdG5(+6FRs>}z6VjWbouez zW(Ac?FsEBsB=z{k#p6%5;h1}%N5T&)Nzrk*=aFfD8K|LwsS zRM9+thj5XZhTCo)p2y1;FS)Pq{pu6k2p=4+&DCMYm++_jQTi!DCVanCKm#Q>5Vw55 zQAfQcbbo+w+eJG%NTs5W7BrgU6DL ztmVAcOvuWbOnNTE>$^bc79wG%JarO(fN|?snq+%}!y`8GGDJBl{k_XynV!(`md^pv zRk}X{q>oyLs8OQdBB>Q`0>2U^V9gMtKg{+jFpJh*`?oi)GN%!k?{ z%|CE9Ibcwa4^dN^+H{6(_lzykT#a57>b6Qte4x__e!39};<964nKP)6r*GW7-D%_>hFtf{SW0arR-{`S%{L2Q?4S+lBn9#g$v6ruA+#WB;9aAmPtK`|~T@R67F)5mVoDWALF!v)4 z?i{uU+I?S`Ucpi#L>;wd{;B)d2|V(}8#hVOl)jDUS^J)|&+w~6ng4$^8w!|0*7r|i zF#9yKA8V%@D97nByT11U!iZlnBC|`@dNey@f6@bXb{wz)OY?{;?Hz(ea!37ib%yKy z1#ZhEzmyu2n6I_uN=ko!wc1}IszQlx{bO3BZgp(|&s@Xr@ebJV2K6Gz(~aDL8v8cw3AiO!hT>0D# z-p7`Zh=G$TG3R1T^2N<_D0j#4eq^Fj)C^P(;mywH_%cHyyi$sPPIv7g`ZR^+-~7}a zsfP2-tZQ)te{q+1(7gru|99TJx<;-YN}y&M3sX_S##E6a*)!O5$Wghkmppal2lQs2 z7*hw?y|1j>-3UviG{WBngt%yt&ZxePBh-u*$ErO}JB5wu>jT(4LN^5`c>oRb_z6!v z21y<mEDnHuXSBBUb?)ZCwHlI9v7lw%=2=BbgvrKF=A*Jxs~%p+s*^9v&-4^5(z? z6;jw1FSc8F0+4j_2Xs@k*c|582gG9TYSgO0Zy5ht$d?{N0wNTLzcXYKyGhCc^bqMv K{&`~W+5!qMC2nZ| delta 10018 zcmV+-C*9cGh6&V$34nwFv;rp>0lc>;7y@zu63M9Z-BwLfXbv+V5P-lyK>|gWtQi6` zf5I^$11w$21guRzI3_mz3??ld#UX!T_3*_-6ps2L z(zdIN5=ySsu9uS>sa?y4*AlnwTZY*i?9|~6pdv&Q-~#rK9zz{e4+8~*3JhqkW@_dB zfo+k#z`v5km|`}j+QDS&vcfpbg<>{9e`=T*k%J;)+40oXw`jhtah0j9a$BMzH)z74 z=eE+T0+{?}iP&}r!zS!eK&UQdgn-^~;e^Hr0gc7+y&Rm6?|sc*d3b!Uq)Ev4jxiwl zUggKh_eLG>rSzD5FJ`ahd$(vX->ar&=6eBRneRQ5)_gDei}Ss_l6Ss$C~VODe{e)8 z6krTES#(TdKWV;Oe}zebb3xzh{&#cU+>x+HRUB@=`IkC>)!P24jq(-x&8Mi~jRg0L z3G^;v*u)*Pj`T<8lBs9ORZ&ZILUNXX7)VZ$xOAWCgRutBjfAjigf)pG3Y<0L5c+sP z!r{gd)2)gSP~&{!sF9u>nN_N>e{zHq>DSc&0fgF}MuY$@Sx|Bm{)*gd8JZN$t)Qy( zgK>9%FXW9-kbWpsa6ezepH|zSj#*%>w3!``a6Hamze{2Txot$HyF-Xq=V;L7OuI z&?}oL3PGb%x1)>XVc03Ufe`|JJ7fEPAq7z2a z>^RCn48mlVcaY)mDRi0&HyAW3NX;KYq@k9MbE(08Dd6JujtTCYOVS6?ENOp!F{SCm zeM+3Nyp9ps`5ay)n=g|o85se$Rl|ET{i9z9ygj3Hk}o{c}rL^C!+cv5v=7m z48k;A373=K@VY4Ye=EGkZ}^a*rjE7A>quskqNGN8>V(fso;8Y=&CRfpiTt`?ML4Hop7ZyNGe)2d{R4#m&$JWe@Tem?fi)&xJcik=m11& ze73E#+UFc3X$xo9gY`&guRe$}U#b7%gbku)E>meB7Nu2M=9g@;s>M5Oe-93TEWqQG#XQTTtF4((7bei( z5NcM?lv=ZzHHi9##vpTc8n*>vaY7*zz>Iz8!FLrEqIa z^THtyy!tnJqBf&gXusBry4+)^xP(vKe?0nELK?Mtam_E1Y3Q-uBQk>2!&c$7CEF4C z?cL=se^;T!-O0C`+UD|*$({G8YF#8qY=55y?dAvjGEDsgI)DU>L>O3O#UdvS z+N5DeYfkCLqp*brBkcPbx#-r+S#d_4z_U8lKGrnw2-a-4=;A%aVgJ~&ImKbhOF!mp zQygBWsZa2fy|_Jsan%V|A8b(J2$&5tBd2@ge{C{r7ncNI22Ih@aPkpd^q&47e{A;h z6xTq!EoJ4#80fzLxSzJ+09%LL!*MbiG15eWP7EI*uoVKCCN($@$7xh8P1U!fY)pm>pk<9D{_Dl&v6% zZp@r|Ae{b$j*|-=|AnplDW%f@zVxO&e+x=z$)&IgrXEg4gtuCN231gcdK&#MLc70X z#4(NA)Xq9O6s&Ig2&FYFZ!!k75CV6FU=@OYYHlX9t$ry;aaciyh$4b4$W@TH;FNaz zYbHuZpXQ`jP{KYMU-;h&@Q;x>Sk|3@9C2RNXpaI{Nj1SG=yrrv)Riu2;PG={Z7$m_CEDHNyZM;i^QSV1O`q<+#LKDF^~dsvwM?tAq?*oYiYE4BV# zn6#S);tdx9JK}_4&*2Gc7!#HH@D?VeUINO(OdC*7G#Lb5j6R+ zfxT6-9`bpuIf2>nI|AL8TgW!le|qz`-+}Lj$KAJ;`ecl@|Ld#(N2~cS$G!`}n*HTfgHAHroI{*g0gl^?=61;r0xg-)Px*hNb?tBKV!4oh`ke-RP@&&bQo zIGoKI%{VNXyv#Vv){7a3-<;OVIE*5+WgKR6Ze<*vY8=Wqtei9%hXVm48HcwuV~LEz zx-^h+m|cKnOC(~9MCuM>ffdC_oaEmKfC)%lnQ&&wy95c`C_S`ycun@&~)rYYfSYVtbusL5hFrk3KJ z$-}dYK^x_VVJCS}twlEgE9G@#QDneScjcXZuTGv_2PK;bl8hJ8f4ExDNV1ye6}w?> z;aRoW(`zXhh8P7vEOqHJ72R<8fBj@{FAjE$_jVs5nt&jl zzA$9NY}(YMsfuL$@@=CVCNJC+QN@bmxp3pr4eO&?`i)4LKUhmEs#d6@@R^mw-bDnLFx_NT zRij?MIrdZUe>PkQft3(%!y>XZ!zuGNe6j$t)!VRuf!TZaHhiLc^nqM`ncZMmR%iWX zq{c}81@cLUkCDM;EOY(~J)1{2E^cQX-J~8VFVZoVO{v}f=@OukoVGqzb-wC2ca z?aA-7hMBQd5m}Oe$=assp=aizekp8oI@00J1-ShJQ=4E|xi?`rX3mb-@d(V^ae+$ziT58KXKGF>VH zvp4aM`!g|!q$2N8&Ma)O2kp$4VDAQY-jPZtq zFyZM@qHlzS{rbm$mXnji+++!qD84KA5tYN$4Y{~ke8ICNpxMx@=8pNiDS~PEGoU~B zd1$qb=CKu;1UFp((U>&wr5v?+m8lzH`(jupe^KbNnK~OCk0ypcbbS{NnX#4IN(sd_ zORR|g{gl?cOdIcm`*E3Q$|*uPZKyL=I7Q$7oBTDc!bAa&e z1Ev51|87UuRMEH`tS~%8TKPxkb4ge^HX^+W`szLOj#vgAcDIzFqeoCPxyd6WLYtul zU|F0DO9d7-t)c2rFbAmm9swJdd$Z$oe|k?#QB3volf^4|u5A!c$^l!;9yDmX9mPA4 z$V{(sh6W9nHT{eYV8r8?K_$V4U(2m^Ajsv!$K6iXvNsiH6!`jg4h--=il6tYGE#1N z6CYPMHJw!|4=Mj~7tq^;!7^+H1(oYdM-y)Si%kb_=!#?P5=5XiG_&HK_sV_(e+US& zjfP19U)F_-0^L&i2KXF_R)(s^MR7er;ZQ!YkV?UVBE&Y|{JtD@f(kHq!T-uS;zMG*^7`L01cvP|! zv`vjN8L>5SYas#h4DgQq0}EnWf8Q@B`5=)mcQpM;y%KOd>ozqQWqriog#lEM$(JK8 zIc2aI%CuhGE~IN7QgH$8;z7262p*{D>D{z}A0-Uv_wFbUxf3qH3ARvWdM$8~nL(8W ztOX1118+d`+?c2zdgLDs-Ira^iix=xrw7HuZJ+{UCLApDjBR|yiC%Kme^6NqAA{_8 zEYK|F#b$OU*a?gnX*UeyJ0BH{o40@oPRD9NqczJ@*_vH~TFl;}H>`9<2}+3i=Eq{Dr%okA-;e33_>4t%rR}SxLbg zz4QZOAC!SWaO0#8Al%1Me`N?AA1L5M0{;WZD*b+fhS4Pdc+Iwf*sSnr+<)WxxRb-dtpkckSICf7o_Ur@XNcUBcf2 z%lJZ$)`F=jkHiwpoAH9m=a}ci#Y}XP93O}Hr5Pt&8jF?SA{q*)sRsB7`vbJ-{&m#< zvkl(WJ}Kd&w-mxK-~1bT};^J(mfiu=eFOjx8bW-*Z5e!a2gc+vb#j z6vVfa>*NXHPZY7W_g&(unqW3z*{S)p2{%CyVsT#)r>vRwfBbn4{;Uo)rW(55oC~Q( zZ@v8`T2s?#1j@*5(&MONINF`*`A=oJR@F0^X5hA9iOf%#)AF7gk@BcRmddUpTU)>9 z0c8tvlk1}u<&^=K1jor5N_38cge*r{`0QjP4VMA`WzG7b*27Hq6X}RGDYFaz_Q$vB802z_MDehYZyp&FC z=Lukdvy124>1NEj)lUfP5|rK6S+e?_tV)>D(bbW6{4eesc!%ws51O{Ju?4pR+wp+T zK~nxl?iWt}4^*=T&-*;OVA$jQU((v=!R?buMgq^Ae-SEt-L~(%50NB33n>rmj<7k* z-tqW1jL`Y=?It3Iwq`y=s-XSmbG;`lRDZaJwQ{8ji*DOW?SZurUXp~n^zdg}Nl~WU zK|Bg1bW9~1P$((ua`X1#hF9R%7u@Nu4zq4Xo(+t3yPc~Zws=~*abW?>YMBS{>Hef` zs!`Owe{I;&whiJmRYhM~^G0)u6jor9EcHMeZ-Sbp<-@~C&9SfIOY|srSv+5GX{XB8 z^#rvBDBD9{O*fELtyZjmJ^s99`k1|*=oyi1)f~r0u{R|DR%#%h7kfX52Ou!pINUpW z&lhFYB`3Sj0>45i#)u=wQr5mpTjt9G043MOe{UCxrOI6{`VSHvo8o!nB?nGUHMPY- zM)eM*KkZ#rgyMpjUlj!vXIX&~BlhODH}6{Q457@^a{bN(+ZDSVCMgD1IySf{6=Ubu z^ZZ5Tp?Wh^#1h(XIIy3AOZWg8nipo@fN2Q&L?nJR1R5>|jcNHC)WpjdFvue-at(uU ze>dPXboKb+!u$qX{V@y_`~v&7ih%&re+)zXiH@u^m;l(J5+>R!Fd0~{#AV`+8?SeN zMI2ay4T+rKO$n2$z7q7WHxa8C8fT1to<*!$8R_b#tD+PN7z1)>Yh?(OtQ#y8JyA)5d`YR_L})D$H6n}!5 z^Xt0rJj;3c+^1Q+j&^%w71+ac*qd}nd)3SQ^5s3DV_ z$Bh`tV~HxogKv=ufn}7E01WSeF0DBUbU!;Hf3&IO`EDXpm)DTBgOl$3+OGxVx%a*p zpBCnZnEGKoaqzk-jda4Og_Smop?^hS(d`PaaXNy80U3H^IJg;AF;7?o2kf*5;Uest z_Kf6a2&u+ugV{{YB}@}UFFD@)xFg{~B6_F9pjFhBY^cCx_8{_ZCW0CkRq4pY4*o0m z!nLZ^Q9F>@(7F89^sN3yiJ|qBUokajq|IR5{Lw z#P9m(rFC0B!pa#V5+qC<5vd;uJR2DuizUf8IkuGGYB!WA%A3`1GZmHp|9$Vb%Pu*81u2qlrU`+HcU1Ce~7^5Kwvj2 zz=ka_l<%Uw za=z}22kYs+md3SMP*-%%)3yl3WCe@R!*e-FMy)&VcOT{cFvs1jfK2Fcx6(u_WQ{Iv zcdnpV!Lvt-yh;6`njefFuRFAMmu&!|G7wftB;b2Z&=m1N9okPD@GW$I6pit5-S@iG z!0>cGLyUo##93w}4COoRQ6Ma3hBb{6+Q#!18~$}aZ}>_p#^upr}eZN*&Jd|cs}!0 zMb6VcpPBJNais=9pIIM&6m2PUon<2m@v>3)%+Rsn0hKrB)xy>TKE0&a3(4$(ignJO zEYl^?rVJ-3u_M5wXNF$f63`Cw(B@8;7(Kjj3Yi~)!CDP4G`3$y6Nj060mNa3E+(dj z#1O-PR9mB57mD zv>!4@1Nx*^Ro@E211hAU-dV~njTe;O+1Ffp$20K_A{&DXm@0&m%@0%8AY6Y8Zp_l? z8{uk&TPImp1RaJQV;7<_IvUNO`=mN{bq0TneSUWFa)ff3Y7w{kLtxbYurpIlp@Xt$D&<);(1~Mv z^gTl|f>|3;tYdF;$D_{fU}IO5EX83eN8hV#1h&2l+T%q+STM>g-L)QN0H*CQq|`=< z)a;|{U5JmhzNvXbf_B~v^lT+5o8plYgaJmGJcr?CPosa)h7)cNeg1dDh z`C*@4-&8avZ}gc#t-%R~ED=D2?GQ|?-CMcy@d zyI0wGXMU^in{`6Ld)R=z=Bq^wdIOxJh}?MEK()FmD#I8Y%UJCa7Bt$74g=bn^uNb~ z;;|f$B3Q<$!#md63R#A|zygF#p`G4i`Saeb3qF=zlT3wDbI^a@G~1s%d!G8wM{`2C z+JAoWJ8f-pMs?W}C{G4!=#4gv&dFv-yWi;HE()GW>mAQo6MJ-Meu=DRBPxn zbb(?MezravD(8UeL*n)E#Y5-wkH#Ge52@$!4X^Z%%)HWnFhl9}a~{J& zkY)NeJ>8U8JQRP6sLmvHm`|*LwdBKUn=R5TK^0|4KLx=b*OE+N`_ru^#Z?9PPF4QP;i?e^v)f5gX^(Yo{S6qL0V;1er zSC~Fx;a6`VjG?6%(r6+5tA@{PAJ>=dn7*f6mP(D^P>7Nq>hhl)2Nz_uTW?iCE}82CKqyw*pf{HjQF& z3?#mGRC=D;4h9YW1j>Jy{SP!+tmR(y`W#X9@|w}9$2`?EWvRV+dEnn`IO_1mD%x{b zJ8LMp5o`VUlZIGZp#1}`wDvZx7Dv1w5u=c1Nt5-|XOu@=#<)m~+yo)J1&N9A)D!UF z7RT$T;_e1l*;4>%<72h^&0bJY8g%W|%mm_m{}sMKc)F2%+YUNuFAsPGe;CSzf-e-XM2sc_|nQZ?q7jC>Tb94A#*$*k91+m|akF>V!Fs2yq2NbP7@X930g zIFAdOM4lco%XC?ui)SE0BMOS6aza6I5Wl^9M#}mV6sN&@TTvo^20*Q37O_BZ)5U$F z_0U}gp&5X6a6>1HsD$c&BQS*B>7pmECZyf+C3eURKmnj3Qg62K?`XK5Ob*0aH{J^#|Jh9rJ@l|W;~rV$;feUfOMo?kf|qZ$XBuu zAZh=Vvra+k-|!4bGIsrfFjzHa@TnXmPVD&edun3MWpG75W6Ed%BpWw>%!U=*!03FV zB>i;wDzm2?Yk*c%scITNhf0)gps+I-f? z9!zj{aWDfODItB#n6F`^Z<0);%rT(p%SgIK6bJ>F7A}^;10V;vHNWva`t1#@IK;Id zrQz^mSopsU3|H%0mI{ttV{X>Hh}x-uV*9|ArURRQNKJ=eZw{~no*+mVeu4Ch*$Rix z%_wIqBvlv|++P-nZm;bd^l^04uP25g^8VFY>&YSCKUEQ(G5%-XFRs>}h6Pcxbiwc3 zX1lSZz{k#p6%5;h1}%N5e542~rk*=aFfD8K|Fypt zbjUn^hj5XZhTCo)p2y3GF1hdC{pu6kdLA6D&GlHvm++_jQTpORCVanCKm!Rk5Vw55 zQAfQcbbo+w+eJG%<276n&HG7I4_azurrs*{9aMPL`+a#Wa44QXEqp&B**N6ypR!`Y#aEoHP&kQQq@{@Dk|H&oDKBrPc1aTNf$^a(6e3}ZJ9UzOTyg7Ieq(!sd?PmUGDJBl{k_XynV!(`Cd&cQ zRh~Wqq>oyLs8KT8BB>Q`0>2U^(##N}vg>sbV2FUmDrP>g{$lC^Jh*`?oi)GN%oo-o z&9QGbIbcv%4N;Sb+I0AA_skp7TzEX8lm(9;f4{_$0M7iMOU2B^qQZOaRSp0RX}GQ-FM;0y|@W7I}|b@RG$4+Un4GEZ1w{ zyTWx4uu6V~-74|cjvB)bp~|>U>hFtf{SW08r0g=LS%`MBm#<3O9xuop#VpXPFT<^_W{C) zU+N*VOP+W%17UyCb#-bRCge&cT0f9xbHT`sThU<<2Zp$Pwlp2$mueId7Nq@E4 zUm~hPiEsU5>Yr|vWC727y6^EW&+sPfBFWQ@+=1i%y_+RkG_nx7s)jEFMSlyQsX91) z__~Nv^Gd2jG+Yt?-V8*KeC`R#po+1m#*84mIlo-_+zj4Vl8}gjlS(M(VodUX#XoZ> zKgID5V4_mg3^)zpmCWaAGD9Q0Qi|Sk?IHR!g`eO2)E%i7@69}CaRc#im+H^GmGu92 z-n_cFtsT6dW?K1DQNhMkks?_T*mTHIxv!V5bmh, /// Capture, replay, and snapshot selection share the immutable session mode. guardian_context_mode: GuardianContextMode, + retain_inherited_user_messages: bool, /// Bumped whenever history is rewritten, such as compaction or rollback. history_version: u64, /// Monotonic user-input/reset revision, independent of compaction's history generation. @@ -169,6 +175,7 @@ impl ContextManager { review_history: None, retained_context: Arc::default(), guardian_context_mode: GuardianContextMode::Legacy, + retain_inherited_user_messages: false, history_version: 0, user_message_revision: 0, token_info: TokenUsageInfo::new_or_append( @@ -194,9 +201,15 @@ impl ContextManager { &self.retained_context } - pub(crate) fn with_guardian_context_mode(guardian_context_mode: GuardianContextMode) -> Self { + pub(crate) fn with_guardian_context_mode( + guardian_context_mode: GuardianContextMode, + source: &SessionSource, + ) -> Self { Self { guardian_context_mode, + retain_inherited_user_messages: guardian_context_mode + == GuardianContextMode::ThreadOwned + && !source.is_non_root_agent(), ..Self::new() } } @@ -213,10 +226,6 @@ impl ContextManager { true } - pub(crate) fn restore_retained_context(&mut self, checkpoint: Option<&RetainedContext>) { - Arc::make_mut(&mut self.retained_context).restore(checkpoint, &self.items); - } - pub(crate) fn guardian_history_checkpoint(&self) -> Option { if self.guardian_context_mode == GuardianContextMode::ThreadOwned { return None; @@ -369,54 +378,11 @@ impl ContextManager { review_history.record(&processed.item); } Arc::make_mut(&mut self.items).push(processed); - if crate::context::is_user_authorization_message(item) { - if self.guardian_context_mode == GuardianContextMode::Legacy { - Arc::make_mut(&mut self.retained_context).mark_user_messages_incomplete(); - } else if !metadata.is_some_and(|metadata| metadata.inherited_user_message) - && let ResponseItem::Message { - content, - internal_chat_message_metadata_passthrough, - .. - } = item - { - let mut complete = internal_chat_message_metadata_passthrough - .as_ref() - .and_then(|metadata| metadata.content_item_kinds.as_ref()) - .is_some_and(|kinds| { - kinds.len() == content.len() - && kinds.iter().all(|kind| kind.0.starts_with("user.")) - }); - let text = content - .iter() - .filter_map(|content| match content { - ContentItem::InputText { text } | ContentItem::OutputText { text } => { - Some(text.as_str()) - } - _ => { - complete = false; - None - } - }) - .collect::>() - .join("\n"); - // Keep the same bounded text that child reviewers receive before - // compaction, instead of letting storage discard a large source. - // Local instruction sections still omit incomplete originals whole. - let (text, truncated) = - guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS); - complete &= !truncated; - Arc::make_mut(&mut self.retained_context).record_user_message( - codex_history::RetainedUserMessage { - turn_id: item.turn_id().unwrap_or_default().to_owned(), - message_id: item.id().map(|id| id.as_str().to_owned()), - text, - complete, - }, - metadata.and_then(|metadata| metadata.user_input_order), - ); - } - self.user_message_revision = self.user_message_revision.saturating_add(1); - } + self.record_user_authorization( + item, + metadata, + user_authorization::UserMessageSource::Original, + ); } } @@ -590,10 +556,7 @@ impl ContextManager { let first_removed_message_id = snapshot[cut_idx] .id() .map(codex_protocol::ResponseItemId::as_str); - let acceptance_order = snapshot[cut_idx] - .metadata - .as_ref() - .and_then(|metadata| metadata.user_input_order); + let source = RetainedInputSource::from(snapshot[cut_idx].metadata.as_ref()); let mut review_history = self.review_history.take(); if let Some(history) = &mut review_history { history.truncate_before(&snapshot[cut_idx].item); @@ -638,7 +601,7 @@ impl ContextManager { Arc::make_mut(&mut retained_context).rollback( &removed_turns, first_removed_message_id, - acceptance_order, + source, ); } else { Arc::make_mut(&mut retained_context).retain_answers(|answer| { diff --git a/codex-rs/core/src/context_manager/history_tests.rs b/codex-rs/core/src/context_manager/history_tests.rs index f634b0049e..bfbe5d162b 100644 --- a/codex-rs/core/src/context_manager/history_tests.rs +++ b/codex-rs/core/src/context_manager/history_tests.rs @@ -1269,7 +1269,10 @@ fn drop_last_n_user_turns_preserves_prefix() { // A steered message shares its source turn, but rollback must keep the earlier // instruction and answer as complete evidence, including after the next compaction. - let mut history = ContextManager::with_guardian_context_mode(GuardianContextMode::ThreadOwned); + let mut history = ContextManager::with_guardian_context_mode( + GuardianContextMode::ThreadOwned, + &codex_protocol::protocol::SessionSource::Exec, + ); let mut expected = None; for (id, text) in [ ("restriction", "Never publish publicly."), diff --git a/codex-rs/core/src/context_manager/history_user_authorization.rs b/codex-rs/core/src/context_manager/history_user_authorization.rs new file mode 100644 index 0000000000..a3de08dd0f --- /dev/null +++ b/codex-rs/core/src/context_manager/history_user_authorization.rs @@ -0,0 +1,105 @@ +//! Captures local authorization and adopts copied instructions when a worker becomes a root. +//! Inherited history remains excluded for workers; roots also recover it from checkpoints. +//! Oversized originals retain bounded, incomplete excerpts for root review. +//! Checkpoint copies cannot establish original completeness, even when their text is short. + +use std::sync::Arc; + +use super::ContextManager; +use crate::context::GuardianContextMode; +use crate::guardian::GUARDIAN_MAX_ROOT_MESSAGE_TOKENS; +use crate::guardian::guardian_truncate_text; +use codex_history::CodexHarnessMetadata; +use codex_history::RetainedContext; +use codex_history::RetainedUserMessage; +use codex_protocol::models::ContentItem; +use codex_protocol::models::ResponseItem; + +/// Checkpoint items may have been shortened while preserving their original metadata. +pub(super) enum UserMessageSource { + Original, + Checkpoint, +} + +impl ContextManager { + pub(crate) fn restore_retained_context(&mut self, checkpoint: Option<&RetainedContext>) { + Arc::make_mut(&mut self.retained_context).restore(checkpoint, &self.items); + if self.retain_inherited_user_messages + && !self.retained_context.has_inherited_user_messages() + { + // Worker checkpoints intentionally omit copied instructions. A standalone + // root adopts the surviving prefix without duplicating an adopted checkpoint. + let items = Arc::clone(&self.items); + for envelope in items.iter().filter(|envelope| { + envelope + .metadata + .as_ref() + .is_some_and(|metadata| metadata.inherited_user_message) + }) { + self.record_user_authorization( + &envelope.item, + envelope.metadata.as_ref(), + UserMessageSource::Checkpoint, + ); + } + } + } + + pub(super) fn record_user_authorization( + &mut self, + item: &ResponseItem, + metadata: Option<&CodexHarnessMetadata>, + source: UserMessageSource, + ) { + if !crate::context::is_user_authorization_message(item) { + return; + } + let inherited = metadata.is_some_and(|metadata| metadata.inherited_user_message); + if self.guardian_context_mode == GuardianContextMode::Legacy { + Arc::make_mut(&mut self.retained_context).mark_user_messages_incomplete(); + } else if (!inherited || self.retain_inherited_user_messages) + && let ResponseItem::Message { + content, + internal_chat_message_metadata_passthrough, + .. + } = item + { + let mut complete = matches!(source, UserMessageSource::Original) + && internal_chat_message_metadata_passthrough + .as_ref() + .and_then(|metadata| metadata.content_item_kinds.as_ref()) + .is_some_and(|kinds| { + kinds.len() == content.len() + && kinds.iter().all(|kind| kind.0.starts_with("user.")) + }); + let text = content + .iter() + .filter_map(|content| match content { + ContentItem::InputText { text } | ContentItem::OutputText { text } => { + Some(text.as_str()) + } + _ => { + complete = false; + None + } + }) + .collect::>() + .join("\n"); + // Keep the same bounded text that child reviewers receive before + // compaction, instead of letting storage discard a large source. + // Local instruction sections still omit incomplete originals whole. + let (text, truncated) = guardian_truncate_text(&text, GUARDIAN_MAX_ROOT_MESSAGE_TOKENS); + complete &= !truncated; + Arc::make_mut(&mut self.retained_context).record_user_message( + RetainedUserMessage { + turn_id: item.turn_id().unwrap_or_default().to_owned(), + message_id: item.id().map(|id| id.as_str().to_owned()), + text, + complete, + }, + metadata.into(), + ); + } + self.user_message_revision = self.user_message_revision.saturating_add(1); + } +} diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 4ee1497ba1..38ca95c8ee 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -339,7 +339,10 @@ impl Session { ) .unwrap_or(u64::MAX); - let mut history = ContextManager::with_guardian_context_mode(self.guardian_context_mode); + let mut history = ContextManager::with_guardian_context_mode( + self.guardian_context_mode, + &turn_context.session_source, + ); let mut saw_legacy_compaction_without_replacement_history = false; if let Some(checkpoint) = base_compaction && let Some(items) = &checkpoint.compacted.replacement_history diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index b283528024..a3e6b17a2c 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1267,7 +1267,10 @@ impl Session { let mut state = SessionState::new_with_auto_compact_window_ids( session_configuration.clone(), initial_auto_compact_window_ids, - guardian_context_mode, + ContextManager::with_guardian_context_mode( + guardian_context_mode, + &session_configuration.session_source, + ), ); state.base_instructions_provenance = base_instructions_provenance.clone(); let managed_network_requirements_configured = config diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 9e4f5992c3..116e450033 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -8786,9 +8786,10 @@ where ); let mut state = SessionState::new(session_configuration.clone()); - state.history = ContextManager::with_guardian_context_mode(GuardianContextMode::from_features( - &config.features, - )); + state.history = ContextManager::with_guardian_context_mode( + GuardianContextMode::from_features(&config.features), + &session_configuration.session_source, + ); let (environment_manager, resolved_turn_environments) = resolved_environments_for_configuration(&session_configuration, &default_environments) .await; diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index d4d04d37aa..5333661a6f 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -12,7 +12,6 @@ use super::AdditionalContextStore; use super::auto_compact_window::AutoCompactWindow; use super::auto_compact_window::AutoCompactWindowIds; use super::auto_compact_window::AutoCompactWindowSnapshot; -use crate::context::GuardianContextMode; use crate::context_manager::ContextManager; use crate::context_manager::HistoryReplacement; use crate::session::PreviousTurnSettings; @@ -65,16 +64,15 @@ impl SessionState { Self::new_with_auto_compact_window_ids( session_configuration, AutoCompactWindowIds::new_initial(), - GuardianContextMode::Legacy, + ContextManager::new(), ) } pub(crate) fn new_with_auto_compact_window_ids( session_configuration: SessionConfiguration, auto_compact_window_ids: AutoCompactWindowIds, - guardian_context_mode: GuardianContextMode, + history: ContextManager, ) -> Self { - let history = ContextManager::with_guardian_context_mode(guardian_context_mode); Self { session_configuration, base_instructions_provenance: None, diff --git a/codex-rs/core/tests/suite/guardian_retained_context.rs b/codex-rs/core/tests/suite/guardian_retained_context.rs index 9b4e632d34..f1079403a5 100644 --- a/codex-rs/core/tests/suite/guardian_retained_context.rs +++ b/codex-rs/core/tests/suite/guardian_retained_context.rs @@ -9,6 +9,7 @@ use std::time::Duration; use anyhow::Context; use anyhow::Result; use codex_core::CodexThread; +use codex_core::ForkSnapshot; use codex_core::TurnInputRequest; use codex_core::config::Config; use codex_core::config::ThreadStoreConfig; @@ -35,7 +36,9 @@ use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::request_user_input::RequestUserInputAnswer; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; +use codex_thread_store::ForkBoundary; use codex_thread_store::LoadThreadHistoryParams; +use codex_thread_store::PrepareForkParams; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_function_call; @@ -893,6 +896,225 @@ enum LifecycleBoundary { ChildFork, } +// Cover live copied history and a truncated referenced checkpoint. Parent-answer +// omission is already exercised by retained_answers_cross_real_session_boundaries. +#[test_case(ThreadHistoryMode::Legacy; "copied worker history")] +#[test_case(ThreadHistoryMode::Paginated; "referenced truncated checkpoint")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn standalone_fork_retains_inherited_user_instructions( + history_mode: ThreadHistoryMode, +) -> Result<()> { + let compacted = history_mode == ThreadHistoryMode::Paginated; + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let mut test = test_codex() + .with_history_mode(history_mode) + .with_config(|config| { + config.experimental_thread_store = ThreadStoreConfig::Local; + config + .features + .disable(Feature::TokenBudget) + .expect("use local compaction for worker checkpoint"); + config.model_provider.name = "Local compaction test provider".to_owned(); + for feature in [ + Feature::GuardianApproval, + Feature::GuardianThreadContext, + Feature::Collab, + Feature::MultiAgentV2, + Feature::DefaultModeRequestUserInput, + ] { + config + .features + .enable(feature) + .expect("enable test feature"); + } + }) + .build_with_auto_env(&server) + .await?; + let instruction = if compacted { + format!( + "You may publish the release. {} Delegate an inspection.", + "Original project detail. ".repeat(500) + ) + } else { + "You may publish the release. Delegate an inspection.".to_owned() + }; + let mut created = test.thread_manager.subscribe_thread_created(); + mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_function_call_with_namespace( + "spawn", "collaboration", "spawn_agent", + &json!({"task_name": "worker", "message": "Inspect the project.", "fork_turns": "all"}).to_string(), + ), + ev_completed("spawn-response"), + ]), + sse(vec![ev_completed("root-complete")]), + sse(vec![ev_completed("worker-complete")]), + ], + ).await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: instruction.clone(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + let worker = test + .thread_manager + .get_thread(created.recv().await?) + .await?; + wait_for_event(&worker, |event| matches!(event, EventMsg::TurnComplete(_))).await; + test.codex.shutdown_and_wait().await?; + let local_instruction = if compacted { + // Leave less than Guardian's 900-token budget for the inherited source + // within local compaction's 20K user-message budget. + format!("Ask me before publishing. {}", "x".repeat(78_000)) + } else { + "Ask me before publishing.".to_owned() + }; + mount_sse_sequence(&server, vec![sse(vec![ev_completed("local-instruction")])]).await; + worker + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: local_instruction.clone(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&worker, |event| matches!(event, EventMsg::TurnComplete(_))).await; + if compacted { + // Local compaction keeps the inherited source message in replacement history. + mount_sse_sequence( + &server, + vec![sse(vec![ + ev_assistant_message("worker-summary", "Inspected the project."), + ev_completed("worker-compacted"), + ])], + ) + .await; + compact_and_assert_answers(&test, &worker, &[]).await?; + } + let worker_context = worker.conversation_history_snapshot().await; + let inherited_text = worker_context + .items() + .find_map(|item| match item { + ResponseItem::Message { role, content, .. } if role == "user" => { + let [ContentItem::InputText { text }] = content.as_slice() else { + return None; + }; + text.starts_with("You may publish the release.") + .then(|| text.clone()) + } + _ => None, + }) + .context("inherited source in worker history")?; + if compacted { + assert_ne!(inherited_text, instruction); + assert!(inherited_text.contains("tokens truncated")); + assert!(inherited_text.len() < 900 * 4); + } + // The adopted root later compacts through the mechanical path, which can drop originals. + test.config + .features + .enable(Feature::TokenBudget) + .expect("use mechanical root compaction"); + let fork = if history_mode == ThreadHistoryMode::Paginated { + let prepared = test + .thread_store + .prepare_fork(PrepareForkParams { + thread_id: worker.session_configured().thread_id, + boundary: ForkBoundary::Latest, + }) + .await?; + test.thread_manager + .fork_prepared_thread( + test.config.clone(), + prepared, + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + } else { + test.thread_manager + .fork_thread_from_history( + ForkSnapshot::Interrupted, + test.config.clone(), + InitialHistory::Resumed(ResumedHistory { + conversation_id: worker.session_configured().thread_id, + history: Arc::new(load_context(&test, &worker).await?), + rollout_path: None, + }), + /*thread_source*/ None, + /*parent_trace*/ None, + ClientMcpExtensions::default(), + /*reserved_thread_id*/ None, + ) + .await? + }; + assert!(fork.thread.guardian_root_snapshot().await.is_none()); + let history = fork.thread.conversation_history_snapshot().await; + let retained = history.retained_context().context("root context")?; + let answers = codex_guardian_context::render_verified_answers(retained); + assert_eq!( + (answers.fragments, answers.complete), + ( + vec![codex_guardian_context::GuardianRootMessage::IncompleteVerifiedAnswers.render()], + false + ), + ); + assert_eq!( + retained + .ordered_entries() + .map(|(_, entry)| match entry { + codex_history::RetainedContextEntry::UserMessage(message) => + (message.text.clone(), message.complete), + codex_history::RetainedContextEntry::VerifiedAnswer(_) => + panic!("no answers before root adoption"), + }) + .collect::>(), + vec![ + (inherited_text, !compacted), + ( + codex_guardian_context::truncate_text(&local_instruction, /*max_tokens*/ 900), + !compacted, + ), + ], + ); + let after = record_answer( + &fork.thread, + &server, + "root-action", + "Do not publish.", + /*acceptance_order*/ 2, + ) + .await?; + let expected = fork + .thread + .conversation_history_snapshot() + .await + .retained_context() + .context("root after local answer")? + .clone(); + assert!(!expected.verified_answers_complete()); + let root = fork.thread; + compact_and_assert_answers(&test, &root, &[after]).await?; + let root = resume(&test, &root).await?; + assert_eq!( + root.conversation_history_snapshot() + .await + .retained_context(), + Some(&expected) + ); + root.shutdown_and_wait().await?; + worker.shutdown_and_wait().await?; + Ok(()) +} + #[test_case(false, true; "enabled without checkpoint")] #[test_case(true, true; "enabled after checkpoint")] #[test_case(false, false; "legacy without checkpoint")] diff --git a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs index df828c99dd..e11b940801 100644 --- a/codex-rs/core/tests/suite/guardian_subagent_authorization.rs +++ b/codex-rs/core/tests/suite/guardian_subagent_authorization.rs @@ -657,11 +657,32 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( assert!(answer_position < queued_position); let mut partial = serde_json::to_value(history.retained_context().expect("retained root context"))?; + let mut inherited_instruction = partial["user_messages"] + .as_array() + .expect("retained user-message records") + .iter() + .find(|entry| entry["text"] == INITIAL_PROMPT) + .expect("initial instruction") + .clone(); + inherited_instruction["inherited"] = json!(true); + inherited_instruction["order"] = json!( + original_history[queued_position] + .metadata + .as_ref() + .expect("queued input metadata") + .user_input_order + .expect("queued acceptance order") + ); partial["user_messages"] .as_array_mut() .expect("retained user-message records") .retain(|entry| entry["text"] == USER_APPROVAL); partial["user_messages_incomplete"] = json!(true); + let mut inherited_prefix = partial.clone(); + inherited_prefix["user_messages"] + .as_array_mut() + .expect("retained user-message records") + .push(inherited_instruction); let mut expected_authorization = expected_messages .into_iter() .filter(|message| !matches!(message, GuardianRootMessage::Assistant(_))) @@ -684,12 +705,39 @@ async fn guardian_subagent_review_preserves_late_root_user_authorization( (partial, MissingCheckpointSource::RootInstruction, false), // Rebuild the local counter even when all retained metadata is absent. (Value::Null, MissingCheckpointSource::None, true), + // Prefix and local orders can collide numerically. Recovery must keep + // the inherited instruction first without losing the queued local grant. + (inherited_prefix, MissingCheckpointSource::None, true), ] { let mut expected = expected_authorization.clone(); + let inherited_message_id = retained["user_messages"] + .as_array() + .and_then(|entries| entries.iter().find(|entry| entry["inherited"] == true)) + .and_then(|entry| entry["message_id"].as_str()); + if inherited_message_id.is_some() { + expected.insert( + /*index*/ 2, + GuardianRootMessage::IncompleteVerifiedAnswers, + ); + } if retained.is_null() { expected.retain(|message| !matches!(message, GuardianRootMessage::UserInput(_))); } let mut replacement_history = original_history.clone(); + if let Some(id) = inherited_message_id { + let metadata = replacement_history + .iter_mut() + .find(|envelope| { + envelope + .item + .id() + .is_some_and(|item_id| item_id.as_str() == id) + }) + .and_then(|envelope| envelope.metadata.as_mut()) + .expect("inherited input metadata"); + metadata.inherited_user_message = true; + metadata.user_input_order = Some(100); + } match missing_source { MissingCheckpointSource::None => {} MissingCheckpointSource::VerifiedAnswer => { diff --git a/codex-rs/guardian-context/src/retained_instructions_tests.rs b/codex-rs/guardian-context/src/retained_instructions_tests.rs index 9e21ece663..fa78ddd875 100644 --- a/codex-rs/guardian-context/src/retained_instructions_tests.rs +++ b/codex-rs/guardian-context/src/retained_instructions_tests.rs @@ -1,5 +1,6 @@ use super::*; use codex_history::RetainedContextEvent; +use codex_history::RetainedInputSource; use codex_history::RetainedUserMessage; use codex_history::VerifiedAnswer; use codex_history::VerifiedQuestionAnswer; @@ -28,7 +29,7 @@ fn instructions_preserve_source_order_and_whole_records() { text: "Do not publish after all.".to_owned(), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let rendered = render_retained_instructions(&context); assert_eq!( @@ -52,7 +53,7 @@ fn instructions_preserve_source_order_and_whole_records() { text: "Permission is conditional. ".repeat(200), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let rendered = render_retained_instructions(&context); assert_eq!(rendered.len(), 2); diff --git a/codex-rs/history/src/lib.rs b/codex-rs/history/src/lib.rs index dd1b9ef05f..258c32291f 100644 --- a/codex-rs/history/src/lib.rs +++ b/codex-rs/history/src/lib.rs @@ -173,6 +173,8 @@ pub use reconciled_retained_context::ReconciledRetainedContext; pub use retained_context::RetainedContext; pub use retained_context::RetainedContextEntry; pub use retained_context::RetainedContextEvent; +pub use retained_context::RetainedContextOrder; +pub use retained_context::RetainedInputSource; pub use retained_context::RetainedUserMessage; pub use retained_context::VerifiedAnswer; pub use retained_context::VerifiedQuestionAnswer; diff --git a/codex-rs/history/src/reconciled_retained_context.rs b/codex-rs/history/src/reconciled_retained_context.rs index 3754044767..1343be6347 100644 --- a/codex-rs/history/src/reconciled_retained_context.rs +++ b/codex-rs/history/src/reconciled_retained_context.rs @@ -6,12 +6,13 @@ use std::collections::HashSet; use crate::RetainedContext; use crate::RetainedContextEntry; +use crate::RetainedContextOrder; use crate::RetainedUserMessage; /// Retained evidence supplemented by surviving user messages without changing the checkpoint. pub struct ReconciledRetainedContext<'a> { retained_context: Option<&'a RetainedContext>, - recovered: Vec<(u64, RetainedUserMessage)>, + recovered: Vec<(RetainedContextOrder, RetainedUserMessage)>, /// Instructions were already missing, or a recovered source could not be ordered. pub missing_user_messages: bool, } @@ -19,6 +20,7 @@ pub struct ReconciledRetainedContext<'a> { impl<'a> ReconciledRetainedContext<'a> { /// Reconciles eligible live user messages with retained sources in acceptance order. /// The caller supplies original source identities and filters out non-user context. + /// Inherited sources must supply no local order; retained identities match before ordering. /// Live messages are consumed only when retained user instructions are incomplete. pub fn new( retained_context: Option<&'a RetainedContext>, @@ -57,7 +59,11 @@ impl<'a> ReconciledRetainedContext<'a> { } // Queued steering can enter raw history after a later-accepted answer. // Compare their persisted sequence numbers, including checkpoint-only answers. - let Some(order) = order.filter(|order| used_orders.insert(*order)) else { + // Parent counters cannot establish an adopted prefix position. + let Some(order) = order + .map(RetainedContextOrder::Local) + .filter(|order| used_orders.insert(*order)) + else { missing_user_messages = true; continue; }; @@ -74,7 +80,7 @@ impl<'a> ReconciledRetainedContext<'a> { /// Returns retained and recovered evidence in persisted acceptance order. pub fn ordered_entries( &self, - ) -> impl DoubleEndedIterator)> { + ) -> impl DoubleEndedIterator)> { let mut entries = self .retained_context .into_iter() diff --git a/codex-rs/history/src/reconciled_retained_context_tests.rs b/codex-rs/history/src/reconciled_retained_context_tests.rs index b851f23e19..baeb79d0dc 100644 --- a/codex-rs/history/src/reconciled_retained_context_tests.rs +++ b/codex-rs/history/src/reconciled_retained_context_tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::RetainedContextEvent; +use crate::RetainedInputSource; use crate::VerifiedAnswer; use crate::VerifiedQuestionAnswer; use pretty_assertions::assert_eq; @@ -25,8 +26,8 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { ..retained_excerpt.clone() }; let mut retained = RetainedContext::default(); - retained.record_user_message(initial.clone(), Some(0)); - retained.record_user_message(retained_excerpt, Some(1)); + retained.record_user_message(initial.clone(), RetainedInputSource::Local(Some(0))); + retained.record_user_message(retained_excerpt, RetainedInputSource::Local(Some(1))); retained.record(&RetainedContextEvent::VerifiedAnswer { answer: VerifiedAnswer { turn_id: "answer-turn".to_owned(), @@ -70,12 +71,15 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { ), ( vec![ - (0, "Inspect the deployment."), - (1, ""), - (2, "Make the deployment public."), - (3, "Never publicly."), - (4, "Do not deploy after all."), - (5, "Inspect the deployment."), + (RetainedContextOrder::Local(0), "Inspect the deployment."), + (RetainedContextOrder::Local(1), ""), + ( + RetainedContextOrder::Local(2), + "Make the deployment public." + ), + (RetainedContextOrder::Local(3), "Never publicly."), + (RetainedContextOrder::Local(4), "Do not deploy after all."), + (RetainedContextOrder::Local(5), "Inspect the deployment."), ], true, ), @@ -86,7 +90,10 @@ fn recovery_preserves_source_identity_acceptance_order_and_checkpoint_gaps() { #[test] fn recovery_marks_missing_and_conflicting_orders_incomplete() { let mut retained = RetainedContext::default(); - retained.record_user_message(instruction("Inspect the deployment."), Some(1)); + retained.record_user_message( + instruction("Inspect the deployment."), + RetainedInputSource::Local(Some(1)), + ); assert!(!retained.has_missing_user_messages()); let checkpoint = retained.clone(); @@ -111,7 +118,10 @@ fn recovery_marks_missing_and_conflicting_orders_incomplete() { reconciled.missing_user_messages, ), ( - vec![(1, "Inspect the deployment."), (2, "Do not deploy.")], + vec![ + (RetainedContextOrder::Local(1), "Inspect the deployment."), + (RetainedContextOrder::Local(2), "Do not deploy.") + ], true, ), "invalid order: {invalid_order:?}", diff --git a/codex-rs/history/src/retained_context.rs b/codex-rs/history/src/retained_context.rs index 8c1f368819..ecc5c364c9 100644 --- a/codex-rs/history/src/retained_context.rs +++ b/codex-rs/history/src/retained_context.rs @@ -1,5 +1,7 @@ //! Bounded host-owned facts outside the model's compaction contract. //! Checkpoints and live recording use the same admission rules; rendering is a consumer concern. +//! Adopted parent instructions precede local facts without sharing the local acceptance counter. +//! Their parent verified answers are unavailable, so adopted authorization stays incomplete. use std::collections::VecDeque; @@ -7,6 +9,7 @@ use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; +use crate::CodexHarnessMetadata; use crate::ResponseItemEnvelope; const MAX_FAMILY_RECORDS: usize = 8; @@ -39,6 +42,33 @@ pub struct RetainedUserMessage { pub complete: bool, } +/// Local facts use their acceptance counter; copied parent instructions use prefix order. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RetainedInputSource { + Local(Option), + Inherited, +} + +impl RetainedInputSource { + /// Parent counters never establish order within the owning thread. + pub fn acceptance_order(self) -> Option { + match self { + Self::Local(order) => order, + Self::Inherited => None, + } + } +} + +impl From> for RetainedInputSource { + fn from(metadata: Option<&CodexHarnessMetadata>) -> Self { + if metadata.is_some_and(|metadata| metadata.inherited_user_message) { + Self::Inherited + } else { + Self::Local(metadata.and_then(|metadata| metadata.user_input_order)) + } + } +} + impl std::fmt::Debug for RetainedUserMessage { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { formatter @@ -50,12 +80,32 @@ impl std::fmt::Debug for RetainedUserMessage { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)] struct Ordered { + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + inherited: bool, #[serde(default)] order: u64, #[serde(flatten)] value: T, } +impl Ordered { + fn key(&self) -> RetainedContextOrder { + if self.inherited { + RetainedContextOrder::Inherited(self.order) + } else { + RetainedContextOrder::Local(self.order) + } + } +} + +/// Inherited prefix order precedes the owning thread's local acceptance order. +/// The counters have separate scopes and must not be compared without their origin. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)] +pub enum RetainedContextOrder { + Inherited(u64), + Local(u64), +} + /// Borrowed host evidence across retained families. pub enum RetainedContextEntry<'a> { UserMessage(&'a RetainedUserMessage), @@ -108,7 +158,7 @@ fn legacy_user_messages_incomplete() -> bool { fn bound_family(items: &mut VecDeque>, incomplete: &mut bool) { // Queued instructions can be recorded after later-accepted answers. Evict by // acceptance order, not by the order in which persistence happened to finish. - items.make_contiguous().sort_by_key(|entry| entry.order); + items.make_contiguous().sort_by_key(Ordered::key); while items.len() > MAX_FAMILY_RECORDS || serde_json::to_vec(items).map_or(true, |bytes| bytes.len() > MAX_FAMILY_BYTES) { @@ -196,24 +246,29 @@ impl RetainedContext { self.user_messages_incomplete = true; } - /// Returns retained evidence with its persisted order, shared across both families. - /// Explicit acceptance order survives delayed recording; legacy records use recording order. + /// Whether a root checkpoint already contains its adopted instruction prefix. + pub fn has_inherited_user_messages(&self) -> bool { + self.user_messages.iter().any(|entry| entry.inherited) + } + + /// Returns retained evidence with its origin and persisted order across both families. + /// Explicit acceptance order survives delayed recording; inherited prefix order stays separate. pub fn ordered_entries( &self, - ) -> impl DoubleEndedIterator)> { + ) -> impl DoubleEndedIterator)> { let mut entries = self .verified_answers .iter() .map(|entry| { ( - entry.order, + entry.key(), RetainedContextEntry::VerifiedAnswer(&entry.value), ) }) .chain( self.user_messages .iter() - .map(|entry| (entry.order, RetainedContextEntry::UserMessage(&entry.value))), + .map(|entry| (entry.key(), RetainedContextEntry::UserMessage(&entry.value))), ) .collect::>(); entries.sort_by_key(|(order, _)| *order); @@ -222,22 +277,41 @@ impl RetainedContext { /// Records a delivered user item with its acceptance order. Legacy items without /// this metadata use recording order; checkpoint/suffix replay uses the same path. + /// Inherited instructions use prefix order because their original counters belong to parents. pub fn record_user_message( &mut self, mut message: RetainedUserMessage, - acceptance_order: Option, + source: RetainedInputSource, ) { message.bound(); + let inherited = source == RetainedInputSource::Inherited; + // Worker forks omit parent answer records. Adopting their instructions + // cannot establish whether an omitted answer restricted an inherited grant. + self.verified_answers_incomplete |= inherited; if let Some(index) = self.user_messages.iter().position(|entry| { message.message_id.is_some() && entry.value.message_id == message.message_id }) { - if self.user_messages[index].value == message { + if self.user_messages[index].value == message + && self.user_messages[index].inherited == inherited + { return; } self.user_messages.remove(index); } - let order = self.record_order(acceptance_order); + // Parent and worker counters have different scopes. Adopt the copied prefix + // in history order without advancing the worker's local acceptance counter. + let order = if inherited { + self.user_messages + .iter() + .filter(|entry| entry.inherited) + .map(|entry| entry.order.saturating_add(1)) + .max() + .unwrap_or_default() + } else { + self.record_order(source.acceptance_order()) + }; self.user_messages.push_back(Ordered { + inherited, order, value: message, }); @@ -283,6 +357,7 @@ impl RetainedContext { } let order = self.record_order(acceptance_order); self.verified_answers.push_back(Ordered { + inherited: false, order, value: answer, }); @@ -315,7 +390,11 @@ impl RetainedContext { } for entry in &mut self.user_messages { entry.value.bound(); - self.next_order = self.next_order.max(entry.order.saturating_add(1)); + // Also repair checkpoints written before adoption recorded the answer gap. + self.verified_answers_incomplete |= entry.inherited; + if !entry.inherited { + self.next_order = self.next_order.max(entry.order.saturating_add(1)); + } } for order in surviving_items .iter() @@ -345,18 +424,19 @@ impl RetainedContext { &mut self, turn_ids: &[&str], first_removed_message_id: Option<&str>, - acceptance_order: Option, + source: RetainedInputSource, ) { - if let Some(order) = acceptance_order.or_else(|| { + let boundary = source.acceptance_order().map(RetainedContextOrder::Local); + if let Some(boundary) = boundary.or_else(|| { first_removed_message_id.and_then(|id| { self.user_messages .iter() .find(|message| message.value.message_id.as_deref() == Some(id)) - .map(|message| message.order) + .map(Ordered::key) }) }) { - self.verified_answers.retain(|entry| entry.order < order); - self.user_messages.retain(|entry| entry.order < order); + self.verified_answers.retain(|entry| entry.key() < boundary); + self.user_messages.retain(|entry| entry.key() < boundary); return; } self.user_messages_incomplete |= first_removed_message_id.is_some() @@ -364,10 +444,14 @@ impl RetainedContext { .user_messages .iter() .any(|message| turn_ids.contains(&message.value.turn_id.as_str())); - self.verified_answers - .retain(|answer| !turn_ids.contains(&answer.value.turn_id.as_str())); - self.user_messages - .retain(|message| !turn_ids.contains(&message.value.turn_id.as_str())); + self.verified_answers.retain(|answer| { + source != RetainedInputSource::Inherited + && !turn_ids.contains(&answer.value.turn_id.as_str()) + }); + self.user_messages.retain(|message| { + (source != RetainedInputSource::Inherited || message.inherited) + && !turn_ids.contains(&message.value.turn_id.as_str()) + }); } } diff --git a/codex-rs/history/src/retained_context_tests.rs b/codex-rs/history/src/retained_context_tests.rs index d3c01cbf6d..9b5bf65f99 100644 --- a/codex-rs/history/src/retained_context_tests.rs +++ b/codex-rs/history/src/retained_context_tests.rs @@ -31,7 +31,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() text: String::new(), complete: false, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let mut expected = context.clone(); expected.user_messages[0].value.text = "Do not publish after all.".to_owned(); @@ -63,7 +63,7 @@ fn retained_evidence_preserves_order_through_recovery_checkpoint_and_rollback() restored.rollback( &["revocation-turn"], Some("revocation"), - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); assert_eq!( restored, @@ -99,7 +99,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { context.rollback( &["turn-2"], /*first_removed_message_id*/ None, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); assert_eq!(context.verified_answers().count(), 0); assert!(!context.verified_answers_complete()); @@ -135,7 +135,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "Keep the repository private.".to_owned(), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); } assert!(!restored.user_messages_complete()); @@ -154,7 +154,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "An older queued instruction.".to_owned(), complete: true, }, - Some(0), + RetainedInputSource::Local(Some(0)), ); assert_eq!( restored, recent, @@ -167,7 +167,7 @@ fn retained_families_enforce_storage_limits_without_changing_snapshots() { text: "restriction ".repeat(MAX_RECORD_BYTES), complete: true, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); let Some((_, RetainedContextEntry::UserMessage(message))) = restored.ordered_entries().next_back() @@ -188,7 +188,7 @@ fn recovered_excerpts_obey_record_and_family_limits() { text: String::new(), complete: false, }, - /*acceptance_order*/ None, + RetainedInputSource::Local(None), ); } let unchanged = context.clone(); @@ -281,10 +281,13 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { text: "Keep the repository private.".to_owned(), complete: true, }; - context.record_user_message(instruction.clone(), Some(steer_order)); + context.record_user_message( + instruction.clone(), + RetainedInputSource::Local(Some(steer_order)), + ); let mut resumed = RetainedContext::default(); resumed.restore(Some(&checkpoint), &[]); - resumed.record_user_message(instruction, Some(steer_order)); + resumed.record_user_message(instruction, RetainedInputSource::Local(Some(steer_order))); assert_eq!(resumed, context); assert_eq!( resumed @@ -297,15 +300,29 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { }) .collect::>(), vec![ - (steer_order, "Keep the repository private."), - (answer_order, "Yes, but never publicly."), + ( + RetainedContextOrder::Local(steer_order), + "Keep the repository private." + ), + ( + RetainedContextOrder::Local(answer_order), + "Yes, but never publicly." + ), ], ); // This checkpoint predates model delivery of the queued instruction. Rollback // still removes its later-accepted answer using the persisted boundary order. let mut before_delivery = checkpoint; - before_delivery.rollback(&["turn-1"], Some("steer"), Some(steer_order)); - resumed.rollback(&["turn-1"], Some("steer"), Some(steer_order)); + before_delivery.rollback( + &["turn-1"], + Some("steer"), + RetainedInputSource::Local(Some(steer_order)), + ); + resumed.rollback( + &["turn-1"], + Some("steer"), + RetainedInputSource::Local(Some(steer_order)), + ); assert_eq!(before_delivery, resumed); assert_eq!( resumed, @@ -315,3 +332,57 @@ fn accepted_order_survives_delayed_recording_and_checkpoint_replay() { } ); } + +#[test] +fn adopted_instructions_preserve_local_order_and_rollback_scope() { + let mut context = RetainedContext::default(); + context.record(&publish_answer()); + for index in 0..2 { + let message = RetainedUserMessage { + turn_id: "parent-turn".to_owned(), + message_id: Some(format!("parent-{index}")), + text: format!("Parent instruction {index}"), + complete: true, + }; + context.record_user_message(message.clone(), RetainedInputSource::Inherited); + context.record_user_message(message, RetainedInputSource::Inherited); + } + assert_eq!(context.reserve_order(), 1); + assert!(!context.verified_answers_complete()); + assert_eq!( + context + .ordered_entries() + .map(|(order, _)| order) + .collect::>(), + vec![ + RetainedContextOrder::Inherited(0), + RetainedContextOrder::Inherited(1), + RetainedContextOrder::Local(0) + ], + ); + let checkpoint = serde_json::from_slice(&serde_json::to_vec(&context).unwrap()).unwrap(); + let mut restored = RetainedContext::default(); + restored.restore(Some(&checkpoint), &[]); + assert_eq!(restored, context); + // Older roots adopted complete instructions without recording the missing parent answers. + let mut legacy_checkpoint = serde_json::to_value(&checkpoint).unwrap(); + legacy_checkpoint["incomplete"] = serde_json::json!(false); + let legacy_checkpoint = serde_json::from_value(legacy_checkpoint).unwrap(); + restored.restore(Some(&legacy_checkpoint), &[]); + assert_eq!(restored, context); + restored.rollback( + &["turn-1"], + /*first_removed_message_id*/ None, + RetainedInputSource::Local(Some(0)), + ); + let mut expected = context; + expected.verified_answers.clear(); + assert_eq!(restored, expected); + restored.rollback( + &["parent-turn"], + Some("parent-1"), + RetainedInputSource::Inherited, + ); + expected.user_messages.pop_back(); + assert_eq!(restored, expected); +} diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 390b1d08b6..462d0f0260 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -31,6 +31,7 @@ pub use codex_history::ResponseItemEnvelope; pub use codex_history::ResumedHistory; pub use codex_history::RetainedContextEntry; pub use codex_history::RetainedContextEvent; +pub use codex_history::RetainedInputSource; pub use codex_history::RolloutItem; pub use codex_history::RolloutLine; pub(crate) use codex_protocol::protocol; diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs index 6da7264cd3..588ac37787 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs @@ -17,6 +17,7 @@ use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::UserMessageEvent; use codex_rollout::CompactedItem; use codex_rollout::RetainedContextEntry; +use codex_rollout::RetainedInputSource; use codex_rollout::RolloutItem; use codex_rollout::RolloutLine; @@ -42,7 +43,7 @@ struct PendingUserResponse { struct InstructionBoundary { record_index: usize, message_id: Option, - acceptance_order: Option, + input_source: RetainedInputSource, alive: bool, } @@ -161,10 +162,7 @@ impl RollbackPlanner { } else if rollback::counts_as_boundary(&response.item) { let boundary = self.start_boundary(index); self.boundaries[boundary].message_id = response.id().cloned(); - self.boundaries[boundary].acceptance_order = response - .metadata - .as_ref() - .and_then(|metadata| metadata.user_input_order); + self.boundaries[boundary].input_source = response.metadata.as_ref().into(); if let ResponseItem::Message { role, content, .. } = &response.item && role == "user" { @@ -323,7 +321,7 @@ impl RollbackPlanner { self.boundaries.push(InstructionBoundary { record_index: index, message_id: None, - acceptance_order: None, + input_source: RetainedInputSource::Local(None), alive: true, }); let had_prior_boundary = !self.boundary_stack.is_empty(); @@ -380,7 +378,8 @@ impl RollbackPlanner { } if let Some(boundary) = first_removed_boundary { let source = &self.boundaries[boundary]; - if let Some(order) = source.acceptance_order { + let acceptance_order = source.input_source.acceptance_order(); + if let Some(order) = acceptance_order { // An answer may have been persisted before the queued instruction // accepted ahead of it. Both are removed at that acceptance boundary. for fact in &self.retained_fact_sources { @@ -411,10 +410,10 @@ impl RollbackPlanner { // including in checkpoints. Legacy evidence uses the recorded boundary. // Later checkpoints have not been observed yet and already reflect this rollback. for frame in self.compactions.iter_mut().rev().take_while(|frame| { - source.acceptance_order.is_some() || frame.record_index >= source.record_index + acceptance_order.is_some() || frame.record_index >= source.record_index }) { if let Some(context) = &mut frame.item.retained_context { - if source.acceptance_order.is_some() + if acceptance_order.is_some() || context .ordered_entries() .any(|(_, entry)| matches!(entry, RetainedContextEntry::UserMessage(_))) @@ -422,7 +421,7 @@ impl RollbackPlanner { context.rollback( &removed_turns, source.message_id.as_ref().map(ResponseItemId::as_str), - source.acceptance_order, + source.input_source, ); } else { // Checkpoints written without instruction retention keep the legacy