@@ -96,21 +96,6 @@ def __init__(self, controller: Any) -> None:
9696 self ._source_control_discovery = FileChatDiscovery ()
9797 self .actions : Any | None = None
9898
99- @staticmethod
100- def _completed_work_review (row : dict [str , Any ]) -> dict [str , Any ]:
101- review = row .get ("context_recovery" ) or {}
102- if review .get ("work_attempt" ) == 1 :
103- review = (row .get ("context_recovery_history" ) or [{}])[- 1 ]
104- return review if review .get ("read_only_review" ) is True and review .get ("unfinished_work" ) else {}
105-
106- @staticmethod
107- def _has_later_equivalent_input (row : dict [str , Any ], original : dict [str , Any ],
108- messages : Sequence [dict [str , Any ]], * , internal_turns : set [str ]) -> bool :
109- created = datetime .fromisoformat (original ["created_at" ])
110- return any (item .get ("role" ) == "user" and item .get ("turn_id" ) not in internal_turns
111- and item .get ("text" ) == row .get ("message" ) and item .get ("origin" ) in {None , "lark" , "web" , "external" }
112- and datetime .fromisoformat (item ["created_at" ]) > created for item in messages )
113-
11499 def _retained_result_review (self , row : dict [str , Any ]) -> dict [str , Any ] | None :
115100 """Read correlation only; the typed owner still checks continuation eligibility."""
116101 reservation = row .get ("context_recovery" ) or {}
@@ -167,13 +152,6 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
167152 anchors = [event ["payload" ]["native_input" ] for event in events
168153 if event .get ("kind" ) == "turn.started" and event .get ("payload" , {}).get ("native_input" )]
169154 session = self ._original_native_session (session , anchors )
170- original = None
171- review = self ._completed_work_review (row )
172- review_turn = store .load_turn (session ["session_id" ], row ["turn_id" ])
173- if review and review_turn and review ["client_turn_id" ] == review_turn ["client_turn_id" ]:
174- original = store .load_turn (session ["session_id" ], review ["failed_turn_id" ])
175- if original is not None :
176- events += store .events_after (session ["session_id" ], original ["turn_id" ], None )
177155 unfinished = turn .get ("status" ) == "completed" and bool ((turn .get ("response" ) or {}).get ("unfinished_work" ))
178156 live_work_owner = False
179157 if unfinished :
@@ -202,15 +180,14 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
202180 {"event_id" : item ["event_id" ], "kind" : item ["kind" ],
203181 ** {key : item .get ("payload" , {})[key ] for key in ("phase" , "schema_version" , "error_code" , "delivery_mode" , "message" )
204182 if key in item .get ("payload" , {})}} for item in events ]}
205- if preparation is not None or unfinished or original is not None :
206- work_turn = original or turn
207- messages = (discovery_reads .user_messages (session ["session_id" ])
208- if discovery_reads is not None else store .messages (session ["session_id" ]))
209- internal_turns = {turn ["turn_id" ], work_turn ["turn_id" ]}
210- if retained_review is not None :
211- internal_turns .add (retained_review ["turn" ]["turn_id" ])
212- later_equivalent = self ._has_later_equivalent_input (row , work_turn , messages ,
213- internal_turns = internal_turns )
183+ if preparation is not None or unfinished :
184+ created = datetime .fromisoformat (turn ["created_at" ])
185+ later_equivalent = any (item .get ("role" ) == "user" and item .get ("turn_id" ) != turn ["turn_id" ]
186+ and (retained_review is None or item .get ("turn_id" ) != retained_review ["turn" ]["turn_id" ])
187+ and item .get ("text" ) == row .get ("message" ) and item .get ("origin" ) in {None , "lark" , "web" }
188+ and datetime .fromisoformat (item ["created_at" ]) > created
189+ for item in (discovery_reads .user_messages (session ["session_id" ])
190+ if discovery_reads is not None else store .messages (session ["session_id" ])))
214191 from ...chat_manager import steward_machine_defaults
215192 return effect_runtime_result ("collaboration.conversation.request" , {
216193 "request" : {key : value for key , value in row .items () if key != "attachments" },
@@ -234,7 +211,6 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
234211 "stop_observed" : self ._source_stop_observed (row , turn , events ,
235212 source_controls = discovery_reads .source_controls if discovery_reads is not None else None ),
236213 "preparation_observation" : preparation ,
237- "reviewed_original_turn" : original ,
238214 "later_equivalent_input_present" : later_equivalent ,
239215 "preparation_scope_available" : preparation_scope_available ,
240216 "later_source_input_present" : any (item .get ("binding_id" ) == row ["binding_id" ]
@@ -399,9 +375,7 @@ def context_recovery_candidate(self, request_ref: str, *,
399375 turn = self .controller .store .load_turn (row ["session_id" ], row ["turn_id" ])
400376 if not turn or turn .get ("status" ) not in {"failed" , "completed" }:
401377 return False
402- review = self ._completed_work_review (row )
403- if (turn .get ("status" ) == "completed" and not (turn .get ("response" ) or {}).get ("unfinished_work" )
404- and review .get ("client_turn_id" ) != turn .get ("client_turn_id" )):
378+ if turn .get ("status" ) == "completed" and not (turn .get ("response" ) or {}).get ("unfinished_work" ):
405379 from ...chat import _review_openings
406380 _ , open_literal , _ = _review_openings (str ((turn .get ("response" ) or {}).get ("message" ) or "" ), 0 )
407381 invalid_metadata = ((turn .get ("response" ) or {}).get ("gate" ) or {}).get ("kind" ) == "response_metadata_invalid"
@@ -489,16 +463,11 @@ def source_requests() -> list[dict[str, Any]]:
489463 if result_review or plan .get ("work_attempt" ):
490464 with self .controller .lock :
491465 adapter = self .controller .adapters .get (row ["session_id" ])
492- original_work_id = plan .get ("work_original_turn_id" )
493- work_turn = (self .controller .store .load_turn (row ["session_id" ], original_work_id )
494- if original_work_id else turn )
495- if not result_review and work_turn is not None and isinstance (adapter , CodexAppServerAdapter ):
466+ if not result_review and isinstance (adapter , CodexAppServerAdapter ):
496467 # Only the original live owner can prove writable effects.
497468 proof = adapter .completed_work_observation (
498- thread_ids = plan ["native_thread_ids" ], turn_id = work_turn ["upstream_turn_id" ],
469+ thread_ids = plan ["native_thread_ids" ], turn_id = turn ["upstream_turn_id" ],
499470 known_turn_ids = self ._settled_native_turn_ids (session ["session_id" ]))
500- if original_work_id and proof is None :
501- return row # The delivered review is retained; no second review or cold proof.
502471 if proof is None and (adapter is None or isinstance (adapter , CodexAppServerAdapter )
503472 and adapter .upstream_thread_id in plan ["native_thread_ids" ]
504473 and adapter .answer_review_replaceable ()):
@@ -613,8 +582,6 @@ def source_requests() -> list[dict[str, Any]]:
613582 recovery .update (work_attempt = 1 , unfinished_work = plan ["unfinished_work" ],
614583 prior_result_delivery_verified = row ["delivery_verified" ],
615584 presentation_key = f"{ MANAGER_CONTEXT_VERSION } .work.1" , not_before = plan ["not_before" ])
616- if plan .get ("work_original_turn_id" ):
617- recovery ["work_original_turn_id" ] = plan ["work_original_turn_id" ]
618585 if plan .get ("completed_review_turn_id" ):
619586 recovery .update (completed_review_turn_id = plan ["completed_review_turn_id" ],
620587 original_turn = deepcopy (turn ), binding_snapshot = selected ["binding" ])
@@ -1229,33 +1196,6 @@ def _checked_completed_result_review(self, session: dict[str, Any], turn: dict[s
12291196 raise CodexChatAgentError ("The original completed result changed; review held." , error_code = "host_gate" , gate = None )
12301197 return review
12311198
1232- def _check_reviewed_work (self , session : dict [str , Any ], turn : dict [str , Any ],
1233- continuation : dict [str , Any ]) -> None :
1234- from ...chat_agent import CodexChatAgentError
1235-
1236- row = self .read_request (continuation ["request_ref" ])
1237- review = self ._completed_work_review (row )
1238- store = self .controller .store
1239- session_id = session ["session_id" ]
1240- original = store .load_turn (session_id , continuation ["work_original_turn_id" ])
1241- previous = store .load_turn (session_id , continuation ["failed_turn_id" ])
1242- current = store .load_session (session_id ) or {}
1243- binding = self .bindings .resolve (binding_id = row ["binding_id" ], ** row ["source" ])["binding" ]
1244- if (original is None or original != review .get ("original_turn" )
1245- or previous is None or previous ["status" ] != "completed"
1246- or previous ["client_turn_id" ] != review .get ("client_turn_id" )
1247- or binding != review .get ("binding_snapshot" ) or current .get ("active_turn_id" ) != turn ["turn_id" ]
1248- or row ["turn_id" ] not in {turn ["turn_id" ], previous ["turn_id" ]}):
1249- raise CodexChatAgentError ("The original result, review, source or grant changed; continuation held." ,
1250- error_code = "host_gate" , gate = None )
1251- events = [event for target in (original , previous , turn )
1252- for event in store .events_after (session_id , target ["turn_id" ], None )]
1253- if (self ._source_stop_observed (row , original , events ) or any (item .get ("kind" ) == "turn.steered" for item in events )
1254- or self ._has_later_equivalent_input (row , original , store .messages (session_id , strict = True ),
1255- internal_turns = {original ["turn_id" ], previous ["turn_id" ], turn ["turn_id" ]})):
1256- raise CodexChatAgentError ("The original source stopped or has newer input; continuation held." ,
1257- error_code = "host_gate" , gate = None )
1258-
12591199 def prepare_continuation_input (self , session : dict [str , Any ], turn : dict [str , Any ],
12601200 native : Any , message : str ) -> str :
12611201 """Recheck reserved unfinished work at dispatch; keep the stored input intact."""
@@ -1281,8 +1221,6 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An
12811221 continuation = self .recovery_context (session , turn )
12821222 if not continuation or continuation .get ("work_attempt" ) != 1 :
12831223 return message
1284- if continuation .get ("work_original_turn_id" ):
1285- self ._check_reviewed_work (session , turn , continuation )
12861224 proof = continuation .get ("native_observation" ) or {}
12871225 if continuation .get ("completed_review_turn_id" ):
12881226 row = self .read_request (continuation ["request_ref" ])
0 commit comments