wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: subprocess-based remote sync, pipeline status, skill updates

Browse Source 
- 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>
This commit is contained in:
hupenglong1
2026-05-28 14:35:35 +08:00
parent 6364fb437c
commit 98b3e586de
7 changed files with 492 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/server.py
+63 -50
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import asyncio
from contextlib import asynccontextmanager
import csv
import sys
import hashlib
import shlex
import json
@@ -2024,9 +2025,9 @@ def create_app(state: AgentState) -> FastAPI:
) -> Response:
sessions_dir = state.account_paths(account_id)['sessions']
state_path = _session_state_path(sessions_dir, session_id)
await asyncio.to_thread(
_sync_remote_program_state, state, account_id, session_id, state_path,
)
binding_path = _jupyter_binding_path(sessions_dir, session_id) if session_id else None
if binding_path and binding_path.is_file():
await _subprocess_sync_remote(binding_path, state_path)
state_entries, _ = _read_program_state(state_path)
trigger = _extract_trigger_from_state(state_entries) or _extract_trigger_from_session(
sessions_dir, session_id
@@ -2118,9 +2119,9 @@ def create_app(state: AgentState) -> FastAPI:
break
now_mono = time.monotonic()
if now_mono - last_remote_sync >= 5.0:
await asyncio.to_thread(
_sync_remote_program_state, state, account_id, session_id, state_path,
)
_bp = _jupyter_binding_path(sessions_dir, session_id) if session_id else None
if _bp and _bp.is_file():
await _subprocess_sync_remote(_bp, state_path)
last_remote_sync = now_mono
state_entries, state_mtime = _read_program_state(state_path)
iter_count = await asyncio.to_thread(
@@ -5902,9 +5903,29 @@ def _scan_for_watchers(state: AgentState) -> None:
)
_SYNC_SCRIPT = str(Path(__file__).resolve().parent.parent.parent / 'scripts' / 'sync_remote_state.py')
async def _subprocess_sync_remote(binding_path: Path, local_state_path: Path) -> None:
"""Sync remote program-state via subprocess. Killable on timeout."""
try:
proc = await asyncio.create_subprocess_exec(
sys.executable, _SYNC_SCRIPT,
str(binding_path), str(local_state_path),
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
await asyncio.wait_for(proc.wait(), timeout=10.0)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
except OSError:
pass
async def _scan_for_watchers_async(state: AgentState) -> None:
"""Async-friendly scan: runs blocking I/O (remote sync, LLM target_set
extraction) in worker threads so the event loop stays responsive for SSE."""
"""Scan program-state.jsonl files for watch declarations and register
watchers. Uses subprocess for remote sync (killable, no thread pool)."""
accounts_root = state.session_directory.parent / 'accounts'
if not accounts_root.is_dir():
return
@@ -5916,7 +5937,7 @@ async def _scan_for_watchers_async(state: AgentState) -> None:
if not account_dir.is_dir():
continue
account_id = account_dir.name
sessions_dir = account_dir / 'sessions'
sessions_dir = state.account_paths(account_id)['sessions']
if not sessions_dir.is_dir():
continue
try:
@@ -5926,44 +5947,16 @@ async def _scan_for_watchers_async(state: AgentState) -> None:
for session_dir in session_dirs:
session_id = session_dir.name
state_path = session_dir / 'output' / 'program-state.jsonl'
# Remote sync runs HTTP/WS — blocking. Off the loop.
try:
await asyncio.to_thread(
_sync_remote_program_state,
state,
account_id,
session_id,
state_path,
)
except Exception as exc: # noqa: BLE001
print(
f'[scanner] remote sync failed for {account_id}/{session_id}: {exc}',
flush=True,
)
binding_path = session_dir / 'jupyter_workspace.json'
# Subprocess remote sync only for sessions with a jupyter binding
if binding_path.is_file():
await _subprocess_sync_remote(binding_path, state_path)
if not state_path.is_file():
continue
try:
entries, _ = await asyncio.to_thread(
_read_program_state, state_path
)
entries, _ = _read_program_state(state_path)
except Exception: # noqa: BLE001
continue
# Refresh LLM-derived target_set cache (also off the loop).
trigger_text = await asyncio.to_thread(
_extract_trigger_from_session, sessions_dir, session_id
)
if trigger_text:
try:
await asyncio.to_thread(
_refresh_target_set_cache_blocking,
state.model_config_for(account_id),
trigger_text,
)
except Exception as exc: # noqa: BLE001
print(
f'[scanner] target_set LLM refresh failed: {exc}',
flush=True,
)
# Watch registration is fast (just spawns asyncio tasks); stay on loop.
last_status_per_step: dict[str, str] = {}
for entry in entries:
@@ -6565,6 +6558,21 @@ def _build_pipeline_items(
return all_items
def _has_active_watch_for_step(state_entries: list[dict[str, Any]], step: str) -> bool:
"""True if state_entries contain a watch for this step that hasn't resolved."""
last_status: dict[str, str] = {}
has_watch = False
for entry in state_entries:
s = entry.get('step')
st = entry.get('status')
if isinstance(s, str) and isinstance(st, str):
last_status[s] = st
watch = entry.get('watch')
if isinstance(watch, dict) and watch.get('step') == step:
has_watch = True
return has_watch and last_status.get(step) not in ('complete', 'failed', 'cancelled')
def _compute_in_flight(
state_entries: list[dict[str, Any]],
iteration_log_count: int,
@@ -6664,7 +6672,8 @@ def _compute_in_flight(
if status not in ('complete', 'running'):
continue
if status == 'running' and gate_running_for_run:
status = 'waiting'
if not _has_active_watch_for_step(state_entries, step):
status = 'waiting'
meta = _CARD_META.get((step, target_section), {
'icon': 'clock',
'title': step,
@@ -6848,14 +6857,18 @@ def _apply_program_state(
statuses = list(latest_per_step.values())
if not statuses:
payload['status']['state'] = 'pending'
elif non_gate_running and not gate_running:
elif non_gate_running:
payload['status']['state'] = 'running'
elif gate_running:
# HiTL gate is running — agent has handed control back to the user.
# Even if a parent step (e.g. augment) is still technically 'running',
# the gate blocks progress. Surface as 'waiting' so the header pill
# says "等待人工" instead of the spinning "Running" badge.
payload['status']['state'] = 'waiting'
# Exception: if a non-gate step has an active watcher (e.g. cml eval
# running on remote), the session is still progressing — show running.
watched_running = any(
status == 'running' and _STEP_KIND.get(step) != 'gate'
and _has_active_watch_for_step(state_entries, step)
for step, status in latest_per_step.items()
)
payload['status']['state'] = 'running' if watched_running else 'waiting'
elif all(s == 'complete' for s in statuses):
if run_active:
payload['status']['state'] = 'running'
@@ -6946,7 +6959,7 @@ def _chat_root_for_runtime(
) -> str | None:
"""Resolve the per-chat autoresearch root as a REMOTE path string.
Returns the NFS path under /mnt/wangsenhao/autoresearch-zk-users/<account>/<session>/
Returns the NFS path under /mnt/<account>/autoresearch-zk-users/<session>/
so jupyter pod and SFT training pod both read/write the same directory.
Returns None when no runtime is bound — caller decides whether to fall
back to the legacy global path."""
@@ -8033,7 +8046,7 @@ def _step_artifact_paths(
Per-chat artifacts (results/, ai-planning/, output/) live under the chat
workspace root on shared NFS at
`/mnt/wangsenhao/autoresearch-zk-users/<account>/<session>/`. Reads still
`/mnt/<account>/autoresearch-zk-users/<session>/`. Reads still
go through the bound jupyter runtime since the path is only mounted there.
metric_diff (cml step) stays on the global RUN_HISTORY_DIR mount that is
shared across chats. When no runtime is bound, falls back to the legacy