hupenglong1
98b3e586de
- Replace thread-based remote state sync with subprocess (sync_remote_state.py) to prevent thread pool exhaustion hanging the API - Fix pipeline showing 'waiting' when a real step is running alongside a gate - Fix watcher scanner path for linux accounts mode - Disable label-master semantic review (Step A.5 + §4.5) per user request - Correct field names: use is_model_correct_dev for accuracy, origin_predict_dev for model output (not cleaned_predict) - Add prohibition against putting test set data in relabel_candidates Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Subprocess-safe remote program-state sync.
|
|
|
|
Reads jupyter binding JSON, fetches program-state.jsonl via jupyter
|
|
Contents API (HTTP GET), writes to local path. Designed to be called
|
|
via asyncio.create_subprocess_exec with a hard timeout — if the pod is
|
|
unreachable, the parent kills this process cleanly.
|
|
|
|
Usage: python3 sync_remote_state.py <binding_json_path> <local_state_path>
|
|
"""
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
import urllib.error
|
|
import ssl
|
|
|
|
def main():
|
|
if len(sys.argv) != 3:
|
|
sys.exit(1)
|
|
binding_path = sys.argv[1]
|
|
local_path = sys.argv[2]
|
|
|
|
try:
|
|
with open(binding_path, encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
except (OSError, json.JSONDecodeError):
|
|
sys.exit(1)
|
|
|
|
binding = data.get('binding', {})
|
|
base_url = binding.get('base_url', '').rstrip('/')
|
|
api_path = binding.get('workspace_api_path', '')
|
|
if not base_url or not api_path:
|
|
sys.exit(1)
|
|
|
|
cookies_list = data.get('cookies', [])
|
|
cookie_str = '; '.join(f"{c['name']}={c['value']}" for c in cookies_list)
|
|
xsrf = data.get('xsrf_token', '')
|
|
|
|
file_api_path = f"{api_path}/output/program-state.jsonl"
|
|
url = f"{base_url}/api/contents/{file_api_path}?content=1&type=file"
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_peer = False
|
|
|
|
req = urllib.request.Request(url)
|
|
req.add_header('Cookie', cookie_str)
|
|
if xsrf:
|
|
req.add_header('X-XSRFToken', xsrf)
|
|
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=8, context=ctx)
|
|
body = json.loads(resp.read())
|
|
except (urllib.error.URLError, OSError, json.JSONDecodeError, TimeoutError):
|
|
sys.exit(1)
|
|
|
|
content = body.get('content', '')
|
|
if not content or not content.strip():
|
|
sys.exit(0)
|
|
|
|
# Preserve local-only _synthetic entries
|
|
local_synthetics = []
|
|
try:
|
|
with open(local_path, encoding='utf-8') as f:
|
|
for line in f:
|
|
if '"_synthetic"' in line:
|
|
try:
|
|
obj = json.loads(line)
|
|
if obj.get('_synthetic'):
|
|
local_synthetics.append(line.rstrip('\n'))
|
|
except (json.JSONDecodeError, ValueError):
|
|
pass
|
|
except OSError:
|
|
pass
|
|
|
|
merged = content.rstrip('\n')
|
|
if local_synthetics:
|
|
merged += '\n' + '\n'.join(local_synthetics)
|
|
merged += '\n'
|
|
|
|
# Skip rewrite if unchanged
|
|
try:
|
|
with open(local_path, encoding='utf-8') as f:
|
|
if f.read() == merged:
|
|
sys.exit(0)
|
|
except OSError:
|
|
pass
|
|
|
|
import os
|
|
os.makedirs(os.path.dirname(local_path), exist_ok=True)
|
|
with open(local_path, 'w', encoding='utf-8') as f:
|
|
f.write(merged)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|