11"""The existing scheduler invocation shared by native benchmark transports."""
22
3+ import argparse
4+ from contextlib import ExitStack
5+ import hashlib
6+ import json
7+ import math
8+ import os
9+ from pathlib import Path
10+ import re
311import shlex
12+ import time
413
514# Private process transport between the benchmark wake and its shell scheduler.
615# The benchmark worker remains the owner of phase-budget exhaustion.
@@ -13,11 +22,167 @@ def phase_work_reserve(mode: str) -> int:
1322 return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS
1423
1524
25+ def _physical_state (path : Path ) -> Path :
26+ if path .is_symlink () or path .resolve () != path :
27+ raise ValueError ("Scheduler phase state must be a physical path" )
28+ return path
29+
30+
31+ def _phase_states (root : Path ) -> list [Path ]:
32+ states = [root , * sorted (root .parent .glob (f"{ root .stem } { root .suffix } ))]
33+ return [_physical_state (path ) for path in states if path .exists ()]
34+
35+
36+ def _state_object (path : Path ) -> dict | None :
37+ try :
38+ value = json .loads (path .read_text ())
39+ except (OSError , UnicodeError , json .JSONDecodeError ):
40+ return None
41+ return value if isinstance (value , dict ) else None
42+
43+
44+ def _latest_phase_state (root : Path , states : list [Path ]) -> Path :
45+ """Follow explicit resume links, then use mtime only for legacy phase files."""
46+ documents = {path : state for path in states if (state := _state_object (path )) is not None }
47+ children : dict [Path , list [Path ]] = {}
48+ for path , state in documents .items ():
49+ parent = state .get ("resume_from" )
50+ if not isinstance (parent , str ):
51+ continue
52+ parent_path = Path (parent )
53+ if (not parent_path .is_absolute () or parent_path .resolve () != parent_path
54+ or parent_path not in documents ):
55+ continue
56+ children .setdefault (parent_path , []).append (path )
57+ current = root if root in documents else max (states , key = lambda path : path .stat ().st_mtime_ns )
58+ seen = set ()
59+ while current not in seen and children .get (current ):
60+ seen .add (current )
61+ current = max (children [current ], key = lambda path : path .stat ().st_mtime_ns )
62+ unreadable = [path for path in states if path not in documents ]
63+ if unreadable :
64+ newest = max (unreadable , key = lambda path : path .stat ().st_mtime_ns )
65+ if newest .stat ().st_mtime_ns > current .stat ().st_mtime_ns :
66+ current = newest # Unreadable current evidence must be held, not skipped.
67+ return current
68+
69+
70+ def _phase_receipt (state_file : Path , wake_log_dir : str | None ) -> dict | None :
71+ if wake_log_dir is None :
72+ return None
73+ # The scheduler writes its current stop after the worker receipt. Later
74+ # phases share this log directory, so the state mtime remains the fence.
75+ stop_time = state_file .stat ().st_mtime_ns
76+ receipts = []
77+ for receipt in Path (wake_log_dir ).glob ("*/receipt.json" ):
78+ if receipt .is_symlink () or receipt .resolve () != receipt :
79+ raise ValueError ("Source wake receipt must be a physical file" )
80+ if receipt .stat ().st_mtime_ns <= stop_time :
81+ receipts .append (receipt )
82+ if not receipts :
83+ return None
84+ latest = max (receipts , key = lambda item : item .stat ().st_mtime_ns )
85+ try :
86+ value = json .loads (latest .read_text ())
87+ except (OSError , UnicodeError , json .JSONDecodeError ):
88+ return None
89+ return value if isinstance (value , dict ) else None
90+
91+
92+ def _create_resume_state (target : Path , predecessor : Path ) -> None :
93+ if target .is_symlink ():
94+ raise ValueError ("Resumed scheduler state must not be a symlink" )
95+ try :
96+ with target .open ("x" , encoding = "utf-8" ) as stream :
97+ json .dump ({"resume_from" : str (predecessor )}, stream )
98+ except FileExistsError :
99+ return
100+ target .chmod (0o600 )
101+
102+
103+ def resume_state_file (state_file : str , * , phase_deadline : float , agent_id : str ,
104+ wake_log_dir : str | None = None ) -> str :
105+ """Select a new transport phase only after an explicit budget continuation.
106+
107+ Kernel state and native sessions stay in place. The original scheduler
108+ record and its independent IO witness are retained, including failures.
109+ The caller owns authorization and the unchanged finite phase deadline.
110+ """
111+ if (isinstance (phase_deadline , bool ) or not isinstance (phase_deadline , (int , float ))
112+ or not math .isfinite (phase_deadline ) or phase_deadline <= time .time ()):
113+ raise ValueError ("Budget continuation requires an open finite phase deadline" )
114+ path = _physical_state (Path (state_file ).absolute ())
115+ from loopx .file_lock import try_exclusive_file_lock
116+ from scripts .external_scheduler_worker import HOST_LOCK_OPERATION , _host_attempt , _host_pending
117+
118+ with ExitStack () as locks :
119+ # The root serializes phase selection. Locking the existing phase files
120+ # as well keeps a running/failed phase from moving under that decision.
121+ if locks .enter_context (try_exclusive_file_lock (
122+ path , agent_id = agent_id , operation = HOST_LOCK_OPERATION )) is None :
123+ return str (path )
124+ for phase in _phase_states (path ):
125+ if phase == path :
126+ continue
127+ if locks .enter_context (try_exclusive_file_lock (
128+ phase , agent_id = agent_id , operation = HOST_LOCK_OPERATION )) is None :
129+ return str (phase )
130+ states = _phase_states (path )
131+ if not states :
132+ return str (path )
133+ predecessor = _latest_phase_state (path , states )
134+ state = _state_object (predecessor )
135+ if state is None :
136+ return str (predecessor )
137+ prior = state .get ("host_attempt" )
138+ if not isinstance (prior , dict ) or prior .get ("status" ) != "stopped" :
139+ return str (predecessor )
140+ # Validate the captured record with its existing transport owner.
141+ scope = prior .get ("scope" )
142+ if not isinstance (scope , str ) or not re .fullmatch (r"[0-9a-f]{64}" , scope ):
143+ return str (predecessor )
144+ try :
145+ _host_attempt (state , scope )
146+ except (TypeError , ValueError ):
147+ return str (predecessor )
148+ if (prior .get ("stop_exit_code" ) != BUDGET_EXHAUSTED_EXIT_CODE
149+ or "host_failure" in prior or _host_pending (predecessor ) is None
150+ or state .get ("last_wake_rc" ) != BUDGET_EXHAUSTED_EXIT_CODE
151+ or state .get ("last_wake_status" ) != "wake_requested_stop" ):
152+ return str (predecessor )
153+ # Only the immutable phase clock names the new state. Todo/reset-token
154+ # changes cannot clear a failure already recorded in this new phase.
155+ suffix = hashlib .sha256 (float (phase_deadline ).hex ().encode ()).hexdigest ()[:20 ]
156+ target = path .with_name (f"{ path .stem } { suffix } { path .suffix } )
157+ if target .is_symlink ():
158+ raise ValueError ("Resumed scheduler state must not be a symlink" )
159+ if target .exists ():
160+ return str (target ) # Never replace a retry/failure in this phase.
161+ # Only a stop before native dispatch admits this shortcut. An
162+ # interrupted native Turn still needs its original recovery/settlement.
163+ # Restrict legacy receipt discovery to the source scheduler snapshot;
164+ # later phases share the log directory but cannot replace this proof.
165+ result = _phase_receipt (predecessor , wake_log_dir )
166+ if (not isinstance (result , dict ) or result .get ("budget_exhausted" ) is not True
167+ or result .get ("host_invoked" ) is not False or result .get ("mode" ) != "heartbeat"
168+ or result .get ("stage" ) != "execute" ):
169+ return str (predecessor )
170+ _create_resume_state (target , predecessor )
171+ return str (target )
172+
173+
16174def worker_command (env : dict [str , str ], * , python : str , source : str ,
17- state_file : str , host_timeout : float ) -> list [str ]:
175+ state_file : str , host_timeout : float ,
176+ resume_budget_stop : bool = False ) -> list [str ]:
177+ if type (resume_budget_stop ) is not bool :
178+ raise ValueError ("resume_budget_stop must be a boolean" )
18179 wake = [python , "-m" , "benchmark.runtime.worker" ]
19180 if env ["LOOPX_EXECUTION_MODE" ] not in "heartbeat" , "turn" }:
20181 return wake
182+ if resume_budget_stop :
183+ return [python , "-m" , "benchmark.runtime.scheduler" , "--python" , python ,
184+ "--source" , source , "--state-file" , state_file ,
185+ "--host-timeout" , str (host_timeout )]
21186 return [
22187 python , f"{ source } ,
23188 "--cli-bin" , env ["LOOPX_CLI" ], "--registry" , env ["LOOPX_REGISTRY" ],
@@ -30,3 +195,23 @@ def worker_command(env: dict[str, str], *, python: str, source: str,
30195 "--quota-timeout-seconds" , "30" , "--error-backoff-seconds" , "15" ,
31196 * (["--wake-host-failure" ] if env ["LOOPX_EXECUTION_MODE" ] == "heartbeat" else []),
32197 ]
198+
199+
200+ def main () -> int :
201+ parser = argparse .ArgumentParser (description = "Resume one explicitly continued benchmark phase." )
202+ parser .add_argument ("--python" , required = True )
203+ parser .add_argument ("--source" , required = True )
204+ parser .add_argument ("--state-file" , required = True )
205+ parser .add_argument ("--host-timeout" , required = True , type = float )
206+ args = parser .parse_args ()
207+ state = resume_state_file (args .state_file ,
208+ phase_deadline = float (os .environ ["LOOPX_PHASE_DEADLINE_EPOCH" ]),
209+ agent_id = os .environ ["LOOPX_AGENT_ID" ], wake_log_dir = os .environ .get ("LOOPX_WAKE_LOG_DIR" ))
210+ command = worker_command (dict (os .environ ), python = args .python , source = args .source ,
211+ state_file = state , host_timeout = args .host_timeout )
212+ os .execvpe (command [0 ], command , os .environ )
213+ return 0
214+
215+
216+ if __name__ == "__main__" :
217+ raise SystemExit (main ())
0 commit comments